Refactor with task.Feed

To avoid getting stuck when an error occurs in a downstream

pipe/fanout.go

@@ -35,8 +35,7 @@ func FanoutWithMap[S, I, T any](mapFn FanoutMapFn[I, T]) *FanoutOp[S, I, T] {
 		if err != nil {
 			return err
 		}
-		out <- o
-		return nil
+		return task.Feed(ctx, out, o)
 	}
 	return &FanoutOp[S, I, T]{aggregate: aggFn}
 }

pipe/map.go

@@ -39,7 +39,9 @@ func (op *MapOp[S, T]) pipeFn() PipeFn[S, T] {
 			if err != nil {
 				return err
 			}
-			out <- mapped
+			if err := task.Feed(ctx, out, mapped); err != nil {
+				return err
+			}
 		}
 		return nil
 	}

pipe/map_test.go

@@ -31,7 +31,7 @@ func TestMap(t *testing.T) {
 		out := make([]string, 0)
 		sin := sink.ToSlice(&out)
 
-		mapped := task.Connect(src.AsTask(), mappingTask, 2)
+		mapped := task.Connect(src.AsTask(), mappingTask, 0)
 		toSlice := task.Connect(mapped, sin.AsTask(), 2)
 
 		err := flow.FromTask(toSlice).Run(context.Background())

source/pull.go

@@ -55,9 +55,11 @@ func (op *PullOp[T]) AsSource() Source[T] {
 			for _, msg := range resp.Messages {
 				el, err := op.marshalSpec.Unmarshal(msg.Value)
 				if err != nil {
-					return nil
+					return err
+				}
+				if err := task.Feed(ctx, out, el); err != nil {
+					return err
 				}
-				out <- el
 			}
 		}
 		return nil

source/slice.go

@@ -16,7 +16,9 @@ func FromSlice[T any](items []T) *SliceSourceOp[T] {
 	return &SliceSourceOp[T]{
 		run: func(ctx context.Context, out chan<- T) error {
 			for _, item := range items {
-				out <- item
+				if err := task.Feed(ctx, out, item); err != nil {
+					return err
+				}
 			}
 			return nil
 		},

task/feed.go

@@ -0,0 +1,12 @@
+package task
+
+import "context"
+
+func Feed[T any](ctx context.Context, out chan<- T, el T) error {
+	select {
+	case out <- el:
+		return nil
+	case <-ctx.Done():
+		return ctx.Err()
+	}
+}