ReadableStream.tee({ synchronized: true })

https://github.com/whatwg/streams/issues/1157

by MattiasBuelens

JavaScript

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) {
      await pullPromise2;
    }
    const result = await reader.read();
    reading = false;
    if (result.done) {
      if (!canceled1) {
        controller1.close();
      }
      if (!canceled2) {
        controller2.close();
      }
      if (!canceled1 || !canceled2) {
        resolveCancelPromise();
      }
    } else {
      if (!canceled1) {
        controller1.enqueue(result.value);
      }
      if (!canceled2) {
        controller2.enqueue(result.value);
      }
      pullPromise1 = new Promise(resolve => {
        resolvePullPromise1 = resolve;
      });
      pullPromise2 = new Promise(resolve => {
        resolvePullPromise2 = resolve;
      });
    }
  }
  const branch1 = new ReadableStream({
    start(c) {
      controller1 = c;
    },
    pull() {
      resolvePullPromise1();
      return pullAlgorithm();
    },
    cancel(reason) {
      canceled1 = true;
      reason1 = reason;
      resolvePullPromise1();
      if (canceled2) {
        const compositeReason = [reason1, reason2];
        const cancelResult = reader.cancel(compositeReason);
        resolveCancelPromise(cancelResult);
      }
      return cancelPromise;
    }
  }, {
    highWaterMark: 0
  });
  const branch2 = new ReadableStream({
    start(c) {
     ...