Flattening operators and behaviors ideation

[benlesh]

I have an experimental operator that basically "does it all" (merge, concat, switch, exhaust) with concurrency. I'm likely to implementing all of our current flattening operators (mergeMap, concatMap, switchMap, exhaustMap) on top of this, because it's lighter-weight that having separate concatMap and switchMap implementations, as an example.

It also means you could do switchMap with concurrency, which has been a request several times. (This would mean interrupting the "oldest" subscription when you hit the concurrency limit).

API Idea

I think I'm going to pitch this with the native observable proposal as well, but basically the API I have right now is as follows:

declare const flatMap = (project: (value: In, index: number) => ObservableInput<Out>, config?: { concurrency?: number, behavior?: 'merge' | 'switch' | 'exhaust' }) => OperatorFunction<In, Out>;

// Where this is basically how it would work:

// behavior defaults to "merge"
// concurrency defaults to 1

flatMap(fn); // concatMap(fn) | mergeMap(fn)
flatMap(fn, { concurrency: Infinity }); // mergeMap(fn)
flatMap(fn, { behavior: 'switch' }); // switchMap(fn)
flatMap(fn, { behavior: 'exhaust' }); // exhaustMap(fn)

// New abilities:
// subscribe to 2 at a time, when you get a 3rd value, unsubscribe from
// the oldest inner subscription and subscribe to the 3rd one.
flatMap(fn, { behavior: 'switch', concurrency: 2 });

// subscribe to 2 at a time, when you get a 3rd value, if you
// already have two inner subscriptions, ignore it.
flatMap(fn, { behavior: 'exhaust', concurrency: 2 });

Why flatMap?

We're flattening. In a general sense. Also, flatMap on a native observable is basically concatMap. Making it configurable is interesting and powerful.

Names

The biggest problem I have with this is honestly naming things. The concurrency aspect adds a wrinkle. Does switch still make sense? Or exhaust? What about concat vs merge? I just don't know.

I'm looking for ideas/proposals there.

[ducin]

@benlesh I like the idea of naming it flatMap very much! Historically, mergeMap==flatMap was super confusing, since each flattener was a flatMap. Now it makes things clear.

I'd consider:

• dropping the default, make the parameter required/explicit. Explicit requires more to write, but I think this saving is not worth potential confusion. Especially that, historically, flatMap used to be == mergeMap.
• (very subjective) nit-pick: strategy sounds better than behavior

[benlesh]

In general, I prefer explicit parameters... however, the native observable will have a flatMap method that is basically flatMap(fn, { concurrent: 1, behavior: 'merge' }). So I'd like to align with that, if possible.

[timdp]

Personally, I like the general idea.

I do wonder if the args shouldn't be reversed, so that:

• a multi-line projection function isn't followed by a dangling options object, which is usually easy to miss;
• you could support partial application for improved reusability and composability:

  const switchMap = flatMap({ behavior: 'switch' })

I know options objects generally go last because they're, well, optional, but I think this might warrant an exception.

[benlesh]

I have a goal of aligning with the native Observable's flatMap method. which takes 1 argument, that is a function (and behaves like our concatMap).

[josepot]

I'd like to propose an enhancement to the FlatMapConfig interface . The idea is to add a new ordered option, defaulting to false, as shown below:

interface FlatMapConfig {
  concurrency?: number;
  behavior?: 'merge' | 'switch' | 'exhaust';
  ordered?: boolean;
}

This additional ordered option would allow for more versatile stream operations. Specifically, it paves the way for implementing the concatMapEager operator, which is available in rxjs-etc and has also been requested by the community a few times. The implementation would look something like this:

flatMap(fn, { concurrency: Infinity, ordered: true });

The concatMapEager operator is particularly useful for scenarios where maintaining the order of emissions is crucial while still processing multiple observables concurrently. In my experience, although I've never officially requested this feature, I've frequently used a custom implementation of it. I'm aware of the backpressure implications it introduces, but I ensure to use it responsibly.

[josepot]

