push-based source sync-tee w/back-pressure

by jib1

HTML

<div id="log1"></div>
<div id="log2"></div>
<div id="log3"></div>
<div id="div"></div>

JavaScript

const console = { log: msg => div.innerHTML += msg + "<br>" };
const wait = ms => new Promise(resolve => setTimeout(resolve, ms));

function FrameDropper() {
  return new TransformStream({
    async transform(frame, controller) {
      if (controller.desiredSize < 2) {
        while (controller.desiredSize < 2) await wait(1);
        return;
      }
      controller.enqueue(frame);
    }
  }, {highWaterMark: 1}, {highWaterMark: 2});
}

let counter = 0;
const r = new ReadableStream({
   start(controller) {
     setInterval(() => {
       this.chunk = ++counter;
       if (!this.paused) {
         const {chunk} = this;
         this.chunk = undefined;
         log1.innerHTML += " " + chunk;
         this.paused = true;
         controller.enqueue(chunk);
       }
     }, 500);
   },
   pull(controller) {
     this.paused = false;
   },
});

const [r1, r2] = teeSynchronized(r);
r1
  .pipeThrough(new FrameDropper())
  .pipeTo(new WritableStream({
    write(c) {
        log2.innerHTML += " " + c;
        return new Promise(r => setTimeout(r, 2000));
    }
}));
r2
  .pipeThrough(new FrameDropper())
  .pipeTo(new WritableStream({
    write(c) {
        log3.innerHTML += " " + c;
        return new Promise(r => setTimeout(r, 4000));
    }
}));

function teeSynchronized(readable) {
  const reader = readable.getReader();
  let canceled1 = false;
  let canceled2 = false;
  let reading = false;
  let reason1;
  let reason2;
  let controller1;
  let controller2;
  let resolvePullPromise1;
  let pullPromise1 = new Promise(resolve => {
    resolvePullPromise1 = resolve;
  });
  let resolvePullPromise2;
  let pullPromise2 = new Promise(resolve => {
    resolvePullPromise2 = resolve;
  });
  let resolveCancelPromise;
  const cancelPromise = new Promise(resolve => {
    resolveCancelPromise = resolve;
  });
  async function pullAlgorithm() {
    if (reading) {
      return;
    }
    reading = true;
    if (!canceled1) {
      await pullPromise1;
    }
    if (!canceled2) {
     ...