أحداث البث: الدليل النهائي

اطّلِع على كيفية استخدام أحداث البث القابلة للقراءة والقابلة للكتابة وتحويلها باستخدام Streams API.

Thomas Steiner

تتيح لك Streams API الوصول آليًا إلى مصادر البيانات المُستلَمة عبر الشبكة. أو تم إنشاؤها بأي وسيلة محلية باستخدام JavaScript. يتضمن البث تقسيم المورد الذي تريد استلامه أو إرساله أو تحويله إلى أجزاء صغيرة، ثم معالجة هذه الأجزاء قليلاً. أثناء البث أمر تفعل المتصفحات على أي حال عند تلقي مواد عرض مثل HTML أو مقاطع الفيديو لعرضها على صفحات الويب، فإن هذا لم يتم توفيره بلغة JavaScript قبل تقديم fetch مع مجموعات البث في عام 2015.

ملاحظة: كان البث ممكنًا من الناحية الفنية باستخدام XMLHttpRequest، إلا أنّه لم يكن جيّدًا.

في السابق، إذا أردت معالجة مورد من نوع ما (سواء كان فيديو أو ملفًا نصيًا وما إلى ذلك)، سيكون عليك تنزيل الملف بأكمله، والانتظار حتى يتم إلغاء تسلسله إلى تنسيق مناسب، ثم معالجتها. مع إمكانية البث JavaScript، كل ذلك يتغير. يمكنك الآن معالجة البيانات الأوليّة باستخدام JavaScript تدريجيًا فور توفّره على العميل، بدون الحاجة إلى إنشاء مخزن مؤقت أو سلسلة أو كائن الثنائي الكبير ويتيح ذلك إمكانية الكشف عن بعض حالات الاستخدام، وأدرجتُ بعضها أدناه:

• تأثيرات الفيديو: تضمين فيديو مضمّن قابل للقراءة من خلال بث تحويل يطبّق التأثيرات في الوقت الفعلي.
• ضغط البيانات (de): ربط مصدر بيانات من خلال مسار تحويل يتم بشكل انتقائي (de) يضغطها.
• فك ترميز الصور: تضمين تدفق استجابة HTTP من خلال تدفق تحويل يفك ترميز وحدات البايت إلى بيانات صور نقطية، ثم من خلال دفق تحويل آخر يترجم الصور النقطية إلى ملفات PNG. في حال حذف التي تم تثبيتها داخل معالج fetch لمشغّل الخدمات، ما يتيح لك إضافة رموز polyfill بشفافية بتنسيقات جديدة للصور مثل AVIF.

دعم المتصفح

ReadableStream وWritableStream

دعم المتصفح

• 43
• 14
• 65
• 10.1

المصدر

TransformStream

دعم المتصفح

• 67
• 79
• 102
• 14.1

المصدر

ملاحظة: للحصول على دعم إضافي، تتوفَّر خدمة polyfill. إذا أمكن، حمِّل رمز polyfill بشكل مشروط وفقط إذا كانت الميزة المضمّنة غير متاحة.

المفاهيم الأساسية

قبل أن أتطرق إلى التفاصيل حول الأنواع المختلفة لمصادر البيانات، دعوني أقدم لكم بعض المفاهيم الأساسية.

كتلة

المقطع هو جزء واحد من البيانات تتم كتابته في مصدر بيانات أو قراءته منه. يمكن أن يكون من أي type; يمكن أن تحتوي ساحات المشاركات على أجزاء من أنواع مختلفة. وفي معظم الأحيان، لن يكون الجزء الأكثر ذرية مصدر بيانات معيّن. على سبيل المثال، قد يحتوي دفق البايت على أجزاء تتكون من 16 Uint8Array كيبيبايت بدلاً من وحدات بايت فردية

أحداث البث القابلة للقراءة

