Validate arguments at the public API boundary

batch.go

@@ -20,6 +20,8 @@ import (
 //
 // See the package documentation for more information on non-blocking ordered functions and error handling.
 func Batch[A any](in <-chan Try[A], size int, timeout time.Duration) <-chan Try[[]A] {
+	validateMinSize(size, 1)
+
 	values, errs := ToChans(in)
 	batches := core.Batch(values, size, timeout)
 	return FromChans(batches, errs)

consume.go

@@ -15,6 +15,9 @@ import (
 //
 // See the package documentation for more information on blocking unordered functions and error handling.
 func ForEach[A any](in <-chan Try[A], n int, f func(A) error) error {
+	validateN(n)
+	validateNilFunc(f == nil)
+
 	// The n = 1 path is an internal contract, not just an optimization.
 	// Other sinks (Any, Reduce) build their n = 1 behavior on it and rely on:
 	//   - items processed sequentially, in stream order
@@ -96,6 +99,9 @@ var errFound = errors.New("found")
 //
 // See the package documentation for more information on blocking unordered functions and error handling.
 func Any[A any](in <-chan Try[A], n int, f func(A) (bool, error)) (bool, error) {
+	validateN(n)
+	validateNilFunc(f == nil)
+
 	err := ForEach(in, n, func(a A) error {
 		ok, err := f(a)
 		if err != nil {
@@ -122,6 +128,9 @@ func Any[A any](in <-chan Try[A], n int, f func(A) (bool, error)) (bool, error)
 //
 // See the package documentation for more information on blocking unordered functions and error handling.
 func All[A any](in <-chan Try[A], n int, f func(A) (bool, error)) (bool, error) {
+	validateN(n)
+	validateNilFunc(f == nil)
+
 	err := ForEach(in, n, func(a A) error {
 		ok, err := f(a)
 		if err != nil {

iter.go

@@ -19,6 +19,8 @@ func FromSeq[A any](seq iter.Seq[A], err error) <-chan Try[A] {
 		return out
 	}
 
+	validateNilFunc(seq == nil)
+
 	out := make(chan Try[A])
 	go func() {
 		for val := range seq {
@@ -31,6 +33,8 @@ func FromSeq[A any](seq iter.Seq[A], err error) <-chan Try[A] {
 
 // FromSeq2 converts an iterator of value-error pairs into a stream.
 func FromSeq2[A any](seq iter.Seq2[A, error]) <-chan Try[A] {
+	validateNilFunc(seq == nil)
+
 	out := make(chan Try[A])
 	go func() {
 		for val, err := range seq {

iter_test.go

@@ -112,6 +112,17 @@ func TestFromSeq(t *testing.T) {
 
 		th.ExpectSlice(t, outSlice, expectedSlice)
 	})
+
+	t.Run("error with nil iterator", func(t *testing.T) {
+		out := FromSeq[int](nil, errors.New("some error"))
+
+		outSlice := toItemSlice(out)
+
+		var expectedSlice []Item[int]
+		expectedSlice = appendErr(expectedSlice, errors.New("some error"))
+
+		th.ExpectSlice(t, outSlice, expectedSlice)
+	})
 }
 
 func TestFromSeq2(t *testing.T) {

reduce.go

@@ -122,6 +122,9 @@ func reduceStage[A any](in Stream[A], n int, f func(A, A) (A, error)) Stream[A]
 //
 // See the package documentation for more information on blocking unordered functions and error handling.
 func Reduce[A any](in <-chan Try[A], n int, f func(A, A) (A, error)) (result A, hasResult bool, err error) {
+	validateN(n)
+	validateNilFunc(f == nil)
+
 	var zero A
 
 	if n == 1 {
@@ -247,6 +250,11 @@ func mapReduceStage[A any, K comparable, V any](in Stream[A], nm int, mapper fun
 //
 // See the package documentation for more information on blocking unordered functions and error handling.
 func MapReduce[A any, K comparable, V any](in <-chan Try[A], nm int, mapper func(A) (K, V, error), nr int, reducer func(V, V) (V, error)) (map[K]V, error) {
+	validateN(nm)
+	validateNilFunc(mapper == nil)
+	validateN(nr)
+	validateNilFunc(reducer == nil)
+
 	if nm == 1 && nr == 1 {
 		m := make(map[K]V)
 		err := ForEach(in, 1, func(a A) error {

transform.go

@@ -12,6 +12,9 @@ import (
 //
 // See the package documentation for more information on non-blocking unordered functions and error handling.
 func Map[A, B any](in <-chan Try[A], n int, f func(A) (B, error)) <-chan Try[B] {
+	validateN(n)
+	validateNilFunc(f == nil)
+
 	return core.FilterMap(in, n, func(a Try[A]) (Try[B], bool) {
 		if a.Error != nil {
 			return Try[B]{Error: a.Error}, true
@@ -28,6 +31,9 @@ func Map[A, B any](in <-chan Try[A], n int, f func(A) (B, error)) <-chan Try[B]
 
 // OrderedMap is the ordered version of [Map].
 func OrderedMap[A, B any](in <-chan Try[A], n int, f func(A) (B, error)) <-chan Try[B] {
+	validateN(n)
+	validateNilFunc(f == nil)
+
 	return core.OrderedFilterMap(in, n, func(a Try[A]) (Try[B], bool) {
 		if a.Error != nil {
 			return Try[B]{Error: a.Error}, true
@@ -50,6 +56,9 @@ func OrderedMap[A, B any](in <-chan Try[A], n int, f func(A) (B, error)) <-chan
 //
 // See the package documentation for more information on non-blocking unordered functions and error handling.
 func Filter[A any](in <-chan Try[A], n int, f func(A) (bool, error)) <-chan Try[A] {
+	validateN(n)
+	validateNilFunc(f == nil)
+
 	return core.FilterMap(in, n, func(a Try[A]) (Try[A], bool) {
 		if a.Error != nil {
 			return a, true // never filter out errors
@@ -66,6 +75,9 @@ func Filter[A any](in <-chan Try[A], n int, f func(A) (bool, error)) <-chan Try[
 
 // OrderedFilter is the ordered version of [Filter].
 func OrderedFilter[A any](in <-chan Try[A], n int, f func(A) (bool, error)) <-chan Try[A] {
+	validateN(n)
+	validateNilFunc(f == nil)
+
 	return core.OrderedFilterMap(in, n, func(a Try[A]) (Try[A], bool) {
 		if a.Error != nil {
 			return a, true // never filter out errors
@@ -89,6 +101,9 @@ func OrderedFilter[A any](in <-chan Try[A], n int, f func(A) (bool, error)) <-ch
 //
 // See the package documentation for more information on non-blocking unordered functions and error handling.
 func FilterMap[A, B any](in <-chan Try[A], n int, f func(A) (B, bool, error)) <-chan Try[B] {
+	validateN(n)
+	validateNilFunc(f == nil)
+
 	return core.FilterMap(in, n, func(a Try[A]) (Try[B], bool) {
 		if a.Error != nil {
 			return Try[B]{Error: a.Error}, true
@@ -105,6 +120,9 @@ func FilterMap[A, B any](in <-chan Try[A], n int, f func(A) (B, bool, error)) <-
 
 // OrderedFilterMap is the ordered version of [FilterMap].
 func OrderedFilterMap[A, B any](in <-chan Try[A], n int, f func(A) (B, bool, error)) <-chan Try[B] {
+	validateN(n)
+	validateNilFunc(f == nil)
+
 	return core.OrderedFilterMap(in, n, func(a Try[A]) (Try[B], bool) {
 		if a.Error != nil {
 			return Try[B]{Error: a.Error}, true
@@ -127,6 +145,9 @@ func OrderedFilterMap[A, B any](in <-chan Try[A], n int, f func(A) (B, bool, err
 //
 // See the package documentation for more information on non-blocking unordered functions and error handling.
 func FlatMap[A, B any](in <-chan Try[A], n int, f func(A) <-chan Try[B]) <-chan Try[B] {
+	validateN(n)
+	validateNilFunc(f == nil)
+
 	if in == nil {
 		return nil
 	}
@@ -150,6 +171,9 @@ func FlatMap[A, B any](in <-chan Try[A], n int, f func(A) <-chan Try[B]) <-chan
 
 // OrderedFlatMap is the ordered version of [FlatMap].
 func OrderedFlatMap[A, B any](in <-chan Try[A], n int, f func(A) <-chan Try[B]) <-chan Try[B] {
+	validateN(n)
+	validateNilFunc(f == nil)
+
 	if in == nil {
 		return nil
 	}
@@ -185,6 +209,9 @@ func OrderedFlatMap[A, B any](in <-chan Try[A], n int, f func(A) <-chan Try[B])
 //
 // See the package documentation for more information on non-blocking unordered functions and error handling.
 func Catch[A any](in <-chan Try[A], n int, f func(error) error) <-chan Try[A] {
+	validateN(n)
+	validateNilFunc(f == nil)
+
 	return core.FilterMap(in, n, func(a Try[A]) (Try[A], bool) {
 		if a.Error == nil {
 			return a, true
@@ -201,6 +228,9 @@ func Catch[A any](in <-chan Try[A], n int, f func(error) error) <-chan Try[A] {
 
 // OrderedCatch is the ordered version of [Catch].
 func OrderedCatch[A any](in <-chan Try[A], n int, f func(error) error) <-chan Try[A] {
+	validateN(n)
+	validateNilFunc(f == nil)
+
 	return core.OrderedFilterMap(in, n, func(a Try[A]) (Try[A], bool) {
 		if a.Error == nil {
 			return a, true

util.go

@@ -1,6 +1,10 @@
 package rill
 
-import "github.com/destel/rill/internal/core"
+import (
+	"fmt"
+
+	"github.com/destel/rill/internal/core"
+)
 
 // Drain consumes and discards all items from an input channel, blocking until the channel is closed.
 func Drain[A any](in <-chan A) {
@@ -31,5 +35,25 @@ func DrainNB[A any](in <-chan A) {
 //	// Now work with the users channel as usual.
 //	// Up to 100 users can be buffered if subsequent stages of the pipeline are slow.
 func Buffer[A any](in <-chan A, size int) <-chan A {
+	validateMinSize(size, 1)
+
 	return core.Buffer(in, size)
 }
+
+func validateN(n int) {
+	if n < 1 {
+		panic(fmt.Sprintf("rill: n must be at least 1, got %d", n))
+	}
+}
+
+func validateMinSize(size int, minSize int) {
+	if size < minSize {
+		panic(fmt.Sprintf("rill: size must be at least %d, got %d", minSize, size))
+	}
+}
+
+func validateNilFunc(fIsNil bool) {
+	if fIsNil {
+		panic("rill: function must not be nil")
+	}
+}

util_test.go

@@ -44,3 +44,39 @@ func TestBuffer(t *testing.T) {
 		Drain(out)
 	})
 }
+
+func TestValidations(t *testing.T) {
+	t.Run("ok", func(t *testing.T) {
+		validateN(1)
+		validateMinSize(5, 5)
+		validateNilFunc(false)
+	})
+
+	t.Run("n too small", func(t *testing.T) {
+		defer func() {
+			if r := recover(); r == nil {
+				t.Errorf("expected panic")
+			}
+		}()
+
+		validateN(0)
+	})
+
+	t.Run("size too small", func(t *testing.T) {
+		defer func() {
+			if r := recover(); r == nil {
+				t.Errorf("expected panic")
+			}
+		}()
+		validateMinSize(5, 6)
+	})
+
+	t.Run("function is nil", func(t *testing.T) {
+		defer func() {
+			if r := recover(); r == nil {
+				t.Errorf("expected panic")
+			}
+		}()
+		validateNilFunc(true)
+	})
+}

wrap.go

@@ -209,6 +209,8 @@ func ToChans[A any](in <-chan Try[A]) (<-chan A, <-chan error) {
 //		stream <- rill.Try[int]{Error: someError}
 //	}()
 func Generate[A any](f func(send func(A), sendErr func(error))) <-chan Try[A] {
+	validateNilFunc(f == nil)
+
 	out := make(chan Try[A])
 	go func() {
 		defer close(out)