สตรีม - คู่มือฉบับสมบูรณ์

ดูวิธีใช้สตรีมที่อ่านได้ เขียนได้ และแปลงได้ด้วย Streams API

Thomas Steiner

Streams API ช่วยให้คุณเข้าถึงสตรีมข้อมูลที่ได้รับผ่านเครือข่ายหรือสร้างขึ้นด้วยวิธีใดก็ตามในเครื่อง และประมวลผลด้วย JavaScript ได้แบบเป็นโปรแกรม การสตรีมเกี่ยวข้องกับการแบ่งทรัพยากรที่คุณต้องการรับ ส่ง หรือแปลง เป็นก้อนเล็กๆ แล้วประมวลผลก้อนเหล่านี้ทีละก้อน แม้ว่าการสตรีมจะเป็นสิ่งที่เบราว์เซอร์ทำอยู่แล้วเมื่อได้รับชิ้นงาน เช่น HTML หรือวิดีโอที่จะแสดงในหน้าเว็บ แต่ความสามารถนี้ไม่เคยมีให้ใช้งานใน JavaScript มาก่อนจนกระทั่งมีการเปิดตัว fetch พร้อมสตรีมในปี 2015

หมายเหตุ ในทางเทคนิคแล้ว คุณสามารถสตรีมด้วย XMLHttpRequest ได้ แต่ก็ถือเป็นวิธีที่ไม่ค่อยดีนัก นี่คือ GitHub Gist ที่แสดงตัวอย่างที่ดีของ "XMLHttpRequest"

ก่อนหน้านี้ หากต้องการประมวลผลทรัพยากรบางอย่าง (ไม่ว่าจะเป็นวิดีโอ ไฟล์ข้อความ ฯลฯ) คุณจะต้องดาวน์โหลดไฟล์ทั้งหมด รอให้ระบบยกเลิกการซีเรียลไลซ์เป็นรูปแบบที่เหมาะสม แล้วจึงประมวลผล แต่เมื่อสตรีมพร้อมใช้งานใน JavaScript ทุกอย่างก็เปลี่ยนไป ตอนนี้คุณสามารถประมวลผลข้อมูลดิบด้วย JavaScript ได้ทีละรายการทันทีที่ข้อมูลพร้อมใช้งานในไคลเอ็นต์ โดยไม่ต้องสร้างบัฟเฟอร์ สตริง หรือ Blob ซึ่งจะช่วยให้คุณใช้ฟีเจอร์นี้ได้ในหลายกรณี ดังที่ระบุไว้ด้านล่าง

• เอฟเฟกต์วิดีโอ: การส่งสตรีมวิดีโอที่อ่านได้ผ่านสตรีมการแปลงที่ใช้เอฟเฟกต์ แบบเรียลไทม์
• (ยกเลิก)การบีบอัดข้อมูล: การส่งสตรีมไฟล์ผ่านสตรีมการแปลงที่จะ(ยกเลิก)บีบอัดอย่างเลือกสรร
• การถอดรหัสรูปภาพ: การส่งสตรีมการตอบกลับ HTTP ผ่านสตรีมการแปลงที่ถอดรหัสไบต์ เป็นข้อมูลบิตแมป แล้วส่งผ่านสตรีมการแปลงอีกรายการที่แปลงบิตแมปเป็น PNG หากติดตั้งภายในตัวแฮนเดิล fetch ของ Service Worker คุณจะสามารถ Polyfill รูปภาพรูปแบบใหม่ เช่น AVIF ได้อย่างโปร่งใส

การสนับสนุนเบราว์เซอร์

ReadableStream และ WritableStream

TransformStream

แนวคิดหลัก

ก่อนที่จะลงรายละเอียดเกี่ยวกับสตรีมประเภทต่างๆ เรามาดูแนวคิดหลักกันก่อน

จับเป็นก้อน

Chunk คือข้อมูลชิ้นเดียวที่เขียนลงในหรืออ่านจากสตรีม โดยอาจเป็นประเภทใดก็ได้ สตรีมอาจมีแม้กระทั่งก้อนข้อมูลประเภทต่างๆ โดยส่วนใหญ่แล้ว Chunk จะไม่ใช่หน่วยข้อมูลที่เล็กที่สุด สำหรับสตรีมที่กำหนด เช่น สตรีมไบต์อาจมีกลุ่มที่ประกอบด้วยหน่วย 16 KiB Uint8Array แทนที่จะเป็นไบต์เดียว

สตรีมที่อ่านได้

สตรีมที่อ่านได้แสดงถึงแหล่งข้อมูลที่คุณอ่านได้ กล่าวคือ ข้อมูลออกมาจากสตรีมที่อ่านได้ กล่าวอย่างเจาะจงคือ Readable Stream เป็นอินสแตนซ์ของคลาส ReadableStream

สตรีมที่เขียนได้

สตรีมที่เขียนได้แสดงถึงปลายทางของข้อมูลที่คุณเขียนได้ กล่าวคือ ข้อมูลจะเข้าไปในสตรีมที่เขียนได้ กล่าวอย่างเจาะจงคือ สตรีมที่เขียนได้คืออินสแตนซ์ของคลาส WritableStream

เปลี่ยนรูปแบบสตรีม

สตรีมการแปลงประกอบด้วยสตรีม 2 รายการ ได้แก่ สตรีมที่เขียนได้ ซึ่งเรียกว่าด้านที่เขียนได้ และสตรีมที่อ่านได้ ซึ่งเรียกว่าด้านที่อ่านได้ คำอุปมาในชีวิตจริงสำหรับฟีเจอร์นี้คือล่ามแปลภาษาแบบเรียลไทม์ ที่แปลจากภาษาหนึ่งเป็นอีกภาษาหนึ่งในทันที ในลักษณะเฉพาะของสตรีมการแปลง การเขียนไปยังด้านที่เขียนได้จะทําให้ข้อมูลใหม่พร้อมให้อ่านจากด้านที่อ่านได้ กล่าวอย่างเจาะจงคือ ออบเจ็กต์ใดก็ตามที่มีพร็อพเพอร์ตี้ writable และพร็อพเพอร์ตี้ readable จะใช้เป็นสตรีมการเปลี่ยนรูปแบบได้ อย่างไรก็ตาม TransformStream คลาสมาตรฐานจะช่วยให้สร้าง คู่ดังกล่าวที่พันกันอย่างเหมาะสมได้ง่ายขึ้น

โซ่ท่อ

โดยจะใช้สตรีมเป็นหลักด้วยการส่งผ่านสตรีมไปยังกันและกัน คุณสามารถส่งสตรีมที่อ่านได้ไปยังสตรีมที่เขียนได้โดยตรง โดยใช้เมธอด pipeTo() ของสตรีมที่อ่านได้ หรือจะส่งผ่านสตรีมการแปลงอย่างน้อย 1 รายการก่อนก็ได้โดยใช้เมธอด pipeThrough() ของสตรีมที่อ่านได้ ชุดของสตรีมที่เชื่อมต่อกันในลักษณะนี้เรียกว่าเชนไปป์

แรงดันย้อนกลับ

