msacore / pipe Goto Github PK

I would like to see some generic functions to construct my own pipeline functions. In concept, it might look like:

import "github.com/msacore/pipe/builder"

...

fn := builder.
  WithInputs(in1, in2).
  WithOutputs(out1, out2).
  WithCloser(closer.Some). // const
  WithCapacity(capacity.Same). // const
  WithStrategy(strategy.Parallel). // func
  WithHandler(???).
  Build(ctx)

...

func Split(...) {
  return fn(...)
}

Of course, we need some more research about how to implement and use different atomic strategies with atomic handlers. I mean, how the handler should work and the Builder should understand for different functions like Split and Merge for example.

These functions have conceptually different algorithms where Split just waits for a single input, and Merge waits for all inputs.

I guess, this problem can be solved with different handler functions builder, like WithHandlerAll and WithHandlerSome and WithHandlerSingle (specific case of WithHandlerAll)

Is your feature request related to a problem? Please describe.
It can be tedious to declare a new structure every time we need to transfer various related data between the conveyor elements.

Describe the solution you'd like
Now we need to declare a new structure to communicate between pipes, I suggest next:

type msgOneToTwo = pipe.Tuple2[string, int]
type msgTwoToThree = pipe.Tuple2[int, string]

msg1 := make(chan msgOneToTwo, 2>>8)

msg2 := pipe.Map(..., func(msg msgOneToTwo) msgTwoToThree {
  return pipe.T2(msg.B, msg.A)
}, msg1)

... := pipe.Map(..., msg2)

Describe alternatives you've considered
I found a convenient solution to this problem using the samber/lo. It would be good to introduce tuples into the current module.

Additional context
none

Is your feature request related to a problem? Please describe.
Pipe aborting requires a bit of configuration, but we can reduce it.

Describe the solution you'd like
Now, we can abort operation with context, but we need to implement our own mechanism of aborting.
I suggest implement this feature as core feature for all pipe functions. I mean, we can stop calling of pipe function handlers in case of ctx.Done, and all rest and unread data for input channels must be read in the background.

Describe alternatives you've considered
A clear and concise description of any alternative solutions or features you've considered.

Additional context
none