RXJS Work Queue

work-queue.ts

import * as rx from 'rxjs';
import { concatMap, onErrorResumeNext, tap } from 'rxjs/operators';

export class WorkQueue<T = unknown> {
  private queue = new rx.Subject<[rx.Observable<T>, rx.Subscriber<T>]>();

  constructor() {
    this.queue.pipe(concatMap(this.doWork)).subscribe();
  }

  enqueue(work: rx.Observable<T>): rx.Observable<T> {
    return new rx.Observable<T>((obs) => {
      this.queue.next([work, obs]);
    });
  }

  private doWork = ([work, sub]: [rx.Observable<T>, rx.Subscriber<T>]) => {
    if (sub.closed) {
      return rx.EMPTY;
    }

    return work.pipe(
      tap({
        next: (x) => sub.next(x),
        error: (x) => sub.error(x),
        complete: () => sub.complete(),
      }),
      onErrorResumeNext(rx.EMPTY)
    );
  };
}