Channel Pipeline (Go)
Topics: channels, goroutines, generics, context
Problem
Feed items from src through a sequence of Pipe stages. Each stage reads from an input channel,
transforms items, and writes to a new output channel; stages run concurrently — stage N+1 starts
consuming as soon as stage N starts producing.
type Pipe[T any] func(ctx context.Context, in <-chan T) <-chan T
func Run[T any](ctx context.Context, src []T, pipes ...Pipe[T]) []T
- Turn
src into a source channel (a goroutine sends each item, then closes it).
- Chain the pipes: the output channel of stage N becomes the input of stage N+1.
- Collect and return the final channel's values in the original order.
- Pass
ctx into each stage so the pipeline can stop early on cancellation.
- An empty
src (or no pipes) returns a non-nil empty slice.
src=[1,2,3], pipes=[] → [1,2,3]
src=[1,2,3], pipes=[double] → [2,4,6]
src=[1,2,3], pipes=[double, addOne] → [3,5,7]
Threading the stages is a single loop: ch = pipe(ctx, ch). The discipline that makes it work is
closing: each Pipe must close its output channel when in is drained or ctx is done, so the
next stage's range loop ends and shutdown cascades from the source down to the collector.
Sign in to submit your solution.