async tee with progress sync workaround

by jib1

HTML

<script src="https://jan-ivar.github.io/dummy/progressbar.js"></script>
<div id="container"></div>
<div id="div"></div>

JavaScript

const base = performance.now();
const now = () => (performance.now() - base).toFixed();
const console = {log: msg => div.innerHTML += `${now()}: ${msg}<br>`};
const waitFrame = () => new Promise(r => requestAnimationFrame(r));
const waitFrames = async n => { while (n--) await waitFrame(); }

const size = 100;

(async url => {
  try {
    const bar1 = makeProgressBar({size});
    const bar2 = makeProgressBar({size});

    const rs1 = new Producer({highWaterMark: 0});
    const [rs1a, rs1b] = rs1.tee();
    const rs2a = rs1a.pipeThrough(new Processor({delay: 2, bar: bar1}));
    const rs2b = rs1b.pipeThrough(new Processor({delay: 1, bar: bar2}));

    await Promise.all([
      rs2a.pipeTo(new WritableStream()),
      rs2b.pipeTo(new WritableStream()),
    ]);

  } catch (e) {
    console.log(`${e.name}: ${e.message}`);
  }
})();

function Producer(options) {
  return new ReadableStream({
    chunks: [],
    start() {
      for (let i = 0; i < size; i++) {
        this.chunks[i] = i+1;
      }
    },
    pull(controller) {
      const chunk = this.chunks.shift();
      if (chunk) {
        controller.enqueue(chunk);
      } else {
        console.log("Empty pull");
      }
    },
  }, options);
}

const map = new Map();

function Processor({delay, bar}) {
  return new TransformStream({
    async transform(chunk, controller) {
      if (!map.has(chunk)) map.set(chunk, []);
      const promises = map.get(chunk);
      let finish;
      promises.push(new Promise(r => finish = r));
      await waitFrames(delay);
      finish();
      await Promise.all(promises);
      bar.progress(chunk);
      controller.enqueue(chunk);
    }
  });
}