I mean, in part the problem is that concurrent: Infinity trumps the "keep/strategy" option. In the sense that the "strategy" is only relevant when there is a limit to the number of concurrent subscriptions.

So, perhaps the concurrent should have a different default value depending on the value of "keep".

At least intuitively, I would expect the default value of concurrent to be Infinity when keep is set to "all", while being defaulted to 1 when keep is set to either "new"/"old".

I mean, it's definitely not easy to get this API to be intuitive :/

[josepot]

Another way to put it is that what makes this API a bit unintuitive is the fact that:

• keep/strategy/behaviour is only relevant when concurrency < Infinity
• preserveSourceOrder is only relevant when concurrency > 1

[josepot]

Another half-baked idea:

What about 3 different operators:

• flatMapAll, the only config option would be preserveSourceOrder defaulting to false. Therefore, no need for a config object, just an optional argument would suffice.
• flatMapMono: its concurrency is always 1, so the only config option would be keep/strategy/behavior. Same deal: no need for a config object, just an optional argument would do.
• flatMapSome: this would be an advanced operator with a config option admitting all the possible permutations of keep, concurrency (perhaps it could be named limit?) and preserveSourceOrder. Obviously, the other 2 operators are sugar for the common permutations produced by this advanced operator.

[LcsGa]

If we keep going with flatMap, don't you think that we could handle the use cases you've raised with typescript overloads?
The cons would be that we couldn't let the users set a concurrency count except for the mergeMap config (since infinity is a number), but is there a need for such an option on switchMap or exhaustMap?

Thus if we still create a flatMap operator but keep the current behavior we could have:

flatMap(fn) // mergeMap
flatMap(fn, { concurrency: 5 }) // still a mergeMap
flatMap(fn, { concurrency: 1 }) // concatMap
flatMap(fn, { strategy: 'queue' }) // concatMap too, but could probably be omitted
flatMap(fn, { strategy: 'switch' }) // switchMap
flatMap(fn, { strategy: 'ignore' }) // exhaustMap

flatMap(fn, { strategy: 'queue', concurrency: 2 }) // would raise a TS error as it wouldn't be in the overload
flatMap(fn, { strategy: 'switch', concurrency: 2 }) // TS error
flatMap(fn, { strategy: 'ignore', concurrency: 2 }) // TS error

=> It wouldn't add any new feature but there would be only one operator now, without (normally) the unintuitiveness issue.

[LcsGa]

Or, we could also simply let anybody add the concurrency option, no matter what the straregy is, and if they set it to infinity on switch or ignore that would be like a mergeMap, which wouldn't be the most optimal, but that wouldn't be a real mistake either.

[kbrilla]

What about something that tells what is actually happening like whenFull : wait | drop | ignore (merge | switch | exhaust ) and ordered would add exit strategy. Just loud thinking. W dniu śr., 17.01.2024 o 01:25 Ben Lesh ***@***.***> napisał(a):

…

Okay, I think the challenge here is how to effectively write this API that it's self-explanatory *what* it's doing. ordered is really ensuring that the values from each subscription are concatenated or that their order is maintained relative to the the values that were mapped into each subscription. behavior (or strategy) is also probably too generic, because really it's defining "what happens when you're already at your concurrency limit, and you get a new value from the source?". merge, exhaust, and switch *only really mean something to RxJS users* and not really to everyone else. — Reply to this email directly, view it on GitHub <#7429 (reply in thread)>, or unsubscribe <https://github.com/notifications/unsubscribe-auth/AFKBQQJWOWJ4RVXCG5TJG43YO4K7NAVCNFSM6AAAAABB5K6UGCVHI2DSMVQWIX3LMV43SRDJONRXK43TNFXW4Q3PNVWWK3TUHM4DCNJQGYZDE> . You are receiving this because you commented.Message ID: ***@***.***>

[kbrilla]

One more though: what if instead of drop so standard switchMap we add function that actually can decide which concurrent stream it should replace? Cause drop probably would replace first concurrent stream but what if someone would like to replace last one or mayby middle one? W dniu śr., 17.01.2024 o 01:32 Krzysztof Brilla ***@***.***> napisał(a):

