feat(operator): add concurrency parameter to switchMap

[staltz]

This issue is just a heads-up about an operator that could be added to RxJS Next. I can also see this used through lift instead of bundling it straight in the library.

Basically flatMapLatestWithMaxConcurrency:

http://stackoverflow.com/questions/32233408/how-to-implement-rxjs-flatmaplatesttwo

What do you think?

[benlesh]

Hmm... I think flatMap and flatMap latest should support concurrency limits regardless. Considering they're basically sugar for map(x => obs).mergeAll() and merge definitely supports concurrency limits.

[staltz]

Oh cool! I didn't know mergeAll() had a concurrency param. It doesn't in Reactive-Extensions/RxJS. That's great.

[benlesh]

So flatMap has a concurrency argument already... It seems more like we don't have flatMapLatest implemented... that surprises me a bit as I thought we had that checked off on #100. That one is a top priority for me.

[benlesh]

so it should be:

function flatMapLatest<T, R>(project: (value:T) => Observable<R>, concurrent: number = Number.POSITIVE_INFINITY): Observable<R>

again this is a top priority for what I need, so if someone doesn't pick this up in the next few minutes I'm all over it.

[staltz]

Bikeshedding: why the concurrent: number as the last parameter?

[xgrommx]

As I understood:

flatMap => flatMapWithMaxConcurrent(Number.POSITIVE_INFINITY)
flatMapConcat => flatMapWithMaxConcurrent(1)

More you can read in our discussion with @paulpdaniels Reactive-Extensions/RxJS#793 (comment)

[benlesh]

So digging into this... what is the point of the concurrent parameter in flatMapLatest? How can you concurrently be subscribed to inner observables if you're only ever subscribed to the latest one.

I'm not sure how I missed that, I was more freaking out that it hadn't been implemented yet.

[staltz]

How can you concurrently be subscribed to inner observables if you're only ever subscribed to the latest one.

with concurrent = 2 you would be subscribed to the latest two inner observables.

[benlesh]

with concurrent = 2 you would be subscribed to the latest two inner observables.

Oh... interesting. Yeah, I don't see why not. I guess it should default to one then ;) haha.

[xgrommx]

Hi @Blesh. Where can I take a look on all implemented methods(some list of methods)?

[benlesh]

So, we have flatMapLatest, it's just been conflated into switchLatest for whatever reason. I didn't catch that before. This goes back to bike-shedding those names to make them more predictable. It doesn't have the concurrency bit though, I can probably add that.

[benlesh]

@xgrommx you can find them all either in src/operators or in src/Rx.ts or src/Observable.ts

[staltz]

Oh yeah how could I forget that flatMapLatest was renamed to switchLatest. We talked so much about mergeMap/switchMap/concatMap. By the way, will switchLatest be named switchMap?

[benlesh]

By the way, will switchLatest be named switchMap?

I think this requires a lot more discussion, honestly.

Digging into this issue, to do it properly is non-trivial. I've found a few other issues I'd like to refactor before I tackle this. I'll need to think about this a little more.

[trxcllnt]

@Blesh @staltz since SwitchLatestSubscriber extends FlatMapSubscriber, and FlatMapSubscriber extends MergeSubscriber, we could extend the concurrent behavior to SwitchLatestSubscriber (it currently defaults to Number.MAX_VALUE).

At first glance, I think we'd only need to make the following four changes:

1. track inner subscriptions in MergeSubscriber locally
2. change _innerComplete to accept the inner Subscription, so we can remove it from the local subscriptions list (we're already planning to do this so we can remove it from the shared Subscription)
3. change when SwitchLatestSubscriber switches to a new inner Observable
4. parameterize how we dequeue the next inner Observable (SwitchLatestSubscriber would want to pop instead of shift)

[benlesh]

I've renamed this issue to make it more accurate as to the request.

[benlesh]

Closing in favor of #347