Delay batch of observables with RxJS

user_276439 0
Viewed 574

I perform http requests to my db and have noticed that if I send all the requests at once, some of them will get a timeout errors. I'd like to add a delay between calls so the server doesn't get overloaded. I'm trying to find the RxJS solution to this problem and don't want to add a setTimeout.

Here is what I currently do:

let observables = [];
for(let int = 0; int < 10000; int++){
   observables.push(new Observable((observer) => {
      db.add(doc[int], (err, result)=>{
         observer.next();
         observer.complete();
      })
   }))
}

forkJoin(observables).subscribe(
   data => {
   },
   error => {
      console.log(error);
   },
   () => {
      db.close();
   }
);
rxjs delay
3 Answers

You can indeed achieve this with Rxjs quite nicely. You'll need higher order observables, which means you'll emit an observable into an observable, and the higher order observable will flatten this out for you.

The nice thing about this approach is that you can easily run X requests in // without having to manage the pool of requests yourself.

Here's the working code:

import { Observable, Subject } from "rxjs";
import { mergeAll, take, tap } from "rxjs/operators";

// this is just a mock to demonstrate how it'd behave if the API was
// taking 2s to reply for a call
const mockDbAddHtppCall = (id, cb) =>
  setTimeout(() => {
    cb(null, `some result for call "${id}"`);
  }, 2000);

// I have no idea what your response type looks like so I'm assigning
// any but of course you should have your own type instead of this
type YourRequestType = any;

const NUMBER_OF_ITEMS_TO_FETCH = 10;

const calls$$ = new Subject<Observable<YourRequestType>>();

calls$$
  .pipe(
    mergeAll(3),
    take(NUMBER_OF_ITEMS_TO_FETCH),
    tap({ complete: () => console.log(`All calls are done`) })
  )
  .subscribe(console.log);

for (let id = 0; id < NUMBER_OF_ITEMS_TO_FETCH; id++) {
  calls$$.next(
    new Observable(observer => {
      console.log(`Starting a request for ID "${id}""`);
      mockDbAddHtppCall(id, (err, result) => {
        if (err) {
          observer.error(err);
        } else {
          observer.next(result);
          observer.complete();
        }
      });
    })
  );
}

And a live demo on Stackblitz: https://stackblitz.com/edit/rxjs-z1x5m9

Please open the console of your browser and note that the console log showing when a call is being triggered starts straight away for 3 of them, and then wait for 1 to finish before picking up another one.

user_23985930

Looks like you could use an initial timer to trigger the http calls. e.g.

timer(delayTime).pipe(combineLatest(()=>sendHttpRequest()));

This would only trigger the sendHttpRequest() method after the timer observable had completed.

So with your solution. You could do the following...

   observables.push(
     timer(delay + int).pipe(combineLatest(new Observable((observer) => {
      db.add(doc[int], (err, result)=>{
         observer.next();
         observer.complete();
      }))
   }))

Where delay could probably start off at 0 and you could increase it using the int index of your loop by some margin.

Timer docs: https://www.learnrxjs.io/learn-rxjs/operators/creation/timer

Combine latest docs: https://www.learnrxjs.io/learn-rxjs/operators/combination/combinelatest

user_89740880

merge with concurrent value:

mergeAll and mergeMap both allow you to define the max number of subscribed observables. mergeAll(1)/mergeMap(LAMBDA, 1) is basically concatAll()/concatMap(LAMBDA).

merge is basically just the static mergeAll

Here's how you might use that:

let observables = [...Array(10000).keys()].map(intV => 
  new Observable(observer => {
    db.add(doc[intV], (err, result) => {
      observer.next();
      observer.complete();
    });
  })
);

const MAX_CONCURRENT_REQUESTS = 10;

merge(...observables, MAX_CONCURRENT_REQUESTS).subscribe({
   next: data => {},
   error: err => console.log(err),
   complete: () => db.close()
});

Of note: This doesn't batch your calls, but it should solve the problem described and it may be a bit faster than batching as well.

mergeMap with concurrent value:

Perhaps a slightly more RxJS way using range and mergeMap

const MAX_CONCURRENT_REQUESTS = 10;

range(0, 10000).pipe(
  mergeMap(intV => 
    new Observable(observer => {
      db.add(doc[intV], (err, result) => {
        observer.next();
        observer.complete();
      });
    }),
    MAX_CONCURRENT_REQUESTS
  )
).subscribe({
   next: data => {},
   error: err => console.log(err),
   complete: () => db.close()
});
user_135009860
Related