编写一个main函数来测试我下面这个流水线

package main

import (
	"context"
	"errors"
	"fmt"
	"sync"
	"time"
)

// Task 定义任务的数据结构
type Task struct {
	ID      int         // 原始任务ID（批次ID）
	Data    interface{} // 输入数据
	Result  interface{} // 处理结果
	Retries int         // 重试次数
}

// TaskFunc 定义任务处理函数类型，返回多个结果
type TaskFunc func(ctx context.Context, data interface{}) ([]interface{}, error)
type CustomFunc func(ctx context.Context, data interface{}, resultChan <-chan interface{}) error

type Layer interface {
	InitLayer(workerCount int, timeout time.Duration, maxRetries int) Layer
	ProcessTask(ctx context.Context, task *Task) (interface{}, int, error)
	GetWorkerNum() int
}

// StandardLayer 定义流水线的每一层
type StandardLayer struct {
	workerCount int
	timeout     time.Duration
	maxRetries  int
	taskFunc    TaskFunc
}

func (l *StandardLayer) InitLayer(workerCount int, timeout time.Duration, maxRetries int) Layer {
	l.workerCount = workerCount
	l.timeout = timeout
	l.maxRetries = maxRetries
	return l
}

func (l *StandardLayer) ProcessTask(ctx context.Context, task *Task) (interface{}, int, error) {
	ctx, cancel := context.WithTimeout(ctx, l.timeout)
	defer cancel()
	for retries := task.Retries; retries <= l.maxRetries; retries++ {
		// 创建带超时的上下文
		results, err := l.taskFunc(ctx, task.Data)
		if err != nil {
			if retries >= l.maxRetries {
				// 超过重试次数，记录失败
				fmt.Printf("Task ID %d failed after %d retries: %v\n", task.ID, retries, err)
				return nil, retries, err
			}
			// 记录重试信息
			fmt.Printf("Task ID %d retry %d due to error: %v\n", task.ID, retries+1, err)
			continue
		}
		return results, retries, nil
	}
	// 不应到达此处
	return nil, l.maxRetries, errors.New("unreachable code in processTask")
}

func (l *StandardLayer) GetWorkerNum() int {
	return l.workerCount
}

// Layer 定义流水线的每一层
type CustomsLayer struct {
	workerCount int
	timeout     time.Duration
	maxRetries  int
	taskFunc    CustomFunc
}

func (l *CustomsLayer) InitLayer(workerCount int, timeout time.Duration, maxRetries int) Layer {
	l.workerCount = workerCount
	l.timeout = timeout
	l.maxRetries = maxRetries
	return l
}

func (l *CustomsLayer) ProcessTask(ctx context.Context, task *Task) (interface{}, int, error) {
	// 将channel传给下一层
	resultChan := make(chan interface{})
	go func() {
		err := l.taskFunc(ctx, task.Data, resultChan)
		if err != nil {

		}
	}()
	return resultChan, 0, nil
}

func (l *CustomsLayer) GetWorkerNum() int {
	return l.workerCount
}

// Pipeline 定义整个流水线结构
type Pipeline struct {
	layers []Layer
}

// NewPipeline 创建一个新的流水线实例
func NewPipeline() *Pipeline {
	return &Pipeline{
		layers: make([]Layer, 0),
	}
}

// AddLayer 向流水线中添加一层
func (p *Pipeline) AddStandardLayer(taskFunc TaskFunc, workerCount int, timeout time.Duration, maxRetries int) {
	layer := &StandardLayer{}
	p.layers = append(p.layers, layer.InitLayer(workerCount, timeout, maxRetries))
}

// AddLayer 向流水线中添加一层
func (p *Pipeline) AddCustomLayer(taskFunc TaskFunc, workerCount int, timeout time.Duration, maxRetries int) {
	layer := &CustomsLayer{}
	p.layers = append(p.layers, layer.InitLayer(workerCount, timeout, maxRetries))
}

// processLayer 处理单层的任务分发和结果收集
func (p *Pipeline) processLayer(ctxPipeline context.Context, layer Layer, in <-chan *Task, taskNum, nextLayerNum int) chan *Task {
	out := make(chan *Task)
	var wg sync.WaitGroup
	slidingWindow := NewSlidingWindow(layer.GetWorkerNum(), taskNum)
	// 启动工作协程
	for w := 0; w < layer.GetWorkerNum(); w++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			for task := range in {
				slidingWindow.Process(task.ID)
				results, retries, err := layer.ProcessTask(ctxPipeline, task)
				task.Retries = retries
				if err != nil {
					// 任务失败，传递失败信息
					task.Result = nil
					out <- task
					continue
				}
				task.Result = results
				// 根据结果数量创建子任务或直接传递
				p.handleTaskResults(task, results, out)
			}
		}()
	}

	// 等待所有工作协程完成后关闭输出通道
	go func() {
		wg.Wait()
		close(out)
	}()

	return out
}

