package pipe

// 流水线
type PipeLineCommonChan chan interface{}

// 生产者,生产数据
type PipeLineProducer func(dataSourceChan PipeLineCommonChan)

// return nil表示这个数据不再进行处理
type PipeLineFuncWorker func(chanItem interface{}) interface{}

// 流水线入口
// todo 每一个处理步骤可以设置处理的worker数量
func PipeLineDo(producer PipeLineProducer, userFns ...PipeLineFuncWorker) PipeLineCommonChan {
	dataSource := commonProducer(producer)
	for _, f := range userFns {
		//这里datasource会更新成下一个chan，传递给下一个函数
		dataSource = commonStep(dataSource, f)
	}
	return dataSource
}

// 处理步骤
func commonStep(ch PipeLineCommonChan, userFunc PipeLineFuncWorker) PipeLineCommonChan {
	out := make(PipeLineCommonChan, 0)
	go func() {
		defer close(out)
		for val := range ch {
			//这里做用户函数逻辑处理
			res := userFunc(val)
			if res != nil {
				out <- res
			}
		}
	}()
	return out
}

// 内部生产者，将原始数据塞入流水线管道，开始进行处理
func commonProducer(userFunc PipeLineProducer) PipeLineCommonChan {
	dataSource := make(PipeLineCommonChan, 0)
	//开协程进行处理
	go func() {
		defer close(dataSource)
		userFunc(dataSource)
	}()
	return dataSource
}