Streams — The definitive guide

מידע על השימוש בזרמי נתונים שניתן לקרוא, לכתוב ולשנות באמצעות Streams API

Thomas Steiner

‫Streams API מאפשר לכם לגשת באופן פרוגרמטי לזרמי נתונים שמתקבלים ברשת או שנוצרים באמצעים מקומיים כלשהם, ולעבד אותם באמצעות JavaScript. סטרימינג כולל פירוק של משאב שרוצים לקבל, לשלוח או לשנות לחלקים קטנים, ואז עיבוד של החלקים האלה בזה אחר זה. הדפדפנים ממילא מבצעים סטרימינג כשהם מקבלים נכסים כמו HTML או סרטונים שמוצגים בדפי אינטרנט, אבל האפשרות הזו לא הייתה זמינה ל-JavaScript לפני שנוספה התמיכה בסטרימינג ב-fetch בשנת 2015.

הערה: מבחינה טכנית, אפשר להשתמש ב-XMLHttpRequest כדי להפעיל סטרימינג, אבל זה לא פתרון אופטימלי. כאן אפשר לראות GitHub gist שמציג דוגמה טובה ל-XMLHttpRequest.

בעבר, אם רציתם לעבד משאב כלשהו (סרטון, קובץ טקסט וכו'), הייתם צריכים להוריד את הקובץ כולו, לחכות עד שהוא יעבור דה-סריאליזציה לפורמט מתאים ואז לעבד אותו. השינויים האלה מאפשרים להשתמש בזרמים ב-JavaScript. מעכשיו אפשר לעבד נתונים גולמיים באמצעות JavaScript באופן הדרגתי ברגע שהם זמינים בלקוח, בלי ליצור מאגר זמני, מחרוזת או blob. האפשרות הזו מאפשרת לכם להשתמש במגוון תרחישים, חלקם מפורטים בהמשך:

• אפקטים של וידאו: העברת סטרימינג של וידאו שניתן לקריאה דרך סטרימינג של טרנספורמציה שמחיל אפקטים בזמן אמת.
• דחיסה או ביטול דחיסה של נתונים: העברת קובץ דרך מקור נתונים של טרנספורמציה שדוחס או מבטל את הדחיסה שלו באופן סלקטיבי.
• פענוח תמונה: העברת זרם של תגובת HTTP דרך זרם טרנספורמציה שמפענח בייטים לנתוני מפת סיביות, ואז דרך זרם טרנספורמציה נוסף שמתרגם מפות סיביות ל-PNG. אם התוסף מותקן בתוך fetch handler של service worker, הוא מאפשר לכם לבצע polyfill באופן שקוף לפורמטים חדשים של תמונות כמו AVIF.

תמיכה בדפדפנים

‫ReadableStream ו-WritableStream

TransformStream

מושגי ליבה

לפני שאפרט על הסוגים השונים של הזרמות, אציג כמה מושגי ליבה.

גושים

‫Chunk הוא חלק נתונים יחיד שנכתב לזרם או נקרא ממנו. הוא יכול להיות מכל סוג; זרמים יכולים אפילו להכיל נתחים מסוגים שונים. ברוב המקרים, נתח לא יהיה יחידת הנתונים האטומית ביותר עבור זרם נתונים נתון. לדוגמה, יכול להיות שזרם של בייטים יכיל נתחים שמורכבים מיחידות של 16‎ KiB Uint8Array, במקום בייטים בודדים.

מקורות נתונים שאפשר לקרוא

מקור נתונים לקריאה מייצג מקור נתונים שממנו אפשר לקרוא. במילים אחרות, הנתונים יוצאים מזרם נתונים שניתן לקריאה. במילים אחרות, זרם קריא הוא מופע של המחלקה ReadableStream.

מקורות נתונים שאפשר לכתוב בהם

מקור נתונים עם הרשאת כתיבה מייצג יעד לנתונים שאפשר לכתוב בו. במילים אחרות, הנתונים נכנסים למקור נתונים שאפשר לכתוב בו. בפועל, זרם שניתן לכתיבה הוא מופע של המחלקה WritableStream.

שינוי השידורים

זרם טרנספורמציה מורכב מזוג זרמים: זרם שניתן לכתיבה, שנקרא הצד שניתן לכתיבה, וזרם שניתן לקריאה, שנקרא הצד שניתן לקריאה. מטאפורה מהעולם האמיתי לתרגום כזה היא מתורגמן סימולטני שמתרגם משפה אחת לשפה אחרת תוך כדי דיבור. בצורה ספציפית לזרם הטרנספורמציה, כתיבה לצד שניתן לכתיבה גורמת לכך שנתונים חדשים יהיו זמינים לקריאה מהצד שניתן לקריאה. במילים אחרות, כל אובייקט עם מאפיין writable ומאפיין readable יכול לשמש כזרם טרנספורמציה. עם זאת, המחלקה הסטנדרטית TransformStream מקלה על יצירת זוג כזה ששזורים זה בזה בצורה נכונה.

שרשראות לצינורות

השימוש העיקרי בזרמים הוא העברה שלהם בצינורות אחד לשני. אפשר להעביר נתונים מזרם קריא ישירות לזרם שניתן לכתיבה באמצעות השיטה pipeTo() של הזרם הקריא, או להעביר אותם דרך זרם טרנספורמציה אחד או יותר באמצעות השיטה pipeThrough() של הזרם הקריא. קבוצה של זרמי נתונים שמחוברים יחד באמצעות צינורות נקראת שרשרת צינורות.

לחץ חוזר

אחרי שיוצרים שרשרת צינורות, היא מעבירה אותות לגבי המהירות שבה נתונים צריכים לעבור דרכה. אם שלב כלשהו בשרשרת לא יכול לקבל עדיין נתחים, הוא מעביר אות אחורה דרך שרשרת הצינורות, עד שבסופו של דבר המקור המקורי מקבל הוראה להפסיק לייצר נתחים כל כך מהר. התהליך הזה של נרמול הזרימה נקרא לחץ חוזר.

הנחת כדור הגולף על טי

אפשר להשתמש בשיטה tee() של זרם קריא כדי לפצל אותו (השם נגזר מהצורה של האות T באנגלית). הפעולה הזו תנעל את הזרם, כלומר לא יהיה יותר אפשר להשתמש בו ישירות. עם זאת, היא תיצור שני זרמים חדשים, שנקראים ענפים, שאפשר להשתמש בהם בנפרד. ההסתעפות חשובה גם כי אי אפשר להריץ אחורה או להפעיל מחדש את הזרמים. נרחיב על כך בהמשך.

איך עובד עדכון תוכן שניתן לקריאה

מקור נתונים שניתן לקריאה מיוצג ב-JavaScript על ידי אובייקט ReadableStream שמוזרם ממקור בסיסי. הקונסטרוקטור ReadableStream() יוצר ומחזיר אובייקט של זרם קריא מהמטפלים שצוינו. יש שני סוגים של מקורות בסיסיים:

• מקורות מסוג Push שולחים לכם נתונים כל הזמן כשאתם ניגשים אליהם, ואתם יכולים להתחיל, להשהות או לבטל את הגישה למקור הנתונים. לדוגמה, שידורי וידאו חיים, אירועים שנשלחים מהשרת או WebSockets.
• מקורות מסוג Pull מחייבים אתכם לבקש מהם נתונים באופן מפורש אחרי שמתחברים אליהם. דוגמאות כוללות פעולות HTTP באמצעות קריאות fetch() או XMLHttpRequest.

נתונים ממקורות סטרימינג נקראים ברצף בחלקים קטנים שנקראים chunks. אומרים שהחלקים שמוצבים בסטרימינג enqueued. כלומר, הם ממתינים בתור לקריאה. תור פנימי עוקב אחרי החלקים שעדיין לא נקראו.

שיטת תור היא אובייקט שקובע איך הזרם צריך לסמן לחץ חוזר על סמך המצב של התור הפנימי שלו. אסטרטגיית התור מקצה גודל לכל נתח, ומשווה את הגודל הכולל של כל הנתחים בתור למספר שצוין, שנקרא high water mark.

החלקים בתוך הזרם נקראים על ידי קורא. הקורא הזה מאחזר את הנתונים בחלקים, ומאפשר לכם לבצע עליהם כל פעולה שתרצו. הקורא בתוספת קוד העיבוד האחר שמשויך אליו נקרא צרכן.

ההגדרה הבאה בהקשר הזה נקראת בקר. לכל זרם קריא יש בקר משויך, שמאפשר לכם לשלוט בזרם, כפי שהשם מרמז.

רק קורא אחד יכול לקרוא נתונים מזרם בכל פעם. כשיוצרים קורא ומתחילים לקרוא נתונים מזרם (כלומר, הוא הופך לקורא פעיל), הוא ננעל לזרם הזה. אם רוצים שקורא אחר יקרא את הנתונים מהזרם, בדרך כלל צריך לשחרר את הקורא הראשון לפני שעושים משהו אחר (אפשר גם לפצל זרמים).

יצירת סטרימינג שניתן לקריאה

כדי ליצור מקור נתונים לקריאה, קוראים לבונה שלו ReadableStream(). ל-constructor יש ארגומנט אופציונלי underlyingSource, שמייצג אובייקט עם שיטות ומאפיינים שמגדירים את אופן הפעולה של מופע הסטרים שנוצר.

underlyingSource

אפשר להשתמש בשיטות האופציונליות הבאות שמוגדרות על ידי המפתח:

• ‫start(controller): מופעל באופן מיידי כשהאובייקט נוצר. השיטה יכולה לגשת למקור הסטרימינג ולבצע כל פעולה אחרת שנדרשת להגדרת פונקציונליות הסטרימינג. אם התהליך הזה צריך להתבצע באופן אסינכרוני, השיטה יכולה להחזיר אובייקט promise כדי לסמן הצלחה או כישלון. הפרמטר controller שמועבר לשיטה הזו הוא ReadableStreamDefaultController.
• ‫pull(controller): אפשר להשתמש בפרמטר הזה כדי לשלוט בסטרימינג כשמאחזרים עוד נתונים. הפונקציה נקראת שוב ושוב כל עוד התור הפנימי של חלקי הנתונים בשידור לא מלא, עד שהתור מגיע לסימן הגובה המקסימלי שלו. אם התוצאה של הקריאה ל-pull() היא אובייקט promise, לא תתבצע קריאה נוספת ל-pull() עד שהאובייקט 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

הארגומנט השני של הקונסטרוקטור ReadableStream(), שהוא אופציונלי, הוא queuingStrategy. זהו אובייקט שאפשר להגדיר בו אסטרטגיית תור להפעלה של הסטרימינג. האובייקט מקבל שני פרמטרים:

• ‫highWaterMark: מספר לא שלילי שמציין את נקודת השיא של הזרם באמצעות אסטרטגיית התור הזו.
• ‫size(chunk): פונקציה שמחשבת ומחזירה את הגודל הסופי הלא שלילי של ערך המקטע הנתון. התוצאה משמשת לקביעת לחץ חוזר, שמוצג באמצעות המאפיין המתאים ReadableStreamDefaultController.desiredSize. היא גם קובעת מתי מתבצעת קריאה ל-method‏ pull() של המקור הבסיסי.

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

השיטות getReader() ו-read()

כדי לקרוא מזרם שניתן לקריאה, צריך קורא, שהוא ReadableStreamDefaultReader. השיטה getReader() של הממשק ReadableStream יוצרת קורא ונועלת את הזרם עבורו. בזמן שהזרם נעול, אף קורא אחר לא יכול להשיג אותו עד שהוא ישוחרר.

השיטה read() של הממשק ReadableStreamDefaultReader מחזירה הבטחה שמספקת גישה לחלק הבא בתור הפנימי של הזרם. הפונקציה מחזירה תוצאה בהתאם למצב של הזרם. אלה האפשרויות השונות:

• אם יש נתח זמין, ההבטחה תתממש עם אובייקט מהצורה
  { value: chunk, done: false }.
• אם הזרם ייסגר, ההבטחה תתממש עם אובייקט מהצורה
  { value: undefined, done: true }.
• אם הסטרימינג ייכשל, ההבטחה תידחה עם השגיאה הרלוונטית.

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() חותמת זמן כל שנייה במשך עשר שניות. לבסוף, הוא מורה לבקר close() את הזרם. כדי לצרוך את הנתונים מהשידור הזה, צריך ליצור קורא באמצעות ה-method‏ getReader() ולהפעיל את ה-method‏ 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));