يمثل البث القابل للقراءة مصدرًا للبيانات يمكنك القراءة من خلاله. بعبارة أخرى، تأتي البيانات من تدفق قابل للقراءة. في الواقع، تعتبر ساحة المشاركات القابلة للقراءة مثيلاً لـ ReadableStream الصف.

أحداث بث قابلة للكتابة

البث القابل للكتابة هو وجهة للبيانات التي يمكنك الكتابة فيها. بعبارة أخرى، توفر البيانات يدخل إلى بث قابل للكتابة. في الواقع، يعتبر البث القابل للكتابة أحد الأمثلة على صف واحد (WritableStream).

تحويل أحداث البث المباشر

تتكوّن مجموعة البث التحويلية من مجموعة بث: مجموعة بث قابلة للكتابة، تُعرف باسم مصدر البيانات القابل للكتابة. ودفق قابل للقراءة، يُعرف باسم الجانب القابل للقراءة. وقد يكون الاستعارة في العالم الحقيقي لهذا مترجم فوري يترجم من لغة إلى أخرى بشكل فوري. وبطريقة خاصة بتدفق التحويل، يمكن كتابة إلى نتائج الجانب القابل للكتابة، فيتم توفير بيانات جديدة للقراءة من الجانب القابل للقراءة. في الواقع، يمكن عرض أي كائن له السمة writable والسمة readable. كتدفق تحويل. مع ذلك، فإنّ فئة TransformStream العادية تسهّل إنشاء. مثل هذا الزوج المترابط بشكل صحيح.

سلاسل الأنابيب

يتم استخدام ساحات المشاركات بشكل أساسي من خلال ربطها ببعضها. يمكن توجيه تدفق قابل للقراءة مباشرةً إلى مصدر قابل للكتابة، باستخدام طريقة pipeTo() الخاصة بالبث القابل للقراءة، أو يمكن إرساله من خلال فلتر أو أكثر، تحويل ساحة المشاركات أولاً، باستخدام طريقة pipeThrough() للبث القابل للقراءة. مجموعة من وهذه الطريقة يُشار إليها بسلسلة الممرات معًا.

الضغط العكسي

بمجرد إنشاء سلسلة أنابيب، ستعمل على نشر الإشارات المتعلقة بكيفية تدفق الأجزاء بسرعة من خلالها. إذا لم تتمكن أي خطوة في السلسلة من قبول مقاطع بعد، فإنها تنشر الإشارة إلى الخلف عبر سلسلة الأنابيب، حتى تتم مطالبة المصدر الأصلي في النهاية بالتوقف عن إنتاج الأجزاء بسرعة. وتُسمى عملية تسوية التدفق هذه عملية الضغط العكسي.

تزلُّج

يمكن ربط ساحة مشاركات قابلة للقراءة (تسمّى بها حرف T كبير) باستخدام طريقة tee(). سيؤدي ذلك إلى قفل البث، ما يمنع استخدامه بشكل مباشر. لكنّه سينشئ قسمين جديدين المجاري المائية، تُسمى الفروع، والتي يمكن استهلاكها بشكل مستقل. من المهم أيضًا ربط البث لأنّه لا يمكن ترجيع البث أو إعادة تشغيله، وسنتحدث عن ذلك لاحقًا.

سلسلة ممر

آليات ساحة المشاركات القابلة للقراءة

البث القابل للقراءة هو مصدر بيانات يتم تمثيله في JavaScript بواسطة ReadableStream الذي البيانات من مصدر أساسي. تشير رسالة الأشكال البيانية ReadableStream() تنشئ الدالة الإنشائية كائن دفق قابل للقراءة من المعالِجات المحددة وتُرجعه. هناك خياران أنواع المصادر الأساسية:

