Pipeline

The problem

You have a stream of data and several transformations to apply. Doing them in one big loop is simple but serial — and mixes the steps together. A pipeline gives each stage its own goroutine, connected by channels, so stages run concurrently, stay decoupled, and get back-pressure and cancellation for free.

Structure

graph LR
G["generator<br/>goroutine"] -->|chan int| S1["square<br/>goroutine"]
S1 -->|chan int| S2["square<br/>goroutine"]
S2 -->|chan int| M["main<br/>range loop"]
D["done chan"] -.cancels.-> G
D -.cancels.-> S1
D -.cancels.-> S2

Each stage follows the same shape: take an inbound channel, return a new outbound channel, run a goroutine that reads → transforms → sends, and close the output when the input is exhausted.

Idiomatic Go

A generator turns values into a stream; each stage reads one channel and writes the next; main ranges the final channel. The done channel lets everything cancel cleanly. Edit and Run:

package main

import "fmt"

// generator turns a list of ints into a stream.
func generator(done <-chan struct{}, nums ...int) <-chan int {
out := make(chan int)
go func() {
	defer close(out) // this stage owns out, so it closes out
	for _, n := range nums {
		select {
		case <-done:
			return
		case out <- n:
		}
	}
}()
return out
}

// square is a stage: read ints, emit their squares.
func square(done <-chan struct{}, in <-chan int) <-chan int {
out := make(chan int)
go func() {
	defer close(out)
	for n := range in {
		select {
		case <-done:
			return
		case out <- n * n:
		}
	}
}()
return out
}

func main() {
done := make(chan struct{})
defer close(done) // cancels every stage on return

// Compose stages: numbers -> square -> square
stream := square(done, square(done, generator(done, 1, 2, 3, 4)))

for n := range stream {
	fmt.Println(n) // 1, 16, 81, 256
}
}

Generators & take

A more composable style, popularized by Concurrency in Go, uses a repeatFn that produces an infinite stream and a take that pulls just the first N — both honoring done:

done := make(chan any)
defer close(done)

repeat := repeatFn(done, func() any { return rand.Int() }) // infinite
for n := range take(done, repeat, 10) {                    // bounded
	fmt.Println(n)
}

Because every stage selects on <-done, closing done unwinds even the infinite generator — no leaked goroutines.

From done channel to context

In production you’ll usually swap the done channel for a context.Context — the select pattern is identical, but you also get timeouts, deadlines, and propagation across API boundaries:

func square(ctx context.Context, in <-chan int) <-chan int {
	out := make(chan int)
	go func() {
		defer close(out)
		for n := range in {
			select {
			case <-ctx.Done(): // timeout, deadline, or cancel — all in one
				return
			case out <- n * n:
			}
		}
	}()
	return out
}

Where it leads

The pipeline is the foundation for the rest of Go’s concurrency toolkit:

• Parallelize a slow stage with Fan-out / Fan-in.
• Bound how many items are in flight with a Worker Pool.
• Replace the done channel with context.Context for deadlines and request scope.
• Wrap a stream so it always stops on cancel with the Or-done channel.

Pitfalls

When to use it — and when not

Related patterns