Make Each take a sequence

all.go

@@ -2,6 +2,7 @@ package flowmatic
 
 import (
 	"context"
+	"errors"
 )
 
 // All runs each task concurrently
@@ -33,3 +34,43 @@ func All(ctx context.Context, tasks ...func(context.Context) error) error {
 		return nil
 	})
 }
+
+// eachN starts numWorkers concurrent workers (or GOMAXPROCS workers if numWorkers < 1)
+// and starts a task for each number from 0 to numItems.
+// Errors returned by a task do not halt execution,
+// but are joined into a multierror return value.
+// If a task panics during execution,
+// the panic will be caught and rethrown in the parent Goroutine.
+func eachN(numWorkers, numItems int, task func(int) error) error {
+	type void struct{}
+	inch, ouch := TaskPool(numWorkers, func(pos int) (void, error) {
+		return void{}, task(pos)
+	})
+	var (
+		panicVal any
+		errs     []error
+	)
+	_ = Do(
+		func() error {
+			for i := 0; i < numItems; i++ {
+				inch <- i
+			}
+			close(inch)
+			return nil
+		},
+		func() error {
+			for r := range ouch {
+				if r.Panic != nil && panicVal == nil {
+					panicVal = r.Panic
+				}
+				if r.Err != nil {
+					errs = append(errs, r.Err)
+				}
+			}
+			return nil
+		})
+	if panicVal != nil {
+		panic(panicVal)
+	}
+	return errors.Join(errs...)
+}

each.go

@@ -2,6 +2,7 @@ package flowmatic
 
 import (
 	"errors"
+	"iter"
 )
 
 // Each starts numWorkers concurrent workers (or GOMAXPROCS workers if numWorkers < 1)
@@ -10,48 +11,35 @@ import (
 // but are joined into a multierror return value.
 // If a task panics during execution,
 // the panic will be caught and rethrown in the parent Goroutine.
-func Each[Input any](numWorkers int, items []Input, task func(Input) error) error {
-	return eachN(numWorkers, len(items), func(pos int) error {
-		return task(items[pos])
-	})
-}
-
-// eachN starts numWorkers concurrent workers (or GOMAXPROCS workers if numWorkers < 1)
-// and starts a task for each number from 0 to numItems.
-// Errors returned by a task do not halt execution,
-// but are joined into a multierror return value.
-// If a task panics during execution,
-// the panic will be caught and rethrown in the parent Goroutine.
-func eachN(numWorkers, numItems int, task func(int) error) error {
+func Each[Input any](numWorkers int, seq iter.Seq[Input], task func(Input) error) error {
 	type void struct{}
-	inch, ouch := TaskPool(numWorkers, func(pos int) (void, error) {
-		return void{}, task(pos)
+
+	inch, ouch := TaskPool(numWorkers, func(in Input) (void, error) {
+		return void{}, task(in)
 	})
-	var (
-		panicVal any
-		errs     []error
-	)
+
+	var errs []error
+
 	_ = Do(
 		func() error {
-			for i := 0; i < numItems; i++ {
-				inch <- i
+			defer close(inch)
+
+			for in := range seq {
+				inch <- in
 			}
-			close(inch)
 			return nil
 		},
 		func() error {
-			for r := range ouch {
-				if r.Panic != nil && panicVal == nil {
-					panicVal = r.Panic
+			for out := range ouch {
+				if out.Panic != nil {
+					panic(out.Panic)
 				}
-				if r.Err != nil {
-					errs = append(errs, r.Err)
+				if err := out.Err; err != nil {
+					errs = append(errs, err)
 				}
 			}
 			return nil
-		})
-	if panicVal != nil {
-		panic(panicVal)
-	}
+		},
+	)
 	return errors.Join(errs...)
 }

each_example_test.go

@@ -2,17 +2,18 @@ package flowmatic_test
 
 import (
 	"fmt"
+	"slices"
 	"time"
 
 	"github.com/earthboundkid/flowmatic/v2"
 )
 
 func ExampleEach() {
-	times := []time.Duration{
+	times := slices.Values([]time.Duration{
 		50 * time.Millisecond,
 		100 * time.Millisecond,
 		200 * time.Millisecond,
-	}
+	})
 	start := time.Now()
 	err := flowmatic.Each(3, times, func(d time.Duration) error {
 		time.Sleep(d)

each_test.go

@@ -2,6 +2,7 @@ package flowmatic_test
 
 import (
 	"errors"
+	"slices"
 	"testing"
 
 	"github.com/earthboundkid/flowmatic/v2"
@@ -10,7 +11,7 @@ import (
 func TestEach_err(t *testing.T) {
 	a := errors.New("a")
 	b := errors.New("b")
-	errs := flowmatic.Each(1, []int{1, 2, 3}, func(i int) error {
+	errs := flowmatic.Each(1, slices.Values([]int{1, 2, 3}), func(i int) error {
 		switch i {
 		case 1:
 			return a

panic_test.go

@@ -3,6 +3,7 @@ package flowmatic_test
 import (
 	"context"
 	"fmt"
+	"slices"
 	"sync/atomic"
 	"testing"
 
@@ -47,7 +48,7 @@ func TestEach_panic(t *testing.T) {
 		err error
 	)
 	r := try(func() {
-		err = flowmatic.Each(1, []int64{1, 2, 3},
+		err = flowmatic.Each(1, slices.Values([]int64{1, 2, 3}),
 			func(delta int64) error {
 				if delta == 2 {
 					panic("boom")