Implementing an http request queue with RXJS

Question

I'm still new to RXJS and trying to get my head around it. The task is as follows:

A user can add assets to a collection, and he might click faster than the backend can handle the requests, so that an error might occur because the lockVersion of the collection conflicts with the one sent by the client (not updated yet). A hacky solution would be to reload the collection when the 409 error occurs, but I thought a more elegant solution would a be a request queue.

Also, we will will have to handle justified 409 errors, when two users are meddling with the same collection, so a queue will come in handy then (reload the collection, re-add the item to the queue).

Here's my naive approach, which actually works:

export class RequestQueue<T> {
  private items: T[] = [];
  private queue$: Subject<boolean> = new Subject<boolean>();
  private dispatcher: Subject<T> = new Subject<T>();

  constructor(
    private readonly requestHandler: (v: T) => Observable<unknown>,
    private readonly errorHandler: (e: HttpErrorResponse) => Observable<boolean>
  ) {
    this.initQueue();
  }

  public add(item: T) {
    if (this.items.push(item) == 1) {
      this.dispatcher.next(this.items.shift());
    }
    return this.queue$;
  }

  public destroy () {
    this.dispatcher.complete();
  }

  private initQueue() {
    this.dispatcher.pipe(
      flatMap(item => {
        return this.requestHandler(item)
      }),
      tap(r => {
        if (this.items.length > 0) {
          this.dispatcher.next(this.items.shift());
        } else {
          this.queue$.next(true);
        }
      }),
      catchError((error: HttpErrorResponse) => {
        let error$ = this.errorHandler(error);
        this.initQueue(); //<--- please note this line
        return error$;
      })
    ).subscribe();
  }
}

As you can see, I am holding an array which can be filled up with items as much as the user feels like clicking. When the first item is added, the dispatcher emits it as its first event (value). After every successful request (collection updated), it shifts the next item from the array and emits it. This happens until the array is empty. The queue$ observer then emits 'true' to notify the caller of the queue, that it is done.

This actually works, and the 409 error never happens again because of a fast-clicking user. The trouble starts, when another error (e.g. 404) occurs. The error handler pops up a message, the user clicks ok, and the pipe of the dispatcher is dead. The next click to add an item to the collection won't cause anything, unless I initialize the queue (re-pipe to the dispatcher) again.

This doesn't feel like an elegant solution at all. I'm not quite sure why this happens, but I would really like to understand it. Any explanation or suggestion for a better solution is highly appreciated.

frido frido · Accepted Answer · 2020-10-07T19:29:02

A different approach would be to have your add function (I name it request) return an Observable that actually emits what the http request emits and can be used to cancel the request.

private dispatcher: Subject<Observable<unknown>> = new Subject<Observable<unknown>>();

public request(item: T) {
  return new Observable(observer => {
    const cancel = new Subject();
    const request$ = this.requestHandler(item).pipe(
      catchError(this.errorHandler),
      takeUntil(cancel), // cancel the request on cancel events
      tap(observer) // attach the observer to the request
    ));
    this.dispatcher.next(request$) // add the request to the queue
    return () => { // send a cancel event on unsubscribe
      cancel.next();
      cancel.complete();
    }
  });
}

private initQueue() {
  this.dispatcher.pipe(concatAll()).subscribe();
}

Use it like

cancel = new Subject();

send(item: T) {
  request(item).pipe(
    takeUntil(this.cancel)
  ).subscribe()
}

cancelAll() {
  this.cancel.next()
}

https://stackblitz.com/edit/rxjs-3xvo8g?file=index.ts

Implementing an http request queue with RXJS

2 Answers