feat(sequence): joins many processors with the same input and output

sequence.go

@@ -0,0 +1,26 @@
+package pipeline
+
+import "context"
+
+type sequence[A any] []Processor[A, A]
+
+func (s sequence[A]) Process(ctx context.Context, a A) (A, error) {
+	var zero A
+	var in = a
+	for _, p := range s {
+		if out, err := p.Process(ctx, in); err != nil {
+			p.Cancel(in, err)
+			return zero, err
+		} else {
+			in = out
+		}
+	}
+	return in, nil
+}
+
+func (s sequence[A]) Cancel(_ A, _ error) {}
+
+// Sequence connects many processors sequentially where the inputs are the same outputs
+func Sequence[A any](ps ...Processor[A, A]) Processor[A, A] {
+	return sequence[A](ps)
+}

sequence_test.go

@@ -0,0 +1,25 @@
+package pipeline
+
+import (
+	"context"
+	"testing"
+)
+
+func TestSequence(t *testing.T) {
+	// Create a step that increments the integer by 1
+	inc := NewProcessor(func(_ context.Context, i int) (int, error) {
+		i++
+		return i, nil
+	}, nil)
+	// Add 5 to every number
+	inc5 := Sequence(inc, inc, inc, inc, inc)
+	// Verify the sequence ads 5 to each number
+	var i int
+	want := []int{5, 6, 7, 8, 9}
+	for o := range Process(context.Background(), inc5, Emit(0, 1, 2, 3, 4)) {
+		if want[i] != o {
+			t.Fatalf("[%d] = %d, want %d", i, o, want[i])
+		}
+		i++
+	}
+}