• تدفع المصادر الفورية البيانات إليك باستمرار عند الوصول إليها، والأمر متروك لك بدء البث أو إيقافه مؤقتًا أو إلغاؤه تتضمن الأمثلة مجموعات بث الفيديو المباشر، والأحداث التي يرسلها الخادم، أو WebSockets.
• تتطلب منك مصادر السحب طلب البيانات صراحةً من هذه المصادر بعد الاتصال بها. أمثلة تضمين عمليات HTTP من خلال استدعاءات fetch() أو XMLHttpRequest.

تُقرأ بيانات البث بشكل تسلسلي في أجزاء صغيرة تُعرف باسم الأجزاء. ويُشار إلى المقاطع الموضوعة في ساحة المشاركات بإدراجها في قائمة الانتظار. هذا يعني أنّهم ينتظرون في قائمة الانتظار. جاهزة للقراءة. تتبع قائمة الانتظار الداخلية المقاطع التي لم تتم قراءتها بعد.

استراتيجية إضافة الإشارات إلى قائمة الانتظار: هي عنصر يحدّد كيفية إشارة البث المباشر إلى الضغط العكسي استنادًا إلى حالة قائمة الانتظار الداخلية. تعين استراتيجية إضافة إلى قائمة الانتظار حجمًا لكل مقطع، وتقارن إجمالي حجم كل القطع في قائمة الانتظار إلى رقم محدد، يُعرف باسم العلامة المائية العالية.

ويقرأ قارئ المجموعات داخل ساحة المشاركات. يسترد هذا القارئ البيانات مقطعًا واحدًا الوقت، مما يتيح لك إجراء أي نوع من العمليات التي ترغب في إجرائها. القارئ وغير ذلك رمز المعالجة الذي يتوافق معه يسمى المستهلك.

ويُطلق على البنية التالية في هذا السياق اسم وحدة التحكّم. ويكون لكل ساحة مشاركات قابلة للقراءة وحدة تحكم، كما يوحي اسمها، تتيح لك التحكم في البث.

يمكن لقارئ واحد فقط قراءة البث في كل مرة. عند إنشاء قارئ وبدء قراءة مجموعة بث (أي يصبح قارئًا نشطًا)، أي أنّه مقفل عليه. إذا أردت أن يتولى قارئ آخر عند قراءة البث، عادةً ما تحتاج إلى إطلاق القارئ الأول قبل اتخاذ أي إجراء آخر (ومع ذلك، يمكنك بدء البث).

إنشاء ساحة مشاركات قابلة للقراءة

يمكنك إنشاء ساحة مشاركات قابلة للقراءة من خلال استدعاء الدالة الإنشائية الخاصة بها. ReadableStream() تتضمن الدالة الإنشائية وسيطة اختيارية underlyingSource، والتي تمثل كائنًا باستخدام الطرق والخصائص التي تحدد سلوك مثيل ساحة المشاركات الذي تم إنشاؤه.

فريق underlyingSource

يمكن أن يستخدم هذا الإجراء الطرق الاختيارية التالية التي يحدّدها المطوِّر:

• start(controller): يتم استدعاؤه فورًا عند إنشاء العنصر. تشير رسالة الأشكال البيانية الوصول إلى مصدر البث وتنفيذ أي إجراء آخر المطلوبة لإعداد وظيفة البث. إذا كانت هذه العملية تتم بشكل غير متزامن، فيمكن أن بالوعد بالإشارة إلى النجاح أو الفشل. معلمة controller التي تم تمريرها إلى هذه الطريقة هي CANNOT TRANSLATE ReadableStreamDefaultController
• pull(controller): يمكن استخدام هذه السمة للتحكّم في البث حيث يتم جلب المزيد من المقاطع. أُنشأها جون هنتر، الذي كان متخصصًا يتم استدعاء الدالة بشكل متكرر طالما أنّ قائمة الانتظار الداخلية الخاصة بالبث غير ممتلئة حتى قائمة الانتظار عند وصوله إلى علامته المائية العالية. إذا كانت نتيجة الاتصال بالرقم pull() وعدًا، لن يتم الاتصال بـ "pull()" مرة أخرى إلى أن يتم استيفاء الوعد المذكور. في حال رفض الوعد، سيتم عرض رسالة خطأ.
• 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 المناسبة. ويتحكم أيضًا في حالات استدعاء طريقة pull() للمصدر الأساسي.

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

