backpressure-pages-queue.ts

import { EMPTY, Observable, Subscription } from "rxjs";
import { concatMap, expand, map } from "rxjs/operators";
import { NotificationQueue } from "rxjs-etc";
import { get } from "./github";

function pages<T>(
  uri: string,
  notifier: Observable<any>
): Observable<T[]> {
  return new Observable<any>(observer => {
    const notificationQueue = new NotificationQueue(notifier);
    const subscription = new Subscription();
    subscription.add(notificationQueue.connect());
    subscription.add(
      get<T[]>(uri).pipe(
        expand(({ next }) => {
          if (next) {
            return notificationQueue.pipe(
              concatMap(() => get(next))
            );
          }
          return EMPTY;
        }),
        map(({ content }) => content)
      ).subscribe(observer)
    );
    return subscription;
  });
}