myntra / pipeline Goto Github PK

The simple example shows a sequential stage processing. It is unclear from the documentation and the example if work could be processed concurrently by the pipeline. If not, what would be required to support concurrent processing.

Hey there,

Thanks for the great library, the Go market really lacked it.

Currently I am working on a data processing pipeline which imports data from various sources. This means that I have a data source with rows of data which I would like to process.

Any ideas how I could do it with this library?

(I know https://github.com/dailyburn/ratchet but I just can't get used to it's syntax)

Thanks in advance!

When I set up concurrent execution for the steps of a stage, I find that program always execute the last step.
Looking at the source code, I found closure functions used for concurrent execution, which I guess is where the bug came from.

// file  stage.go
for _, step := range st.Steps {
			step.Status("begin")
			g.run(func() *Result {

				defer step.Status("end")

				//disables strict mode. g.run will wait for all steps to finish
				if st.DisableStrictMode {
					return step.Exec(request)
				}
				fmt.Println(step.getCtx().index, step.getCtx().name)
				resultChan := make(chan *Result, 1)

in result_group.go func run()
when Error == nil should g.result = result
and in stage.go func run()
many steps results should merge

Hi! Thanks for Pipeline! I mentioned on your reddit post that this is definitely something I intend to use!

I'm trying to test a Step I implemented. I wrote the step, which simply shells out to the console and stores the stdout into the Result.Data field. My problem is that this isn't testable because after running the pipeline, a nil *Result is returned on success.

I propose pipeline.Run() always return a *Result on both success and failure. Simply return the *Result from the final Stage.

I'm going to fork this repo and implement the change for my own purposes. Let me know if you like the proposal and want me to PR back upstream. :)

Have a great day!

I'd like to use pipeline to manage my sequence tasks, in my situation, user can manually kill the task, but pipeline do not export a Cancel function.