diff options
| author | sangtran <tranquangsang12.7@gmail.com> | 2026-09-03 23:10:33 +0700 |
|---|---|---|
| committer | sangtran <tranquangsang12.7@gmail.com> | 2026-09-03 23:10:33 +0700 |
| commit | 02123df8cc15896f97c1580e49d8e2a991e37ddd (patch) | |
| tree | d0c00bc3f65e5f5493a9dff8034c80640cb12960 /pipeline/pipeline.go | |
| signature | Verified sangtran <tranquangsang12.7@gmail.com> Key: 8DD79823D235E1BB ultimate trust | |
| download | quacker-02123df8cc15896f97c1580e49d8e2a991e37ddd.tar.gz quacker-02123df8cc15896f97c1580e49d8e2a991e37ddd.zip | |
Diffstat (limited to 'pipeline/pipeline.go')
| -rw-r--r-- | pipeline/pipeline.go | 21 |
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 @@ | |||
| 1 | // Package pipeline provides composable, type-safe stream processing with backpressure. | ||
| 2 | package pipeline | ||
| 3 | |||
| 4 | import "context" | ||
| 5 | |||
| 6 | // Stage transforms an input channel into an output channel. | ||
| 7 | // The stage owns the output channel and closes it when the input | ||
| 8 | // is exhausted or the context is cancelled. | ||
| 9 | type Stage[T any] func(ctx context.Context, in <-chan T) <-chan T | ||
| 10 | |||
| 11 | // Run connects stages into a pipeline and returns the final output channel. | ||
| 12 | func Run[T any](ctx context.Context, input <-chan T, stages ...Stage[T]) <-chan T { | ||
| 13 | if input == nil { | ||
| 14 | panic("pipeline: input channel must not be nil") | ||
| 15 | } | ||
| 16 | ch := input | ||
| 17 | for _, stage := range stages { | ||
| 18 | ch = stage(ctx, ch) | ||
| 19 | } | ||
| 20 | return ch | ||
| 21 | } | ||