איטרציה אסינכרונית

יכול להיות שממשק ה-API שבודק בכל איטרציה של לולאת read() אם הסטרים הוא done לא הכי נוח. למזלנו, בקרוב תהיה דרך טובה יותר לעשות את זה: איטרציה אסינכרונית.

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 מכינה את הזרם הנוכחי שניתן לקריאה, ומחזירה מערך עם שני רכיבים שמכיל את שני הענפים שנוצרו כמופעים חדשים של ReadableStream. כך שני קוראים יכולים לקרוא נתונים מזרם בו-זמנית. לדוגמה, אפשר לעשות את זה בקובץ service worker אם רוצים לאחזר תגובה מהשרת ולהזרים אותה לדפדפן, אבל גם להזרים אותה למטמון של קובץ ה-service worker. אי אפשר להשתמש בגוף התגובה יותר מפעם אחת, ולכן צריך שני עותקים כדי לעשות את זה. כדי לבטל את הזרם, צריך לבטל את שני הענפים שנוצרו. בדרך כלל, כשמכינים זרם לשידור, הוא ננעל למשך השידור כדי למנוע מקוראים אחרים לנעול אותו.

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). ההטמעה שמוגדרת כברירת מחדל יכולה לתת מגוון של פלטים שונים, כמו מחרוזות או מאגרי מערכים במקרה של WebSockets, בעוד שזרמי בייטים מבטיחים פלט בייטים. בנוסף, קוראים שמביאים ספריה משלהם נהנים מיתרונות של יציבות. הסיבה לכך היא שאם מאגר נתונים זמני מנותק, אפשר להבטיח שלא תתבצע כתיבה לאותו מאגר נתונים זמני פעמיים, וכך נמנעים מצבי מירוץ. קוראי BYOB יכולים להפחית את מספר הפעמים שהדפדפן צריך להפעיל את 'איסוף הזבל', מכיוון שהוא יכול לעשות שימוש חוזר במאגרי מידע.

