स्ट्रीम — पूरी जानकारी

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

Thomas Steiner

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

ध्यान दें कि XMLHttpRequest की मदद से स्ट्रीमिंग की जा सकती है. हालांकि, यह सबसे सही तरीका नहीं है. यहां GitHub gist में, 'XMLHttpRequest' का एक अच्छा उदाहरण दिखाया गया है.

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

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

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

ReadableStream और WritableStream

TransformStream

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

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

बड़े हिस्से

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

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

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

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

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

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

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

पाइप चेन

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

बैकप्रेशर

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

टीइंग

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

रीडेबल स्ट्रीम का तरीका

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

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

Writable स्ट्रीम का कोड सैंपल

नीचे दिए गए कोड सैंपल में, सभी चरणों को दिखाया गया है.

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 ऑब्जेक्ट स्वीकार करता है. यह 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 क्यूइंग की रणनीतियां हैं. इन्हें रीडेबल और राइटेबल स्ट्रीम सेक्शन में बताया गया है.

स्ट्रीम के कोड को बदलने का सैंपल

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

// 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() के "shouting" वर्शन को लागू करने का तरीका बताया गया है. इसमें, जवाब के प्रॉमिस को स्ट्रीम के तौर पर इस्तेमाल करके, सभी टेक्स्ट को बड़े अक्षरों में बदला जाता है. साथ ही, हर हिस्से को बड़े अक्षरों में बदला जाता है. इस तरीके का फ़ायदा यह है कि आपको पूरे दस्तावेज़ के डाउनलोड होने का इंतज़ार नहीं करना पड़ता. बड़ी फ़ाइलों के मामले में, इससे काफ़ी फ़र्क़ पड़ सकता है.

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, जो मेडली, और ऐडम राइस. Jake Archibald की ब्लॉग पोस्ट से मुझे स्ट्रीम को समझने में बहुत मदद मिली. कुछ कोड सैंपल, GitHub उपयोगकर्ता @bellbind के एक्सप्लोरेशन से लिए गए हैं. साथ ही, इस लेख के कुछ हिस्से MDN Web Docs on Streams पर आधारित हैं. Streams Standard के लेखकों ने इस स्पेसिफ़िकेशन को लिखने में शानदार काम किया है.