स्ट्रीम—खास गाइड

Streams API की मदद से, पढ़ने, लिखने, और बदलने वाली स्ट्रीम इस्तेमाल करने का तरीका जानें.

Thomas Steiner

Streams API की मदद से, नेटवर्क पर मिले डेटा की स्ट्रीम को प्रोग्रामैटिक तरीके से ऐक्सेस किया जा सकता है. इसके अलावा, इसे स्थानीय तौर पर किसी भी तरीके से बनाया जा सकता है और JavaScript की मदद से प्रोसेस किया जा सकता है. स्ट्रीमिंग में, जिस संसाधन को पाना, भेजना या बदलना है उसे छोटे-छोटे हिस्सों में बाँटा जाता है. इसके बाद, इन हिस्सों को एक-एक करके प्रोसेस किया जाता है. स्ट्रीमिंग एक ऐसी सुविधा है जो ब्राउज़र, वेबपेजों पर दिखाने के लिए एचटीएमएल या वीडियो जैसे ऐसेट पाने के दौरान पहले से ही करते हैं. हालांकि, यह सुविधा JavaScript के लिए कभी उपलब्ध नहीं थी. इसे 2015 में fetch के साथ स्ट्रीम की सुविधा के तौर पर पेश किया गया था.

पहले, किसी संसाधन (जैसे, वीडियो या टेक्स्ट फ़ाइल वगैरह) को प्रोसेस करने के लिए, आपको पूरी फ़ाइल डाउनलोड करनी पड़ती थी. इसके बाद, आपको फ़ाइल को सही फ़ॉर्मैट में बदलने के लिए इंतज़ार करना पड़ता था. इसके बाद, फ़ाइल को प्रोसेस किया जाता था. JavaScript के लिए स्ट्रीम उपलब्ध होने के बाद, यह सब बदल जाता है. अब क्लाइंट पर उपलब्ध होते ही, JavaScript की मदद से रॉ डेटा को प्रोसेस किया जा सकता है. इसके लिए, बफ़र, स्ट्रिंग या ब्लॉब जनरेट करने की ज़रूरत नहीं होती. इससे कई तरह के इस्तेमाल के उदाहरण मिलते हैं. इनमें से कुछ उदाहरण यहां दिए गए हैं:

• वीडियो इफ़ेक्ट: ट्रांसफ़ॉर्म स्ट्रीम के ज़रिए, पढ़ने लायक वीडियो स्ट्रीम को पाइप करना. इससे रीयल टाइम में इफ़ेक्ट लागू होते हैं.
• डेटा (डी)कंप्रेशन: किसी फ़ाइल स्ट्रीम को ट्रांसफ़ॉर्म स्ट्रीम के ज़रिए पाइप करना. इससे फ़ाइल स्ट्रीम को चुनिंदा तौर पर (डी)कंप्रेस किया जाता है.
• इमेज डिकोडिंग: एचटीटीपी रिस्पॉन्स स्ट्रीम को ट्रांसफ़ॉर्म स्ट्रीम के ज़रिए पाइप करना. यह स्ट्रीम, बाइट को बिटमैप डेटा में डिकोड करती है. इसके बाद, इसे दूसरी ट्रांसफ़ॉर्म स्ट्रीम के ज़रिए पाइप करना. यह स्ट्रीम, बिटमैप को पीएनजी में बदलती है. अगर इसे किसी सर्विस वर्कर के fetch हैंडलर में इंस्टॉल किया जाता है, तो इससे आपको AVIF जैसे नए इमेज फ़ॉर्मैट को पारदर्शी तरीके से पॉलीफ़िल करने की सुविधा मिलती है.

ब्राउज़र समर्थन

ReadableStream और WritableStream

TransformStream

मुख्य सिद्धांत

अलग-अलग तरह की स्ट्रीम के बारे में बताने से पहले, आइए कुछ मुख्य कॉन्सेप्ट के बारे में जान लें.

