Pipelines in Go: Small Stages, Composable Workflows

A pipeline splits a task into stages and connects them so each stage feeds the next. Anyone who has used a Unix shell has already worked this way: small commands are joined together, and the combination does more than any single command. The same idea shows up in streaming systems, functional programming, and API gateways that orchestrate microservice calls. In Go, pipelines are useful because they let code follow single-responsibility boundaries: each stage can stay small, cohesive, and loosely coupled, then be assembled into a larger workflow.

Wrapping HTTP handlers

A set of decorators such as WithServerHeader(), WithBasicAuth(), and WithDebugLog() can be composed for an API. Direct nesting looks like this:

http.HandleFunc("/v1/hello", WithServerHeader(WithAuthCookie(hello)))
http.HandleFunc("/v2/hello", WithServerHeader(WithBasicAuth(hello)))
http.HandleFunc("/v3/hello", WithServerHeader(WithBasicAuth(WithDebugLog(hello))))

Add a proxy function:

type HttpHandlerDecorator func(http.HandlerFunc) http.HandlerFunc
func Handler(h http.HandlerFunc, decors ...HttpHandlerDecorator) http.HandlerFunc {
    for i := range decors {
        d := decors[len(decors)-1-i] // iterate in reverse
        h = d(h)
    }
    return h
}

The loop walks the decorators in reverse, so the first decorator listed ends up closest to the original handler. That removes repeated nesting:

http.HandleFunc("/v4/hello", Handler(hello,
                WithServerHeader, WithBasicAuth, WithDebugLog))

Channel stages

A generic pipeline framework is not trivial to write by hand, and Go generation can help, but Go's goroutines and channels are already strong building blocks for this pattern. Rob Pike's Go Concurrency Patterns: Pipelines and cancellation describes the approach.

Start with an echo() function that puts an integer slice into a channel and returns that channel:

func echo(nums []int) <-chan int {
  out := make(chan int)
  go func() {
    for _, n := range nums {
      out <- n
    }
    close(out)
  }()
  return out
}

Following the same shape, a square stage:

func sq(in <-chan int) <-chan int {
  out := make(chan int)
  go func() {
    for n := range in {
      out <- n * n
    }
    close(out)
  }()
  return out
}

A stage that filters odd numbers:

func odd(in <-chan int) <-chan int {
  out := make(chan int)
  go func() {
    for n := range in {
      if n%2 != 0 {
        out <- n
      }
    }
    close(out)
  }()
  return out
}

And a summing stage:

func sum(in <-chan int) <-chan int {
  out := make(chan int)
  go func() {
    var sum = 0
    for n := range in {
      sum += n
    }
    out <- sum
    close(out)
  }()
  return out
}

The caller can now compose them directly:

var nums = []int{1, 2, 3, 4, 5, 6, 7, 8, 9, 10}
for n := range sum(sq(odd(echo(nums)))) {
  fmt.Println(n)
}

That has the same shape as a shell pipeline such as echo $nums | sq | sum. If the nesting becomes too deep, a proxy function can assemble the stages:

type EchoFunc func ([]int) (<- chan int)
type PipeFunc func (<- chan int) (<- chan int)

func pipeline(nums []int, echo EchoFunc, pipeFns ... PipeFunc) <- chan int {
  ch  := echo(nums)
  for i := range pipeFns {
    ch = pipeFns[i](ch)
  }
  return ch
}

Then the call site becomes:

var nums = []int{1, 2, 3, 4, 5, 6, 7, 8, 9, 10}
for n := range pipeline(nums, gen, odd, sq, sum) {
    fmt.Println(n)
  }

The stages shown above are deliberately similar. They could be merged with a map/reduce pattern or generated code if the repetition becomes a problem.

Fan in and fan out

Goroutines and channels also make it possible to write one-to-many and many-to-one pipelines: fan-out and fan-in. A fan-in example is a concurrent prime sum over a long array. The array is split across workers, the partial sums are computed, and the results are merged.

The main function:

func makeRange(min, max int) []int {
  a := make([]int, max-min+1)
  for i := range a {
    a[i] = min + i
  }
  return a
}

func main() {
  nums := makeRange(1, 10000)
  in := echo(nums)

  const nProcess = 5
  var chans [nProcess]<-chan int
  for i := range chans {
    chans[i] = sum(prime(in))
  }

  for n := range sum(merge(chans[:])) {
    fmt.Println(n)
  }
}

The prime stage:

func is_prime(value int) bool {
  for i := 2; i <= int(math.Floor(float64(value) / 2)); i++ {
    if value%i == 0 {
      return false
    }
  }
  return value > 1
}

func prime(in <-chan int) <-chan int {
  out := make(chan int)
  go func ()  {
    for n := range in {
      if is_prime(n) {
        out <- n
      }
    }
    close(out)
  }()
  return out
}

The flow is:

• Build an array from 1 to 10000.
• Echo the whole array into a channel, in.
• Create five channels, each running sum(prime(in)); every sum goroutine starts computing.
• Merge and sum all partial results to get the final value.

The merge function:

func merge(cs []<-chan int) <-chan int {
  var wg sync.WaitGroup
  out := make(chan int)

  wg.Add(len(cs))
  for _, c := range cs {
    go func(c <-chan int) {
      for n := range c {
        out <- n
      }
      wg.Done()
    }(c)
  }
  go func() {
    wg.Wait()
    close(out)
  }()
  return out
}

This structure keeps each stage focused and lets the runtime handle the concurrency between them. The same approach applies whenever a stream of work can be divided into independent stages and reassembled.