เมื่อสร้างไปป์เชนแล้ว ระบบจะเผยแพร่สัญญาณเกี่ยวกับความเร็วที่ควรส่งผ่านก้อนข้อมูล ผ่านไปป์เชน หากขั้นตอนใดในเชนยังรับก้อนข้อมูลไม่ได้ ระบบจะส่งสัญญาณย้อนกลับ ผ่านเชนไปป์จนกว่าแหล่งที่มาต้นฉบับจะได้รับแจ้งให้หยุดสร้างก้อนข้อมูลเร็ว เกินไป กระบวนการการทำให้โฟลว์เป็นปกตินี้เรียกว่าแรงดันย้อนกลับ

การที

คุณสามารถแยกสตรีมที่อ่านได้ (ตั้งชื่อตามรูปร่างของตัว "T" พิมพ์ใหญ่) โดยใช้เมธอด tee() การดำเนินการนี้จะล็อกสตรีม ซึ่งหมายความว่าสตรีมจะใช้โดยตรงไม่ได้อีกต่อไป แต่จะสร้างสตรีมใหม่ 2 รายการ ที่เรียกว่ากิ่ง ซึ่งสามารถใช้แยกกันได้ นอกจากนี้ การเตรียมตัวยังมีความสำคัญเนื่องจากคุณไม่สามารถกรอหรือเริ่มสตรีมใหม่ได้ เราจะพูดถึงเรื่องนี้ในภายหลัง

กลไกของสตรีมที่อ่านได้

สตรีมที่อ่านได้คือแหล่งข้อมูลที่แสดงใน JavaScript โดยออบเจ็กต์ ReadableStream ซึ่ง ไหลจากแหล่งที่มาพื้นฐาน ตัวสร้าง ReadableStream() จะสร้างและแสดงผลออบเจ็กต์สตรีมที่อ่านได้จากแฮนเดิลที่ระบุ แหล่งที่มาพื้นฐานมี 2 ประเภท ดังนี้

• แหล่งข้อมูลแบบพุชจะพุชข้อมูลมาให้คุณอย่างต่อเนื่องเมื่อคุณเข้าถึงแหล่งข้อมูลเหล่านั้น และคุณเป็นผู้มีสิทธิ์ เริ่ม หยุดชั่วคราว หรือยกเลิกการเข้าถึงสตรีม ตัวอย่างเช่น สตรีมวิดีโอสด เหตุการณ์ที่เซิร์ฟเวอร์ส่ง หรือ WebSocket
• แหล่งที่มาแบบดึงกำหนดให้คุณต้องขอข้อมูลจากแหล่งที่มาอย่างชัดเจนเมื่อเชื่อมต่อแล้ว ตัวอย่าง เช่น การดำเนินการ HTTP ผ่านการเรียกใช้ fetch() หรือ XMLHttpRequest

ระบบจะอ่านข้อมูลสตรีมตามลำดับเป็นชิ้นเล็กๆ ที่เรียกว่าก้อน โดยก้อนข้อมูลที่วางในสตรีมจะเรียกว่าenqueued ซึ่งหมายความว่าข้อความเหล่านั้นกำลังรออยู่ในคิว พร้อมให้คุณอ่าน คิวภายในจะติดตามก้อนข้อมูลที่ยังไม่ได้อ่าน

กลยุทธ์การจัดคิวคือออบเจ็กต์ที่กำหนดวิธีที่สตรีมควรส่งสัญญาณแรงดันย้อนกลับตาม สถานะของคิวภายใน กลยุทธ์การจัดคิวจะกำหนดขนาดให้กับแต่ละก้อน และเปรียบเทียบ ขนาดรวมของก้อนทั้งหมดในคิวกับตัวเลขที่ระบุ ซึ่งเรียกว่าเครื่องหมายระดับสูง

โปรแกรมอ่านจะอ่านก้อนข้อมูลภายในสตรีม โปรแกรมอ่านนี้จะดึงข้อมูลทีละก้อน ทำให้คุณดำเนินการใดๆ กับข้อมูลได้ตามต้องการ โปรแกรมอ่านและโค้ดประมวลผลอื่นๆ ที่มาพร้อมกันเรียกว่าผู้ใช้

โครงสร้างถัดไปในบริบทนี้เรียกว่าคอนโทรลเลอร์ สตรีมที่อ่านได้แต่ละรายการจะมีตัวควบคุมที่เชื่อมโยงอยู่ ซึ่งช่วยให้คุณควบคุมสตรีมได้ตามชื่อ

มีผู้อ่านได้เพียงคนเดียวที่อ่านสตรีมในแต่ละครั้ง เมื่อสร้างผู้อ่านและเริ่มอ่านสตรีม (นั่นคือกลายเป็นผู้อ่านที่ใช้งานอยู่) ระบบจะล็อกผู้อ่านนั้นไว้ หากต้องการให้ผู้อ่านคนอื่นอ่านสตรีมของคุณ คุณมักจะต้องปล่อยผู้อ่านคนแรกก่อนที่จะทำอย่างอื่น (แม้ว่าคุณจะแยกสตรีมได้ก็ตาม)

การสร้างสตรีมที่อ่านได้

คุณสร้างสตรีมที่อ่านได้โดยการเรียกเครื่องมือสร้างของสตรีมนั้น ReadableStream() ตัวสร้างมีอาร์กิวเมนต์ที่ไม่บังคับ underlyingSource ซึ่งแสดงถึงออบเจ็กต์ ที่มีเมธอดและพร็อพเพอร์ตี้ที่กำหนดลักษณะการทำงานของอินสแตนซ์สตรีมที่สร้างขึ้น

underlyingSource

ซึ่งสามารถใช้วิธีการต่อไปนี้ที่นักพัฒนาแอปกำหนดได้ (ไม่บังคับ)

• start(controller): เรียกใช้ทันทีเมื่อสร้างออบเจ็กต์ เมธอดสามารถเข้าถึงแหล่งที่มาของสตรีมและทำสิ่งอื่นๆ ที่จำเป็นในการตั้งค่าฟังก์ชันการทำงานของสตรีมได้ หากต้องการดำเนินการนี้แบบไม่พร้อมกัน เมธอดจะ ส่งคืน Promise เพื่อส่งสัญญาณความสำเร็จหรือความล้มเหลวได้ controller พารามิเตอร์ที่ส่งไปยังเมธอดนี้คือ ReadableStreamDefaultController
• pull(controller): ใช้เพื่อควบคุมสตรีมเมื่อมีการดึงข้อมูลก้อนข้อมูลเพิ่มเติม ระบบจะเรียกใช้ฟังก์ชันนี้ซ้ำๆ ตราบใดที่คิวภายในของสตรีมยังไม่เต็ม จนกว่าคิวจะถึงเครื่องหมายระดับสูงสุด หากผลลัพธ์ของการเรียกใช้ pull() เป็น Promise ระบบจะไม่เรียกใช้ pull() อีกจนกว่า Promise ดังกล่าวจะเสร็จสมบูรณ์ หาก Promise ปฏิเสธ สตรีมจะเกิดข้อผิดพลาด
• cancel(reason): เรียกใช้เมื่อผู้ใช้สตรีมยกเลิกสตรีม

