package main

import (
	"fmt"
	"time"
)

func gen() <-chan int {
	ch := make(chan int, 100)

	go func() {
		defer close(ch)
		for i := 0; i < 20; i++ {
			fmt.Printf("gen %d\n", i)
			ch <- i
			time.Sleep(100 * time.Millisecond)
		}
	}()

	return ch
}

func mut(in <-chan int) <-chan int {
	ch := make(chan int, 200)

	go func() {
		defer close(ch)
		for v := range in {
			ch <- v * 10
		}
	}()

	return ch
}

func sub(ch <-chan int) {
	const BufLimit = 5
	buf := make([]int, 0, BufLimit)

	for v := range ch {
		fmt.Printf("sub %d\n", v)
		buf = append(buf, v)
		if len(buf) == BufLimit {
			flush(buf)
			buf = buf[:0]

		}
		time.Sleep(300 * time.Millisecond)
	}
}

func flush(buf []int) {
	fmt.Print("flush ")
	for _, v := range buf {
		fmt.Printf("%d ", v)
	}
	fmt.Println()
}

func main() {
	ch1 := gen()
	ch2 := mut(ch1)
	sub(ch2)
}