玩转 stream API

const readableStream = new ReadableStream(
  {
    start(controller) {
      // 实例化时立即执行
      // 主要是做一些事前准备
      // 如果要在这里 enqueue 数据的话必须考虑 desiredSize
    },

    pull(controller) {
      // 简单来说就是当队列有空余的时候，下游会从上游拉数据
      // 这里说到队列策略再解释
      controller.enqueue("chunk");
    },

    cancel(reason) {
      // 取消回调
    },
  },
  new CountQueuingStrategy({ highWaterMark: 5 }), // 非必填
);

const reader = readableStream.getReader();

reader.read().then(
  ({ value, done }) => {
    if (done) {
      console.log("The stream was already closed!");
    } else {
      console.log(value);
    }
  },
  (e) => console.error("The stream became errored and cannot be read from!", e),
);

const writableStream = new WritableStream(
  {
    start(controller) {
      // 实例化时立即执行
      // 主要是做一些事前准备
    },

    write(chunk, controller) {
      // chunk 写入的最终处理
    },

    close(controller) {
      // 关闭回调
    },

    abort(reason) {
      // 报错回调
    },
  },
  new CountQueuingStrategy({ highWaterMark: 5 }),
);

// https://streams.spec.whatwg.org/#example-manual-write-dont-await
async function writeSuppliedBytesForever(writableStream, getBytes) {
  const writer = writableStream.getWriter();

  while (true) {
    await writer.ready;

    const bytes = getBytes();
    writer.write(bytes).catch(() => {});
  }
}

const transformStream = new TransformStream(
  // 这块名为 transformer
  {
    start(controller) {
      /* … */
    },

    transform(chunk, controller) {
      /* … */
    },

    flush(controller) {
      /* 所有任务完成的收尾工作 */
    },
  },
  writableStrategy,
  readableStrategy,
);

readableStream.pipeThrough(transformStream1).pipeThrough(transformStream2).pipeTo(writableStream);

// ...
{
  highWaterMark: 1024,
  size(chunk) {
    return chunk.length
  },
},
// ...

const rQueuingStrategy = new CountQueuingStrategy({ highWaterMark: 15 })
const readableStream = new ReadableStream(
  {
    start(controller) {
      button.onclick = () => {
        const log =
          index++ + ' start: ' + randomChars() + controller.desiredSize
        logInDiv(readableDiv!, log)
        controller.enqueue(log)
        console.log('controller.desiredSize', controller.desiredSize)
      }
    },
    pull(controller) {
      const log = index++ + ' pull: ' + randomChars() + controller.desiredSize
      logInDiv(readableDiv!, log)
      controller.enqueue(log)
      console.log('controller.desiredSize', controller.desiredSize)
    },
  },
  rQueuingStrategy,
)

const wQueuingStrategy = new CountQueuingStrategy({ highWaterMark: 10 })
const writableStream = new WritableStream(
  {
    async write(chunk) {
      await sleep(5000)
      logInDiv(writableDiv!, chunk)
    },
  },
  wQueuingStrategy,
)

const bridge = async function () {
  const reader = readableStream.getReader()
  const writer = writableStream.getWriter()
  while (true) {
    const { value, done } = await reader.read()
    if (value) {
      console.log('writer.desiredSize', writer.desiredSize)
      await writer.ready
      writer.write(value) // 中间列的输出
      logInDiv(transformDiv!, value)
    }
    if (done) {
      break
    }
  }
}

app.get("/novel", async (req, res) => {
  // get file
  const file = fs.createReadStream("les-miserable.txt", { highWaterMark: 1024 * 25 });
  // set header
  res.set("Content-Type", "message");
  const encryptStream = new Transform({
    transform(chunk, encoding, callback) {
      setTimeout(() => {
        const encrypted = [];
        chunk.forEach((byte) => {
          encrypted.push(map[byte]);
        });
        const encryptedBuffer = Buffer.from(encrypted);
        this.push(encryptedBuffer);
        callback();
      }, 10);
    },
  });
  file.pipe(encryptStream).pipe(res);
});

class Decrypter extends TransformStream {
  constructor() {
    super({
      transform(chunk, controller) {
        const decrypted = chunk.map((byte) => decryptMap[byte]);
        controller.enqueue(decrypted);
      },
    });
  }
}
class DomWriter extends WritableStream {
  constructor() {
    super({
      write(chunk) {
        resultsContainer.append(chunk);
      },
    });
  }
}

let ac = null;
const getNovelStream = async () => {
  if (ac) {
    ac.abort("refetch");
  }
  resultsContainer.innerHTML = "";
  const start = performance.now();
  // prevent wrong callback
  ac = new AbortController();
  const signal = ac.signal;
  // signal = null
  const response = await fetch(`/novel`, { signal });

  let firstResponse = performance.now();
  console.log(`Done! 🔥 Took ${firstResponse - start}ms`);

  await response.body
    .pipeThrough(new Decrypter())
    .pipeThrough(new TextDecoderStream())
    .pipeTo(new DomWriter());
  console.log(`Streaming done! 🔥 ${performance.now() - firstResponse}ms after request.`);
  ac = null;
};

const getNovel = async () => {
  if (ac) {
    ac.abort("fetch all");
  }
  resultsContainer.innerHTML = "";

  let firstResponse = performance.now();
  const response = await fetch(`/novel`);
  const arrayBuffer = await response.arrayBuffer();
  const view = new Uint8Array(arrayBuffer);
  for (let i = 0; i < view.byteLength; i++) {
    view[i] = decryptMap[view[i]];
  }
  const decoder = new TextDecoder();
  resultsContainer.innerHTML = decoder.decode(view);
  console.log(`Streaming done! 🔥 ${performance.now() - firstResponse}ms after request.`);
};

class LineSplitStream extends TransformStream {
  constructor() {
    let leftover = "";
    super({
      transform(chunk, controller) {
        console.log(chunk);
        leftover += chunk;
        const lines = leftover.split("\n");
        leftover = lines.pop();
        controller.enqueue(lines);
      },
      flush(controller) {
        if (leftover) {
          controller.enqueue([leftover]);
        }
      },
    });
  }
}