चंक

चंक, डेटा का एक हिस्सा होता है. इसे स्ट्रीम में लिखा या उससे पढ़ा जाता है. यह किसी भी टाइप का हो सकता है. स्ट्रीम में अलग-अलग टाइप के चंक भी शामिल हो सकते हैं. ज़्यादातर मामलों में, किसी स्ट्रीम के लिए कोई चंक, डेटा की सबसे छोटी यूनिट नहीं होता. उदाहरण के लिए, बाइट स्ट्रीम में सिंगल बाइट के बजाय, 16 कि॰बा॰ Uint8Array यूनिट वाले चंक शामिल हो सकते हैं.

पढ़ी जा सकने वाली स्ट्रीम

रीडेबल स्ट्रीम, डेटा के उस सोर्स को दिखाती है जिससे डेटा पढ़ा जा सकता है. दूसरे शब्दों में, डेटा को पढ़ने लायक स्ट्रीम से बाहर निकाला जाता है. आसान शब्दों में कहें, तो रीडेबल स्ट्रीम, ReadableStream क्लास का एक इंस्टेंस होती है.

लिखने के लिए स्ट्रीम

लिखने की अनुमति वाली स्ट्रीम, डेटा के लिए एक डेस्टिनेशन होती है. इसमें डेटा लिखा जा सकता है. दूसरे शब्दों में कहें, तो डेटा को राइटेबल स्ट्रीम में भेजा जाता है. आसान शब्दों में कहें, तो राइटेबल स्ट्रीम, WritableStream क्लास का एक इंस्टेंस होता है.

स्ट्रीम को बदलना

ट्रांसफ़ॉर्म स्ट्रीम में स्ट्रीम का एक जोड़ा होता है: एक राइटेबल स्ट्रीम, जिसे राइटेबल साइड कहा जाता है और एक रीडेबल स्ट्रीम, जिसे रीडेबल साइड कहा जाता है. असल दुनिया में, इसकी तुलना एक साथ अनुवाद करने वाले व्यक्ति से की जा सकती है. यह व्यक्ति, एक भाषा से दूसरी भाषा में तुरंत अनुवाद करता है. ट्रांसफ़ॉर्म स्ट्रीम के हिसाब से, लिखने की अनुमति वाली स्ट्रीम में डेटा लिखने से, पढ़ने की अनुमति वाली स्ट्रीम में नया डेटा उपलब्ध हो जाता है. असल में, writable और readable प्रॉपर्टी वाला कोई भी ऑब्जेक्ट, ट्रांसफ़ॉर्म स्ट्रीम के तौर पर काम कर सकता है. हालांकि, स्टैंडर्ड TransformStream क्लास की मदद से, इस तरह का पेयर बनाना आसान हो जाता है.

पाइप चेन

स्ट्रीम का इस्तेमाल मुख्य तौर पर, एक-दूसरे को पाइप करके किया जाता है. पढ़ने लायक स्ट्रीम को सीधे तौर पर लिखने लायक स्ट्रीम में पाइप किया जा सकता है. इसके लिए, पढ़ने लायक स्ट्रीम के pipeTo() तरीके का इस्तेमाल करें. इसके अलावा, इसे पहले एक या उससे ज़्यादा ट्रांसफ़ॉर्म स्ट्रीम में पाइप किया जा सकता है. इसके लिए, पढ़ने लायक स्ट्रीम के pipeThrough() तरीके का इस्तेमाल करें. इस तरह से एक साथ पाइप की गई स्ट्रीम के सेट को पाइप चेन कहा जाता है.

बैकप्रेशर

