Back pressure handling with operator and subject

[richbai90]

I know that handling back pressure has been a recurring topic here with many issues opened and closed on the matter, and yet for all the discussion, I haven't been able to find a solid implementation anywhere, even unofficially. That's why I'm starting this discussion, not because I expect any of the maintainers to change their mind on the topic, (it would be nice, but I understand their reasoning), but to provide a reasonable solution I recently had to implement for a project, get feedback on it, and to begin a discussion around the best way to handle back pressure in RxJS in lieu of an official operator or Observer type.

import { Observable, Observer, Subject, takeUntil, tap, timer } from "rxjs";

/**
 * @param size The chunk size to return from the buffer
 * @param cooldownPeriod How long to wait before clearing the buffer
 * @param queue An observer that will emit with the values in the buffer
 * @param maxSize the maximum size the buffer may be before an error is thrown. Defaults to zero (unbounded)
 * @returns a new observable that emits values from the provided queue in a given number of chunks.
 * The value passed into the buffer is passed straight through, so any transformations required
 * will need to be done ahead of the buffer operator, or as part of the queue pipeline
 */
function buffer(
  size: number,
  cooldownPeriod: number,
  queue: Observer<any>,
  maxSize = 0
) {
  const _buffer = [];
  const notifier = new Subject();
  let subscriptionComplete = false;
  // uses a timer -- suits my purposes but if preferred, pass in a notifier instead of cooldownPeriod
  const checkBuffer = timer(cooldownPeriod, cooldownPeriod).pipe(
    // will not complete until the queue is empty. If that is not desired behavior, just call notifier.next in the inner subscription.
    tap(() => {
      // handle potential memory overflow issues. Throwing an error is nice because it can easily be handled with existing operators
      if (maxSize && _buffer.length > maxSize) {
        throw "buffer has exceeded the maximum allowable size";
      }
      if (!_buffer.length && subscriptionComplete) {
        notifier.next(true);
      } else if (_buffer.length) {
        queue.next(_buffer.splice(0, size));
      }
    }),
    takeUntil(notifier)
  );

  checkBuffer.subscribe();

  return function <T>(source: Observable<T>): Observable<T> {
    return new Observable((subscriber) => {
      return source.subscribe({
        next(value) {
          _buffer.push(value);
          return value;
        },
        complete() {
          subscriptionComplete = true;
          subscriber.complete();
        },

        error(error) {
          subscriber.error(error);
        },
      });
    });
  };
}

export default buffer;

[voliva]

I'm not sure I understand what are you solving here - As far as I understood, back pressure is something that some operators on RxJS may have, so you need to be careful when using them.

The majority of the operators will never have backpressure (e.g. map(), filter(), tap(), etc.), whereas a few of them do (I can think of zip and shareReplay()) where if you are not careful then they might eat through your resources really quick.

buffer is a bit special - It can have backpressure, but I think it's easily solvable, because every time that the notifier emits you get the whole buffer cleaned up, so it sounds like it's unlikely to be used in a way where it could be problematic. And we already have the bufferCount, bufferTime, bufferToggle and bufferWhen.

I think your buffer implementation is doing too much, and breaks conventions on RxJS operators. I honestly don't know when I would need this specific operator... so it's a buffer so that I can give it a size of the arrays I want to receive, but that's actually just the maximum elements I will receive, because I'm also setting a cooldownPeriod which will emit the values it has received so far up to size (but if it received only one value I will get an array with one element), but also, instead of giving me those values through the observable it returns it just .next() them into a queue observer I pass through a parameter. And oh, I can also set a parameter maxSize so that if the internal buffer ever reaches that value then the observable blows up.

It's just... too much? Why not break it down in separate steps and use existing conventions?

queue$ = source$.pipe(
  bufferTime(cooldownPeriod),
  tap(arr => {
    if (maxValue && arr.length > maxValue) {
      throw new Error("buffer has exceeded the maximum allowable size");
    }
  }),
  map(arr => arr.slice(0, size))
)

Isn't this chain equivalent to your buffer? This can now be easily customised to each own's needs.

[richbai90]

The specific challenge is feeding the results of an API call that cannot be paginated, into an API that enforces a limitation on the number of requests that can be made within a certain period. The first API returns a result set of say 100, the 2nd API will only support 3 requests per minute, I need a way of processing the requests from the first in batches to the 2nd. The first should not complete until it has emitted all of the observables, but it needs to be throttled. Like this:

items$ = fromFetch('https://unPaginatedApi/items').pipe(
    concatAll(),
     // collect 3 requests,
    mergeMap(items => fromFetch('https://2ndApi/items'), {method: 'POST', ...}),
   // wait for some time
  // collect 3 more requests
 // repeat until all requests are processed
)

I think that a combination of bufferTime and bufferCount might work for this use case, I looked through the documentation and didn't see those operators, and researching rxjs back pressure handling led to onversations like these:

https://stackoverflow.com/questions/48493924/how-can-i-apply-timed-back-pressure-in-rxjs5
#71
https://javascript.plainenglish.io/lossless-back-pressure-for-rxjs-using-batches-of-data-534752fda666

I think that you are right about this doing too much, it was a working solution to a real problem I needed solve, but if I can make bufferTime/bufferCount work, I'm all for dropping this implementation. I am curious why these operators haven't come up in discussions relating to back pressure. I understand that rxjs4/5 are outdated, but I find it odd that nothing has been mentioned about the issue as far as I could find for the current supported versions.