Feature request: Add Flow#switchMap operator.

[He-Pin]

Motivation:
I find that Pekko stream doen't has the switchmap operator, it's called as this in both fs2 and flux, I think we need to take the same name and same semantic.

Usercase:
A Stream of Post, eg twitter/X 's Post, and every Post has a Stream of Comments.

[Post 1] ---> [Comment 1] ... [Comment 2] ... [Comment 3] ...
.
.
.
[Post 2] ---> [Comment 1] ... [Comment 2] ... 
.
.
.
[Post 3] ---> [Comment 1] ... [Comment 2] ... [Comment 3] ... [Comment 4]..
.
.
.
\|/ User scrolling down.

Emits when: sub stream emits and downstream is available
Backpressures when: sub stream backpressure.
cancels when: down stream cancels
completes when: upstream completes and the current substream completes.

In some implementation, the upstream is never backpressured, I think that's not safe. I think the upstream should be backpressured if there is no downstream demands.

Result:
We have rich operators set.

[He-Pin]

@pjfanning @mdedetrich what do you think about his one, I find it would be nice for live streaming.

[He-Pin]

Implementation this shold not be that hard but seems no much interest, will defer to later version. it's a kind of stream of streams

[mdedetrich]

I think we have enough on our plate right now with all of the new features, we can always add this into 1.1.0-M2 (if a new M2 is needed) or in 1.2.x.

[He-Pin]

I will back my hometown soon, if anyone submited a pr about this, then this can be added to 1.1.0-M2 I think, before that, it's safe for later version.

I was expecting we stating remove code in 1.2.x but not adding features.

I just update what I thought about this operator.

[He-Pin]

refs: #1660

[He-Pin]

https://github.com/spring-projects/spring-ai/blob/3c539a37a38e009e11e2a8870a1708787831069b/models/spring-ai-anthropic/src/main/java/org/springframework/ai/anthropic/AnthropicChatModel.java#L283-L288

Flux<ChatResponse> chatResponseFlux = response.switchMap(chatCompletionResponse -> {

It's been used in spring-ai

[MartinHH]

I used to use this operator a lot when dealing with various applications related to monitoring live data and always missed it when working with akka streams in the past. I'm currently not working on any of such projects - if I were, I'd happily submit a PR (and might as well still do that if I find the time).

One thing though: from my experience (I might be biased due to the "live data" use cases I used it for), I'd strongly advocate for "Backpressures when: never." The way this operator works, only the very latest value from upstream should cause items to be emitted to downstream, so there is no sense in keeping anything but that one latest value (and the resulting substream) in the pipeline.

[He-Pin]

That's true; this kind of operator will never do any backpressure.
I was planning to add one, but I have no scenario in my $Work, if you submit one, I would be pleased to review it.