पाइप चेन बनने के बाद, यह इस बारे में सिग्नल भेजेगी कि इसमें से डेटा के छोटे-छोटे हिस्से कितनी तेज़ी से गुज़रने चाहिए. अगर चेन में मौजूद कोई भी चरण, अभी तक चंक स्वीकार नहीं कर सकता, तो यह पाइप चेन के ज़रिए पीछे की ओर एक सिग्नल भेजता है. ऐसा तब तक होता है, जब तक ओरिजनल सोर्स को चंक को इतनी तेज़ी से जनरेट करने से रोकने के लिए नहीं कहा जाता. फ़्लो को सामान्य करने की इस प्रोसेस को बैकप्रेशर कहा जाता है.

टीइंग

रीडेबल स्ट्रीम को tee() तरीके का इस्तेमाल करके टी किया जा सकता है. इसे कैपिटल लेटर 'T' के आकार के हिसाब से नाम दिया गया है. इससे स्ट्रीम लॉक हो जाएगी. इसका मतलब है कि अब इसका सीधे तौर पर इस्तेमाल नहीं किया जा सकेगा. हालांकि, इससे दो नई स्ट्रीम बनेंगी, जिन्हें ब्रांच कहा जाता है. इनका इस्तेमाल अलग-अलग किया जा सकता है. टीइंग भी ज़रूरी है, क्योंकि स्ट्रीम को रिवाइंड या फिर से शुरू नहीं किया जा सकता. इसके बारे में बाद में ज़्यादा जानकारी दी जाएगी.

पढ़ी जा सकने वाली स्ट्रीम का तरीका

रीडेबल स्ट्रीम, एक डेटा सोर्स है. इसे JavaScript में ReadableStream ऑब्जेक्ट के तौर पर दिखाया जाता है. यह ऑब्जेक्ट, किसी सोर्स से मिलता है. ReadableStream() कंस्ट्रक्टर, दिए गए हैंडलर से एक रीडेबल स्ट्रीम ऑब्जेक्ट बनाता है और उसे दिखाता है. डेटा सोर्स दो तरह के होते हैं:

• पुश सोर्स से, ऐक्सेस करने पर आपको लगातार डेटा मिलता रहता है. स्ट्रीम का ऐक्सेस शुरू करना, रोकना या रद्द करना आपके हाथ में होता है. उदाहरण के लिए, लाइव वीडियो स्ट्रीम, सर्वर से भेजे गए इवेंट या वेबसॉकेट.
• पुल सोर्स से कनेक्ट होने के बाद, आपको उनसे डेटा का अनुरोध करना होगा. उदाहरण के लिए, fetch() या XMLHttpRequest कॉल के ज़रिए एचटीटीपी कार्रवाइयां.

स्ट्रीम किए गए डेटा को क्रम से पढ़ा जाता है. इसे छोटे-छोटे हिस्सों में बांटा जाता है, जिन्हें चंक कहा जाता है. स्ट्रीम में रखे गए चंक को enqueued कहा जाता है. इसका मतलब है कि वे पढ़ने के लिए तैयार हैं और एक कतार में इंतज़ार कर रहे हैं. इंटरनल क्यू, उन चंक का ट्रैक रखता है जिन्हें अब तक नहीं पढ़ा गया है.

कतार में लगाने की रणनीति एक ऐसा ऑब्जेक्ट है जो यह तय करता है कि स्ट्रीम को बैकप्रेशर के बारे में कैसे सिग्नल देना चाहिए. यह इस बात पर निर्भर करता है कि उसकी इंटरनल कतार की स्थिति क्या है. कतार में लगाने की रणनीति, हर चंक को एक साइज़ असाइन करती है. साथ ही, कतार में मौजूद सभी चंक के कुल साइज़ की तुलना, तय की गई संख्या से करती है. इस संख्या को हाई वॉटर मार्क कहा जाता है.

स्ट्रीम में मौजूद चंक को रीडर पढ़ता है. यह रीडर, डेटा को एक बार में एक हिस्सा वापस लाता है. इससे आपको डेटा पर अपनी पसंद के मुताबिक कोई भी कार्रवाई करने की सुविधा मिलती है. रीडर और उससे जुड़ा प्रोसेसिंग कोड, उपयोगकर्ता कहलाता है.