const readableStream = new ReadableStream({
  start(controller) {
    /* … */
  },

  pull(controller) {
    /* … */
  },

  cancel(reason) {
    /* … */
  },
});

ReadableStreamDefaultController รองรับวิธีการต่อไปนี้

• ReadableStreamDefaultController.close() จะปิดสตรีมที่เชื่อมโยง
• ReadableStreamDefaultController.enqueue() จัดคิวก้อนข้อมูลที่ระบุในสตรีมที่เชื่อมโยง
• ReadableStreamDefaultController.error() จะทำให้การโต้ตอบในอนาคตกับสตรีมที่เชื่อมโยงเกิดข้อผิดพลาด

/* … */
start(controller) {
  controller.enqueue('The first chunk!');
},
/* … */

queuingStrategy

อาร์กิวเมนต์ที่ 2 ของตัวสร้าง ReadableStream() ซึ่งจะใส่หรือไม่ใส่ก็ได้คือ queuingStrategy เป็นออบเจ็กต์ที่กําหนดกลยุทธ์การจัดคิวสําหรับสตรีมได้ (ไม่บังคับ) ซึ่งมีพารามิเตอร์ 2 รายการดังนี้

• highWaterMark: ตัวเลขที่ไม่เป็นลบซึ่งระบุเครื่องหมายระดับสูงสุดของสตรีมโดยใช้กลยุทธ์การจัดคิวนี้
• size(chunk): ฟังก์ชันที่คำนวณและแสดงผลขนาดที่ไม่เป็นลบแบบจำกัดของค่าก้อนที่ระบุ ระบบจะใช้ผลลัพธ์เพื่อกำหนดแรงดันย้อนกลับ ซึ่งจะแสดงผ่านพร็อพเพอร์ตี้ ReadableStreamDefaultController.desiredSize ที่เหมาะสม นอกจากนี้ยังควบคุมเวลาที่เรียกใช้เมธอด pull() ของแหล่งที่มาพื้นฐานด้วย

const readableStream = new ReadableStream({
    /* … */
  },
  {
    highWaterMark: 10,
    size(chunk) {
      return chunk.length;
    },
  },
);

วิธีการ getReader() และ read()

หากต้องการอ่านจากสตรีมที่อ่านได้ คุณต้องมีโปรแกรมอ่านซึ่งจะเป็น ReadableStreamDefaultReader เมธอด getReader() ของอินเทอร์เฟซ ReadableStream จะสร้างโปรแกรมอ่านและล็อกสตรีมไว้กับโปรแกรมอ่านนั้น ในขณะที่สตรีมล็อกอยู่ คุณจะซื้อสตรีมอื่นไม่ได้จนกว่าจะปลดล็อกสตรีมนี้

เมธอด read() ของอินเทอร์เฟซ ReadableStreamDefaultReader จะแสดงผล Promise ที่ให้สิทธิ์เข้าถึง ก้อนข้อมูลถัดไปในคิวภายในของสตรีม โดยจะดำเนินการหรือปฏิเสธพร้อมผลลัพธ์ตามสถานะของ สตรีม ความเป็นไปได้ต่างๆ มีดังนี้

• หากมี Chunk อยู่ ระบบจะทำตามสัญญาด้วยออบเจ็กต์ในรูปแบบ
  { value: chunk, done: false }
• หากสตรีมปิดลง สัญญาจะได้รับการตอบสนองด้วยออบเจ็กต์ในรูปแบบ
  { value: undefined, done: true }
• หากสตรีมเกิดข้อผิดพลาด ระบบจะปฏิเสธ Promise พร้อมข้อผิดพลาดที่เกี่ยวข้อง

const reader = readableStream.getReader();
while (true) {
  const { done, value } = await reader.read();
  if (done) {
    console.log('The stream is done.');
    break;
  }
  console.log('Just read a chunk:', value);
}

พร็อพเพอร์ตี้ locked

คุณสามารถตรวจสอบว่าสตรีมที่อ่านได้ล็อกอยู่หรือไม่โดยเข้าถึงพร็อพเพอร์ตี้ ReadableStream.locked ของสตรีม

const locked = readableStream.locked;
console.log(`The stream is ${locked ? 'indeed' : 'not'} locked.`);

ตัวอย่างโค้ดสตรีมที่อ่านได้

ตัวอย่างโค้ดต่อไปนี้แสดงขั้นตอนทั้งหมดในการดำเนินการ ก่อนอื่น ให้สร้าง ReadableStream ซึ่งในอาร์กิวเมนต์ underlyingSource (นั่นคือคลาส TimestampSource) จะกำหนดเมธอด start() วิธีนี้จะบอกให้สตรีม controller enqueue() ประทับเวลาทุกวินาทีเป็นเวลา 10 วินาที สุดท้ายคือบอกให้ตัวควบคุมclose()สตรีม คุณใช้สตรีมนี้ได้โดยการสร้างเครื่องอ่านผ่านเมธอด getReader() และเรียกใช้ read() จนกว่าสตรีมจะdone

class TimestampSource {
  #interval

  start(controller) {
    this.#interval = setInterval(() => {
      const string = new Date().toLocaleTimeString();
      // Add the string to the stream.
      controller.enqueue(string);
      console.log(`Enqueued ${string}`);
    }, 1_000);

    setTimeout(() => {
      clearInterval(this.#interval);
      // Close the stream after 10s.
      controller.close();
    }, 10_000);
  }

  cancel() {
    // This is called if the reader cancels.
    clearInterval(this.#interval);
  }
}

const stream = new ReadableStream(new TimestampSource());

async function concatStringStream(stream) {
  let result = '';
  const reader = stream.getReader();
  while (true) {
    // The `read()` method returns a promise that
    // resolves when a value has been received.
    const { done, value } = await reader.read();
    // Result objects contain two properties:
    // `done`  - `true` if the stream has already given you all its data.
    // `value` - Some data. Always `undefined` when `done` is `true`.
    if (done) return result;
    result += value;
    console.log(`Read ${result.length} characters so far`);
    console.log(`Most recently read chunk: ${value}`);
  }
}
concatStringStream(stream).then((result) => console.log('Stream complete', result));

การวนซ้ำแบบอะซิงโครนัส

การตรวจสอบว่าสตรีมเป็น done ในแต่ละread()การวนซ้ำของลูปอาจไม่ใช่ API ที่สะดวกที่สุด โชคดีที่ในเร็วๆ นี้จะมีวิธีที่ดีกว่าในการทำสิ่งนี้ นั่นคือการวนซ้ำแบบไม่พร้อมกัน

for await (const chunk of stream) {
  console.log(chunk);
}

วิธีแก้ปัญหาชั่วคราวในการใช้การวนซ้ำแบบอะซิงโครนัสในปัจจุบันคือการใช้ลักษณะการทำงานกับ Polyfill

if (!ReadableStream.prototype[Symbol.asyncIterator]) {
  ReadableStream.prototype[Symbol.asyncIterator] = async function* () {
    const reader = this.getReader();
    try {
      while (true) {
        const {done, value} = await reader.read();
        if (done) {
          return;
          }
        yield value;
      }
    }
    finally {
      reader.releaseLock();
    }
  }
}

การแยกสตรีมที่อ่านได้

