modernise with go fix, drop deprecated rand.Seed

README.md

@@ -34,10 +34,10 @@ running handlers gracefully and won't keep any goroutine running/leaking.
 ```go
 // ReaderHandler creates flow.Handler, reading strings from any io.Reader
 func ReaderHandler(reader io.Reader) Handler {
-	return func(ctx context.Context, ch chan interface{}) (chan interface{}, func() error) {
+	return func(ctx context.Context, ch chan any) (chan any, func() error) {
 		metrics := flow.GetMetrics(ctx) // metrics collects how many records read with "read" key.
 
-		readerCh := make(chan interface{}, 1000)
+		readerCh := make(chan any, 1000)
 		readerFn := func() error {
 			defer close(readerCh)
 
@@ -70,8 +70,8 @@ func ExampleFlow_flow() {
 	f.Add(     // add handlers. Note: handlers can be added directly in New
 
 		// first handler, generate 100 initial values.
-		func(ctx context.Context, in chan interface{}) (out chan interface{}, runFn func() error) {
-			out = make(chan interface{}, 100) // example of non-async handler
+		func(ctx context.Context, in chan any) (out chan any, runFn func() error) {
+			out = make(chan any, 100) // example of non-async handler
 			for i := 1; i <= 100; i++ {
 				out <- i
 			}
@@ -80,8 +80,8 @@ func ExampleFlow_flow() {
 		},
 
 		// second handler - picks odd numbers only and multiply
-		func(ctx context.Context, in chan interface{}) (out chan interface{}, runFn func() error) {
-			out = make(chan interface{}) // async handler makes its out channel
+		func(ctx context.Context, in chan any) (out chan any, runFn func() error) {
+			out = make(chan any) // async handler makes its out channel
 			runFn = func() error {
 				defer close(out) // handler should close out channel
 				for e := range in {
@@ -102,8 +102,8 @@ func ExampleFlow_flow() {
 		},
 
 		// final handler - sum all numbers
-		func(ctx context.Context, in chan interface{}) (out chan interface{}, runFn func() error) {
-			out = make(chan interface{}, 1)
+		func(ctx context.Context, in chan any) (out chan any, runFn func() error) {
+			out = make(chan any, 1)
 			runFn = func() error {
 				defer close(out)
 				sum := 0
@@ -145,8 +145,8 @@ func ExampleFlow_parallel() {
 	f.Add(
 
 		// generate 100 initial values in single handler
-		func(ctx context.Context, in chan interface{}) (out chan interface{}, runFn func() error) {
-			out = make(chan interface{}, 100) // example of non-async handler
+		func(ctx context.Context, in chan any) (out chan any, runFn func() error) {
+			out = make(chan any, 100) // example of non-async handler
 			for i := 1; i <= 100; i++ {
 				out <- i
 			}
@@ -155,8 +155,8 @@ func ExampleFlow_parallel() {
 		},
 
 		// multiple all numbers in 10 parallel handlers
-		f.Parallel(10, func(ctx context.Context, in chan interface{}) (out chan interface{}, runFn func() error) {
-			out = make(chan interface{}) // async handler makes its out channel
+		f.Parallel(10, func(ctx context.Context, in chan any) (out chan any, runFn func() error) {
+			out = make(chan any) // async handler makes its out channel
 			runFn = func() error {
 				defer close(out) // handler should close out channel
 				for e := range in {
@@ -173,7 +173,7 @@ func ExampleFlow_parallel() {
 		}),
 
 		// print all numbers
-		func(ctx context.Context, in chan interface{}) (out chan interface{}, runFn func() error) {
+		func(ctx context.Context, in chan any) (out chan any, runFn func() error) {
 			runFn = func() error {
 				defer close(out)
 				sum := 0
@@ -221,7 +221,7 @@ terminated early, by an error with the default fail-fast or by a canceled contex
 ### worker function
 
 Worker function passed by user and runs in multiple workers (goroutines) concurrently. 
-This is the function: `type WorkerFn func(ctx context.Context, inpRec interface{}, sender SenderFn, store WorkerStore) error`
+This is the function: `type WorkerFn func(ctx context.Context, inpRec any, sender SenderFn, store WorkerStore) error`
 
 It takes `inp` parameter, does the job and optionally send result(s) with `SenderFn` to the common results channel. 
 Error will terminate all workers unless `ContinueOnError` set.
@@ -235,8 +235,8 @@ is not synchronised internally, it doesn't need to be as only the owning worker
 
 ```go
 type WorkerStore interface {
-	Set(key string, val interface{})
-	Get(key string) (interface{}, bool)
+	Set(key string, val any)
+	Get(key string) (any, bool)
 	GetInt(key string) int
 	GetFloat(key string) float64
 	GetString(key string) string
@@ -251,7 +251,7 @@ _alternatively state can be kept outside of workers as a slice of values and acc
 ### usage
 
 ```go
-    p := pool.New(8, func(ctx context.Context, v interface{}, sendFn pool.SenderFn, ws pool.WorkerStore) error {
+    p := pool.New(8, func(ctx context.Context, v any, sendFn pool.SenderFn, ws pool.WorkerStore) error {
         // worker function gets input v, processes it and sends result(s) to the common results channel
 
         input, ok := v.(string) // in this case it gets string as input
@@ -286,7 +286,7 @@ _alternatively state can be kept outside of workers as a slice of values and acc
         p.Close() // indicates completion of all inputs
     }()
 
-    var v interface{}
+    var v any
     for cursor.Next(ctx, &v) {
         log.Print(v)  // print value
     }

_example/main.go

@@ -21,7 +21,7 @@ func main() {
 
 	// seed channel with the list of input files. Usually seeding implemented as a separate handler but for our toy example
 	// filling a buffered channel will do it.
-	seedCh := make(chan interface{}, 4)
+	seedCh := make(chan any, 4)
 	seedCh <- "data/input-1.txt"
 	seedCh <- "data/input-2.txt"
 	seedCh <- "data/input-3.txt"
@@ -41,10 +41,10 @@ func main() {
 }
 
 // lineSplitHandler gets file names and sends lines of text
-func lineSplitHandler(ctx context.Context, ch chan interface{}) (chan interface{}, func() error) {
+func lineSplitHandler(ctx context.Context, ch chan any) (chan any, func() error) {
 	log.Print("make line split handler")
 	metrics := flow.GetMetrics(ctx)
-	lineCh := make(chan interface{}, 100)
+	lineCh := make(chan any, 100)
 	lineFn := func() error {
 		log.Printf("start line split handler %d", flow.CID(ctx))
 		defer close(lineCh)
@@ -89,10 +89,10 @@ type wordsInfo struct {
 // wordsHandler reads lines of text from the input and sends wordsInfo
 func wordsHandler(minSize int) flow.Handler {
 	log.Printf("make words handler with minsize=%d", minSize)
-	return func(ctx context.Context, ch chan interface{}) (chan interface{}, func() error) {
+	return func(ctx context.Context, ch chan any) (chan any, func() error) {
 		log.Printf("start words handler %d with minsize=%d", flow.CID(ctx), minSize)
 		metrics := flow.GetMetrics(ctx)
-		wordsCh := make(chan interface{}, 1000)
+		wordsCh := make(chan any, 1000)
 		wordsFn := func() error {
 			defer close(wordsCh)
 			count := 0
@@ -120,8 +120,8 @@ func wordsHandler(minSize int) flow.Handler {
 }
 
 // sumHandler reduces all inputs with wordsInfo to the final figures and prints it as result
-func sumHandler(_ context.Context, ch chan interface{}) (chan interface{}, func() error) {
-	nopCh := make(chan interface{})
+func sumHandler(_ context.Context, ch chan any) (chan any, func() error) {
+	nopCh := make(chan any)
 	sumFn := func() error {
 		log.Printf("start sum handler")
 		defer close(nopCh)

flow.go

@@ -31,8 +31,8 @@ type Flow struct {
 	group *errgroup.Group // all handlers runs in this errgroup
 	ctx   context.Context // context used for cancellation
 
-	lastCh chan interface{} // last channel in flow
-	funcs  []func() error   // all runnable functions
+	lastCh chan any       // last channel in flow
+	funcs  []func() error // all runnable functions
 
 	fanoutBuffer int       // buffer size for fanout
 	activateOnce sync.Once // prevents multiple activations of flow
@@ -43,7 +43,7 @@ type Flow struct {
 // fn will be executed in a separate goroutine. fn is thread-safe and may have mutable state. It will live
 // all flow lifetime and usually implements read->process->write cycle. If fn returns != nil it indicates
 // critical failure and will stop, with canceled context, all handlers in the flow.
-type Handler func(ctx context.Context, in chan interface{}) (out chan interface{}, runFn func() error)
+type Handler func(ctx context.Context, in chan any) (out chan any, runFn func() error)
 
 // New creates flow object with context and common errgroup. This errgroup used to schedule and cancel all handlers.
 // options defines non-default parameters.
@@ -91,9 +91,9 @@ func (f *Flow) Parallel(concurrent int, handler Handler) Handler {
 		return handler
 	}
 
-	return func(ctx context.Context, ch chan interface{}) (chan interface{}, func() error) {
-		var outChs []chan interface{}
-		for n := 0; n < concurrent; n++ {
+	return func(ctx context.Context, ch chan any) (chan any, func() error) {
+		outChs := make([]chan any, 0, concurrent)
+		for n := range concurrent {
 			ctxWithID := context.WithValue(ctx, CidContextKey, n) // put n as id to context for parallel handlers
 			out, fn := handler(ctxWithID, ch)                     // all parallel handlers read from the same lastCh
 			f.funcs = append(f.funcs, fn)                         // register runnable with flow executor
@@ -115,15 +115,15 @@ func (f *Flow) FanOut(handler Handler, handlers ...Handler) Handler {
 		return handler
 	}
 
-	return func(ctx context.Context, ch chan interface{}) (chan interface{}, func() error) {
+	return func(ctx context.Context, ch chan any) (chan any, func() error) {
 
 		handlers = append([]Handler{handler}, handlers...) // add head handler to head
 
-		inChs := make([]chan interface{}, len(handlers))  // input channels for forked input from ch
-		outChs := make([]chan interface{}, len(handlers)) // output channels for merging
+		inChs := make([]chan any, len(handlers))  // input channels for forked input from ch
+		outChs := make([]chan any, len(handlers)) // output channels for merging
 
 		for i := 0; i < len(handlers); i++ {
-			inChs[i] = make(chan interface{}, f.fanoutBuffer)     // buffered to allow async readers
+			inChs[i] = make(chan any, f.fanoutBuffer)             // buffered to allow async readers
 			ctxWithID := context.WithValue(ctx, CidContextKey, i) // keep i as ID for handler in context
 			out, fn := handlers[i](ctxWithID, inChs[i])           // handle forked input
 			f.funcs = append(f.funcs, fn)                         // register runnable with flow executor
@@ -171,7 +171,7 @@ func (f *Flow) Wait() error {
 
 // Channel returns last (final) channel in flow. Usually consumers don't need this channel, but can be used
 // to return some final result(s)
-func (f *Flow) Channel() chan interface{} {
+func (f *Flow) Channel() chan any {
 	return f.lastCh
 }
 
@@ -181,15 +181,14 @@ func (f *Flow) Metrics() *Metrics {
 }
 
 // merge gets multiple channels and fan-in to a single output channel
-func (f *Flow) merge(ctx context.Context, chs []chan interface{}) (mergeCh chan interface{}, mergeFn func() error) {
+func (f *Flow) merge(ctx context.Context, chs []chan any) (mergeCh chan any, mergeFn func() error) {
 
-	mergeCh = make(chan interface{})
+	mergeCh = make(chan any)
 	mergeFn = func() error {
 		defer close(mergeCh)
 
 		gr, ctxGroup := errgroup.WithContext(ctx)
 		for _, ch := range chs {
-			ch := ch
 			gr.Go(func() error {
 				for e := range ch {
 					if err := Send(ctxGroup, mergeCh, e); err != nil {
@@ -216,7 +215,7 @@ func CID(ctx context.Context) int {
 
 // Send entry to channel or returns error if context canceled.
 // Shortcut for send-or-fail-on-cancel most handlers implement.
-func Send(ctx context.Context, ch chan interface{}, e interface{}) error {
+func Send(ctx context.Context, ch chan any, e any) error {
 	select {
 	case ch <- e:
 		return nil
@@ -227,7 +226,7 @@ func Send(ctx context.Context, ch chan interface{}, e interface{}) error {
 
 // Recv gets entry from the channel or returns error if context canceled.
 // Shortcut for read-or-fail-on-cancel most handlers implement.
-func Recv(ctx context.Context, ch chan interface{}) (interface{}, error) {
+func Recv(ctx context.Context, ch chan any) (any, error) {
 	select {
 	case val := <-ch:
 		return val, nil