feat: add chain writer

Co-authored-by: Dmitry Fedotov <dmitry@uint32.ru>
Co-committed-by: Dmitry Fedotov <dmitry@uint32.ru>

.gitignore

@@ -1,2 +1,3 @@
 cmd
 coverage.out
+example

LICENSE

@@ -0,0 +1,22 @@
+Copyright 2025 Dmitry Fedotov
+
+Permission is hereby granted, free of charge, to any person
+obtaining a copy of this software and associated documentation 
+files (the “Software”), to deal in the Software without 
+restriction, including without limitation the rights to use, 
+copy, modify, merge, publish, distribute, sublicense, and/or 
+sell copies of the Software, and to permit persons to whom 
+the Software is furnished to do so, 
+subject to the following conditions:
+
+The above copyright notice and this permission notice shall 
+be included in all copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED “AS IS”, WITHOUT WARRANTY OF ANY KIND, 
+EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES 
+OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND 
+NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT
+HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY,
+WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
+FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE
+OR OTHER DEALINGS IN THE SOFTWARE.

README.md

@@ -0,0 +1,137 @@
+# package script
+
+This package is intended as a collection of helper functions and tools fow quick construction of scripts that process .csv tables.
+
+Note that this is a work in progress. API is not guaranteed to be stable.
+
+## Example usage
+
+This demonstrates a very basic case: reading from stdin, checking whether array len is 2 and switching array elements.
+
+```go
+package main
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"os/signal"
+	"syscall"
+
+	"code.uint32.ru/dmitry/script"
+)
+
+func main() {
+	ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGTERM, syscall.SIGINT)
+	defer stop()
+
+	reader := script.NewStdinReader()
+	writer := script.NewStdoutWriter()
+
+	processor := func(_ context.Context, in []string) ([]string, error) {
+		if len(in) != 2 {
+			return nil, errors.New("incorrect input len")
+		}
+
+		return []string{in[1], in[0]}, nil
+	}
+
+	conf := script.RunConfig{
+		Input:     reader,
+		Output:    writer,
+		Processor: processor,
+	}
+
+	if _, err := script.Run(ctx, conf); err != nil {
+		fmt.Println(err)
+	}
+}
+```
+
+A more conplicated example featuring a processor which is itself a struct with external dependency.
+
+```go
+package main
+
+import (
+	"context"
+	"fmt"
+	"os/signal"
+	"syscall"
+
+	"code.uint32.ru/dmitry/script"
+)
+
+// ExternalDep represent an external dependency
+// which your script utilizes. This can be a DB connection etc.
+type ExternalDep interface {
+	QueryID(string) ([]string, error)
+}
+
+type Processor struct {
+	dep ExternalDep
+}
+
+func New(e ExternalDep) *Processor {
+
+	return &Processor{
+		dep: e,
+	}
+}
+
+func (p *Processor) Process(ctx context.Context, in []string) ([]string, error) {
+	var out []string
+
+	result, err := p.dep.QueryID(in[0])
+	if err != nil {
+		// you can as well record the error and continue
+		// processing without yielding the error here
+		return nil, err
+	}
+
+	out = append(out, result...)
+	out = append(out, "processed")
+
+	return out, nil
+}
+
+func main() {
+	ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGTERM, syscall.SIGINT)
+	defer stop()
+
+	r, err := script.NewCSVReader("/tmp/some_file.csv")
+	if err != nil {
+		panic(err)
+	}
+
+	defer func() {
+		if err := r.Close(); err != nil {
+			fmt.Println("err closing reader", err)
+		}
+	}()
+
+	w, err := script.NewCSVWriter("/tmp/some_output.csv")
+	if err != nil {
+		panic(err)
+	}
+
+	defer func() {
+		if err := w.Close(); err != nil {
+			fmt.Println("err closing writer", err)
+		}
+	}()
+
+    // intializing the processor
+	p := New(ExternalDep(nil))
+
+	conf := script.RunConfig{
+		Input:     r,
+		Output:    w,
+		Processor: p.Process, // Process implements script.Processor
+	}
+
+	if _, err := script.Run(ctx, conf); err != nil {
+		fmt.Println(err)
+	}
+}
+```

chain.go

