Parallel rxJs Observables without waiting for all to complete

Question

Is there a way to request many similar requests in parallel using rxJs Observables in parallel without waiting for all requests to complete. I want to handle as each one comes. I don't want to use a loop of requests. Use case is one of the request is slow and there are many of them I tried forkJoin (below) but it waits for all to complete. and mergeMap is not taking an array. Any help? Thanks

/*
  // example of many posts
  const post1 = getPostHTTP(1);
  const post2 = getPostHTTP(2); // A slow post
  const post3 = getPostHTTP(3);
  ...
*/

const promises = [];

for (let i = 0; i < 100; i++) {
    promises.push(getPostHTTP(i))
}

forkJoin(promises).pipe(
    finalize(() => {
        //displayPost
    })).subscribe(posts => {
        posts.forEach((post: any) => {
            displayPost(post)
        })
    })

user_5699993 · Answer

I think that a mix of from and mergeMap is what you are looking for. The code could look like this

for (let i = 0; i < 100; i++) {
    promises.push(getPostHTTP(i))
}

from(promises).pipe(
  mergeMap(post => displayPost(post)),
).subscribe(
  next: result => {// do something with the result of each displayPost operation},
  error: err => {// handle errors},
  complete: () => {// do something when everything is completed}
)

user_7255496 · Answer

You should convert Promise to Observable via from (https://www.learnrxjs.io/learn-rxjs/operators/creation/from) and merge them into one Observable via merge (https://www.learnrxjs.io/learn-rxjs/operators/combination/merge).

It will be like:

merge(...promises.map(p => from(p))
    .subscribe(post => {
        // do smth with post
    });

It will be like:

merge(...promises.map(p => from(p))
    .subscribe(post => {
        // do smth with post
    });

user_6656967 · Answer

You can use Promise API to send multiple requests simultaneously and await them all to resolve.

If you have a condition that you need to run those promises sequentially, then this solution might not work for you.

Excerpt is from MDN.

var p1 = new Promise((resolve, reject) => {
  setTimeout(resolve, 1000, 'one');
});
var p2 = new Promise((resolve, reject) => {
  setTimeout(resolve, 2000, 'two');
});
var p3 = new Promise((resolve, reject) => {
  setTimeout(resolve, 3000, 'three');
});
var p4 = new Promise((resolve, reject) => {
  setTimeout(resolve, 4000, 'four');
});
var p5 = new Promise((resolve, reject) => {
  reject('reject');
});

Promise.all([p1, p2, p3, p4, p5]).then(values => {
  console.log(values);
}, reason => {
  console.log(reason)
});

//From console:
//"reject"

//You can also use .catch
Promise.all([p1, p2, p3, p4, p5]).then(values => {
  console.log(values);
}).catch(reason => {
  console.log(reason)
});

Parallel rxJs Observables without waiting for all to complete

3 Answers