fanout/main.go

package main

import (
	"bufio"
	"fmt"
	"io"
	"os"
	"sync"
	"time"
)

func compute(id int, work <-chan string) <-chan error {
	out := make(chan error)
	go func() {
		for w := range work {
			time.Sleep(1 * time.Second)
			out <- fmt.Errorf("worker %d is done with %q", id, w)
		}
		close(out)
	}()
	return out
}

func source(in io.Reader) <-chan string {
	out := make(chan string)
	s := bufio.NewScanner(in)
	go func() {
		for s.Scan() {
			out <- s.Text()
		}
		if err := s.Err(); err != nil {
			panic(err)
		}
		close(out)
	}()
	return out
}

func merge(cs []<-chan error) <-chan error {
	out := make(chan error)

	var wg sync.WaitGroup

	output := func(c <-chan error) {
		for n := range c {
			out <- n
		}
		wg.Done()
	}

	wg.Add(len(cs))
	for _, c := range cs {
		go output(c)
	}

	go func() {
		wg.Wait()
		close(out)
	}()
	return out
}

func main() {
	work := source(os.Stdin)
	results := []<-chan error{}

	workers := 4
	for w := 0; w < workers; w++ {
		results = append(results, compute(w, work))
	}

	out := merge(results)

	for e := range out {
		fmt.Printf("received: %+v\n", e)
	}
}