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

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

Thomas Steiner

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

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

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

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

ReadableStream और WritableStream

TransformStream

मुख्य कॉन्सेप्ट

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

चंक

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

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

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

लिखी जा सकने वाली स्ट्रीम

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

स्ट्रीम में बदलाव करना

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

पाइप चेन

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

बैकप्रेशर

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

टीइंग

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

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

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

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

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

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

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

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

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

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

इसके कन्स्ट्रक्टर को कॉल करके, पढ़ी जा सकने वाली स्ट्रीम बनाई जाती है 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() 10 सेकंड के दौरान हर सेकंड में एक टाइमस्टैंप जोड़ने के लिए कहा जाता है. आखिर में, यह कंट्रोलर को स्ट्रीम close() करने के लिए कहता है. इस स्ट्रीम का इस्तेमाल करने के लिए, getReader() तरीके से रीडर बनाएं और read() को तब तक कॉल करें, जब तक स्ट्रीम done न हो जाए.

class TimestampSource {
  #interval

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

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

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

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

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

एसिंक्रोनस तरीके से दोहराना

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

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

बाइट दिखाने वाली स्ट्रीम के लिए, पढ़ी जा सकने वाली स्ट्रीम का बड़ा वर्शन उपलब्ध कराया जाता है, ताकि बाइट को बेहतर तरीके से मैनेज किया जा सके. खास तौर पर, कॉपी को कम करके. बाइट स्ट्रीम की मदद से, 'अपना बफ़र खुद लाएं' (BYOB) रीडर हासिल किए जा सकते हैं. डिफ़ॉल्ट तौर पर लागू करने पर, 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() कन्स्ट्रक्टर के पहले आर्ग्युमेंट के तौर पर, transformer को दिखाने वाला वैकल्पिक JavaScript ऑब्जेक्ट स्वीकार किया जाता है. ऐसे ऑब्जेक्ट में, इनमें से कोई भी तरीका शामिल हो सकता है:

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 दिखाता है. इसे पढ़ने पर, ब्लॉब में मौजूद डेटा दिखता है. यह भी याद रखें कि 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);

फ़ाइल सिस्टम ऐक्सेस एपीआई की 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);
}

काम के संसाधन

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

आभार

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