Pipeline Pattern in Golang

A pipeline is a concurrency pattern in Golang where you split a job into a series of stages, connect them with channels, and let each stage run in its own goroutine. Data flows in one direction: the output channel of one stage becomes the input channel of the next, so the stages run at the same time instead of one after another.

1. What is a Pipeline?

A pipeline has three kinds of stages. A source (or generator) produces values and sends them onto the first channel. Any number of middle stages read from an inbound channel, transform each value, and send the result onto an outbound channel. A sink reads the final channel and does something terminal with the values — print them, sum them, write them to a database.

• Every stage receives values from an inbound channel and sends values on an outbound channel
• Each stage runs in its own goroutine, so stage 2 can already be working on the first item while stage 1 is still producing the second
• When a stage's inbound channel is closed and drained, the stage closes its outbound channel and returns — the "close" signal propagates down the line

2. Analogy: A Bakery Production Line

Imagine a bread factory with a single conveyor belt running through several stations. Each station does exactly one job to whatever passes in front of it, then puts it back on the belt for the next station.

• Flour hopper = the source stage, dropping raw batches of dough onto the belt one by one
• Kneading station = stage 1: takes a raw batch, kneads it, puts the kneaded dough back on the belt
• Proofing / shaping station = stage 2: takes kneaded dough, shapes it into a loaf, sends it on
• Oven = stage 3: takes a raw loaf, bakes it, sends out a finished loaf
• Packing table = the sink: takes finished loaves off the belt and boxes them

The important part is that every station works in parallel. While the oven bakes loaf #1, the shaping station is already shaping dough #2, and the kneading station is kneading batch #3. Nobody waits for the whole line to clear before starting the next batch.

• The belt segment between two stations = a channel connecting two stages
• A station worker = a goroutine running one stage
• The hopper running out of flour and being switched off = the source closing its channel
• Each station shutting down once the belt in front of it stops and clears = a stage closing its outbound channel after its inbound channel drains
• The line is fully stopped only when the last loaf reaches the packing table

3. The Source: The Flour Hopper

The source stage takes no inbound channel. It owns its outbound channel, sends every value, and closes the channel when there is nothing left to produce. Closing from the producing side is the rule that makes the rest of the pipeline shut down cleanly.

// hopper drops raw dough batches onto the belt, then switches off
func hopper(batches ...int) <-chan int {
	belt := make(chan int)
	go func() {
		defer close(belt) // flour ran out — stop the belt segment
		for _, b := range batches {
			belt <- b
		}
	}()
	return belt
}

4. Middle Stages: One Job Per Station

Every middle stage has the same shape: accept a receive-only inbound channel, create a new outbound channel, and spawn a goroutine that ranges over the inbound channel until it closes. Because range stops automatically when the channel is closed and drained, each stage's shutdown is triggered by the stage before it.

func knead(in <-chan int) <-chan int {
	out := make(chan int)
	go func() {
		defer close(out)
		for batch := range in {
			time.Sleep(50 * time.Millisecond) // kneading time
			out <- batch * 10 // kneaded dough id
		}
	}()
	return out
}

func shape(in <-chan int) <-chan int {
	out := make(chan int)
	go func() {
		defer close(out)
		for dough := range in {
			time.Sleep(40 * time.Millisecond) // shaping time
			out <- dough + 1 // shaped loaf id
		}
	}()
	return out
}

func bake(in <-chan int) <-chan string {
	out := make(chan string)
	go func() {
		defer close(out)
		for loaf := range in {
			time.Sleep(80 * time.Millisecond) // baking time
			out <- fmt.Sprintf("baked loaf %d", loaf)
		}
	}()
	return out
}

Notice that a stage can change the channel's element type. knead and shape pass int down the belt, but bake turns each loaf into a string. The pipeline does not care, as long as each stage's output type matches the next stage's input type.

5. Putting It Together

Assembling the pipeline is just nesting the stage functions. Each call returns the channel the next stage should read from, and main() acts as the packing table reading the final channel.

func main() {
	start := time.Now()

	// Build the line: hopper -> knead -> shape -> bake -> packing table
	loaves := bake(shape(knead(hopper(1, 2, 3, 4, 5))))

	for loaf := range loaves {
		fmt.Println(loaf)
	}

	fmt.Printf("line finished in %s\n", time.Since(start))
}

Run serially, five batches through three 50–80ms steps would take well over a second. In the pipeline the stages overlap, so the total time is roughly the slowest stage multiplied by the number of items plus a one-item warm-up — close to the time of the oven alone.

6. Stopping the Line Early