…

What about something that tells what is actually happening like whenFull : wait | drop | ignore (merge | switch | exhaust ) and ordered would add exit strategy. Just loud thinking. W dniu śr., 17.01.2024 o 01:25 Ben Lesh ***@***.***> napisał(a): > Okay, I think the challenge here is how to effectively write this API > that it's self-explanatory *what* it's doing. > > ordered is really ensuring that the values from each subscription are > concatenated or that their order is maintained relative to the the values > that were mapped into each subscription. > > behavior (or strategy) is also probably too generic, because really it's > defining "what happens when you're already at your concurrency limit, > and you get a new value from the source?". merge, exhaust, and switch *only > really mean something to RxJS users* and not really to everyone else. > > — > Reply to this email directly, view it on GitHub > <#7429 (reply in thread)>, > or unsubscribe > <https://github.com/notifications/unsubscribe-auth/AFKBQQJWOWJ4RVXCG5TJG43YO4K7NAVCNFSM6AAAAABB5K6UGCVHI2DSMVQWIX3LMV43SRDJONRXK43TNFXW4Q3PNVWWK3TUHM4DCNJQGYZDE> > . > You are receiving this because you commented.Message ID: > ***@***.***> >

[deanrad]

As the creator of a library that gratefully and shamelessly steals these concurrency operators and hides them behind a facade - 𝗥𝘅𝑓𝑥 - I have some well-aged opinions on these names. I'd be honored if you steal back any part of this for helping to clarify the naming in v8:

The words I've used for the concurrency operator modes:

• mergeMap => immediate
• concatMap => queueing
• switchMap => switching (also, replacing)
• exhaustMap => blocking (also, throttling)

But a whole other interesting way to go, is to not use a single name for the mode, but instead to separately encode for the 3 variables that determine each mode:

• Starts New Subscription (Y/N)
• Cancels existing Subscription (Y,N), and
• Queues after existing Subscription (can only be Y if previous answers were Y/N)

flatMap(fn,  { startsNew: true, cancelsExisting: false });  // mergeMap 
flatMap(fn,  { startsNew: true, cancelsExisting: false, queues: true });  // concatMap
flatMap(fn,  { startsNew: true, cancelsExisting: true });  // switchMap
flatMap(fn,  { startsNew: false, cancelsExisting: false });  // exhaustMap

This gets around the challenge of choosing the perfect name.

Incidentally, there's another useful operator strategy it'd be nice to include. It was called keepLatest in Ember Concurrency, and I've called it queueOnlyLatest in my library.. It's a combination of concat and switch strategies - it concats, but switches each newly concatted one, so that in the end, only the latest new subscription is ever queued. Implemention here: https://github.com/deanrad/rxfx/blob/main/operators/src/queueOnlyLatest.ts

[LcsGa]

I love the general idea !!

About my personal opinion:

=> What I love about this operator is:

• the concurrency option could be a cool addition
• having only one operator is "better" for building some special tools that could use any of those strategies => in your custom tool, you would need to import only this operator and that's it, resulting in a smaller bundle size

=> After reading the comments, I'd go for:

• strategy instead of behavior => it feels more "conventional" / "understandable"
• ignore instead of exhaust => Whenever I explain the exhaustMap operator to a beginner I always tell them that "if the inner observable is not done yet, any incoming next notification will be ignored/lost" + it is a very simple word for any non-native English (I guess).

=> About the ordered option, it seems cool but I must admit that I can't find any use case where I could use it:

• Except if we keep a concat / queue strategy, I already don't see any use case where I'd need concurrency => and if I don't need it, then I won't need ordered too
• If I have concurrency, I think that I usually wouldn't care about the order I would receive the next notifications
  => Do you have any example though ?

[mfp22]

'I would like this if 'switch' was default, because I've seen a need for it about 10x more frequently than the other operators. And I get that this would reduce the code in the RxJS library itself, but I'm wondering if I'm downloading the code for all of this behavior even if I only want to do a traditional switchMap.

[benlesh]

Currently, the native Observable's flatMap behaves like concatMap.