golang fan-in fan-out

faninfanout.go

// faninfanout
package main

import (
	"fmt"
	"sync"
)

func source() <-chan int {
	ch1 := make(chan int)

	go func() {
		ch1 <- 3
		ch1 <- 4
		close(ch1)
	}()

	return ch1
}

func sq(chin <-chan int) <-chan int {
	chOut := make(chan int)

	go func() {
		for n := range chin {
			chOut <- n * n
			close(chOut)
		}
	}()

	return chOut
}

func main() {
	cmain := source()
	c1 := sq(cmain)
	c2 := sq(cmain)

	for n := range merge(c1, c2) {
		fmt.Println(n)
	}
}

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

	// Start an output goroutine for each input channel in cs.  output
	// copies values from c to out until c is closed, then calls wg.Done.
	output := func(c <-chan int) {
		for n := range c {
			out <- n
		}
		wg.Done()
	}
	wg.Add(len(cs))
	for _, c := range cs {
		go output(c)
	}

	// Start a goroutine to close out once all the output goroutines are
	// done.  This must start after the wg.Add call.
	go func() {
		wg.Wait()
		close(out)
	}()
	return out
}