ملاحظة: يمكنك تحديد queuingStrategy المخصّص أو استخدام مثيل ByteLengthQueuingStrategy أو CountQueuingStrategy لقيمة هذا العنصر. إذا لم يتم تقديم queuingStrategy، ستكون القيمة التلقائية المستخدَمة هي نفسها CountQueuingStrategy مع highWaterMark من 1.

الطريقتان 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() للبث. أنت تستهلك هذه بدء البث من خلال إنشاء قارئ باستخدام طريقة 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));

تكرار غير متزامن

عند التحقّق من كل تكرار لتكرار 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 جديدة. هذا يسمح قراء اثنين لقراءة مجموعة البث في وقت واحد. يمكنك إجراء ذلك، على سبيل المثال، في عامل الخدمة إذا تريد جلب استجابة من الخادم وبثها إلى المتصفح، ولكن أيضًا بثها إلى ذاكرة التخزين المؤقت لمشغّل الخدمات. نظرًا لعدم إمكانية استهلاك نص الاستجابة أكثر من مرة، تحتاج إلى نسختَين. للقيام بذلك. لإلغاء البث، عليك إلغاء الإصدارَين الفرعيَين الناتجَين. بدء البث سيؤدي ذلك بشكل عام إلى قفله طوال المدة، ما يمنع القراء الآخرين من قفله.

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 بمزايا الاستقرار. هذا هو لأنه إذا انفصل المورد الاحتياطي، فإنه يمكن أن يضمن عدم الكتابة في نفس المورد الاحتياطي مرتين، وبالتالي تجنب ظروف السباق. يمكن لبرامج قراءة BYOB تقليل عدد المرات التي يحتاج فيها المتصفح إلى التشغيل جمع القمامة، لأنها قد تعيد استخدام الموارد الاحتياطية.

إنشاء تدفق بايت قابل للقراءة

يمكنك إنشاء بث بايت قابل للقراءة من خلال تمرير معلَمة type إضافية إلى الدالة الإنشائية ReadableStream().

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

فريق underlyingSource

يتم تحديد ReadableByteStreamController للمصدر الأساسي لتدفق بايت قابل للقراءة التلاعب. تستخدم طريقة ReadableByteStreamController.enqueue() الوسيطة chunk التي تحتوي قيمتها هي ArrayBufferView. تعرض السمة ReadableByteStreamController.byobRequest القيمة الحالية طلب سحب BYOB، أو قيمة خالية إذا لم يكن هناك أي طلب. أَخِيرًا، فِي ReadableByteStreamController.desiredSize الحجم المطلوب لملء قائمة الانتظار الداخلية للبث الخاضع للتحكُّم.

فريق queuingStrategy

الوسيطة الثانية الاختيارية بالمثل للدالة الإنشائية ReadableStream() هي queuingStrategy. وهو كائن يحدّد بشكل اختياري استراتيجية إضافة إلى قائمة الانتظار لمصدر البيانات، حيث يأخذ :

• highWaterMark: عدد غير سالب من وحدات البايت يشير إلى ارتفاع العلامة المائية للبث باستخدام استراتيجية الوضع في "قائمة المحتوى التالي" هذه. ويُستخدَم هذا لتحديد الضغط العكسي، الذي يظهر من خلال السمة ReadableByteStreamController.desiredSize المناسبة. ويتحكم أيضًا في حالات استدعاء طريقة pull() للمصدر الأساسي.

