1 files changed, 21 insertions, 0 deletions

diff --git a/pipeline/pipeline.go b/pipeline/pipeline.go
new file mode 100644
index 0000000..170a04e
--- /dev/null
+++ b/pipeline/pipeline.go
@@ -0,0 +1,21 @@
+// Package pipeline provides composable, type-safe stream processing with backpressure.
+package pipeline
+
+import "context"
+
+// Stage transforms an input channel into an output channel.
+// The stage owns the output channel and closes it when the input
+// is exhausted or the context is cancelled.
+type Stage[T any] func(ctx context.Context, in <-chan T) <-chan T
+
+// Run connects stages into a pipeline and returns the final output channel.
+func Run[T any](ctx context.Context, input <-chan T, stages ...Stage[T]) <-chan T {
+        if input == nil {
+                panic("pipeline: input channel must not be nil")
+        }
+        ch := input
+        for _, stage := range stages {
+                ch = stage(ctx, ch)
+        }
+        return ch
+}