Simple example of fanout.
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
 

77 lines
1.1 KiB

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)
}
}