Extract common things to a separate package (#1)

chans/batch_test.go

@@ -48,7 +48,7 @@ func TestBatch(t *testing.T) {
 		th.ExpectSlice(t, outSlice[3], []int{10})
 	})
 
-	t.Run("slow w/o timeout", func(t *testing.T) {
+	t.Run("slow wo timeout", func(t *testing.T) {
 		in := make(chan int)
 		go func() {
 			defer close(in)
@@ -65,6 +65,19 @@ func TestBatch(t *testing.T) {
 		th.ExpectSlice(t, outSlice[1], []int{5, 6, 7, 8})
 		th.ExpectSlice(t, outSlice[2], []int{9, 10})
 	})
+
+	for _, timeout := range []time.Duration{-1, 10 * time.Second} {
+		t.Run(th.Name("ordering", timeout), func(t *testing.T) {
+			in := th.FromRange(0, 20000)
+
+			out := Batch(in, 1000, timeout)
+
+			ForEach(out, 1, func(batch []int) bool {
+				th.ExpectSorted(t, batch)
+				return !t.Failed()
+			})
+		})
+	}
 }
 
 func TestUnbatch(t *testing.T) {

chans/core.go

@@ -1,94 +1,43 @@
 package chans
 
-func MapAndFilter[A, B any](in <-chan A, n int, f func(A) (B, bool)) <-chan B {
-	if in == nil {
-		return nil
-	}
-
-	out := make(chan B)
-
-	loop(in, out, n, func(x A) {
-		y, keep := f(x)
-		if keep {
-			out <- y
-		}
-	})
-
-	return out
-}
-
-func OrderedMapAndFilter[A, B any](in <-chan A, n int, f func(A) (B, bool)) <-chan B {
-	if in == nil {
-		return nil
-	}
-
-	out := make(chan B)
-	orderedLoop(in, out, n, func(a A, canWrite <-chan struct{}) {
-		y, keep := f(a)
-		<-canWrite
-		if keep {
-			out <- y
-		}
-	})
-
-	return out
-}
+import "github.com/destel/rill/internal/common"
 
 func Map[A, B any](in <-chan A, n int, f func(A) B) <-chan B {
-	return MapAndFilter(in, n, func(a A) (B, bool) {
+	return common.MapOrFilter(in, n, func(a A) (B, bool) {
 		return f(a), true
 	})
 }
 
 func OrderedMap[A, B any](in <-chan A, n int, f func(A) B) <-chan B {
-	return OrderedMapAndFilter(in, n, func(a A) (B, bool) {
+	return common.OrderedMapOrFilter(in, n, func(a A) (B, bool) {
 		return f(a), true
 	})
 }
 
 func Filter[A any](in <-chan A, n int, f func(A) bool) <-chan A {
-	return MapAndFilter(in, n, func(a A) (A, bool) {
+	return common.MapOrFilter(in, n, func(a A) (A, bool) {
 		return a, f(a)
 	})
 }
 
 func OrderedFilter[A any](in <-chan A, n int, f func(A) bool) <-chan A {
-	return OrderedMapAndFilter(in, n, func(a A) (A, bool) {
+	return common.OrderedMapOrFilter(in, n, func(a A) (A, bool) {
 		return a, f(a)
 	})
 }
 
 func FlatMap[A, B any](in <-chan A, n int, f func(A) <-chan B) <-chan B {
-	if in == nil {
-		return nil
-	}
-
-	out := make(chan B)
-
-	loop(in, out, n, func(a A) {
-		for b := range f(a) {
-			out <- b
-		}
+	var zero B
+	return common.MapOrFlatMap(in, n, func(a A) (b B, bb <-chan B, flat bool) {
+		return zero, f(a), true
 	})
-
-	return out
 }
 
 func OrderedFlatMap[A, B any](in <-chan A, n int, f func(A) <-chan B) <-chan B {
-	if in == nil {
-		return nil
-	}
-
-	out := make(chan B)
-	orderedLoop(in, out, n, func(a A, canWrite <-chan struct{}) {
-		bb := f(a)
-		<-canWrite
-		for b := range bb {
-			out <- b
-		}
+	var zero B
+	return common.OrderedMapOrFlatMap(in, n, func(a A) (b B, bb <-chan B, flat bool) {
+		return zero, f(a), true
 	})
-
-	return out
 }
 
 // blocking
@@ -104,12 +53,12 @@ func ForEach[A any](in <-chan A, n int, f func(A) bool) {
 		return
 	}
 
-	in, doBreak := breakable(in)
+	in, earlyExit := common.Breakable(in)
 	done := make(chan struct{})
 
-	loop(in, done, n, func(a A) {
+	common.Loop(in, done, n, func(a A) {
 		if !f(a) {
-			doBreak()
+			earlyExit()
 		}
 	})