Fan-in merges values from multiple channels into one stream, allowing independent concurrent stages to feed a single downstream stage. section 9.2.3
A reliable fan-in implementation:
- Creates one forwarding goroutine for each input channel. section 9.2.3
- Sends every received value to a shared output channel. section 9.2.3
- Uses a sync.WaitGroup so the shared channel closes only after every input channel is drained. section 9.2.3
- Uses a quit channel to let forwarding goroutines stop before consuming all input. section 9.2.3
Accounting streams
A select loop reads interleaved sales and expenses, subtracts expenses from sales, and sets each channel to nil after closure. The loop ends when both channels are nil. This fixed-source form works for two known inputs; dynamic source counts require a separate fan-in function. section 8.1.6
Listing 9.10 Implementing a fan-in function
package listing9_10 import ( "sync" ) func FanIn[K any](quit <-chan int, allChannels ...<-chan K) chan K { wg := sync.WaitGroup{} #1 wg.Add(len(allChannels)) #1 output := make(chan K) #2 for _, c := range allChannels { go func(channel <-chan K) { #3 defer wg.Done() #4 for i := range channel { select { case output <- i: #5 case <-quit: #6 return #6 } } }(c) #7 } go func() { #8 wg.Wait() #8 close(output) #8 }() return output #9 } #1 Creates a waitgroup, setting the size to be equal to the number of input channels #2 Creates the output channel #3 Starts a goroutine for every input channel #4 Once the goroutine terminates, marks the waitgroup as done #5 Forwards each received message to the shared output channel #6 If quit channel is closed, terminates the goroutine #7 Passes one input channel to the goroutine #8 Waits for all the goroutines to finish and then closes the output channel #9 Returns the output channel
Inspect the complete FanIn function, including per-channel forwarding goroutines, cancellation, WaitGroup coordination, and output closure. section 9.2.3