इस कॉन्टेक्स्ट में, अगले कंस्ट्रक्ट को कंट्रोलर कहा जाता है. हर रीडेबल स्ट्रीम से एक कंट्रोलर जुड़ा होता है. जैसा कि नाम से पता चलता है, इसकी मदद से स्ट्रीम को कंट्रोल किया जा सकता है.

एक समय में सिर्फ़ एक रीडर, स्ट्रीम को पढ़ सकता है. जब कोई रीडर बनाया जाता है और वह स्ट्रीम को पढ़ना शुरू करता है, तो वह ऐक्टिव रीडर बन जाता है. इसके बाद, स्ट्रीम को उसके लिए लॉक कर दिया जाता है. अगर आपको किसी दूसरे रीडर को अपनी स्ट्रीम पढ़ने का ऐक्सेस देना है, तो आम तौर पर आपको पहले रीडर को रिलीज़ करना होगा. इसके बाद ही, कोई दूसरा रीडर आपकी स्ट्रीम पढ़ पाएगा. हालांकि, स्ट्रीम को टी किया जा सकता है.

पढ़ने लायक स्ट्रीम बनाना

रीडेबल स्ट्रीम बनाने के लिए, इसके कंस्ट्रक्टर ReadableStream() को कॉल करें. कंस्ट्रक्टर में एक वैकल्पिक आर्ग्युमेंट underlyingSource होता है. यह एक ऐसा ऑब्जेक्ट होता है जिसमें ऐसे तरीके और प्रॉपर्टी होती हैं जिनसे यह तय होता है कि बनाया गया स्ट्रीम इंस्टेंस कैसे काम करेगा.

underlyingSource

यह डेवलपर के तय किए गए इन वैकल्पिक तरीकों का इस्तेमाल कर सकता है:

• start(controller): ऑब्जेक्ट बनाए जाने के तुरंत बाद कॉल किया जाता है. इस तरीके से स्ट्रीम के सोर्स को ऐक्सेस किया जा सकता है. साथ ही, स्ट्रीम की सुविधा को सेट अप करने के लिए ज़रूरी अन्य काम भी किए जा सकते हैं. अगर इस प्रोसेस को एसिंक्रोनस तरीके से पूरा करना है, तो यह तरीका, प्रॉमिस को पूरा होने या न होने का सिग्नल देने के लिए एक प्रॉमिस दिखा सकता है. इस तरीके को पास किया गया controller पैरामीटर, 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;
    },
  },
);

getReader() और read() तरीके

रीडेबल स्ट्रीम से डेटा पढ़ने के लिए, आपको एक रीडर की ज़रूरत होगी. यह ReadableStreamDefaultReader होगा. ReadableStream इंटरफ़ेस का getReader() तरीका, रीडर बनाता है और स्ट्रीम को उसके लिए लॉक कर देता है. स्ट्रीम लॉक होने पर, किसी दूसरे व्यक्ति को तब तक ऐक्सेस नहीं दिया जा सकता, जब तक इस व्यक्ति को ऐक्सेस नहीं दिया जाता.

ReadableStreamDefaultReader इंटरफ़ेस का read() तरीका, एक प्रॉमिस दिखाता है. इससे स्ट्रीम की इंटरनल कतार में मौजूद अगले चंक को ऐक्सेस किया जा सकता है. यह स्ट्रीम की स्थिति के आधार पर, नतीजे को पूरा करता है या खारिज करता है. ऐसा इन वजहों से हो सकता है:

• अगर कोई हिस्सा उपलब्ध है, तो प्रॉमिस को
  { 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() तरीके का इस्तेमाल करके रीडर बनाएं. इसके बाद, स्ट्रीम के done होने तक read() को कॉल करें.

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);
}

