Angular 10, rxjs - looping on subscription?

user_10143290 0
Viewed 614

I'm making multiple concurrent DELETE requests to an API, causing a DeadLock on the server side.

If I had a fixed number of deletions I could do something like this:

myAPI.delete(id).subscribe(res => {
   myAPI.delete(id2).subscribe(res => {
      myAPI.delete(id3).subscribe(res => {
      })
   })
})

Waiting to the previous deletion to be succesfully completed before making a new one (secuential).

The problem is that I can have from 1 to N requests, so I'm looking for something similar to a for instruction on subscription.

Is there any elegant approach for solving this?

rxjs angular
4 Answers

If you have an array of ids that need to be removed, you can use the from operator in combination with concatMap and toArray. The last one is to only call the subscribe listener after all have been completed:

const ids = [ 1, 2, 3 ];

from(ids).pipe(
  concatMap((id) => deleteThis(id)),
  toArray()
).subscribe(() => {
   console.log('all deleted');
})

working stack with mock request observable

user_31069200

There are multiple methods.

Sequential requests

You could use the RxJS range operator with mapping operator concatMap to call each request sequentially.

import { range } from 'rxjs';
import { concatMap } from 'rxjs';

range(1, N).pipe(                      // <-- will emit 1-N numbers
  concatMap(id => myAPI.delete(id))    // <-- set the correct `id` for each request here
).subscribe(
  res => console.log(res),
  err => console.log(err),
);

Since we're using concatMap operator, the request for each id will wait before the previous request is complete.

You could also use other mapping operators like switchMap, flatMap and exhaustMap for different behaviors. Difference b/n them here.

Other methods - Parallel requests

Complete parallel requests

You could make multiple concurrent calls using the RxJS toArray and forkJoin() operators. It subscribes to multiple observables simultaneously.

import { range, forkJoin } from 'rxjs';
import { concatMap, toArray } from 'rxjs/operators';

range(1, N).pipe(                      // <-- will emit 1-N numbers
  concatMap(id => myAPI.delete(id)),   // <-- set the correct `id` for each request here
  toArray(),                           // <-- buffer all notifications and emit once as an array
  concatMap(reqs => forkJoin(reqs))
).subscribe(
  res => console.log(res),
  err => console.log(err),
);

Buffered parallel requests

Most browsers have a hard limit on number of maximum parallel requests to a single domain (eg. Chrome - 6). If you run into this issue you could use RxJS bufferCount operator instead of the toArray operator to control the max number of parallel requests.

import { from, forkJoin } from 'rxjs';
import { concatMap, bufferCount } from 'rxjs';

range(1, N).pipe(                      // <-- will emit 1-N numbers
  concatMap(id => myAPI.delete(id)),   // <-- set the correct `id` for each request here
  bufferCount(6),                      // <-- adjust number of parallel requests here
  concatMap(reqs => forkJoin(reqs))
).subscribe(
  res => console.log(res),
  err => console.log(err),
);
user_65139210

Alternative solution is to use concat operator:

import { concat } from 'rxjs';
import { toArray } from 'rxjs/operators';

concat(
   ...[1, 2, 3].map(id => this.delete(id)),
)
.pipe(toArray())
.subscribe(() => console.log('all deleted'))

or without toArray() operator:

import { concat } from 'rxjs';

concat(...[1, 2, 3].map(id => this.delete(id)))
  .subscribe(
    res => {
      console.log('one deleted')
    },
    err => {},
    () => {
      console.log('all deleted')
    })
user_54851670 
const N = 10;
range(1, N)
  .pipe(concatMap(deleteById), ignoreElements())
  .subscribe({
      complete: () => console.log('Complete')
  });

function deleteById(id: number): Observable<String> {
  return of(`deleted-${id}`).pipe(delay(1000));
}

If you want make multiple request same time:


const N = 10;
const CHUNKED = 3; // (3 requests same time) concat (3 requests same time) ...

range(1, N)
  .pipe(
    bufferCount(CHUNKED),
    concatMap((ids) => forkJoin(ids.map(deleteById))),
    ignoreElements()
  )
  .subscribe({
    complete: () => console.log(`Complete`)
  });
user_111914240
Related