יצירת זרם בייטים שניתן לקריאה

אפשר ליצור זרם של בייטים שניתן לקריאה על ידי העברת פרמטר נוסף type לקונסטרוקטור ReadableStream().

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

underlyingSource

המקור הבסיסי של זרם בייטים שניתן לקריאה מקבל ReadableByteStreamController כדי לבצע בו מניפולציה. השיטה ReadableByteStreamController.enqueue() שלה מקבלת ארגומנט chunk שהערך שלו הוא ArrayBufferView. המאפיין ReadableByteStreamController.byobRequest מחזיר את בקשת המשיכה הנוכחית של BYOB, או null אם אין כזו. לבסוף, המאפיין ReadableByteStreamController.desiredSize מחזיר את הגודל הרצוי כדי למלא את התור הפנימי של הזרם המבוקר.

queuingStrategy

הארגומנט השני של הקונסטרוקטור ReadableStream(), שהוא אופציונלי, הוא queuingStrategy. זהו אובייקט שאופציונלית מגדיר אסטרטגיית תור לסטרימינג, שמקבל פרמטר אחד:

• ‫highWaterMark: מספר לא שלילי של בייטים שמציין את נקודת השיא של הזרם באמצעות אסטרטגיית התור הזו. הערך הזה משמש לקביעת לחץ חוזר, שמוצג באמצעות המאפיין המתאים ReadableByteStreamController.desiredSize. היא גם קובעת מתי מתבצעת קריאה ל-method‏ 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. היא משמשת כהפשטה מעל מאגר בסיסי – מאגר קלט/פלט ברמה נמוכה יותר, שאליו נכתבים נתונים גולמיים.

