go 52 lines · 8 steps

The fan-in pattern in Go channels

How to merge many channels into one and collect from several sources without leaks or deadlocks.

Explained by highlit
Advanced concurrency channels fan-in select waitgroup
1package pipeline
2 
3import "sync"
4 
5func mergeMetrics(sources ...<-chan Metric) <-chan Metric {
6 out := make(chan Metric)
7 var wg sync.WaitGroup
8 
9 drain := func(src <-chan Metric) {
10 defer wg.Done()
11 for m := range src {
12 out <- m
13 }
14 }
15 
16 wg.Add(len(sources))
17 for _, src := range sources {
18 go drain(src)
19 }
20 
21 go func() {
22 wg.Wait()
23 close(out)
24 }()
25 
26 return out
27}
28 
29func collect(primary, fallback <-chan Metric, done <-chan struct{}) []Metric {
30 var collected []Metric
31 for {
32 select {
33 case m, ok := <-primary:
34 if !ok {
35 primary = nil
36 break
37 }
38 collected = append(collected, m)
39 case m, ok := <-fallback:
40 if !ok {
41 fallback = nil
42 break
43 }
44 collected = append(collected, m)
45 case <-done:
46 return collected
47 }
48 if primary == nil && fallback == nil {
49 return collected
50 }
51 }
52}
01 / 01
STEP 01

Walkthrough

Space play ←→ step click any line
Three takeaways
  1. 1A WaitGroup that closes the output channel once all producers finish is the safe way to signal completion in a fan-in.
  2. 2Setting a channel variable to nil in a select disables that case, letting you wind down partial sources cleanly.
  3. 3Combining a done channel with drained-source detection gives you two independent ways to stop collecting.

Related explainers

go 
package streaming
 
import (
	"bufio"

Streaming NDJSON logs over HTTP in Go

http-streaming channels select
Advanced 10 steps
go 
package api
 
import (
	"crypto/sha256"

ETag conditional requests in Gin

http-caching etag conditional-requests
Intermediate 6 steps
go 
func (w *Watcher) resetDebounce(d time.Duration) {
	if !w.timer.Stop() {
		select {
		case <-w.timer.C:

Debouncing a stream of events in Go

debounce timers channels
Advanced 7 steps
go 
package logging
 
import (
	"context"

Deduplicating log attributes in Go's slog

decorator-pattern structured-logging immutability
Intermediate 8 steps
rust 
use std::collections::VecDeque;
use std::sync::{Arc, Condvar, Mutex};
use std::time::{Duration, Instant};
 

Building a counting semaphore in Rust

concurrency synchronization condition-variable
Advanced 9 steps
go 
package middleware
 
import (
	"net/http"

Per-plan export limits in Gin middleware

middleware rate-limiting authorization
Intermediate 7 steps

Share this explainer

Here's the card — post it anywhere.

The fan-in pattern in Go channels — share card
Made with highlit — turn any snippet into a walkthrough like this in about a minute.
Explain your code