RxJS pipeline with zero backpressure.

balanced-pipeline.ts

import { map, Subject, tap } from "rxjs";

function getBalancedPipeline<T>(
  generator: AsyncGenerator<T, void, void>,
  pipeline = new Subject<T>()
) {
  pipeline.subscribe({
    next: async () => {
      const nextInput = await generator.next();
      if (!nextInput.done) {
        pipeline.next(nextInput.value);
      } else {
        console.log("Generator exhausted");
      }
    },
  });
  return pipeline;
}

async function* getData(): AsyncGenerator<string, void, void> {
  const values = ["one", "two", "three", "four"];
  for (const value of values) {
    yield value;
  }
}

(async () => {
  const generator = getData();
  const pipeline = await getBalancedPipeline(generator);
  pipeline
    .pipe(
      map((value: string) => {
        return value.toUpperCase();
      }),
      tap((value) => console.log(value))
    )
    .subscribe();
  const initial = await generator.next();
  if (!initial.done) {
    pipeline.next(initial.value);
  }
})();