Implement debounced chunked async queue with streams

Question

I am new to reactive programming and curious if there might be a more elegant way of implementing a debounced chunked async queue. What is a debounced chunked async queue you ask? Well, maybe there is a better name for it, but the idea is that an async function may be called many times over some period of time. We want to first debounce the calls to the function and batch up all those arguments into one argument. Secondly all executions of that async function should be serialized in a FIFO manner.

Here is my implementation without streams, observables or reactive programming, as I see it:

const debouncedChunkedQueue = <T>(
  fn: (items: T[]) => Promise<void> | void,
  delay = 1000
) => {
  let items: T[] = [];
  let started = false;
  const start = async () => {
    started = true;
    while (items.length) {
      await sleep(delay);
      const chunk = items.splice(0, items.length);
      await fn(chunk);
    }
    started = false;
  };
  const push = (item: T) => {
    items.push(item);
    if (!started) start();
  };
  return { push };
};

https://codesandbox.io/s/priceless-sanne-dkrkw?file=/src/index.ts:87-550

Answer 1

I followed your current implementation 1:1 which resulted in more complicated code than you would achieve with simple bufferTime.

You can test the following code in RxViz.

const { BehaviorSubject, fromEvent, of } = Rx;
const { buffer, delay, first, tap, mergeMap, exhaustMap } = RxOperators;

// Source of the queue - click to emit event
const items = fromEvent(document, 'click');

// a "pushback" source
const processing = new BehaviorSubject(false);

items.pipe(
  // we resubscribe to items to start only when an item is there
  buffer(
    // we take items again so we only
    // start if there are any items
    // to be processed and then we wait
    // 1 second
    items.pipe(
      // we do not start processing until
      // pending async call completes
      exhaustMap(() =>
        processing.pipe(
          // wait for processing end
          first(val => !val),
          // wait for 1 second
          delay(1000),
        ),
      ),
    ),
  ),
  tap(() => processing.next(true)),
  // basic mergeMap is okay, since we control that the next value
  // will come no sooner than this is completed
  mergeMap(items =>
    // async fn simulation
    of(items.length).pipe(delay(300 + Math.random() * 500)),
  ),
  tap(() => processing.next(false)),
);

It's a bummer that we have to keep a state outside the observable, though.

A debouncedChunkedQueue

You can test the following code in stackblitz

import { BehaviorSubject, Subject, of } from "rxjs";
import {
  buffer,
  delay,
  first,
  tap,
  mergeMap,
  exhaustMap
} from "rxjs/operators";

const debouncedChunkedQueue = <T>(
  fn: (items: T[]) => Promise<void> | void,
  delayMs = 1000
) => {
  // Source of the queue - click to emit event
  const items = new Subject<T>();

  // a "pushback" source
  const processing = new BehaviorSubject(false);

  items
    .pipe(
      // we resubscribe to items to start only when an item is there
      buffer(
        // we take items again so we only
        // start if there are any items
        // to be processed and then we wait
        // 1 second
        items.pipe(
          // we do not start processing until
          // pending async call completes
          exhaustMap(() =>
            processing.pipe(
              // wait for processing end
              first(val => !val),
              // wait for 1 second
              delay(delayMs)
            )
          )
        )
      ),
      tap(() => processing.next(true)),
      // basic mergeMap is okay, since we control that the next value
      // will come no sooner than this is completed
      mergeMap(
        items =>
          // TODO: Make sure to catch errors from the fn if you want the queue to recover
          // async fn simulation
          fn(items) || of(null)
      ),
      tap(() => processing.next(false))
    )
    .subscribe();
  return { push: (item: T) => items.next(item) };
};