package main
import (
	"context"
	"log"
	"sync"
	"time"
)
// Task 包含任務(wù)編號(hào)及任務(wù)所需時(shí)長(zhǎng)
type Task struct {
	Number int
	Cost   time.Duration
}
// task channel 生成器
func taskChannelGerenator(ctx context.Context, taskList []Task) <-chan Task {
	taskCh := make(chan Task)
	go func() {
		defer close(taskCh)
		for _, task := range taskList {
			select {
			case <-ctx.Done():
				return
			case taskCh <- task:
			}
		}
	}()
	return taskCh
}
// doTask 處理并返回已處理的任務(wù)編號(hào)作為通道的函數(shù)
func doTask(ctx context.Context, taskCh <-chan Task) <-chan int {
	doneTaskCh := make(chan int)
	go func() {
		defer close(doneTaskCh)
		for task := range taskCh {
			select {
			case <-ctx.Done():
				return
			default:
				log.Printf("do task number: %d\n", task.Number)
				// task 任務(wù)處理
				// 根據(jù)任務(wù)耗時(shí)休眠
				time.Sleep(task.Cost)
				doneTaskCh <- task.Number // 已處理任務(wù)的編號(hào)放入通道
			}
		}
	}()
	return doneTaskCh
}
// `fan-in` 意味著將多個(gè)數(shù)據(jù)流復(fù)用或合并成一個(gè)流。
// merge 函數(shù)接收參數(shù)傳遞的多個(gè)通道 “taskChs”,并返回單個(gè)通道 “<-chan int”
func merge(ctx context.Context, taskChs []<-chan int) <-chan int {
	var wg sync.WaitGroup
	mergedTaskCh := make(chan int)
	mergeTask := func(taskCh <-chan int) {
		defer wg.Done()
		for t := range taskCh {
			select {
			case <-ctx.Done():
				return
			case mergedTaskCh <- t:
			}
		}
	}
	wg.Add(len(taskChs))
	for _, taskCh := range taskChs {
		go mergeTask(taskCh)
	}
	// 等待所有任務(wù)處理完畢
	go func() {
		wg.Wait()
		close(mergedTaskCh)
	}()
	return mergedTaskCh
}
func main() {
	start := time.Now()
	// 使用 context 來(lái)防止 goroutine 泄漏，即使在處理過(guò)程中被中斷
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	// taskList 定義每個(gè)任務(wù)及其成本
	taskList := []Task{
		Task{1, 1 * time.Second},
		Task{2, 7 * time.Second},
		Task{3, 2 * time.Second},
		Task{4, 3 * time.Second},
		Task{5, 5 * time.Second},
		Task{6, 3 * time.Second},
	}
	// taskChannelGerenator 是一個(gè)函數(shù)，它接收一個(gè) taskList 并將其轉(zhuǎn)換為 Task 類型的通道
	// 執(zhí)行結(jié)果（int slice channel）存儲(chǔ)在 worker 中
	// 由于 doTask 的結(jié)果是一個(gè)通道，被分給了多個(gè) worker，這就對(duì)應(yīng)了 fan-out 處理
	taskCh := taskChannelGerenator(ctx, taskList)
	numWorkers := 4
	workers := make([]<-chan int, numWorkers)
	for i := 0; i < numWorkers; i++ {
		workers[i] = doTask(ctx, taskCh)  // doTask 處理并返回已處理的任務(wù)編號(hào)作為通道的函數(shù)
	}
	count := 0
	for d := range merge(ctx, workers) { // merge 從中讀取已處理的任務(wù)編號(hào)
		count++
		log.Printf("done task number: %d\n", d)
	}
	log.Printf("Finished. Done %d tasks. Total time: %fs", count, time.Since(start).Seconds())
}