blob: 170a04eface68d185135780341b75d7031642598 (
plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
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
}
|
