async stream experiment 5

by jib1

HTML

<div id="div"></div>

JavaScript

let base = performance.now();
const now = () => (performance.now() - base).toFixed();
const console = {log: msg => div.innerHTML += `${now()}: ${msg}<br>`};
const wait = ms => new Promise(resolve => setTimeout(resolve, ms));

(async url => {
  try {
    const rs1 = new ChunkProducer({highWaterMark: 5});
    const rs2 = rs1
      .pipeThrough(new ChunkProcessor({highWaterMark: 2}))
      .pipeThrough(new ChunkProcessor({highWaterMark: 1}));
    await wait(20000);
    console.log("--- READ ---");
    await rs2.pipeTo(new WritableConsole());

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

function ChunkProducer(options) {
  return new ReadableStream({
    chunks: [
      {name: "A"},
      {name: "B"},
      {name: "C"},
      {name: "D"},
      {name: "E"},
      {name: "F"},
      {name: "G"},
      {name: "H"},
      {name: "I"},
      {name: "J"},
      {name: "K"},
      {name: "L"},
      {name: "M"},
      {name: "N"},
      {name: "O"},
      {name: "P"},
      {name: "Q"},
      {name: "R"},
      {name: "S"},
      {name: "T"},
    ],
    queued: [],
    pull(controller) {
      const chunk = this.chunks.shift();
      const action = chunk? `Pulling ${chunk.name}` : "Flushing";
      const {desiredSize} = controller;
      console.log(`${action} (desiredSize=${desiredSize}, enqueued=${this.queued.length})`);
      this.queued.push(chunk || {close: () => {}});
      if (chunk) controller.enqueue(chunk);
    },
  }, options);
}

function ChunkProcessor(options) {
  return new TransformStream({
    async transform(chunk, controller) {
      const name = chunk.name + chunk.name[0];
      console.log(`Processing ${chunk.name} to ${name} (desiredSize=${controller.desiredSize})...`);
      await wait(1000);
      console.log(`...done processing ${chunk.name} to ${name}`);
      controller.enqueue({name});
    }
  }, options, options);
}

function WritableConsole() {
  return new WritableStream({
    write: chunk => console.log(`Output:...