הנתונים נכתבים למקור הנתונים באמצעות writer, מקטע אחד בכל פעם. נתח יכול להיות בכל מיני צורות, בדיוק כמו נתחים בקורא. אתם יכולים להשתמש בכל קוד שתרצו כדי ליצור את החלקים שמוכנים לכתיבה. הכותב בתוספת הקוד המשויך נקרא מפיק.

כשיוצרים רכיב כתיבה והוא מתחיל לכתוב לזרם (רכיב כתיבה פעיל), אומרים שהוא ננעל לזרם. רק כותב אחד יכול לכתוב לזרם שניתן לכתיבה בכל פעם. אם רוצים שכותב אחר יתחיל לכתוב לזרם, בדרך כלל צריך לשחרר אותו, לפני שמצרפים אליו כותב אחר.

תור פנימי עוקב אחרי נתחי הנתונים שנכתבו לזרם אבל עדיין לא עברו עיבוד על ידי יעד הבסיס.

שיטת תור היא אובייקט שקובע איך הזרם צריך לסמן לחץ חוזר על סמך המצב של התור הפנימי שלו. אסטרטגיית התור מקצה גודל לכל נתח, ומשווה את הגודל הכולל של כל הנתחים בתור למספר שצוין, שנקרא high water mark.

המבנה הסופי נקרא בקר. לכל זרם שניתן לכתיבה משויך בקר שמאפשר לכם לשלוט בזרם (למשל, לבטל אותו).