If the packing table stops taking loaves — an error, a full truck, a cancelled order — the upstream stages would block forever trying to send onto a belt nobody reads. The fix is a done channel (or a context.Context) that every stage watches with a select, so a close signal can also travel upstream.

func bake(done <-chan struct{}, in <-chan int) <-chan string {
	out := make(chan string)
	go func() {
		defer close(out)
		for loaf := range in {
			time.Sleep(80 * time.Millisecond)
			select {
			case out <- fmt.Sprintf("baked loaf %d", loaf):
			case <-done: // line is being shut down — abandon this loaf
				return
			}
		}
	}()
	return out
}

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

	loaves := bake(done, shape(done, knead(done, hopper(done, 1, 2, 3, 4, 5))))

	for loaf := range loaves {
		fmt.Println(loaf)
		break // stop after the first loaf; deferred close(done) drains the rest
	}
}

With done wired through every stage, closing it once makes each select fall through to the <-done case, every goroutine returns, and every outbound channel is closed by its defer. No goroutine is leaked even when the consumer quits early.

7. When to Use This Pattern

• The work is a sequence of transforms where each step depends on the previous step's output
• Steps take meaningfully different amounts of time and you want them to overlap instead of queueing
• You want each step isolated and testable on its own, wired together only by channel types
• Combine it with fan-out / fan-in when a single stage (like the oven) is the bottleneck and can be safely parallelised
• Avoid it when the steps are trivially fast — channel and goroutine overhead can outweigh a plain loop

8. Practice Exercises

Try solving these on your own before checking the reference solutions. Each one builds a multi-stage pipeline connected only by channels.

// Exercise 1: Text Processing Pipeline
// Build a 3-stage pipeline that processes a list of sentences.
//
// Requirements:
// 1. Stage 1 — Generator: generateSentences(sentences ...string) <-chan string
//    Send each sentence from a slice you create (at least 5 sentences, any
//    topic you like).
// 2. Stage 2 — Filter: filterShort(in <-chan string) <-chan string
//    Only pass through sentences whose length (len()) is greater than 15
//    characters. Short sentences are skipped (not sent to the next stage).
// 3. Stage 3 — Transform: toUpperCase(in <-chan string) <-chan string
//    Turn each sentence that passed through into all upper case
//    (strings.ToUpper).
// 4. In main, wire the three stages together, read the final output, and
//    print every processed sentence.
//
// Extra conditions:
// 1. Print a line before each stage sends its data (like the example above:
//    [generateSentences] ..., [filterShort] ..., [toUpperCase] ...) so you
//    can see the interleaving pattern.
// 2. Make sure sentences skipped in Stage 2 never appear in the final
//    output at all.

// Exercise 2: Sensor Data Validation & Conversion Pipeline
// Build a 4-stage pipeline that processes temperature readings from a sensor
// (in Celsius), then converts them to Fahrenheit — but some readings are
// invalid and must be FLAGGED rather than silently dropped.
//
// Requirements:
// 1. Stage 1 — Generator: generateReadings(temps ...float64) <-chan float64
//    Send each temperature from a slice (at least 8 values, mixing valid ones
//    with nonsensical ones — e.g. below -100°C or above 100°C counts as a
//    sensor error).
// 2. Stage 2 — Validation: validate(in <-chan float64) <-chan Reading
//    Check each temperature and wrap it in a struct:
//        type Reading struct {
//            Celsius  float64
//            Valid    bool
//            ErrorMsg string
//        }
//    Rule: a reading is valid when -100 <= celsius <= 100. If it is invalid,
//    fill in ErrorMsg (e.g. "temperature out of reasonable range"), but STILL
//    send it to the next stage — unlike the filter exercise, here every
//    reading continues down the line, just marked valid or not.
// 3. Stage 3 — Conversion: convertToFahrenheit(in <-chan Reading) <-chan Reading
//    Add a new field:
//        type Reading struct {
//            Celsius    float64
//            Fahrenheit float64 // new field
//            Valid      bool
//            ErrorMsg   string
//        }
//    Only convert (C * 9/5 + 32) when Valid == true. When Valid == false,
//    leave Fahrenheit at 0 (do not compute it, since the data is nonsense).
// 4. In main, read from the last stage, print every Reading (both valid and
//    invalid, with a different format so it is obvious which ones are errors),
//    then at the end print the total count of valid vs invalid readings.

Conclusion

A pipeline turns a sequence of steps into a set of goroutines connected by channels, each doing one job and passing the result along. Keep the bakery in mind: a single belt, one job per station, every station working at once, and a done signal that can stop the whole line from either end.