How to Implement a Fan-In Pattern in Go

Fan-In Pattern

Fan-in is a concurrency pattern that allows merging input of several goroutines into one. For example, we have two goroutines Alice and Bob, and each goroutine sends messages and we want to read it in one function.

Let’s implement it in 2 steps.

A generator function

First, we define a generator function that returns a channel.

package main

import "fmt"

func Say(who string, msgs ...string) <-chan string {
  c := make(chan string)
  go func() {
    for _, msg := range msgs {
      c <- fmt.Sprintf("%s said, %s", who, msg)
    }
    close(c)
  }()
  return c
}

A fan-in function

Next, implement a function that merges data from channels.

package main

func FanIn(msgs ...<-chan string) chan string {
	out := make(chan string)

	for _, each := range msgs {

		go func(input <-chan string) {
			for {
				// read value from input
				val, ok := <-input
				// break if channel is closed
				if !ok {
					break
				}
				out <- val
			}
		}(each)

	}
	return out
}

Main

Now, we can read data from several goroutines.

package main

import "fmt"

func main() {
	Alice := Say(
		"Alice", "hi", "how'r you doing?", "i'm good",
	)
	Bob := Say(
		"Bob", "hey", "what's up", "great!",
	)

	all := FanIn(Alice, Bob)

	for i := 0; i < 6; i++ {
		fmt.Println(<-all)
	}
}

Result

➜  fanin git:(main) ✗ go run .
Alice said, hi
Alice said, how'r you doing?
Bob said, hey
Bob said, what's up
Alice said, i'm good
Bob said, great!

Happy coding!

Code examples can be found in my GitHub repo pavel-fokin/go-patterns/concurrency/fanin.