יצירת מקור נתונים שאפשר לכתוב בו

ממשק WritableStream של Streams API מספק הפשטה סטנדרטית לכתיבת נתוני סטרימינג ליעד, שנקרא sink. האובייקט הזה מגיע עם לחץ חוזר וניהול תורים מובנים. יוצרים זרם שניתן לכתיבה על ידי קריאה לבונה שלו WritableStream(). יש לו פרמטר אופציונלי underlyingSink שמייצג אובייקט עם שיטות ומאפיינים שמגדירים את אופן הפעולה של מופע הסטרים שנוצר.

underlyingSink

ה-underlyingSink יכול לכלול את השיטות האופציונליות הבאות שהמפתחים מגדירים. הפרמטר controller שמועבר לחלק מהשיטות הוא WritableStreamDefaultController.

• ‫start(controller): המערכת קוראת לשיטה הזו באופן מיידי כשהאובייקט נוצר. התוכן של השיטה הזו צריך להיות מכוון לקבלת גישה ליעד הבסיסי. אם התהליך הזה צריך להתבצע באופן אסינכרוני, הוא יכול להחזיר הבטחה כדי לסמן הצלחה או כישלון.
• ‫write(chunk, controller): השיטה הזו תופעל כשנתח נתונים חדש (שמצוין בפרמטר chunk) יהיה מוכן לכתיבה ליעד הבסיסי. היא יכולה להחזיר אובייקט promise כדי לציין הצלחה או כישלון של פעולת הכתיבה. ה-method הזו תופעל רק אחרי שפעולות הכתיבה הקודמות יסתיימו בהצלחה, ואף פעם לא אחרי שהסטרים ייסגר או יבוטל.
• ‫close(controller): השיטה הזו תופעל אם האפליקציה תאותת שהיא סיימה לכתוב נתונים לחלקים בזרם. התוכן צריך לבצע את כל הפעולות שנדרשות כדי לסיים את הכתיבה ל-sink הבסיסי ולשחרר את הגישה אליו. אם התהליך הזה הוא אסינכרוני, הוא יכול להחזיר הבטחה (promise) כדי לסמן הצלחה או כישלון. השיטה הזו תופעל רק אחרי שכל הפעולות של כתיבה בתור יסתיימו בהצלחה.
• ‫abort(reason): השיטה הזו תופעל אם האפליקציה תאותת שהיא רוצה לסגור את הסטרימינג באופן פתאומי ולהעביר אותו למצב שגיאה. היא יכולה למחוק משאבים מושהים, בדומה ל-close(), אבל abort() תיקרא גם אם פעולות כתיבה ממתינות בתור. החלקים האלה יימחקו. אם התהליך הזה הוא אסינכרוני, הוא יכול להחזיר הבטחה כדי לסמן הצלחה או כישלון. הפרמטר reason מכיל את הערך DOMString שמתאר למה הופסקה ההעברה.

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

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

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

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

