aboutsummaryrefslogtreecommitdiff
path: root/pipeline/pipeline.go
diff options
context:
space:
mode:
authorsangtran <tranquangsang12.7@gmail.com>2026-09-03 23:10:33 +0700
committersangtran <tranquangsang12.7@gmail.com>2026-09-03 23:10:33 +0700
commit02123df8cc15896f97c1580e49d8e2a991e37ddd (patch)
treed0c00bc3f65e5f5493a9dff8034c80640cb12960 /pipeline/pipeline.go
signatureVerified sangtran <tranquangsang12.7@gmail.com> Key: 8DD79823D235E1BB ultimate trust
downloadquacker-02123df8cc15896f97c1580e49d8e2a991e37ddd.tar.gz
quacker-02123df8cc15896f97c1580e49d8e2a991e37ddd.zip
Verified Moving all legacy quacker source code from github to cgitHEADmain
Diffstat (limited to 'pipeline/pipeline.go')
-rw-r--r--pipeline/pipeline.go21
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.
2package pipeline
3
4import "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.
9type 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.
12func 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}