เมธอด tee() ของอินเทอร์เฟซ ReadableStream จะตั้งค่าสตรีมที่อ่านได้ปัจจุบัน โดยจะแสดงผลอาร์เรย์ 2 องค์ประกอบ ซึ่งมีกิ่งก้าน 2 กิ่งที่ได้เป็นอินสแตนซ์ ReadableStream ใหม่ ซึ่งจะช่วยให้ ผู้อ่าน 2 คนอ่านสตรีมพร้อมกันได้ คุณอาจทำเช่นนี้ใน Service Worker หากต้องการดึงข้อมูลการตอบกลับจากเซิร์ฟเวอร์และสตรีมไปยังเบราว์เซอร์ แต่ก็สตรีมไปยังแคชของ Service Worker ด้วย เนื่องจากใช้เนื้อหาการตอบกลับได้เพียงครั้งเดียว คุณจึงต้องมีสำเนา 2 ชุดเพื่อดำเนินการนี้ หากต้องการยกเลิกสตรีม คุณจะต้องยกเลิกทั้ง 2 สาขาที่ได้ โดยทั่วไปแล้ว การปักหมุดสตรีม จะล็อกสตรีมนั้นไว้ตลอดระยะเวลาที่ปักหมุด เพื่อป้องกันไม่ให้ผู้อ่านคนอื่นๆ ล็อกสตรีม

const readableStream = new ReadableStream({
  start(controller) {
    // Called by constructor.
    console.log('[start]');
    controller.enqueue('a');
    controller.enqueue('b');
    controller.enqueue('c');
  },
  pull(controller) {
    // Called `read()` when the controller's queue is empty.
    console.log('[pull]');
    controller.enqueue('d');
    controller.close();
  },
  cancel(reason) {
    // Called when the stream is canceled.
    console.log('[cancel]', reason);
  },
});

// Create two `ReadableStream`s.
const [streamA, streamB] = readableStream.tee();

// Read streamA iteratively one by one. Typically, you
// would not do it this way, but you certainly can.
const readerA = streamA.getReader();
console.log('[A]', await readerA.read()); //=> {value: "a", done: false}
console.log('[A]', await readerA.read()); //=> {value: "b", done: false}
console.log('[A]', await readerA.read()); //=> {value: "c", done: false}
console.log('[A]', await readerA.read()); //=> {value: "d", done: false}
console.log('[A]', await readerA.read()); //=> {value: undefined, done: true}

// Read streamB in a loop. This is the more common way
// to read data from the stream.
const readerB = streamB.getReader();
while (true) {
  const result = await readerB.read();
  if (result.done) break;
  console.log('[B]', result);
}

สตรีมไบต์ที่อ่านได้

สำหรับสตรีมที่แสดงไบต์ ระบบจะจัดเตรียมสตรีมที่อ่านได้ในเวอร์ชันขยายเพื่อจัดการไบต์อย่างมีประสิทธิภาพ โดยเฉพาะอย่างยิ่งด้วยการลดการคัดลอก สตรีมไบต์ช่วยให้ได้ผู้ใช้ที่นำบัฟเฟอร์ของตนเองมาใช้ (BYOB) การติดตั้งใช้งานเริ่มต้นอาจให้เอาต์พุตที่แตกต่างกัน เช่น สตริงหรือบัฟเฟอร์อาร์เรย์ในกรณีของ WebSocket ในขณะที่สตรีมไบต์จะรับประกันเอาต์พุตไบต์ นอกจากนี้ ผู้อ่าน BYOB ยังได้รับประโยชน์ด้านความเสถียรด้วย เนื่องจากหากบัฟเฟอร์แยกออก ก็จะรับประกันได้ว่าไม่มีการเขียนลงในบัฟเฟอร์เดียวกัน 2 ครั้ง จึงหลีกเลี่ยงสภาวะแข่งได้ โปรแกรมอ่าน BYOB จะช่วยลดจำนวนครั้งที่เบราว์เซอร์ต้องเรียกใช้ระบบจัดการหน่วยความจำที่ไม่ใช้แล้วได้ เนื่องจากสามารถนำบัฟเฟอร์กลับมาใช้ใหม่ได้

การสร้างสตรีมไบต์ที่อ่านได้

คุณสร้างสตรีมไบต์ที่อ่านได้โดยส่งพารามิเตอร์ type เพิ่มเติมไปยังตัวสร้าง ReadableStream()

new ReadableStream({ type: 'bytes' });

underlyingSource

แหล่งที่มาพื้นฐานของสตรีมไบต์ที่อ่านได้จะได้รับ ReadableByteStreamController เพื่อ จัดการ เมธอด ReadableByteStreamController.enqueue() ของคลาสนี้รับอาร์กิวเมนต์ chunk ซึ่งมีค่าเป็น ArrayBufferView พร็อพเพอร์ตี้ ReadableByteStreamController.byobRequest จะแสดงผลคำขอ Pull BYOB ปัจจุบัน หรือค่า Null หากไม่มี สุดท้าย ReadableByteStreamController.desiredSize พร็อพเพอร์ตี้จะส่งคืนขนาดที่ต้องการเพื่อเติมคิวภายในของสตรีมที่ควบคุม

queuingStrategy

อาร์กิวเมนต์ที่ 2 ของตัวสร้าง ReadableStream() ซึ่งจะใส่หรือไม่ใส่ก็ได้คือ queuingStrategy เป็นออบเจ็กต์ที่กำหนดกลยุทธ์การจัดคิวสำหรับสตรีมได้โดยไม่บังคับ ซึ่งมีพารามิเตอร์ 1 รายการ ดังนี้

• highWaterMark: จำนวนไบต์ที่ไม่เป็นลบซึ่งระบุเครื่องหมายระดับสูงของสตรีมโดยใช้กลยุทธ์การจัดคิวนี้ ระบบจะใช้ข้อมูลนี้เพื่อกำหนดแรงดันย้อนกลับ ซึ่งแสดงผ่านพร็อพเพอร์ตี้ ReadableByteStreamController.desiredSize ที่เหมาะสม นอกจากนี้ยังควบคุมเวลาที่เรียกใช้เมธอด pull() ของแหล่งที่มาพื้นฐานด้วย

วิธีการ getReader() และ read()

จากนั้นคุณจะเข้าถึง ReadableStreamBYOBReader ได้โดยตั้งค่าพารามิเตอร์ mode ตามนั้น ReadableStream.getReader({ mode: "byob" }) ซึ่งช่วยให้ควบคุมการจัดสรรบัฟเฟอร์ได้แม่นยำยิ่งขึ้นเพื่อหลีกเลี่ยงการคัดลอก หากต้องการอ่านจากสตรีมไบต์ คุณต้องเรียกใช้ ReadableStreamBYOBReader.read(view) โดยที่ view คือ ArrayBufferView

ตัวอย่างโค้ดสตรีมไบต์ที่อ่านได้

const reader = readableStream.getReader({ mode: "byob" });

let startingAB = new ArrayBuffer(1_024);
const buffer = await readInto(startingAB);
console.log("The first 1024 bytes, or less:", buffer);