@@ -0,0 +1,43 @@
+package script
+
+import "context"
+
+// Chain chains provided Processors.
+// When an error is returned by a Processor in chain, processing
+// stops and the error is retuned without running further stages.
+func ChainProcessor(processors ...Processor) Processor {
+	return func(ctx context.Context, in []string) ([]string, error) {
+		var err error
+		for _, p := range processors {
+			// not checking ctx expiry here,
+			// let the processor handle it
+
+			in, err = p(ctx, in)
+			if err != nil {
+				return nil, err
+			}
+		}
+
+		return in, nil
+	}
+}
+
+type chainWriter struct {
+	w []Writer
+}
+
+func (c *chainWriter) Write(in []string) error {
+	for _, w := range c.w {
+		if err := w.Write(in); err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
+func ChainWriter(writers ...Writer) Writer {
+	return &chainWriter{
+		w: writers,
+	}
+}

chain_test.go

@@ -0,0 +1,49 @@
+package script
+
+import (
+	"context"
+	"slices"
+	"testing"
+)
+
+func TestChainProcessor(t *testing.T) {
+	p := func(_ context.Context, in []string) ([]string, error) {
+		in[0] = in[0] + in[0]
+		return in, nil
+	}
+
+	chain := ChainProcessor(p, p, p)
+
+	in := []string{"a"}
+	want := []string{"aaaaaaaa"}
+
+	res, err := chain(t.Context(), in)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	if !slices.Equal(res, want) {
+		t.Fatalf("slices are not equal, have: %+v, want: %+v", res, want)
+	}
+}
+
+func TestChainWriter(t *testing.T) {
+	w1 := NewMemWriter()
+	w2 := NewMemWriter()
+
+	w := ChainWriter(w1, w2)
+
+	in := []string{"a"}
+
+	if err := w.Write(in); err != nil {
+		t.Fatal(err)
+	}
+
+	if !slices.Equal(w1.Output()[0], in) {
+		t.Fatalf("w1 slices are not equal, have: %+v, want: %+v", w1.Output()[0], in)
+	}
+
+	if !slices.Equal(w2.Output()[0], in) {
+		t.Fatalf("w2 slices are not equal, have: %+v, want: %+v", w2.Output()[0], in)
+	}
+}

input_csv.go

@@ -1,7 +1,6 @@
 package script
 
 import (
-	"context"
 	"encoding/csv"
 	"os"
 )
@@ -16,7 +15,7 @@ type CSVReader struct {
 	rdr *csv.Reader
 }
 
-func (c *CSVReader) Read(context.Context) ([]string, error) {
+func (c *CSVReader) Read() ([]string, error) {
 	return c.rdr.Read()
 }
 

input_csv_test.go

@@ -0,0 +1,29 @@
+package script_test
+
+import (
+	"slices"
+	"testing"
+
+	"code.uint32.ru/dmitry/script"
+)
+
+func TestCSVReader(t *testing.T) {
+	t.Parallel()
+	r, err := script.NewCSVReader("testdata/sample_csv.csv")
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	want := []string{"one", "two", "three"}
+
+	for range 2 {
+		row, err := r.Read()
+		if err != nil {
+			t.Fatal(err)
+		}
+
+		if !slices.Equal(row, want) {
+			t.Fatalf("rows not equal, want: %v, have: %v", want, row)
+		}
+	}
+}

input_mem.go

@@ -1,7 +1,6 @@
 package script
 
 import (
-	"context"
 	"io"
 )
 
@@ -16,7 +15,7 @@ func NewMemReader(records [][]string) *MemReader {
 	}
 }
 
-func (m *MemReader) Read(context.Context) ([]string, error) {
+func (m *MemReader) Read() ([]string, error) {
 	if m.curr == len(m.rows) || len(m.rows) == 0 {
 		return nil, io.EOF
 	}

input_stdin.go

@@ -2,7 +2,6 @@ package script
 
 import (
 	"bufio"
-	"context"
 	"io"
 	"os"
 	"strings"
@@ -23,7 +22,7 @@ func NewStdinReader() *StdinReader {
 // then splits result at "," separator and returns
 // the resulting slice.
 // It returns EOF when nothing left to read.
-func (s *StdinReader) Read(_ context.Context) ([]string, error) {
+func (s *StdinReader) Read() ([]string, error) {
 	if !s.scanner.Scan() {
 		err := s.scanner.Err()
 		if err == nil {

output_csv.go

@@ -1,7 +1,6 @@
 package script
 
 import (
-	"context"
 	"encoding/csv"
 	"errors"
 	"os"
@@ -20,7 +19,7 @@ type CSVWriter struct {
 }
 
 // Write writer row to csv. Writes are buffered
-func (c *CSVWriter) Write(_ context.Context, record []string) error {
+func (c *CSVWriter) Write(record []string) error {
 	err := c.wr.Write(record)
 	c.err = err
 

output_csv_test.go

@@ -0,0 +1,42 @@
+package script
+
+import (
+	"bytes"
+	"os"
+	"testing"
+)
+
+func TestCSVWriter(t *testing.T) {
+	path := "testdata/output_csv.csv"
+
+	os.Remove(path)
+	defer os.Remove(path)
+
+	w, err := newCSVwriter(path)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	want := []byte("one,two,three\none,two,three\n")
+
+	row := []string{"one", "two", "three"}
+
+	for range 2 {
+		if err := w.Write(row); err != nil {
+			t.Fatal(err)
+		}
+	}
+
+	if err := w.Close(); err != nil {
+		t.Fatal(err)
+	}
+
+	b, err := os.ReadFile(path)
+	if err != nil {
+		t.Fatalf("err reading output file: %s", err)
+	}
+
+	if !bytes.Equal(want, b) {
+		t.Errorf("incorrect result, want: %s, have: %s", string(want), string(b))
+	}
+}

output_mem.go

@@ -1,7 +1,5 @@
 package script
 
-import "context"
-
 type MemWriter struct {
 	rows [][]string
 }
@@ -10,7 +8,7 @@ func NewMemWriter() *MemWriter {
 	return new(MemWriter)
 }
 
-func (m *MemWriter) Write(_ context.Context, record []string) error {
+func (m *MemWriter) Write(record []string) error {
 	m.rows = append(m.rows, record)
 
 	return nil

output_nop.go

@@ -1,13 +1,11 @@
 package script
 
-import "context"
-
 type NopWriter struct{}
 
 func NewNopWriter() *NopWriter {
 	return new(NopWriter)
 }
 
-func (d *NopWriter) Write(context.Context, []string) error {
+func (d *NopWriter) Write([]string) error {
 	return nil
 }

output_stdout.go

@@ -1,7 +1,6 @@
 package script
 
 import (
-	"context"
 	"os"
 	"strings"
 )
@@ -12,7 +11,7 @@ func NewStdoutWriter() *StdoutWriter {
 	return new(StdoutWriter)
 }
 
-func (s *StdoutWriter) Write(_ context.Context, in []string) error {
+func (s *StdoutWriter) Write(in []string) error {
 	_, err := os.Stdout.WriteString(strings.Join(in, ",") + "\n")
 	return err
 }

runner.go

@@ -5,13 +5,14 @@ import (
 	"errors"
 	"fmt"
 	"io"
+	"sync/atomic"
 
 	"golang.org/x/sync/errgroup"
 )
 
 var (
 	EOF             error = io.EOF
-	ErrNoProcessors       = errors.New("no processors provided")
+	ErrNoProcessors       = errors.New("script: no processors provided")
 )
 
 var (
@@ -28,18 +29,18 @@ type Reader interface {
 	// Read is called by Processor to obtain input
 	// for processing. Read must return EOF/io.EOF to indicate
 	// that there is no more input.
-	Read(context.Context) ([]string, error)
+	Read() ([]string, error)
 }
 
 type Writer interface {
 	// Write is called by Processor to record
 	// output.
-	Write(context.Context, []string) error
+	Write([]string) error
 }
 
-// Process accepts string slice, does what it should
+// Processor accepts string slice, does what it should
 // and returns output. Non-nil error returned
-// by Processor terminates stops further processing.
+// by Processor stops further processing.
 type Processor func(context.Context, []string) ([]string, error)
 
 type RunConfig struct {
@@ -51,15 +52,22 @@ type RunConfig struct {
 	Concurrency int
 }
 
-// Run starts concurrency threads (goroutines), reads from provided Reader,
-// executes each Processor in the order they were provide and records result
-// with provided Writer.
-// At first Read is called offset times with output of Read being discarded.
+type RunResult struct {
+	Read      int // number of records read without error (offset count not included)
+	Processed int // number of records processed without error
+	Written   int // number of records written to Writer without error
+}
+
+// Run starts the script described by r.
+// First Read is called offset times with output of Read being discarded.
 // Then limit Reads are made and processor is called for each portion
-// of data. If limit is 0 then Runner keep processing input until it receives
+// of data. If limit is 0 then Run keeps processing input until it receives
 // EOF from Reader.
-func Run(ctx context.Context, r RunConfig) error {
-	if r.Concurrency == 0 {
+// Run fails on any error including Reader error, Writer error and Processor error.
+// The returned RunResult is AWAYS VALID and indicates the actual progress of script.
+// Returned error explains why Run failed. It may be either read, process or write error.
+func Run(ctx context.Context, r RunConfig) (RunResult, error) {
+	if r.Concurrency <= 0 {
 		r.Concurrency = 1
 	}
 
@@ -68,27 +76,33 @@ func Run(ctx context.Context, r RunConfig) error {
 	rdch := make(chan []string, r.Concurrency)
 	wrch := make(chan []string, r.Concurrency)
 
+	var read, proc, written uint32
+
 	// read input from Reader and forward to Processor
 	grp.Go(func() error {
-		// closing chan for processor to complete operations
+		// closing chan to Processor
 		defer close(rdch)
 
 		for range r.Offset {
-			_, err := r.Input.Read(ctx)
+			_, err := r.Input.Read()
 			if err != nil {
-				return fmt.Errorf("could not advance to required offset: %w", err)
+				return fmt.Errorf("script: could not advance to required offset (%d): %s", r.Offset, err)
 			}
 		}
 
 		count := 0
+
 		for {
-			inp, err := r.Input.Read(ctx)
-			if err != nil && errors.Is(err, EOF) {
+			inp, err := r.Input.Read()
+			if errors.Is(err, EOF) {
 				return nil
 			} else if err != nil {
-				return err
+				return fmt.Errorf("script: read error: %s", err)
 			}
 
+			// increment read count
+			read++
+
 			select {
 			case rdch <- inp:
 			case <-ctx.Done():
@@ -99,7 +113,7 @@ func Run(ctx context.Context, r RunConfig) error {
 
 			count++
 
-			if count == r.Limit { // will never happen if limit set to 0
+			if count == r.Limit { // will never happen if limit has been set to 0
 				return nil
 			}
 		}
@@ -107,41 +121,63 @@ func Run(ctx context.Context, r RunConfig) error {
 
 	// read output of Processor and write to Writer
 	grp.Go(func() error {
+		defer func() {
+			for range wrch {
+				// NOP to drain channel
+			}
+		}()
+
 		// not paying attention to context here
 		// because we must complete writes
+		// this is run within group so that write
+		// error would cancel group context
 		for outp := range wrch {
-			if err := r.Output.Write(ctx, outp); err != nil {
-				return err
+			if err := r.Output.Write(outp); err != nil {
+				return fmt.Errorf("script: write error: %s", err)
 			}
+
+			//increment write count
+			written++
 		}
 
 		return nil
 	})
 
+	// run processing routines
 	grp.Go(func() error {
-		// will close write chan once
-		// all workers are done
+		// closing chan to Writer
 		defer close(wrch)
+		defer func() {
+			for range rdch {
+				// NOP to drain channel
+			}
+		}()
 
-		workergrp, innrctx := errgroup.WithContext(ctx)
+		workergrp := errgroup.Group{}
 
 		for range r.Concurrency {
 			workergrp.Go(func() error {
-				// not paying attention to context here
-				// because we must complete writes
 				for inp := range rdch {
-					result, err := r.Processor(innrctx, inp)
+					result, err := r.Processor(ctx, inp)
 					if err != nil {
-						return err
+						return fmt.Errorf("script: process error: %s", err)
 					}
 
-					wrch <- result
+					// increment processed count
+					atomic.AddUint32(&proc, 1)
+
+					select {
+					case wrch <- result:
+					case <-ctx.Done():
+						// this case is a must if writer fails
+						// otherwise we'd want to push process result
+						// to wrch
+						return nil
+					}
 				}
 
 				return nil
-
 			})
-
 		}
 
 		if err := workergrp.Wait(); err != nil {
@@ -151,9 +187,11 @@ func Run(ctx context.Context, r RunConfig) error {
 		return nil
 	})
 
-	if err := grp.Wait(); err != nil {
-		return err
-	}
+	err := grp.Wait() // if this is a context expiry then error is nil
 
-	return nil
+	return RunResult{
+		Read:      int(read),
+		Processed: int(proc),
+		Written:   int(written),
+	}, err
 }

runner_test.go

@@ -28,7 +28,8 @@ func TestBasicRun(t *testing.T) {
 		Processor: echoProcessor,
 	}
 
-	if err := script.Run(t.Context(), conf); err != nil {
+	res, err := script.Run(t.Context(), conf)
+	if err != nil {
 		t.Fatal(err)
 	}
 
@@ -37,11 +38,15 @@ func TestBasicRun(t *testing.T) {
 	if !reflect.DeepEqual(input, output) {
 		t.Errorf("incorrect output, want: %v, got: %v", input, output)
 	}
+
+	if res.Read != 1 || res.Processed != 1 || res.Written != 1 {
+		t.Fatal("incorrect process result, want all fields to equal 1")
+	}
 }
 
 type infiniteReader struct{}
 
-func (ir *infiniteReader) Read(_ context.Context) ([]string, error) {
+func (ir *infiniteReader) Read() ([]string, error) {
 	return []string{"infinity", "looks", "like", "this"}, nil
 }
 
@@ -60,7 +65,7 @@ func TestRunnerObeysContext(t *testing.T) {
 	ctx, cancel := context.WithTimeout(t.Context(), time.Millisecond)
 	defer cancel()
 
-	if err := script.Run(ctx, conf); err != nil {
+	if _, err := script.Run(ctx, conf); err != nil {
 		t.Fatal(err)
 	}
 }

testdata/sample_csv.csv

@@ -0,0 +1,2 @@
+one,two,three
+one,two,three