backpressure-pages-take.ts

import { EMPTY, Observable } from "rxjs";
import { expand, map, mergeMap, take } from "rxjs/operators";
import { get } from "./github";

function pages<T>(
  uri: string,
  notifier: Observable<any>
): Observable<T[]> {
  return get<T[]>(uri).pipe(
    expand(({ next }) => {
      if (next) {
        return notifier.pipe(
          take(1),
          mergeMap(() => get(next))
        );
      }
      return EMPTY;
    }),
    map(({ content }) => content)
  );
}