From 02123df8cc15896f97c1580e49d8e2a991e37ddd Mon Sep 17 00:00:00 2001 From: sangtran Date: Thu, 3 Sep 2026 23:10:33 +0700 Subject: Moving all legacy quacker source code from github to cgit --- pipeline/pipeline.go | 21 +++++++++++++++++++++ 1 file changed, 21 insertions(+) create mode 100644 pipeline/pipeline.go (limited to 'pipeline/pipeline.go') 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 +} -- cgit v1.3.1