ממשק WritableStreamDefaultController של Streams API מייצג בקר שמאפשר שליטה במצב של WritableStream במהלך ההגדרה, כשחלקים נוספים נשלחים לכתיבה או בסיום הכתיבה. כשיוצרים WritableStream, יעד הנתונים הבסיסי מקבל מופע WritableStreamDefaultController תואם לשינוי. ל-WritableStreamDefaultController יש רק method אחד: WritableStreamDefaultController.error(), שגורם לשגיאה בכל האינטראקציות העתידיות עם הסטרימינג המשויך. ‫WritableStreamDefaultController תומך גם במאפיין signal שמחזיר מופע של AbortSignal, ומאפשר לעצור פעולה של WritableStream אם צריך.

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

queuingStrategy

הארגומנט השני של הקונסטרוקטור WritableStream(), שהוא אופציונלי, הוא queuingStrategy. זהו אובייקט שאפשר להגדיר בו אסטרטגיית תור להפעלה של הסטרימינג. האובייקט מקבל שני פרמטרים:

• ‫highWaterMark: מספר לא שלילי שמציין את נקודת השיא של הזרם באמצעות אסטרטגיית התור הזו.
• ‫size(chunk): פונקציה שמחשבת ומחזירה את הגודל הסופי הלא שלילי של ערך המקטע הנתון. התוצאה משמשת לקביעת לחץ חוזר, שמוצג באמצעות המאפיין המתאים WritableStreamDefaultWriter.desiredSize.

השיטות getWriter() ו-write()

כדי לכתוב לזרם שאפשר לכתוב בו, צריך writer, שהוא WritableStreamDefaultWriter. השיטה getWriter() של הממשק WritableStream מחזירה מופע חדש של WritableStreamDefaultWriter ונועלת את הזרם למופע הזה. בזמן שהסטרים נעול, אף כותב אחר לא יכול לקבל גישה אליו עד שהכותב הנוכחי משחרר אותו.

השיטה write() method of the WritableStreamDefaultWriter interface writes a passed chunk of data to a WritableStream and its underlying sink, then returns a promise that resolves to indicate the success or failure of the write operation. שימו לב: המשמעות של 'הצלחה' תלויה ביעד הבסיסי. היא יכולה לציין שהנתונים התקבלו, ולא בהכרח שהם נשמרו בבטחה ביעד הסופי.

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 נתון ומחזירה אובייקט promise שמושלם כשתהליך ההעברה מסתיים בהצלחה, או נדחה אם נתקלו בשגיאות.

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]');

יצירת מקור נתונים מסוג transform

ממשק TransformStream של Streams API מייצג קבוצה של נתונים שניתנים לשינוי. יוצרים זרם טרנספורמציה על ידי קריאה לבונה שלו TransformStream(), שיוצר ומחזיר אובייקט של זרם טרנספורמציה מהמטפלים שצוינו. ה-constructor‏ TransformStream() מקבל כארגומנט הראשון שלו אובייקט JavaScript אופציונלי שמייצג את transformer. אובייקטים כאלה יכולים להכיל את כל אחת מהשיטות הבאות:

transformer