async function readInto(buffer) {
  let offset = 0;

  while (offset < buffer.byteLength) {
    const { value: view, done } =
        await reader.read(new Uint8Array(buffer, offset, buffer.byteLength - offset));
    buffer = view.buffer;
    if (done) {
      break;
    }
    offset += view.byteLength;
  }

  return buffer;
}

ฟังก์ชันต่อไปนี้จะแสดงผลสตรีมไบต์ที่อ่านได้ ซึ่งช่วยให้อ่านอาร์เรย์ที่สร้างขึ้นแบบสุ่มได้อย่างมีประสิทธิภาพโดยไม่ต้องคัดลอก แทนที่จะใช้ขนาดก้อนข้อมูลที่กำหนดไว้ล่วงหน้าเป็น 1,024 ระบบจะพยายามเติมบัฟเฟอร์ที่นักพัฒนาแอปจัดหาให้ ซึ่งช่วยให้ควบคุมได้อย่างเต็มที่

const DEFAULT_CHUNK_SIZE = 1_024;

function makeReadableByteStream() {
  return new ReadableStream({
    type: 'bytes',

    pull(controller) {
      // Even when the consumer is using the default reader,
      // the auto-allocation feature allocates a buffer and
      // passes it to us via `byobRequest`.
      const view = controller.byobRequest.view;
      view = crypto.getRandomValues(view);
      controller.byobRequest.respond(view.byteLength);
    },

    autoAllocateChunkSize: DEFAULT_CHUNK_SIZE,
  });
}

กลไกของสตรีมที่เขียนได้

สตรีมที่เขียนได้คือปลายทางที่คุณเขียนข้อมูลได้ ซึ่งแสดงใน JavaScript ด้วยออบเจ็กต์ WritableStream ซึ่งทำหน้าที่เป็น Abstraction เหนือSink พื้นฐาน ซึ่งเป็น Sink I/O ระดับล่างที่ใช้เขียนข้อมูลดิบ

ระบบจะเขียนข้อมูลลงในสตรีมผ่านโปรแกรมเขียนทีละก้อน โดยก้อนข้อมูลจะมี หลายรูปแบบเช่นเดียวกับก้อนข้อมูลในโปรแกรมอ่าน คุณใช้โค้ดใดก็ได้เพื่อสร้าง ก้อนข้อมูลที่พร้อมสำหรับการเขียน โดยนักเขียนและโค้ดที่เกี่ยวข้องจะเรียกว่าโปรดิวเซอร์

เมื่อสร้าง Writer และเริ่มเขียนไปยังสตรีม (Writer ที่ใช้งานอยู่) จะถือว่า Writer ล็อกกับสตรีมนั้น โดยมีผู้เขียนได้เพียงคนเดียวเท่านั้นที่เขียนไปยังสตรีมที่เขียนได้ในครั้งเดียว หากต้องการให้ผู้เขียนคนอื่นเริ่มเขียนไปยังสตรีมของคุณ โดยปกติแล้วคุณจะต้องปล่อยสตรีมก่อน แล้วจึงแนบผู้เขียนคนอื่นเข้ากับสตรีม

คิวภายในจะติดตามก้อนข้อมูลที่เขียนลงในสตรีมแล้ว แต่ยังไม่ได้ ประมวลผลโดย Sink พื้นฐาน

กลยุทธ์การจัดคิวคือออบเจ็กต์ที่กำหนดวิธีที่สตรีมควรส่งสัญญาณแรงดันย้อนกลับตาม สถานะของคิวภายใน กลยุทธ์การจัดคิวจะกำหนดขนาดให้กับแต่ละก้อน และเปรียบเทียบ ขนาดรวมของก้อนทั้งหมดในคิวกับตัวเลขที่ระบุ ซึ่งเรียกว่าเครื่องหมายระดับสูง

โครงสร้างสุดท้ายเรียกว่าคอนโทรลเลอร์ สตรีมที่เขียนได้แต่ละรายการมีตัวควบคุมที่เชื่อมโยงอยู่ ซึ่ง ช่วยให้คุณควบคุมสตรีมได้ (เช่น ยกเลิก)

การสร้างสตรีมที่เขียนได้

อินเทอร์เฟซ WritableStream ของ Streams API มีการแยกข้อมูลมาตรฐานสำหรับการเขียนข้อมูลการสตรีมไปยังปลายทางที่เรียกว่า Sink ออบเจ็กต์นี้มาพร้อมกับแรงดันย้อนกลับและการจัดคิวในตัว คุณสร้างสตรีมที่เขียนได้โดย เรียกใช้ตัวสร้าง WritableStream() โดยมีพารามิเตอร์ underlyingSink ซึ่งเป็นออบเจ็กต์ ที่มีเมธอดและพร็อพเพอร์ตี้ที่กำหนดลักษณะการทำงานของอินสแตนซ์สตรีมที่สร้างขึ้น

underlyingSink

underlyingSink สามารถมีเมธอดที่ไม่บังคับซึ่งนักพัฒนาแอปกำหนดได้ดังต่อไปนี้ controller พารามิเตอร์ที่ส่งไปยังเมธอดบางรายการคือWritableStreamDefaultController

• start(controller): ระบบจะเรียกใช้เมธอดนี้ทันทีเมื่อสร้างออบเจ็กต์ เนื้อหาของเมธอดนี้ควรมีเป้าหมายเพื่อรับสิทธิ์เข้าถึง Sink พื้นฐาน หากต้องการดำเนินการนี้แบบไม่พร้อมกัน ฟังก์ชันจะคืนค่า Promise เพื่อส่งสัญญาณความสำเร็จหรือความล้มเหลว
• write(chunk, controller): ระบบจะเรียกใช้เมธอดนี้เมื่อพร้อมที่จะเขียนข้อมูลก้อนใหม่ (ระบุไว้ในพารามิเตอร์ chunk) ลงใน Sink พื้นฐาน โดยจะคืนค่า Promise เพื่อ ส่งสัญญาณว่าการดำเนินการเขียนสำเร็จหรือไม่ ระบบจะเรียกใช้เมธอดนี้หลังจากที่การเขียนก่อนหน้า สำเร็จแล้วเท่านั้น และจะไม่เรียกใช้หลังจากที่สตรีมปิดหรือถูกยกเลิก
• close(controller): ระบบจะเรียกใช้เมธอดนี้หากแอปส่งสัญญาณว่าเขียน ก้อนข้อมูลลงในสตรีมเสร็จแล้ว เนื้อหาควรทำทุกอย่างที่จำเป็นเพื่อเขียนข้อมูลไปยัง ซิงก์พื้นฐานให้เสร็จสมบูรณ์ และปล่อยสิทธิ์เข้าถึง หากกระบวนการนี้เป็นแบบไม่พร้อมกัน กระบวนการจะคืนค่า Promise เพื่อส่งสัญญาณว่าสำเร็จหรือไม่สำเร็จ ระบบจะเรียกใช้วิธีนี้หลังจากที่การเขียนที่อยู่ในคิวทั้งหมด สำเร็จแล้วเท่านั้น
• abort(reason): ระบบจะเรียกใช้เมธอดนี้หากแอปส่งสัญญาณว่าต้องการปิดสตรีมอย่างกะทันหัน และเปลี่ยนเป็นสถานะข้อผิดพลาด ซึ่งจะล้างทรัพยากรที่ระงับไว้ได้เหมือนกับ close() แต่ระบบจะเรียกใช้ abort() แม้ว่าจะมีการจัดคิวการเขียนก็ตาม ระบบจะทิ้ง ส่วนเหล่านั้น หากกระบวนการนี้เป็นแบบไม่พร้อมกัน ก็จะแสดงผล Promise เพื่อส่งสัญญาณความสำเร็จหรือความล้มเหลว reason พารามิเตอร์มี DOMString ที่อธิบายสาเหตุที่สตรีมถูกยกเลิก

