Go pipelines think

pipelines.go

package main

import (
	"fmt"
	"sync"
	"time"
)

func gen(vals ...int) <-chan int {
	out := make(chan int)
	go func() {
		for _, i := range vals {
			out <- i
		}
		close(out)
	}()
	return out
}

func sq(in <-chan int) <-chan int {
	out := make(chan int)
	go func() {
		for i := range in {
			out <- i * i
		}
		close(out)
	}()
	return out
}

func merge(inputStreams ...<-chan int) <-chan int {
	out := make(chan int)
	var wg sync.WaitGroup
  
	// goroutine Streamer for each incoming stream
        for _, inputStream := range inputStreams {
    
		wg.Add(1) // Hey you make sure the weightGroup knows about you
		go func(inputStream <-chan int) {
			for value := range inputStream {
				out <- value
			}
			wg.Done()  // ...and I'm done. Let the closer goroutine know we can close the channel.
		}(inputStream)
	
        }
 
	// Goroutine to close channels. Will wait for the streamers to be "Done()"
	go func() {
		wg.Wait()
		close(out)
	}()

	return out
}

func main() {
	fmt.Println("Hello, playground")

	r := gen(2, 4, 8, 10)
	r2 := gen(60, 600, 6000)
	for i := range merge(sq(r), r2) {
		fmt.Println(i)
	}

	time.Sleep(30)
}