package main

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

func gen(nums ...int) <-chan int {
	genOut := make(chan int)
	go func() {
		for _, n := range nums {
			genOut <- n
		}
		fmt.Println("Input gen Channel number =>", len(genOut))
		close(genOut)
	}()
	return genOut
}

func sq(done <-chan struct{}, in <-chan int) <-chan int {
	sqOut := make(chan int)
	go func() {
		// 这个 close(sqOut) 一定要先写，执行的时候优先压入栈，待函数执行完成关闭 sqOut channel
		defer close(sqOut)
		for n := range in {
            // 利用 select {} 均衡调度 channel 
			select {
			case sqOut <- n * n:
				fmt.Printf("=> %v <= write into sqOut channel \n", n*n)
			case <-done:
				return
			}
		}
		//fmt.Printf("Wait close the chan => %v\n", len(sqOut))
	}()
	return sqOut
}

// merge Fan-In 函数合并多个结果
func merge(done <-chan struct{}, cs ...<-chan int) <-chan int {
	var wg sync.WaitGroup
	mergeOut := make(chan int, 1)

	output := func(c <-chan int) {
		defer wg.Done()
		for n := range c {
			select {
			case mergeOut <- n:
			case <-done:
				return
			}
		}
	}

	wg.Add(len(cs))

	for _, c := range cs {
		go output(c)
	}

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

}

// pfnumber 计算算数平方数
func pfnumber() {
    // 定义 don channel 用于终止 pipeline
	don := make(chan struct{}, 3)
	don <- struct{}{}
	don <- struct{}{}
	close(don)
    // 传入 don 通知发送方停止发送
	for n := range sq(don, sq(don, gen(3, 4, 2))) {
		fmt.Println("Last result ", n)
	}
	fmt.Println("============================================")
}

func fanInOut() {
	don := make(chan struct{}, 3)
	in := gen(2, 3)
	c1 := sq(don, in)
	c2 := sq(don, in)

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

	don <- struct{}{}
	don <- struct{}{}
	don <- struct{}{}
	fmt.Println("Finish channel len => ", len(don))
	<-don
	close(don)

}

func f1(i chan int) {
	fmt.Println(<-i)
}

func runf1() {
	out := make(chan int)
	go f1(out)
	time.Sleep(2 * time.Second)
	out <- 2
	time.Sleep(2 * time.Second)
}

func main() {
	//runf1()

	pfnumber()
    // FanIn and FanOut
	//fanInOut()

}

简单总结一下如何正确构建一个 pipeline:

• 当所有的发送已完成，stage 应该关闭输出 channel
• stage 应该持续从只读 channel 中读出数据，除非 channel 关闭或主动通知到发送方停止发送

Golang Pipeline Blog 译文

Golang Pipeline Blog

扇出、扇入

扇出模式优先的场景：

• 它不依赖于之前的 stage 计算的值
• 需要运行很长时间，例如：I/O 等待，远程调用，访问 REST full API等

扇入模式优先：

扇入意味着多个数据流复用或者合并成一个流。例如：上文 pipeline 中的 merge 函数，可以通过打开 fanInOut() 函数执行一下试试。

or-done-channel

在防止 goroutine 泄露，pipeline 中我们都在函数执行过程中嵌入了 done channel 以便终止需要停止的 goroutine。我们可以看出他们有个统一的特点，传入 done ,jobChannel ，返回 resultChannel 。那么我们可以把它封装起来，像这样：

orDone := func(done ,c <-chan interface{}) <- chan interface{}{
    valStream := make(chan interface{})
    go func(){
        defer close(valStream)
        for {
            select{
            case <- done:
            case v,ok := <- c:
                if ok == false{
                    return
                }
                select{
                case valStream <- v:
                case <-done:
                }
            }
        }
    }()
    return valStream
}

tee := func(done <- chan interface{},in <-chan interface{}
           )(_,_ <- chan interface{}) { <-chan interface{}) {
    out1 := make(chan interface{})
    out2 := make(chan interface{})
    go func(){
        defer close(out1)
        defer close(out2)
        for val := range orDone(done, in){
            var out1,out2 = out1,out2
            for i:=0;i<2; i++{
                select{
                case <- done:
                case out1 <- val:
                    out1 = nil
                case out2 <- val:
                    out2 = nil
                }
            }
        }
    }()
    return out1,out2
}

type Context interface {
    // 当该 context 工作的 work 被取消时，返回超时时间
    Deadline() (deadline time.Time, ok bool)
    // done 返回停止 pipeline 的 channel 
    Done() <chan struct{}
    // error 一等公民。
    // 如果 context 被取消，超时，返回取消，超时的原因，以 error 形式返回。 
    Err() error
    // 返回与此 context 关联的 key
    Value(key interface{}) interface{}
}