const writableStream = new WritableStream({
  start(controller) {
    /* … */
  },

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

  close(controller) {
    /* … */
  },

  abort(reason) {
    /* … */
  },
});

อินเทอร์เฟซของ Streams API WritableStreamDefaultController แสดงถึงตัวควบคุมที่อนุญาตให้ควบคุมสถานะของ WritableStream ระหว่างการตั้งค่า เมื่อส่งก้อนข้อมูลเพิ่มเติมเพื่อเขียน หรือเมื่อเขียนเสร็จแล้ว เมื่อสร้าง WritableStream ระบบจะให้อินสแตนซ์ WritableStreamDefaultController ที่เกี่ยวข้องแก่ Sink พื้นฐานเพื่อใช้จัดการ WritableStreamDefaultController มีเพียงวิธีเดียวคือ WritableStreamDefaultController.error() ซึ่งจะทำให้การโต้ตอบในอนาคตกับสตรีมที่เชื่อมโยงเกิดข้อผิดพลาด WritableStreamDefaultController ยังรองรับพร็อพเพอร์ตี้ signal ซึ่งจะแสดงผลอินสแตนซ์ของ AbortSignal ทำให้หยุดการดำเนินการ WritableStream ได้หากจำเป็น

/* … */
write(chunk, controller) {
  try {
    // Try to do something dangerous with `chunk`.
  } catch (error) {
    controller.error(error.message);
  }
},
/* … */

queuingStrategy

อาร์กิวเมนต์ที่ 2 ของตัวสร้าง WritableStream() ซึ่งจะใส่หรือไม่ใส่ก็ได้คือ queuingStrategy เป็นออบเจ็กต์ที่กําหนดกลยุทธ์การจัดคิวสําหรับสตรีมได้โดยไม่บังคับ ซึ่งมีพารามิเตอร์ 2 รายการดังนี้

• highWaterMark: ตัวเลขที่ไม่เป็นลบซึ่งระบุเครื่องหมายระดับสูงสุดของสตรีมโดยใช้กลยุทธ์การจัดคิวนี้
• size(chunk): ฟังก์ชันที่คำนวณและแสดงผลขนาดที่ไม่เป็นลบแบบจำกัดของค่าก้อนที่ระบุ ระบบจะใช้ผลลัพธ์เพื่อกำหนดแรงดันย้อนกลับ ซึ่งจะแสดงผ่านพร็อพเพอร์ตี้ WritableStreamDefaultWriter.desiredSize ที่เหมาะสม

วิธีการ getWriter() และ write()

หากต้องการเขียนไปยังสตรีมที่เขียนได้ คุณต้องมี Writer ซึ่งจะเป็น WritableStreamDefaultWriter เมธอด getWriter() ของอินเทอร์เฟซ WritableStream จะแสดงผลอินสแตนซ์ใหม่ของ WritableStreamDefaultWriter และล็อกสตรีมไปยังอินสแตนซ์นั้น ขณะที่ สตรีมถูกล็อก จะไม่มีนักเขียนคนอื่นจนกว่าจะมีการเผยแพร่สตรีมปัจจุบัน

เมธอด write() ของอินเทอร์เฟซ WritableStreamDefaultWriter จะเขียนกลุ่มข้อมูลที่ส่งไปยัง WritableStream และ Sink ที่เกี่ยวข้อง จากนั้นจะแสดงผล Promise ที่แก้ไขเพื่อระบุความสำเร็จหรือความล้มเหลวของการดำเนินการเขียน โปรดทราบว่า "สำเร็จ" หมายถึงอะไรนั้นขึ้นอยู่กับ Sink ที่เกี่ยวข้อง ซึ่งอาจบ่งบอกว่ายอมรับก้อนข้อมูลแล้ว และไม่จำเป็นต้องหมายความว่าระบบได้บันทึกก้อนข้อมูลอย่างปลอดภัยไปยังปลายทางสุดท้าย

const writer = writableStream.getWriter();
const resultPromise = writer.write('The first chunk!');

พร็อพเพอร์ตี้ locked

คุณตรวจสอบได้ว่าสตรีมที่เขียนได้ล็อกอยู่หรือไม่โดยเข้าถึงพร็อพเพอร์ตี้ WritableStream.locked ของสตรีม

const locked = writableStream.locked;
console.log(`The stream is ${locked ? 'indeed' : 'not'} locked.`);

ตัวอย่างโค้ดสตรีมที่เขียนได้

ตัวอย่างโค้ดต่อไปนี้แสดงขั้นตอนทั้งหมดในการดำเนินการ

const writableStream = new WritableStream({
  start(controller) {
    console.log('[start]');
  },
  async write(chunk, controller) {
    console.log('[write]', chunk);
    // Wait for next write.
    await new Promise((resolve) => setTimeout(() => {
      document.body.textContent += chunk;
      resolve();
    }, 1_000));
  },
  close(controller) {
    console.log('[close]');
  },
  abort(reason) {
    console.log('[abort]', reason);
  },
});

const writer = writableStream.getWriter();
const start = Date.now();
for (const char of 'abcdefghijklmnopqrstuvwxyz') {
  // Wait to add to the write queue.
  await writer.ready;
  console.log('[ready]', Date.now() - start, 'ms');
  // The Promise is resolved after the write finishes.
  writer.write(char);
}
await writer.close();

การส่งสตรีมที่อ่านได้ไปยังสตรีมที่เขียนได้

คุณส่งสตรีมที่อ่านได้ไปยังสตรีมที่เขียนได้ผ่านเมธอด pipeTo() ของสตรีมที่อ่านได้ ReadableStream.pipeTo() จะส่ง ReadableStream ปัจจุบันไปยัง WritableStream ที่ระบุ และแสดงผลสัญญาที่ดำเนินการเมื่อกระบวนการส่งต่อเสร็จสมบูรณ์ หรือปฏิเสธหากพบข้อผิดพลาด

const readableStream = new ReadableStream({
  start(controller) {
    // Called by constructor.
    console.log('[start readable]');
    controller.enqueue('a');
    controller.enqueue('b');
    controller.enqueue('c');
  },
  pull(controller) {
    // Called when controller's queue is empty.
    console.log('[pull]');
    controller.enqueue('d');
    controller.close();
  },
  cancel(reason) {
    // Called when the stream is canceled.
    console.log('[cancel]', reason);
  },
});

