NRaf NRaf - 16 days ago 7
Javascript Question

Batching using RxJS?

I'm guessing this should be somewhat easy to achieve but I've having trouble (conceptually, I guess) figuring out how to tackle it.

What I have is an API that returns an array of JSON objects. I need to step through these objects, and, for each object, make another AJAX call. The issue is the system that handles each AJAX call can only handle two active calls at a time (as it's quite a CPU-intensive task that hooks out into a desktop application).

I was wondering how I could achieve this using RxJS (either using version 5 or 4)?

EDIT: In addition, is it possible to have a chain of steps running concurrently. i.e.


Downloading File: 1
Processing File: 1
Converting File: 1
Uploading File: 1
Downloading File: 2
Processing File: 2
Converting File: 2
Uploading File: 2
Downloading File: 3
Processing File: 3
Converting File: 3
Uploading File: 3


I've tried doing something like:

Rx.Observable.fromPromise(start())
.concatMap(arr => Rx.Observable.from(arr))
.concatMap(x => downloadFile(x))
.concatMap((entry) => processFile(entry))
.concatMap((entry) => convertFile(entry))
.concatMap((entry) => UploadFile(entry))
.subscribe(
data => console.log('data', new Date().getTime(), data),
error => logger.warn('err', error),
complete => logger.info('complete')
);


However that doesn't seem to work. The downloadFile, for example doesn't wait for processFile, convertFile and uploadFile to all complete, rather, the next one will run again as soon as the previous one completes.

Answer

Here are 2 approaches, if you want the sequence of requests exactly like this

Downloading File: 1
Processing File: 1
Converting File: 1
Uploading File: 1
Downloading File: 2
Processing File: 2
...

You need to resolve all promises inside single concatMap method, like this

Rx.Observable.fromPromise(getJSONOfAjaxRequests())
  .flatMap(function(x) { return x;})
  .concatMap(function(item) {
    return downloadFile(item)
      .then(processFile)
      .then(convertFile);
  })
  .subscribe(function(data) {
    console.log(data);
  });

see the working plunkr here: https://plnkr.co/edit/iugdlC2PpW3NeNF2yLzS?p=preview This way, the new ajax call will be sent only when the previous is finished.

Another approach is that allow the files to send requests in parallel but the operations 'downloading,processing,converting,uploading' will be in sequence. For this you can get it working by

Rx.Observable.fromPromise(getJSONOfAjaxRequests())
  .flatMap(function(x) { return x;})
  .merge(2)  // in case maximum concurrency required is 2
  .concatMap(function(item) {
    return downloadFile(item);
  })
  .concatMap(function(item) {
    return processFile(item);
  })
  .concatMap(function(item) {
    return convertFile(item)
  })
  .subscribe(function(data) {
    //console.log(data);
  });

see plunkr here: https://plnkr.co/edit/mkDj6Q7lt72jZKQk8r0p?p=preview

Comments