Suggestion: Pipe routing and merging abstraction.

[Enrico2]

As I was working a lot with ro I notice that I often want to split a pipe into multiple subpipes, have different handling for each one, and then merge into an eventual result.

Consider, for example, handling a WebSocket events where each event has a different handler, but at the end of all handlers I converge into the same type.

Right now I do it manually and the code is cumbersome and involves some duplication of data which I'm not a fan of if it can be avoided (Especially when you consider pointers that would be shared by different branches of the pipe).

Here's a sketch of the idea:

import (
  "github.com/samber/mo"
  "github.com/samber/ro"
)

var input ro.Pipe[In, T] = ...

var handleX ro.Pipe[X, Out] = ...
var handleY ro.Pipe[Y, Out] = ...
var handleZ ro.Pipe[Z, Out] = ...

var router ro.PipeOp[In, mo.Either3[X, Y, Z]]

var result ro.Pipe[In, Out] = ro.Route3(
  input,
  router,
  handleX,
  handleY,
  handleZ,
)

[samber]

I wonder if input should be a Subject or ConnectableObservable accepting multiple simultaneous subscribers, each subscriber having a routing method returning true/false.

It would be more "compliant" with the idea of Observables. On the other side, it would be more costly than running a routing function that receives a fixed list of Observer and a single switch/case statement for every "subscribers".

[samber]

If handlers have the same type T, and you need an unbounded list of destinations, then your routing method would be a simple ro.GroupBy(...)

[Enrico2]

I wonder if input should be a Subject or ConnectableObservable accepting multiple simultaneous subscribers, each subscriber having a routing method returning true/false.

It would be more "compliant" with the idea of Observables. On the other side, it would be more costly than running a routing function that receives a fixed list of Observer and a single switch/case statement for every "subscribers".

I think that depending on the use case, the cost might matter?
Ideally, this library should allow for building the most efficient pipelines when those are needed.

I see the value in being compliant to the principles though. Which aspect breaks compliance?

[Enrico2]

If handlers have the same type T, and you need an unbounded list of destinations, then your routing method would be a simple ro.GroupBy(...)

I can see when that would be useful, but the limitation is when the type that we're generating cannot be a key in a map.
The reason my sketch has those X,Y,Z types is that I would only need to implement the routing function once and not have to repeat a filter or something in each of the divergence branches.

e.g. in my use case, I have an interface like ServerEvent and that has a few specific implementations, e.g. AudioDelta, TextDelta, SessionUpdate.
Each of those are handled differently. So audio deltas are sent directly to the client, Text deltas are accumulated to a complete message with ro.Scan, and SessionUpdates essentially do and are filtered.

if I were to GroupBy and use a string for the event type, I would have to recast the type in each branch pipe and rely on the correctness of the string grouping, I'd lose type safety. Does that make sense?