एसिंक्रोनस इटरेशन का इस्तेमाल करने के लिए, पॉलीफ़िल के साथ व्यवहार लागू करना होगा.

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();
    }
  }
}

पढ़ने लायक स्ट्रीम को टी करना

ReadableStream इंटरफ़ेस का tee() तरीका, मौजूदा रीडेबल स्ट्रीम को टीज़ करता है. साथ ही, दो एलिमेंट वाला एक कलेक्शन दिखाता है. इसमें नई 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);
}

रीड की जा सकने वाली बाइट स्ट्रीम

बाइट को दिखाने वाली स्ट्रीम के लिए, पढ़ने लायक स्ट्रीम का एक्सटेंड किया गया वर्शन उपलब्ध कराया जाता है. इससे बाइट को बेहतर तरीके से मैनेज किया जा सकता है. खास तौर पर, कॉपी को कम करके. बाइट स्ट्रीम की मदद से, 'अपने हिसाब से बफ़रिंग की सुविधा इस्तेमाल करें' (बीवाईओबी) का इस्तेमाल करने वाले लोगों को हासिल किया जा सकता है. डिफ़ॉल्ट तौर पर लागू करने पर, अलग-अलग तरह के आउटपुट मिल सकते हैं. जैसे, WebSockets के मामले में स्ट्रिंग या ऐरे बफ़र. वहीं, बाइट स्ट्रीम से बाइट आउटपुट की गारंटी मिलती है. इसके अलावा, BYOB के सदस्यों को सदस्यता की अवधि के दौरान कीमत में बढ़ोतरी से सुरक्षा मिलती है. ऐसा इसलिए है, क्योंकि अगर कोई बफ़र अलग हो जाता है, तो यह गारंटी दी जा सकती है कि एक ही बफ़र में दो बार नहीं लिखा जाएगा. इसलिए, रेस कंडीशन से बचा जा सकता है. BYOB रीडर, ब्राउज़र को गार्बेज कलेक्शन को कम बार चलाने की अनुमति देते हैं, क्योंकि यह बफ़र का फिर से इस्तेमाल कर सकता है.

पढ़ने लायक बाइट स्ट्रीम बनाना

ReadableStream() कंस्ट्रक्टर में एक और type पैरामीटर पास करके, पढ़ने लायक बाइट स्ट्रीम बनाई जा सकती है.

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

underlyingSource

पढ़ने लायक बाइट स्ट्रीम के सोर्स को ReadableByteStreamController दिया जाता है, ताकि उसे बदला जा सके. इसके ReadableByteStreamController.enqueue() तरीके में, chunk आर्ग्युमेंट लिया जाता है. इसकी वैल्यू ArrayBufferView होती है. ReadableByteStreamController.byobRequest प्रॉपर्टी, मौजूदा BYOB पुल अनुरोध दिखाती है. अगर कोई अनुरोध नहीं है, तो यह प्रॉपर्टी null दिखाती है. आखिर में, ReadableByteStreamController.desiredSize प्रॉपर्टी, कंट्रोल की गई स्ट्रीम की इंटरनल कतार को भरने के लिए, तय किया गया साइज़ वापस भेजती है.

queuingStrategy

ReadableStream() कंस्ट्रक्टर का दूसरा आर्ग्युमेंट queuingStrategy है. यह आर्ग्युमेंट देना ज़रूरी नहीं है. यह एक ऑब्जेक्ट है. यह स्ट्रीम के लिए, कतार में लगाने की रणनीति तय करता है. हालांकि, यह ज़रूरी नहीं है. इसमें एक पैरामीटर होता है:

• highWaterMark: यह बाइट की एक ऐसी संख्या है जो शून्य या उससे ज़्यादा होती है. इससे, इस कतार लगाने की रणनीति का इस्तेमाल करके स्ट्रीम के हाई वॉटर मार्क का पता चलता है. इसका इस्तेमाल, बैकप्रेशर का पता लगाने के लिए किया जाता है. यह सही ReadableByteStreamController.desiredSize प्रॉपर्टी के ज़रिए दिखता है. यह इस बात को भी कंट्रोल करता है कि सोर्स के pull() मेथड को कब कॉल किया जाए.

