js Stream API简单使用

ReadableStream

<script>
  // TextDecoder将字节转换为字符串,默认 utf-8 编码
  let uint8Array = new Uint8Array([72, 101, 108, 108, 111]);
  console.log(new TextDecoder().decode(uint8Array)); // Hello

  let uint8Array1 = new Uint8Array([228, 189, 160, 229, 165, 189]);
  console.log(new TextDecoder().decode(uint8Array1)); // 你好

  let uint8Array2 = new Uint8Array([0, 72, 101, 108, 108, 111, 0]);
  let binaryString = uint8Array2.subarray(1, -1);
  console.log(new TextDecoder().decode(binaryString)); // Hello

  // TextEncoder将字符串转换为字节
  console.log(new TextEncoder().encode("Hello")); // Uint8Array 72,101,108,108,111

  //ReadableStream 可读取的二进制流
  // 数据被按序读入到许多小的片段，这些片段被称作分块（chunk）。分块可以是单个字节，也可以是某种更大的数据类型，例如特定大小的类型化数组。单个流的分块可以有不同的大小和类型。
  // 已放入到流中的分块称作已入队（enqueued）——这意味着它们已经在队列中排队等待被读取。流的一个内置队列跟踪了那些尚未读取的分块。
  // 流中的分块由一个 reader 读取——该数据处理过程一次只处理一个分块，允许你对其执行任何类型的操作。
  // 每个 reader 都有一个关联的 controller，用来控制流例如可以将流关闭。
  fetch("http://localhost:3000/html-demos/table.html")
    .then((response) => response.body)
    .then((rb) => {
      const reader = rb.getReader();
      return new ReadableStream({
        start(controller) {
          // 处理每一个数据块
          function push() {
            reader.read().then(({ done, value }) => {
              // 读取完全部数据块
              if (done) {
                console.log("done", done);
                controller.close();
                return;
              }

              // 获取数据value，并通过controller发送给浏览器
              controller.enqueue(value);
              console.log(done, value);
              push();
            });
          }

          push();
        },
      });
    })
    .then((stream) =>
      // 处理流数据
      new Response(stream, { headers: { "Content-Type": "text/html" } }).text()
    )
    .then((result) => {
      console.log(result);
    });
</script>

WritableStream

<script> 

  const list = document.querySelector("ul");
  function sendMessage(message, writableStream) {
    const defaultWriter = writableStream.getWriter();
    const encoder = new TextEncoder();
    const encoded = encoder.encode(message, { stream: true });
    encoded.forEach((chunk) => {
      defaultWriter.ready
        .then(() => {
          return defaultWriter.write(chunk);
        })
        .then(() => {
          console.log(chunk, "Chunk written to sink.");
        })
        .catch((err) => {
          console.log("Chunk error:", err);
        });
    });
    // 重新调用ready确保writer关闭前所有数据已写入
    defaultWriter.ready
      .then(() => {
        defaultWriter.close();
      })
      .then(() => {
        console.log("All chunks written");
      })
      .catch((err) => {
        console.log("Stream error:", err);
      });
  }
  const decoder = new TextDecoder("utf-8");
  // CountQueuingStrategy 接口提供了一个内置的、用于对分块进行计数的队列策略，可以在构造流的时候使用。
  const queuingStrategy = new CountQueuingStrategy({ highWaterMark: 1 });
  let result = "";
  // 一个可写流（Writable stream）是一个可以写入数据的数据终点
  const writableStream = new WritableStream(
    {
      write(chunk) {
        return new Promise((resolve, reject) => {
          var buffer = new ArrayBuffer(1);
          var view = new Uint8Array(buffer);
          view[0] = chunk;
          var decoded = decoder.decode(view, { stream: true });
          var listItem = document.createElement("li");
          listItem.textContent = "Chunk decoded: " + decoded;
          list.appendChild(listItem);
          result += decoded;
          resolve();
        });
      },
      close() {
        var listItem = document.createElement("li");
        listItem.textContent = "[MESSAGE RECEIVED] " + result;
        list.appendChild(listItem);
      },
      abort(err) {
        console.log("Sink error:", err);
      },
    },
    queuingStrategy
  );
  sendMessage("Hello, world.", writableStream);
</script>