How to poll several endpoints sequentially with rxjs?

Updated: Original answer was not on the right track.

What we want to achieve is that on each go around of the interval we poll all the outstanding jobs in order. We yield up any completed jobs to the output observable and we also omit those completed jobs from subsequent polls.

We can do that by using a Subject instead of a static observable of the job IDs. We start our poll interval and we use withLatestFrom to include the latest list of job IDs. We can then add a tap into the output observable when we get a finished job and update the Subject to omit that job.

To end the poller interval we can create an observable that fires when the array of outstanding jobs is empty and use takeUntil with that.

const number$ = new Subject();
const noMoreNumber$ = number$.pipe(skipWhile((numbers) => numbers.length > 0));

const printout$ = interval(5000).pipe(
  withLatestFrom(number$),
  switchMap(([_, numbers]) => {
    return numbers.map((number) => defer(() => doRequest(number)));
  }),
  concatAll(),
  //tap(console.log),
  filter(({ status }) => status === 'FINISHED'),
  withLatestFrom(number$),
  tap(([{ value }, numbers]) =>
    number$.next(numbers.filter((num) => num != value))
  ),
  map(([item]) => item),
  takeUntil(noMoreNumber$)
);

printout$.subscribe({
  next: console.log,
  error: console.error,
  complete: () => console.log('COMPLETE'),
});

number$.next([1, 2, 3, 4, 5, 6, 7, 8, 9, 10]);

The other tweak I would make is to use switchMap instead of mergeMap inside the poller itself. If you use that in combination with fromFetch for performing your HTTP calls then, if there is some long-running HTTP call which gets stuck, on the next poll the previous call will be cancelled before it makes the next HTTP call because switchMap disposes of the previous observable before subscribing to the new one.

Here's a working example: https://stackblitz.com/edit/js-gxrrb3?devToolsHeight=33&file=index.js

Generates console output looking like this...

If I understand the problem right, I would proceed like this.

First of all I would create a poll function that returns an Observable which notifies after a round of pollings, and it emits an array of all numbers for which the call to doRequest returns 'RUNNING'. Such a function would look something like this

const poll = (numbers: number[]) => {
  return from(numbers).pipe(
    concatMap((n) =>
      doRequest(n).pipe(
        filter((resp) => resp.status === 'RUNNING'),
        map((resp) => resp.value)
      )
    ),
    toArray()
  );
};

Then what you need to do is to recursively iterate a call the poll function until the array emitted by the Observable returned by poll is empty.

Recursion in rxjs is obtained typically with the expand operator, and this is the operator which we are going to use also in this case, like this

poll(numbers)
  .pipe(
    expand((numbers) =>
      numbers.length === 0
        ? EMPTY
        : timer(2000).pipe(concatMap(() => poll(numbers)))
    )
  )
  .subscribe(console.log);

A complete example can be seen in this stackblitz.

UPDATE

If the objective is to notify the job ids which have finished with a polling logic, the structure of the solution remains the same (a poll function and recursivity via expand) but the details are different.

The poll function makes sure we emit all the responses of a polling round and it looks like this:

const poll = (
  numbers: number[]
) => {
  console.log(`Polling ${numbers}`);
  return from(numbers).pipe(
    concatMap((n) => doRequest(n)),
    toArray()
  );
};

The recursion logic makes sure that all jobs that are still with "RUNNING" status are polled again but then we filter only the jobs which are FINISHED and passed them downstream. In other words the logic looks like this

poll(start)
  .pipe(
    expand((responses) => {
      const numbers = responses.filter(r => r.status === 'RUNNING').map(r => r.value)
      return numbers.length === 0
        ? EMPTY
        : timer(2000).pipe(concatMap(() => poll(numbers)));
    }),
    map(responses => responses.filter(r => r.status === 'FINISHED')),
    filter(finished => finished.length > 0)
  )
  .subscribe({
    next: responses => console.log(`Job finished ${responses.map(r => r.value)}`),
    complete: () => {console.log('All processed')}
  });

A working example can be seen in this stackblitz.

TRY this

import { delay, EMPTY, from, of, range } from 'rxjs';
import { concatMap, filter, mergeMap, tap, toArray } from 'rxjs/operators';

const number$ = from(range(1, 3));

const doRequest = (input) => {
  const status = Math.random() < 0.15 ? 'FINISHED' : 'RUNNING';

  return of({ status, value: input }).pipe(delay(1000));
};

const poll = (jobs: object[]) => {
  return from(jobs).pipe(
    filter((job) => job['status'] !== 'FINISHED'),
    concatMap((job) => doRequest(job['value'])),
    tap((job) => {
      console.log('polling with................', job);
    }),
    toArray(),
    tap((result) => {
      console.log('curent jobs................', JSON.stringify(result));
    }),
    mergeMap((result) =>
      result.length > 0 ? poll(result) : of('All job completed!')
    )
  );
};

const initiateJob = number$.pipe(
  mergeMap((id) => doRequest(id)),
  toArray(),
  tap((jobs) => {
    console.log('initialJobs: ', JSON.stringify(jobs));
  }),
  concatMap(poll)
);

initiateJob.subscribe({
  next: console.log,
  error: console.log,
  complete: () => console.log('COMPLETED'),
});