// handleTaskResults 根据任务结果创建子任务或直接传递
func (p *Pipeline) handleTaskResults(task *Task, results interface{}, out chan<- *Task) {
	switch results.(type) {
	case []interface{}:
		resultAry := results.([]interface{})
		if len(resultAry) > 1 {
			for _, res := range resultAry {
				subTask := &Task{
					ID:      task.ID, // 保持相同的任务ID
					Data:    res,
					Result:  nil,
					Retries: task.Retries,
				}
				out <- subTask
			}
		} else if len(resultAry) == 1 {
			// 单个结果，直接更新并传递
			task.Data = resultAry[0]
			out <- task
		} else {
			// 无结果，直接传递任务
			out <- task
		}
	case chan interface{}:
		resultChan := results.(chan interface{})
		task.Data = resultChan
		out <- task
	}

}

// Run 启动流水线处理任务，并在每个任务完成时调用回调函数
func (p *Pipeline) Run(input []interface{}, callback func(task *Task)) error {
	if len(p.layers) == 0 {
		return errors.New("pipeline has no layers")
	}

	// 初始化任务通道
	inCh := make(chan *Task)

	// 将输入数据封装成任务并发送到第一个通道
	go func() {
		for idx, data := range input {
			inCh <- &Task{ID: idx, Data: data}
		}
		close(inCh)
	}()

	// 用于取消整个流水线的上下文（仅在严重错误时使用）
	ctxPipeline, cancelPipeline := context.WithCancel(context.Background())
	defer cancelPipeline()

	currentIn := inCh

	// 遍历每一层，构建流水线
	for _, layer := range p.layers {
		currentIn = p.processLayer(ctxPipeline, layer, currentIn, len(input), layer.GetWorkerNum())
	}

	// 最后一层输出处理
	var wgFinal sync.WaitGroup
	wgFinal.Add(1)
	go func() {
		defer wgFinal.Done()
		for task := range currentIn {
			if task.Result == nil {
				// 如果某个任务失败，记录错误信息
				fmt.Printf("Task ID %d failed during processing.\n", task.ID)
				// 根据需要处理失败任务，如记录日志或继续
			} else {
				// 调用回调函数处理每个任务的结果
				callback(task)
			}
		}
	}()
	// 等待所有任务完成
	wgFinal.Wait()

	return nil
}

type SlidingWindow struct {
	mutex       sync.Mutex
	cond        *sync.Cond
	taskSize    int         // 任务大小
	windowStart int         // 当前窗口的起始下标
	windowSize  int         // 窗口大小
	maxIndex    int         // 最大下标值
	processed   map[int]int // 记录已经处理过的下标
}

func NewSlidingWindow(windowSize int, maxIndex int, taskNum int) *SlidingWindow {
	processed := make(map[int]int, maxIndex)
	for i := 0; i < taskNum; i++ {
		processed[i] = taskNum
	}
	w := &SlidingWindow{
		windowStart: 0,
		windowSize:  windowSize,
		maxIndex:    maxIndex,
		processed:   processed,
	}
	w.cond = sync.NewCond(&w.mutex)
	return w
}

// Process 方法，传入下标，如果不在窗口内则阻塞，直到窗口滑动包含该下标
func (w *SlidingWindow) Process(index int) {
	w.mutex.Lock()
	defer w.mutex.Unlock()

	// 等待下标在窗口范围内
	for index < w.windowStart || index >= w.windowStart+w.windowSize {
		w.cond.Wait()
	}

	// 下标在窗口内，标记为已处理
	w.processed[index] -= 1
	fmt.Printf("下标 %d 在窗口内，开始处理\n", index)

	// 判断窗口是否可以滑动
	// 只有窗口起始下标已处理，才能滑动
	for w.processed[w.windowStart] == 0 {
		delete(w.processed, w.windowStart)
		w.windowStart++

		// 如果达到最大下标，停止滑动
		if w.windowStart >= w.maxIndex {
			break
		}
	}

	// 通知所有等待的协程
	w.cond.Broadcast()
}