getReader() और read() तरीके

इसके बाद, mode पैरामीटर को सेट करके, ReadableStreamBYOBReader को ऐक्सेस किया जा सकता है: 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 ऑब्जेक्ट के तौर पर दिखाया जाता है. यह अंडरलाइंग सिंक के ऊपर एक ऐब्स्ट्रैक्शन के तौर पर काम करता है. यह एक लोअर-लेवल I/O सिंक होता है, जिसमें रॉ डेटा लिखा जाता है.

डेटा को स्ट्रीम में राइटर के ज़रिए लिखा जाता है. इसमें एक बार में एक ही हिस्सा लिखा जाता है. रीडर में मौजूद चंक की तरह ही, किसी चंक के कई रूप हो सकते हैं. लिखने के लिए तैयार किए गए चंक बनाने के लिए, अपनी पसंद का कोई भी कोड इस्तेमाल किया जा सकता है. लेखक और उससे जुड़ा कोड, प्रोड्यूसर कहलाता है.

जब कोई राइटर बनाया जाता है और वह किसी स्ट्रीम में लिखना शुरू करता है, तो उसे ऐक्टिव राइटर कहा जाता है. साथ ही, उसे उस स्ट्रीम के लिए लॉक कर दिया जाता है. एक बार में, सिर्फ़ एक लेखक ही लिखने की सुविधा वाली स्ट्रीम में लिख सकता है. अगर आपको किसी दूसरे लेखक को अपनी स्ट्रीम में लिखने की अनुमति देनी है, तो आम तौर पर आपको स्ट्रीम को रिलीज़ करना होगा. इसके बाद, किसी दूसरे लेखक को स्ट्रीम से जोड़ा जा सकता है.

इंटरनल क्यू, उन चंक को ट्रैक करता है जिन्हें स्ट्रीम में लिखा गया है, लेकिन अब तक सिंक नहीं किया गया है.

कतार में लगाने की रणनीति एक ऐसा ऑब्जेक्ट है जो यह तय करता है कि स्ट्रीम को अपने इंटरनल क्यू की स्थिति के आधार पर बैकप्रेशर को कैसे सिग्नल करना चाहिए. कतार में लगाने की रणनीति, हर चंक को एक साइज़ असाइन करती है. साथ ही, कतार में मौजूद सभी चंक के कुल साइज़ की तुलना, तय की गई संख्या से करती है. इस संख्या को हाई वॉटर मार्क कहा जाता है.

आखिरी कंस्ट्रक्ट को कंट्रोलर कहा जाता है. लिखने की सुविधा वाली हर स्ट्रीम से एक कंट्रोलर जुड़ा होता है. इसकी मदद से, स्ट्रीम को कंट्रोल किया जा सकता है. उदाहरण के लिए, इसे बंद किया जा सकता है.

लिखने के लिए स्ट्रीम बनाना

Streams API का WritableStream इंटरफ़ेस, स्ट्रीमिंग डेटा को डेस्टिनेशन में लिखने के लिए स्टैंडर्ड ऐब्स्ट्रैक्शन उपलब्ध कराता है. इसे सिंक कहा जाता है. इस ऑब्जेक्ट में, बैकप्रेशर और क्यूइंग की सुविधा पहले से मौजूद होती है. राइटेबल स्ट्रीम बनाने के लिए, इसके कंस्ट्रक्टर 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) {
    /* … */
  },
});

Streams API का WritableStreamDefaultController इंटरफ़ेस, कंट्रोलर को दिखाता है. इससे 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 प्रॉपर्टी के ज़रिए दिखता है.

getWriter() और write() तरीके