• ‫start(controller): המערכת קוראת לשיטה הזו באופן מיידי כשהאובייקט נוצר. בדרך כלל, משתמשים בזה כדי להוסיף לתור נתחים של קידומות, באמצעות controller.enqueue(). החלקים האלה ייקראו מהצד שניתן לקריאה, אבל הם לא תלויים בפעולות כתיבה בצד שניתן לכתיבה. אם התהליך הראשוני הזה הוא אסינכרוני, למשל כי נדרש מאמץ מסוים כדי להשיג את נתחי התחילית, הפונקציה יכולה להחזיר אובייקט promise כדי לסמן הצלחה או כישלון. אובייקט promise שנדחה יגרום לשגיאה בזרם. כל החריגים שמועברים יועברו מחדש על ידי הבונה TransformStream().
• ‫transform(chunk, controller): קוראים ל-method הזה כשנתח חדש שנכתב במקור בצד שניתן לכתיבה מוכן להמרה. ההטמעה של הסטרימינג מבטיחה שהפונקציה הזו תיקרא רק אחרי שההמרות הקודמות יסתיימו בהצלחה, ואף פעם לפני ש-start() יסתיים או אחרי ש-flush() ייקרא. הפונקציה הזו מבצעת את העבודה בפועל של שינוי הנתונים בזרם. אפשר להוסיף את התוצאות לתור באמצעות controller.enqueue(). כך, נתח יחיד שנכתב לצד שניתן לכתיבה יכול להניב אפס או כמה נתחים בצד שניתן לקריאה, בהתאם למספר הפעמים שבהן מתבצעת הקריאה ל-controller.enqueue(). אם תהליך ההמרה הוא אסינכרוני, הפונקציה הזו יכולה להחזיר הבטחה כדי לסמן הצלחה או כישלון של ההמרה. הבטחה שנדחתה תגרום לשגיאה גם בצדדים של הזרם הניתן לקריאה וגם בצדדים של הזרם שניתן לכתיבה. אם לא מסופקת שיטת transform(), נעשה שימוש בשינוי הזהות, שמעביר את החלקים ללא שינוי מהצד שניתן לכתיבה לצד שניתן לקריאה.
• ‫flush(controller): השיטה הזו נקראת אחרי שכל החלקים שנכתבו בצד שניתן לכתיבה עברו טרנספורמציה בהצלחה דרך transform(), והצד שניתן לכתיבה עומד להיסגר. בדרך כלל משתמשים בשיטה הזו כדי להוסיף לתור נתחים של סיומות לצד הקריא, לפני שהוא נסגר. אם תהליך הניקוי הוא אסינכרוני, הפונקציה יכולה להחזיר אובייקט promise כדי לציין הצלחה או כישלון. התוצאה תועבר למי שקורא לפונקציה stream.writable.write(). בנוסף, הבטחה שנדחתה תגרום לשגיאה גם בצד הקריא וגם בצד שניתן לכתיבה של הזרם. העלאת חריגה נחשבת זהה להחזרת הבטחה שנדחתה.

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

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

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

אסטרטגיות התור writableStrategy ו-readableStrategy

הפרמטרים האופציונליים השני והשלישי של בנאי TransformStream() הם אופציונליים writableStrategy ושיטות התור readableStrategy. הם מוגדרים כמו שמתואר בקטע על הזרם לקריאה ובקטע על הזרם לכתיבה.

דוגמת קוד של שינוי הזרם

דוגמת הקוד הבאה מציגה transform stream בפעולה.

// 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() שבה כל הטקסט מופיע באותיות רישיות. לשם כך, צריך לצרוך את אובייקט ה-promise של התגובה שמוחזר כזרם ולהפוך את האותיות לרישיות בחלקים. היתרון בגישה הזו הוא שלא צריך לחכות להורדה של כל המסמך, וזה יכול להיות משמעותי מאוד כשמדובר בקבצים גדולים.

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 בהתאמה. בדוגמת הקוד הבאה אפשר לראות איך להוריד את מפרט ה-Streams, לדחוס אותו (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's FileSystemWritableFileStream וfetch() request streams הניסיוני הם דוגמאות לזרמי נתונים שאפשר לכתוב אליהם.

ב-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
• הדגמות נלוות
• Streams polyfill
• 2016 – השנה של מקורות לנתוני האתר
• איטרטורים וגנרטורים אסינכרוניים
• Stream Visualizer

תודות

המאמר הזה נבדק על ידי Jake Archibald,‏ François Beaufort,‏ Sam Dutton,‏ Mattias Buelens,‏ Surma,‏ Joe Medley ו-Adam Rice. הפוסטים בבלוג של Jake Archibald עזרו לי מאוד להבין את הסטרימינג. חלק מדוגמאות הקוד מבוססות על מחקרים של משתמש GitHub ‏@bellbind, וחלקים מהטקסט מבוססים על MDN Web Docs on Streams. היוצרים של Streams Standard עשו עבודה מצוינת בכתיבת המפרט הזה.