Change "concurrent" value dynamically in mergeMap operator

[pnsvk]

Can some one please help with some pointers, insights into, how one can dynamically change the 'concurrent' parameter value to mergeMap operator ?

My use-case is to let the end-users, play with the number of concurrent API-Fetch calls to be made dynamically.. just as a typical download-manager would let a user dynamically change the number of concurrent downloads from a huge list of queued up download-links..

[pnsvk]

FYI, there was exactly the same request on reactive-dotnet long time back, but was closed stating no valid reason for its closure.

dotnet/reactive#48

[benlesh]

Can you describe the use case?

[Sawtaytoes]

Based on available bandwidth, you might want to raise or lower the number of items going through a pipeline.

I think a mergeAllWhen would be in order if this is something you'd wanna add to RxJS proper. In another answer, I explained a way to do this today, but there's a lot of code involved.

[Sawtaytoes]

This is definitely possible, but will cause you to unsubscribe and resubscribe. If you had something like XHR requests, you could pause it. With these steps, you'll have to start over after changing the number of pipelines.

1. You need a running list of all observables.
2. When you dynamically change the number of pipelines, you're going to trigger a subject to fire again.
3. You'll need to filter the observable list to only subscribe to those that did not complete.

downloadInBatches$
.pipe(
  switchMap(({
    numberOfPipelines,
    observables,
  }) => (
    of(observables)
    .pipe(
      mergeAll(
        numberOfPipelines,
      ),
    )
  ))
)

// start
downloadInBatches$
.next({
  numberOfPipelines: 4,
  observables: [obs1$, obs2$],
})

// change occurred
downloadInBatches$
.next({
  numberOfPipelines: 2,
  observables: (
    [obs1$, obs2$]
    .filter(({
      isClosed,
    }) => (
      !isClosed
    )),
})

[pnsvk]

@Sawtaytoes -- Sorry.. I was held up with some priorities.. could not ACK your response right after..

Thanks a lot for taking time to respond and come up with a possible solution.. I will try to leverage this in our case, but couple things:

• How can we pause in-flight XHR requests and resume the same, when we re-subscribe ?
• Is there a better way (if I am not asking for more) to be able to dynamically change the concurrency parameter of merge/map-operator ? -- that would be a lot more readable..

[Sawtaytoes]

To your second point, no likely. This is what I would do, but there'd be a bit more logic to filter down that list of finished observables somewhere else, so when the dynamic pipeline change comes in, I'm only sending observables yet to finish.

I've played around with pausing XHR, and it's possible, but I don't think it's possible with the ajax observable-creator in RxJS. You'd need to create that functionality yourself. It's possible, just not simple.

Here's an example of me writing around XHR with progress: https://github.com/Sawtaytoes/image-viewer/blob/master/src/components/imageLoader/createFileDownloadObservable.js

Hopefully that gives you some idea of how to create a pausing observable. You could probably keep track of the paused XHR object separate from the observable wrapper and when calling downloadInBatches$.next, create a new observable wrapper with the paused version.

[pnsvk]

thank you for being so kind to have given pointers and insight.. will definitely give it a shot. thanks once again.. !!