const writableStream = new WritableStream({
  start(controller) {
    // Called by constructor
    console.log('[start writable]');
  },
  async write(chunk, controller) {
    // Called upon writer.write()
    console.log('[write]', chunk);
    // Wait for next write.
    await new Promise((resolve) => setTimeout(() => {
      document.body.textContent += chunk;
      resolve();
    }, 1_000));
  },
  close(controller) {
    console.log('[close]');
  },
  abort(reason) {
    console.log('[abort]', reason);
  },
});

await readableStream.pipeTo(writableStream);
console.log('[finished]');

การสร้างสตรีมการเปลี่ยนรูปแบบ

TransformStream อินเทอร์เฟซของ Streams API แสดงชุดข้อมูลที่แปลงได้ คุณ สร้างสตรีมการแปลงโดยการเรียกตัวสร้าง TransformStream() ซึ่งจะสร้างและแสดงผล ออบเจ็กต์สตรีมการแปลงจากแฮนเดิลที่ระบุ คอนสตรัคเตอร์ TransformStream() รับออบเจ็กต์ JavaScript ที่ไม่บังคับเป็นอาร์กิวเมนต์แรก ซึ่งแสดงถึง transformer ออบเจ็กต์ดังกล่าวอาจมีเมธอดต่อไปนี้

transformer

• start(controller): ระบบจะเรียกใช้เมธอดนี้ทันทีเมื่อสร้างออบเจ็กต์ โดยปกติแล้ว จะใช้เพื่อจัดคิวกลุ่มคำนำหน้าโดยใช้ controller.enqueue() ระบบจะอ่านข้อมูลเหล่านั้นจากฝั่งที่อ่านได้ แต่จะไม่ขึ้นอยู่กับการเขียนใดๆ ในฝั่งที่เขียนได้ หากกระบวนการเริ่มต้นนี้เป็นแบบไม่พร้อมกัน เช่น เนื่องจากต้องใช้ความพยายามในการรับก้อนข้อมูลของคำนำหน้า ฟังก์ชันจะคืนค่า Promise เพื่อส่งสัญญาณว่าสำเร็จหรือล้มเหลว Promise ที่ถูกปฏิเสธจะทำให้เกิดข้อผิดพลาดในสตรีม ตัวสร้าง TransformStream() จะส่งข้อยกเว้นที่เกิดขึ้นอีกครั้ง
• transform(chunk, controller): ระบบจะเรียกใช้เมธอดนี้เมื่อพร้อมที่จะแปลงก้อนข้อมูลใหม่ที่เขียนลงในด้านที่เขียนได้ การติดตั้งใช้งานสตรีมช่วยให้มั่นใจได้ว่าฟังก์ชันนี้ จะเรียกใช้หลังจากที่การแปลงก่อนหน้าสำเร็จแล้วเท่านั้น และจะไม่เรียกใช้ก่อนที่ start() จะ เสร็จสมบูรณ์หรือหลังจากเรียกใช้ flush() ฟังก์ชันนี้จะทําการแปลงจริง ของสตรีมการแปลง โดยสามารถจัดคิวผลลัพธ์ได้โดยใช้ controller.enqueue() ซึ่งจะ อนุญาตให้เขียนก้อนข้อมูลเดียวไปยังฝั่งที่เขียนได้เพื่อทำให้เกิดก้อนข้อมูล 0 หรือหลายก้อนในฝั่งที่อ่านได้ โดยขึ้นอยู่กับจำนวนครั้งที่เรียกใช้ controller.enqueue() หากกระบวนการ แปลงเป็นแบบไม่พร้อมกัน ฟังก์ชันนี้จะแสดงผล Promise เพื่อส่งสัญญาณความสำเร็จหรือความล้มเหลวของ การแปลง Promise ที่ถูกปฏิเสธจะทำให้เกิดข้อผิดพลาดทั้งด้านที่อ่านได้และเขียนได้ของสตรีมการแปลง หากไม่ได้ระบุtransform()วิธี ระบบจะใช้การแปลงข้อมูลระบุตัวตน ซึ่งจะ จัดคิวกลุ่มข้อมูลจากฝั่งที่เขียนได้ไปยังฝั่งที่อ่านได้โดยไม่มีการเปลี่ยนแปลง
• flush(controller): เมธอดนี้จะเรียกใช้หลังจากที่เขียนก้อนข้อมูลทั้งหมดลงในฝั่งที่เขียนได้แล้ว และผ่านการแปลงโดยผ่าน transform() เรียบร้อยแล้ว และฝั่งที่เขียนได้กำลังจะปิด โดยปกติแล้วจะใช้เพื่อจัดคิวกลุ่มคำต่อท้ายไปยังฝั่งที่อ่านได้ก่อนที่จะปิดฝั่งนั้น หากกระบวนการล้างข้อมูลเป็นแบบไม่พร้อมกัน ฟังก์ชันจะคืนค่า Promise เพื่อ ส่งสัญญาณว่าสำเร็จหรือล้มเหลว และจะแจ้งผลลัพธ์ให้ผู้เรียกใช้ stream.writable.write() ทราบ นอกจากนี้ Promise ที่ถูกปฏิเสธจะทำให้ทั้งด้านที่อ่านได้และด้านที่เขียนได้ของสตรีมเกิดข้อผิดพลาด การส่งข้อยกเว้นจะถือว่าเหมือนกับการส่งคืน Promise ที่ถูกปฏิเสธ

const transformStream = new TransformStream({
  start(controller) {
    /* … */
  },

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

  flush(controller) {
    /* … */
  },
});

กลยุทธ์การจัดคิว writableStrategy และ readableStrategy

พารามิเตอร์ที่ 2 และ 3 ที่ไม่บังคับของตัวสร้าง TransformStream() คือ writableStrategy และreadableStrategy กลยุทธ์การจัดคิว โดยมีการกำหนดตามที่ระบุไว้ในส่วนสตรีมที่อ่านได้และเขียนได้ตามลำดับ

ตัวอย่างโค้ดสตรีมการแปลง

ตัวอย่างโค้ดต่อไปนี้แสดงสตรีมการแปลงที่ใช้งานอยู่

// Note that `TextEncoderStream` and `TextDecoderStream` exist now.
// This example shows how you would have done it before.
const textEncoderStream = new TransformStream({
  transform(chunk, controller) {
    console.log('[transform]', chunk);
    controller.enqueue(new TextEncoder().encode(chunk));
  },
  flush(controller) {
    console.log('[flush]');
    controller.terminate();
  },
});

(async () => {
  const readStream = textEncoderStream.readable;
  const writeStream = textEncoderStream.writable;

  const writer = writeStream.getWriter();
  for (const char of 'abc') {
    writer.write(char);
  }
  writer.close();

  const reader = readStream.getReader();
  for (let result = await reader.read(); !result.done; result = await reader.read()) {
    console.log('[value]', result.value);
  }
})();

การส่งสตรีมที่อ่านได้ผ่านสตรีมการแปลง

เมธอด pipeThrough() ของอินเทอร์เฟซ ReadableStream มีวิธีที่เชื่อมต่อกันได้ในการส่งสตรีมปัจจุบัน ผ่านสตรีมการเปลี่ยนรูปแบบหรือคู่ที่เขียน/อ่านได้อื่นๆ โดยทั่วไป การส่งสตรีมผ่านไปป์จะล็อก สตรีมนั้นตลอดระยะเวลาของไปป์ ซึ่งจะป้องกันไม่ให้ผู้อ่านคนอื่นๆ ล็อกสตรีม