ملاحظة: على عكس استراتيجيات الوضع في "قائمة المحتوى التالي" لأنواع مصادر البيانات الأخرى، لا تتضمّن استراتيجية إضافة المحتوى إلى "قائمة المحتوى التالي" لبث بايت قابل للقراءة الدالة size(chunk). ويتم دائمًا تحديد حجم كل مجموعة من خلال السمة byteLength.

ملاحظة: إذا لم يتم تقديم queuingStrategy، يتم استخدام القيمة التلقائية واحدة مع highWaterMark من 0.

الطريقتان 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;
}

تعرض الدالة التالية تدفقات بايت قابلة للقراءة تسمح بقراءة النسخة الصفرية بشكل فعال صفيفة تم إنشاؤها عشوائيًا. وبدلاً من استخدام حجم مقطع محدد مسبقًا من 1024، يحاول ملء المورد الاحتياطي الذي يوفره المطور، مما يسمح بالتحكم الكامل.

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. هذا النمط يعمل بمثابة تجريد فوق الجزء العلوي من الحوض الأساسي، وهو حوض إدخال/إخراج منخفض المستوى البيانات الأولية.

تتم كتابة البيانات في مصدر البيانات عبر كاتب، بمعدل مقطع واحد في كل مرة. يمكن أن يستغرق المقطع العديد من النماذج، تمامًا مثل الأجزاء الموجودة في القارئ. يمكنك استخدام أي تعليمة برمجية تريد إنتاجها المقاطع الجاهزة للكتابة؛ ويسمّى المؤلف بالإضافة إلى الرمز المرتبط بالمنتج.

عندما يتم إنشاء مؤلف ويبدأ في الكتابة إلى ساحة مشاركات (كاتب نشط)، يُقال إنه محظور عليها. يمكن لكاتب واحد فقط الكتابة إلى سلسلة بيانات قابلة للكتابة عليها في آنٍ واحد. إذا أردت الحصول على آخر لبدء الكتابة في البث، تحتاج عادةً إلى سحبه قبل إرفاقه بكاتب آخر إليها.

تتبع قائمة الانتظار الداخلية المقاطع التي تمت كتابتها في ساحة المشاركات ولكن لم تتم كتابتها بعد. تتم معالجتها بواسطة الحوض السفلي.

استراتيجية إضافة الإشارات إلى قائمة الانتظار: هي عنصر يحدّد كيفية إشارة البث المباشر إلى الضغط العكسي استنادًا إلى حالة قائمة الانتظار الداخلية. تعين استراتيجية إضافة إلى قائمة الانتظار حجمًا لكل مقطع، وتقارن إجمالي حجم كل القطع في قائمة الانتظار إلى رقم محدد، يُعرف باسم العلامة المائية العالية.

يُطلق على البنية النهائية وحدة التحكّم. يشتمل كل بث قابل للكتابة على وحدة تحكم مرتبطة يسمح لك بالتحكم في البث (على سبيل المثال، لإلغائه).

إنشاء ساحة مشاركات قابلة للكتابة

واجهة WritableStream توفر واجهة Streams API تجريدًا عاديًا لكتابة بيانات البث إلى وجهة، على شكل حوض. يتضمّن هذا الجهاز نظامًا مدمَجًا لضغط خلفي ووضعه في قائمة الانتظار. يمكنك إنشاء بث قابل للكتابة من خلال استدعاء الدالة الإنشائية WritableStream() وهي تتضمن مَعلمة underlyingSink اختيارية تمثّل عنصرًا. باستخدام الطرق والخصائص التي تحدد سلوك مثيل ساحة المشاركات الذي تم إنشاؤه.

فريق underlyingSink

يمكن أن تتضمّن underlyingSink الطرق الاختيارية التالية التي يحدِّدها المطوِّر. controller التي يتم تمريرها إلى بعض الطرق هي WritableStreamDefaultController

