如何用Deno的ReadableStream模拟Node的readable.unshift方法？

Deno标准库的ReadableStream没有Node.js中readable.unshift()的直接等效方法，但你可以通过包装原流创建新流的方式模拟该功能——让新流先输出需要放回的数据，再转发原流的后续内容。

实现思路 #

创建一个新的ReadableStream，在启动阶段先将需要unshift的数据写入流，再通过读取原流的内容并转发到新流中，以此实现“数据回流”的效果。

代码实现 #

function unshiftStream(originalStream: ReadableStream<Uint8Array>, data: Uint8Array): ReadableStream<Uint8Array> {
  return new ReadableStream({
    async start(controller) {
      // 优先输出需要放回的数据
      controller.enqueue(data);
      
      // 读取原流并转发数据
      const reader = originalStream.getReader();
      try {
        while (true) {
          const { done, value } = await reader.read();
          if (done) break;
          controller.enqueue(value);
        }
      } catch (err) {
        controller.error(err);
      } finally {
        reader.releaseLock();
        controller.close();
      }
    }
  });
}

使用示例 #

假设你从Deno.Conn获取了可读流，读取了部分数据后需要放回：

// 建立连接
const conn = await Deno.connect({ hostname: "example.com", port: 80 });
const originalStream = conn.readable;

// 模拟读取后需要放回的数据（比如解析HTTP头时多读取了部分内容）
const dataToUnshift = new Uint8Array([0x48, 0x54, 0x54, 0x50, 0x2F]); // "HTTP/"

// 生成包含回流数据的新流
const modifiedStream = unshiftStream(originalStream, dataToUnshift);

// 从新流读取数据（先拿到unshift的内容，再是原流后续数据）
const reader = modifiedStream.getReader();
while (true) {
  const { done, value } = await reader.read();
  if (done) break;
  // 处理读取到的数据
  console.log(new TextDecoder().decode(value));
}

注意事项 #

• 每次unshift都会创建一个新的流，若需要多次回流，可链式调用unshiftStream（新的unshift数据会排在最前面）。
• 确保原流未被其他reader锁定，若已锁定需先释放锁再使用此方法。

内容的提问来源于stack exchange，提问作者onegosmer