लिखने की अनुमति वाली स्ट्रीम में लिखने के लिए, आपको राइटर की ज़रूरत होगी. यह WritableStreamDefaultWriter होगा. WritableStream इंटरफ़ेस का getWriter() तरीका, WritableStreamDefaultWriter का नया इंस्टेंस दिखाता है और स्ट्रीम को उस इंस्टेंस पर लॉक करता है. स्ट्रीम लॉक होने पर, किसी दूसरे लेखक को तब तक नहीं चुना जा सकता, जब तक मौजूदा लेखक को रिलीज़ नहीं कर दिया जाता.

WritableStreamDefaultWriter इंटरफ़ेस का write() तरीका, पास किए गए डेटा के हिस्से को 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]');

ट्रांसफ़ॉर्म स्ट्रीम बनाना

Streams API का TransformStream इंटरफ़ेस, बदलाव किए जा सकने वाले डेटा के सेट को दिखाता है. ट्रांसफ़ॉर्म स्ट्रीम बनाने के लिए, इसके कंस्ट्रक्टर TransformStream() को कॉल करें. यह दिए गए हैंडलर से एक ट्रांसफ़ॉर्म स्ट्रीम ऑब्जेक्ट बनाता है और उसे दिखाता है. TransformStream() कंस्ट्रक्टर, पहले आर्ग्युमेंट के तौर पर एक वैकल्पिक JavaScript ऑब्जेक्ट स्वीकार करता है. यह TransformStream() को दिखाता है.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 क्यूइंग की रणनीतियां हैं. इन्हें readable और 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);
  }
})();

पढ़ने लायक स्ट्रीम को ट्रांसफ़ॉर्म स्ट्रीम के ज़रिए पाइप करना

ReadableStream इंटरफ़ेस का pipeThrough() तरीका, मौजूदा स्ट्रीम को ट्रांसफ़ॉर्म स्ट्रीम या किसी अन्य लिखने/पढ़ने वाले पेयर के ज़रिए पाइप करने का एक तरीका है. किसी स्ट्रीम को पाइप करने पर, आम तौर पर वह पाइप किए जाने की अवधि के लिए लॉक हो जाती है. इससे दूसरे रीडर उसे लॉक नहीं कर पाते.

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 का एक खास टाइप होता है. इसका इस्तेमाल किसी भी ऐसे कॉन्टेक्स्ट में किया जा सकता है जहां किसी 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() अनुरोध स्ट्रीम, वाइल्ड में मौजूद राइटेबल स्ट्रीम के उदाहरण हैं.

सीरियल एपीआई, पढ़ने और लिखने, दोनों तरह की स्ट्रीम का ज़्यादा इस्तेमाल करता है.

// 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);
}

काम के संसाधन

• स्ट्रीम की खास बातें
• साथ में दिए गए डेमो
• Streams polyfill
• 2016—वेब स्ट्रीम का साल
• एसिंक इटरेटर और जनरेटर
• स्ट्रीम विज़ुअलाइज़र

Acknowledgements

इस लेख की समीक्षा इन लोगों ने की है: जेक आर्चीबाल्ड, फ़्रांस्वा ब्यूफ़ोर्ट, सैम डटन, मैटियास ब्यूलेंस, Surma, जो मेडली, और ऐडम राइस. जेक आर्चीबाल्ड की ब्लॉग पोस्ट से मुझे स्ट्रीम को समझने में बहुत मदद मिली. कोड के कुछ सैंपल, GitHub उपयोगकर्ता @bellbind के एक्सप्लोरेशन से लिए गए हैं. साथ ही, गद्य के कुछ हिस्से MDN Web Docs on Streams से लिए गए हैं. Streams Standard के लेखकों ने इस स्पेसिफ़िकेशन को लिखने में बेहतरीन काम किया है. हीरो इमेज, रायन लारा ने Unsplash पर उपलब्ध कराई है.