const transformStream = new TransformStream({
  transform(chunk, controller) {
    console.log('[transform]', chunk);
    controller.enqueue(new TextEncoder().encode(chunk));
  },
  flush(controller) {
    console.log('[flush]');
    controller.terminate();
  },
});

const readableStream = new ReadableStream({
  start(controller) {
    // called by constructor
    console.log('[start]');
    controller.enqueue('a');
    controller.enqueue('b');
    controller.enqueue('c');
  },
  pull(controller) {
    // called read when controller's queue is empty
    console.log('[pull]');
    controller.enqueue('d');
    controller.close(); // or controller.error();
  },
  cancel(reason) {
    // called when rs.cancel(reason)
    console.log('[cancel]', reason);
  },
});

(async () => {
  const reader = readableStream.pipeThrough(transformStream).getReader();
  for (let result = await reader.read(); !result.done; result = await reader.read()) {
    console.log('[value]', result.value);
  }
})();

ตัวอย่างโค้ดถัดไป (ซึ่งอาจดูซับซ้อนเล็กน้อย) แสดงวิธีใช้เวอร์ชัน "ตะโกน" ของ fetch() ซึ่งจะเปลี่ยนข้อความทั้งหมดเป็นตัวพิมพ์ใหญ่โดยใช้สัญญาการตอบกลับที่ส่งคืน เป็นสตรีม และเปลี่ยนเป็นตัวพิมพ์ใหญ่ทีละกลุ่ม ข้อดีของแนวทางนี้คือคุณไม่จำเป็นต้องรอให้ดาวน์โหลดเอกสารทั้งฉบับ ซึ่งอาจสร้างความแตกต่างอย่างมากเมื่อต้องจัดการกับไฟล์ขนาดใหญ่

function upperCaseStream() {
  return new TransformStream({
    transform(chunk, controller) {
      controller.enqueue(chunk.toUpperCase());
    },
  });
}

function appendToDOMStream(el) {
  return new WritableStream({
    write(chunk) {
      el.append(chunk);
    }
  });
}

fetch('./lorem-ipsum.txt').then((response) =>
  response.body
    .pipeThrough(new TextDecoderStream())
    .pipeThrough(upperCaseStream())
    .pipeTo(appendToDOMStream(document.body))
);

สาธิต

การสาธิตด้านล่างแสดงสตรีมที่อ่านได้ เขียนได้ และแปลงได้ในการทำงาน นอกจากนี้ ยังมีตัวอย่าง ของเชนไปป์ pipeThrough() และ pipeTo() รวมถึงแสดงให้เห็นถึง tee() คุณเลือกเรียกใช้เดโมในหน้าต่างของตัวเองหรือดูซอร์สโค้ดได้

ไลฟ์สดที่มีประโยชน์ซึ่งพร้อมให้รับชมในเบราว์เซอร์

เบราว์เซอร์มีสตรีมที่มีประโยชน์หลายอย่างในตัว คุณสร้าง ReadableStreamจาก Blob ได้อย่างง่ายดาย เมธอด stream() ของอินเทอร์เฟซ Blob จะแสดงผล ReadableStream ซึ่งเมื่ออ่านแล้วจะแสดงผลข้อมูลที่อยู่ใน Blob นอกจากนี้ โปรดทราบว่าออบเจ็กต์ File เป็นออบเจ็กต์ Blob ประเภทหนึ่ง และสามารถใช้ในบริบทใดก็ได้ที่ใช้ Blob ได้

const readableStream = new Blob(['hello world'], { type: 'text/plain' }).stream();

รูปแบบการสตรีมของ TextDecoder.decode() และ TextEncoder.encode() เรียกว่า TextDecoderStream และ TextEncoderStream ตามลำดับ

const response = await fetch('https://streams.spec.whatwg.org/');
const decodedStream = response.body.pipeThrough(new TextDecoderStream());

การบีบอัดหรือคลายการบีบอัดไฟล์ทำได้ง่ายๆ ด้วยสตรีมการแปลงCompressionStream และDecompressionStream ตามลำดับ ตัวอย่างโค้ดด้านล่างแสดงวิธีดาวน์โหลดสเปคของสตรีม บีบอัด (gzip) ในเบราว์เซอร์โดยตรง และเขียนไฟล์ที่บีบอัดลงในดิสก์โดยตรง

const response = await fetch('https://streams.spec.whatwg.org/');
const readableStream = response.body;
const compressedStream = readableStream.pipeThrough(new CompressionStream('gzip'));

const fileHandle = await showSaveFilePicker();
const writableStream = await fileHandle.createWritable();
compressedStream.pipeTo(writableStream);

File System Access API FileSystemWritableFileStream และfetch()สตรีมคำขอเวอร์ชันทดลองเป็น ตัวอย่างของสตรีมที่เขียนได้ในสภาพแวดล้อมจริง

Serial API ใช้ทั้งสตรีมที่อ่านได้และสตรีมที่เขียนได้เป็นจำนวนมาก

// Prompt user to select any serial port.
const port = await navigator.serial.requestPort();
// Wait for the serial port to open.
await port.open({ baudRate: 9_600 });
const reader = port.readable.getReader();

// Listen to data coming from the serial device.
while (true) {
  const { value, done } = await reader.read();
  if (done) {
    // Allow the serial port to be closed later.
    reader.releaseLock();
    break;
  }
  // value is a Uint8Array.
  console.log(value);
}

// Write to the serial port.
const writer = port.writable.getWriter();
const data = new Uint8Array([104, 101, 108, 108, 111]); // hello
await writer.write(data);
// Allow the serial port to be closed later.
writer.releaseLock();

สุดท้าย API WebSocketStream จะผสานรวมสตรีมกับ WebSocket API

const wss = new WebSocketStream(WSS_URL);
const { readable, writable } = await wss.connection;
const reader = readable.getReader();
const writer = writable.getWriter();

while (true) {
  const { value, done } = await reader.read();
  if (done) {
    break;
  }
  const result = await process(value);
  await writer.write(result);
}

แหล่งข้อมูลที่มีประโยชน์

• ข้อกำหนดของสตรีม
• การสาธิตประกอบ
• Streams polyfill
• 2016 - ปีแห่งสตรีมบนเว็บ
• ตัววนซ้ำและตัวสร้างแบบไม่พร้อมกัน
• Stream Visualizer

คำขอบคุณ

บทความนี้ได้รับการตรวจสอบโดย Jake Archibald François Beaufort Sam Dutton Mattias Buelens Surma Joe Medley และ Adam Rice บล็อกโพสต์ของ Jake Archibald ช่วยให้ฉันเข้าใจสตรีมได้มาก ตัวอย่างโค้ดบางส่วนได้รับแรงบันดาลใจจากการสำรวจของผู้ใช้ GitHub @bellbind และส่วนของข้อความอิงตามเอกสารประกอบบนเว็บ MDN เกี่ยวกับสตรีม มาตรฐาน Streams ผู้เขียนได้เขียนข้อกำหนดนี้ได้ยอดเยี่ยมมาก