• start(controller): يتم استدعاء هذه الطريقة فورًا عند إنشاء الكائن. تشير رسالة الأشكال البيانية ويجب أن يهدف كل من هذه الطريقة إلى الوصول إلى الحوض الأساسي. إذا كان من المقرر أن تكون هذه العملية إذا تم إجراؤها بشكل غير متزامن، فقد تظهر وعدًا بالإشارة إلى النجاح أو الفشل.
• write(chunk, controller): سيتم استدعاء هذه الطريقة عندما تكون مجموعة البيانات الجديدة (محددة في مَعلمة chunk) جاهزة لتتم كتابتها في الحوض الأساسي. يمكن أن يعود الوعد إلى تشير إلى نجاح أو فشل عملية الكتابة. سيتم استدعاء هذه الطريقة بعد السابقة نجاح البيانات، ولا يحدث ذلك بعد إغلاق البث أو إلغائه.
• close(controller): سيتم استدعاء هذه الطريقة إذا كان التطبيق يشير إلى اكتمال كتابته. أجزاء مهمة في التدفق. يجب أن يفعل المحتوى كل ما هو ضروري لإنهاء الكتابة إلى الحوض الأساسي، ومنح إمكانية الوصول إليه. إذا كانت هذه العملية غير متزامنة، فيمكنها عرض تعدك بالإشارة إلى النجاح أو الفشل. سيتم استدعاء هذه الطريقة فقط بعد أن يتم استدعاء كل عمليات الكتابة في قائمة الانتظار بنجاح.
• 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، فتتضمّن طريقة واحدة فقط: 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 المناسبة.

ملاحظة: يمكنك تحديد queuingStrategy المخصّص، أو استخدام مثيل ByteLengthQueuingStrategy أو CountQueuingStrategy لقيمة العنصر هذه. إذا لم يتم تقديم queuingStrategy، ستكون القيمة التلقائية المستخدَمة هي نفسها CountQueuingStrategy مع highWaterMark من 1.

الطريقتان getWriter() وwrite()

للكتابة إلى تدفق قابل للكتابة، تحتاج إلى كاتب WritableStreamDefaultWriter تعرض الطريقة getWriter() للواجهة WritableStream نسخة جديدة من WritableStreamDefaultWriter وتقوم بقفل ساحة المشاركات لتلك المثيل. وفي حين أن تم قفل البث، ولا يمكن الحصول على كاتب آخر إلا بعد إصدار الكتاب الحالي.

write() طريقة WritableStreamDefaultWriter كتبت واجهة المستخدم مجموعة من البيانات تم تمريرها إلى WritableStream وحوضها الأساسي، ثم تُرجع وعد يتم حله للإشارة إلى نجاح أو فشل عملية الكتابة. لاحظ أن ما "نجاح" والمتوسطة متروكة للحوض الأساسي؛ فقد يشير ذلك إلى أنه قد تم قبول المقطع، وليس بالضرورة أن يتم حفظها بأمان في وجهتها النهائية.

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(). ستتم قراءة هذه المقاطع من الجانب القابل للقراءة ولكنها لا تعتمد على أي عمليات كتابة إلى الجانب القابل للكتابة. إذا كانت هذه القيمة غير متزامنة، على سبيل المثال لأنه يتطلب بعض الجهد للحصول على أجزاء البادئة، يمكن أن ترجع الدالة وعدًا بالإشارة إلى النجاح أو الفشل؛ فإن الوعد المرفوض سوف يُخطئ في دفق. وستعيد الدالة الإنشائية TransformStream() طرح أي استثناءات تم تجاهلها.
• transform(chunk, controller): يتم استدعاء هذه الطريقة عندما تتم كتابة مقطع جديد في الأصل إلى وقابل للكتابة جاهز لتحويله. يضمن تنفيذ البث أن تكون هذه الدالة سيتم استدعاؤه فقط بعد نجاح التحويلات السابقة، ولن يتم استدعاؤه قبل نجاح start() مكتملة أو بعد طلب flush(). تنفذ هذه الدالة التحويل الفعلي لعمل تدفق التحويل. يمكن إدراج النتائج في قائمة الانتظار باستخدام controller.enqueue(). هذا النمط يسمح بمقطع واحد مكتوب على الجانب القابل للكتابة أن ينتج عنه أجزاء صفرية أو متعددة على جانبًا قابلاً للقراءة، بناءً على عدد مرات استدعاء controller.enqueue(). إذا كانت عملية فإن التحويل غير متزامن، فإن هذه الدالة يمكن أن تقدم وعدًا بالإشارة إلى نجاح أو فشل التحويل. يؤدي رفض الوعد إلى خطأ في الجانبين القابل للقراءة والقابل للكتابة تحويل ساحة المشاركات. وإذا لم يتم تقديم طريقة transform()، سيتم استخدام عملية تحويل الهوية، تضيف المجموعات إلى قائمة انتظار بدون تغيير من الجانب القابل للكتابة إلى الجانب القابل للقراءة.
• flush(controller): يتم استدعاء هذه الطريقة بعد حذف جميع الأجزاء المكتوبة إلى الجانب القابل للكتابة عن طريق المرور بنجاح عبر transform()، والجانب القابل للكتابة على وشك الانتهاء مُغْلَق. يُستخدم هذا عادةً لإضافة أجزاء لاحقة إلى قائمة الانتظار الخاصة بالجانب القابل للقراءة، قبل ذلك أيضًا مغلقًا. إذا كانت عملية السحب غير متزامنة، فيمكن للدالة إرجاع نجاح الإشارة أو إخفاقها سيتم إبلاغ المتصل بالنتيجة stream.writable.write() بالإضافة إلى ذلك، سيؤدي الوعد المرفوض إلى خطأ في كل من وقابلة للكتابة من التدفق. يتم التعامل مع طرح استثناء بالطريقة نفسها التي يتم التعامل بها مع عرض نتيجة مرفوضة وعدك.

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

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

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

استراتيجيتا writableStrategy وreadableStrategy لإضافة المحتوى إلى قائمة المحتوى التالي

المعلمتان الاختياريتان الثانية والثالثة للدالة الإنشائية TransformStream() اختياريتان استراتيجيتا writableStrategy وreadableStrategy الإضافة إلى قائمة المحتوى التالي كما يتم تعريفها كما هو موضح في للقراءة وساحة المشاركات writable الأقسام على التوالي.

تحويل نموذج رمز البث

يعرض نموذج الرمز البرمجي التالي تدفق تحويل بسيط قيد التنفيذ.

// 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() على إرجاع يشير ذلك المصطلح إلى ReadableStream الذي يعرض البيانات المضمَّنة في الكائن الثنائي الكبير (blob) عند القراءة. تذكر أيضًا أن الكائن File هو نوع معيّن من 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 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();

أخيرًا، تعمل واجهة برمجة التطبيقات 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);
}

مراجع مفيدة

• مواصفات مجموعات البث
• العروض التوضيحية المصاحبة
• polyfill في أحداث البث
• 2016: عام مصادر بيانات الويب
• المكرّرات وغير المتزامنة
• فيديو البث المباشر

شكر وتقدير

تمت مراجعة هذه المقالة بواسطة جاك أرشيبالد، فرانسوا بوفورت، سام دوتون، ماتياس بولينز، Surma، Joe Medley، آدم رايس. ساعدتني مشاركات مدوّنة جيك أرتشيبالد كثيرًا في فهم جداول البيانات. بعض عينات التعليمات البرمجية مستوحاة من مستخدم GitHub استكشافات @bellbind من النص المكتوب بشكل كبير على مستندات ويب MDN في مجموعات البث. تشير رسالة الأشكال البيانية Streams Standard لقد قام المؤلفون بعمل هائل في وكتابة هذه المواصفات. صورة رئيسية بواسطة راين لارا على إزالة البداية: