স্ট্রিমস — চূড়ান্ত নির্দেশিকা

Streams API ব্যবহার করে কীভাবে পঠনযোগ্য, লিখনযোগ্য এবং রূপান্তরযোগ্য স্ট্রিম ব্যবহার করতে হয় তা শিখুন।

Thomas Steiner

স্ট্রিমস এপিআই আপনাকে নেটওয়ার্কের মাধ্যমে প্রাপ্ত বা স্থানীয়ভাবে যেকোনো উপায়ে তৈরি ডেটার স্ট্রিমগুলোকে প্রোগ্রাম্যাটিকভাবে অ্যাক্সেস করতে এবং জাভাস্ক্রিপ্ট দিয়ে সেগুলোকে প্রসেস করতে দেয়। স্ট্রিমিং-এর মাধ্যমে আপনি যে রিসোর্সটি গ্রহণ, প্রেরণ বা রূপান্তর করতে চান, সেটিকে ছোট ছোট খণ্ডে বিভক্ত করা হয় এবং তারপর এই খণ্ডগুলোকে এক এক করে প্রসেস করা হয়। যদিও ওয়েবপেজে দেখানোর জন্য এইচটিএমএল বা ভিডিওর মতো অ্যাসেট গ্রহণ করার সময় ব্রাউজারগুলো এমনিতেই স্ট্রিমিং করে থাকে, কিন্তু ২০১৫ সালে স্ট্রিমস সহ fetch with streams) চালু হওয়ার আগে জাভাস্ক্রিপ্টে এই সুবিধাটি কখনোই ছিল না।

উল্লেখ্য যে, XMLHttpRequest ব্যবহার করে স্ট্রিমিং করা প্রযুক্তিগতভাবে সম্ভব, কিন্তু এটি একটি সর্বোত্তম সমাধান নয়। এখানে একটি গিটহাব গিস্ট রয়েছে যেখানে 'XMLHttpRequest'-এর একটি ভালো উদাহরণ দেখানো হয়েছে ।

পূর্বে, আপনি যদি কোনো ধরনের রিসোর্স (সেটা ভিডিও হোক বা টেক্সট ফাইল, ইত্যাদি) প্রসেস করতে চাইতেন, তাহলে আপনাকে পুরো ফাইলটি ডাউনলোড করতে হতো, সেটিকে একটি উপযুক্ত ফরম্যাটে ডিসিরিয়ালাইজ হওয়ার জন্য অপেক্ষা করতে হতো এবং তারপর প্রসেস করতে হতো। জাভাস্ক্রিপ্টে স্ট্রিম উপলব্ধ হওয়ায় এই সবকিছু বদলে গেছে। এখন আপনি ক্লায়েন্টে ডেটা উপলব্ধ হওয়ার সাথে সাথেই জাভাস্ক্রিপ্ট দিয়ে পর্যায়ক্রমে তা প্রসেস করতে পারবেন, কোনো বাফার, স্ট্রিং বা ব্লব তৈরি করার প্রয়োজন ছাড়াই। এটি বেশ কিছু নতুন ব্যবহারের সুযোগ তৈরি করেছে, যার কয়েকটি আমি নিচে উল্লেখ করছি:

• ভিডিও এফেক্টস: একটি পাঠযোগ্য ভিডিও স্ট্রিমকে একটি ট্রান্সফর্ম স্ট্রিমের মধ্য দিয়ে প্রবাহিত করা, যা রিয়েল টাইমে বিভিন্ন এফেক্ট প্রয়োগ করে।
• ডেটা (ডি)কম্প্রেশন: একটি ফাইল স্ট্রিমকে এমন একটি ট্রান্সফর্ম স্ট্রিমের মধ্য দিয়ে চালনা করা, যা ফাইলটিকে বেছে বেছে (ডি)কম্প্রেস করে।
• ইমেজ ডিকোডিং: একটি HTTP রেসপন্স স্ট্রিমকে প্রথমে একটি ট্রান্সফর্ম স্ট্রিমের মধ্য দিয়ে পাঠানো হয়, যা বাইটকে বিটম্যাপ ডেটাতে ডিকোড করে, এবং তারপর আরেকটি ট্রান্সফর্ম স্ট্রিমের মধ্য দিয়ে পাঠানো হয়, যা বিটম্যাপকে PNG-তে রূপান্তর করে। যদি কোনো সার্ভিস ওয়ার্কারের fetch হ্যান্ডলারের ভিতরে এটি ইনস্টল করা হয়, তবে এটি আপনাকে AVIF-এর মতো নতুন ইমেজ ফরম্যাটগুলোকে স্বচ্ছভাবে পলিফিল করার সুযোগ দেয়।

ব্রাউজার সমর্থন

পঠনযোগ্য স্ট্রিম এবং লিখনযোগ্য স্ট্রিম

ট্রান্সফর্মস্ট্রিম

মূল ধারণা

বিভিন্ন ধরণের স্ট্রিম সম্পর্কে বিস্তারিত আলোচনার আগে, আমি কিছু মূল ধারণা তুলে ধরতে চাই।

খণ্ড

চাঙ্ক হলো ডেটার একটি একক অংশ যা কোনো স্ট্রিমে লেখা বা পড়া হয়। এটি যেকোনো ধরনের হতে পারে; এমনকি একটি স্ট্রিমে বিভিন্ন ধরনের চাঙ্কও থাকতে পারে। বেশিরভাগ সময়, একটি চাঙ্ক কোনো নির্দিষ্ট স্ট্রিমের জন্য ডেটার সবচেয়ে ক্ষুদ্রতম একক হয় না। উদাহরণস্বরূপ, একটি বাইট স্ট্রিমে একক বাইটের পরিবর্তে ১৬ কিলোবাইট Uint8Array ইউনিট দিয়ে গঠিত চাঙ্ক থাকতে পারে।

পাঠযোগ্য স্ট্রিম

একটি রিডেবল স্ট্রিম হলো ডেটার এমন একটি উৎস যেখান থেকে ডেটা পড়া যায়। অন্য কথায়, একটি রিডেবল স্ট্রিম থেকে ডেটা বেরিয়ে আসে । নির্দিষ্টভাবে বলতে গেলে, একটি রিডেবল স্ট্রিম হলো ReadableStream ক্লাসের একটি ইনস্ট্যান্স।

লেখার যোগ্য স্ট্রিম

একটি রাইটেবল স্ট্রিম হলো ডেটার এমন একটি গন্তব্যস্থল যেখানে আপনি লিখতে পারেন। অন্য কথায়, ডেটা একটি রাইটেবল স্ট্রিমে প্রবেশ করে । নির্দিষ্টভাবে বলতে গেলে, একটি রাইটেবল স্ট্রিম হলো WritableStream ক্লাসের একটি ইনস্ট্যান্স।

স্রোতকে রূপান্তর করুন

একটি ট্রান্সফর্ম স্ট্রিম দুটি স্ট্রিমের একটি জোড়া নিয়ে গঠিত: একটি রাইটেবল স্ট্রিম, যা এর রাইটেবল সাইড নামে পরিচিত, এবং একটি রিডেবল স্ট্রিম, যা এর রিডেবল সাইড নামে পরিচিত। এর একটি বাস্তব-জগতের রূপক হলো একজন সাইমালটেনিয়াস ইন্টারপ্রেটার , যিনি তাৎক্ষণিকভাবে এক ভাষা থেকে অন্য ভাষায় অনুবাদ করেন। ট্রান্সফর্ম স্ট্রিমের একটি নির্দিষ্ট পদ্ধতিতে, রাইটেবল সাইডে লেখার ফলে রিডেবল সাইড থেকে পড়ার জন্য নতুন ডেটা উপলব্ধ হয়। নির্দিষ্টভাবে বলতে গেলে, writable প্রপার্টি এবং readable প্রপার্টি আছে এমন যেকোনো অবজেক্ট একটি ট্রান্সফর্ম স্ট্রিম হিসেবে কাজ করতে পারে। তবে, স্ট্যান্ডার্ড TransformStream ক্লাসটি সঠিকভাবে এনট্যাঙ্গলড এমন একটি জোড়া তৈরি করা সহজ করে তোলে।

পাইপ চেইন

স্ট্রিমগুলো প্রধানত একে অপরের সাথে পাইপ করার মাধ্যমে ব্যবহৃত হয়। একটি রিডেবল স্ট্রিমকে তার pipeTo() মেথড ব্যবহার করে সরাসরি একটি রাইটেবল স্ট্রিমে পাইপ করা যায়, অথবা রিডেবল স্ট্রিমটির pipeThrough() মেথড ব্যবহার করে এটিকে প্রথমে এক বা একাধিক ট্রান্সফর্ম স্ট্রিমের মধ্যে দিয়ে পাঠানো যায়। এইভাবে একসাথে পাইপ করা স্ট্রিমগুলোর একটি সেটকে পাইপ চেইন বলা হয়।

ব্যাকপ্রেশার

একবার একটি পাইপ চেইন তৈরি হয়ে গেলে, এটি তার মধ্য দিয়ে খণ্ডগুলো কত দ্রুত প্রবাহিত হবে সে সম্পর্কে সংকেত প্রেরণ করে। যদি চেইনের কোনো ধাপ তখনও খণ্ড গ্রহণ করতে না পারে, তবে এটি পাইপ চেইনের মধ্য দিয়ে বিপরীত দিকে একটি সংকেত প্রেরণ করে, যতক্ষণ না অবশেষে মূল উৎসকে এত দ্রুত খণ্ড উৎপাদন বন্ধ করতে বলা হয়। প্রবাহকে স্বাভাবিক করার এই প্রক্রিয়াকে ব্যাকপ্রেশার বলা হয়।

টিয়িং

একটি পঠনযোগ্য স্ট্রিমকে তার tee() মেথড ব্যবহার করে 'tee' করা যায় (বড় হাতের 'T' অক্ষরের আকৃতির অনুকরণে এর নামকরণ করা হয়েছে)। এটি স্ট্রিমটিকে লক করে দেয়, অর্থাৎ এটিকে আর সরাসরি ব্যবহারযোগ্য রাখে না; তবে, এটি 'branch' নামক দুটি নতুন স্ট্রিম তৈরি করে, যেগুলো স্বাধীনভাবে ব্যবহার করা যায়। 'Tee' করা এই কারণেও গুরুত্বপূর্ণ যে, স্ট্রিম রিওয়াইন্ড বা রিস্টার্ট করা যায় না, এ বিষয়ে পরে আরও আলোচনা করা হবে।

পাঠযোগ্য স্ট্রিমের কার্যপ্রণালী

একটি রিডেবল স্ট্রিম হলো একটি ডেটা সোর্স যা জাভাস্ক্রিপ্টে একটি ReadableStream অবজেক্ট দ্বারা উপস্থাপিত হয় এবং যা একটি অন্তর্নিহিত উৎস থেকে প্রবাহিত হয়। ReadableStream() কনস্ট্রাক্টরটি প্রদত্ত হ্যান্ডলারগুলো থেকে একটি রিডেবল স্ট্রিম অবজেক্ট তৈরি করে এবং ফেরত দেয়। অন্তর্নিহিত উৎস দুই প্রকারের হয়:

• পুশ সোর্সগুলো আপনি অ্যাক্সেস করলে ক্রমাগত আপনার কাছে ডেটা পাঠাতে থাকে, এবং স্ট্রিমে অ্যাক্সেস শুরু করা, থামানো বা বাতিল করার দায়িত্ব আপনারই। এর উদাহরণ হলো লাইভ ভিডিও স্ট্রিম, সার্ভার-প্রেরিত ইভেন্ট বা ওয়েবসকেটস।
• পুল সোর্সের সাথে একবার সংযুক্ত হলে, আপনাকে সেখান থেকে ডেটার জন্য স্পষ্টভাবে অনুরোধ করতে হবে। উদাহরণস্বরূপ fetch() এর মাধ্যমে HTTP অপারেশন অথবা 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) : একটি ফাংশন যা প্রদত্ত 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() মেথডটি বর্তমান রিডেবল স্ট্রিমটিকে টি (tee) করে এবং দুটি নতুন ReadableStream ইনস্ট্যান্স হিসেবে দুটি শাখা ধারণকারী একটি দুই-উপাদানের অ্যারে ফেরত দেয়। এটি দুটি রিডারকে একই সাথে একটি স্ট্রিম পড়ার সুযোগ দেয়। উদাহরণস্বরূপ, আপনি একটি সার্ভিস ওয়ার্কারে এটি করতে পারেন, যদি আপনি সার্ভার থেকে একটি প্রতিক্রিয়া আনতে চান এবং ব্রাউজারে স্ট্রিম করতে চান, কিন্তু একই সাথে সার্ভিস ওয়ার্কার ক্যাশেও স্ট্রিম করতে চান। যেহেতু একটি প্রতিক্রিয়া বডি একবারের বেশি গ্রহণ করা যায় না, তাই এটি করার জন্য আপনার দুটি কপির প্রয়োজন। স্ট্রিমটি বাতিল করতে, আপনাকে দুটি শাখাই বাতিল করতে হবে। একটি স্ট্রিমকে টি (tee) করলে সাধারণত নির্দিষ্ট সময়ের জন্য এটি লক হয়ে যায়, যা অন্য রিডারদের এটিকে লক করা থেকে বিরত রাখে।

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) রিডারদের অ্যাকোয়ার করার সুযোগ দেয়। ডিফল্ট ইমপ্লিমেন্টেশনটি ওয়েবসকেটের ক্ষেত্রে স্ট্রিং বা অ্যারে বাফারের মতো বিভিন্ন ধরনের আউটপুট দিতে পারে, যেখানে বাইট স্ট্রিমগুলি বাইট আউটপুটের নিশ্চয়তা দেয়। এছাড়াও, 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;
}

নিম্নলিখিত ফাংশনটি পাঠযোগ্য বাইট স্ট্রিম রিটার্ন করে, যা র‍্যান্ডমভাবে তৈরি একটি অ্যারের কার্যকর জিরো-কপি রিডিংয়ের সুযোগ দেয়। এটি ১,০২৪-এর একটি পূর্বনির্ধারিত চাঙ্ক সাইজ ব্যবহার করার পরিবর্তে, ডেভেলপার-সরবরাহকৃত বাফারটি পূরণ করার চেষ্টা করে, যা সম্পূর্ণ নিয়ন্ত্রণের সুযোগ দেয়।

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

লেখনযোগ্য স্ট্রিমের কার্যপ্রণালী

একটি রাইটেবল স্ট্রিম হলো এমন একটি গন্তব্য যেখানে ডেটা লেখা যায়, যা জাভাস্ক্রিপ্টে একটি WritableStream অবজেক্ট দ্বারা উপস্থাপিত হয়। এটি একটি অন্তর্নিহিত সিঙ্কের (sink) উপরে একটি অ্যাবস্ট্রাকশন হিসেবে কাজ করে—সিঙ্ক হলো একটি নিম্ন-স্তরের 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) : একটি ফাংশন যা প্রদত্ত 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() কনস্ট্রাক্টরটি তার প্রথম আর্গুমেন্ট হিসেবে transformer প্রতিনিধিত্বকারী একটি ঐচ্ছিক জাভাস্ক্রিপ্ট অবজেক্ট গ্রহণ করে। এই ধরনের অবজেক্টগুলোতে নিম্নলিখিত যেকোনো মেথড থাকতে পারে:

transformer

• start(controller) : অবজেক্টটি তৈরি হওয়ার সাথে সাথেই এই মেথডটি কল করা হয়। সাধারণত এটি controller.enqueue() ব্যবহার করে প্রিফিক্স চাঙ্ক এনকিউ করার জন্য ব্যবহৃত হয়। এই চাঙ্কগুলো রিডেবল সাইড থেকে পড়া হবে, কিন্তু এগুলো রাইটেবল সাইডে কোনো রাইটের উপর নির্ভরশীল নয়। যদি এই প্রাথমিক প্রক্রিয়াটি অ্যাসিঙ্ক্রোনাস হয়, উদাহরণস্বরূপ প্রিফিক্স চাঙ্কগুলো সংগ্রহ করতে কিছু প্রচেষ্টা লাগার কারণে, ফাংশনটি সাফল্য বা ব্যর্থতা বোঝাতে একটি প্রমিজ রিটার্ন করতে পারে; একটি রিজেক্টেড প্রমিজ স্ট্রিমে এরর তৈরি করবে। যেকোনো থ্রো করা এক্সেপশন TransformStream() কনস্ট্রাক্টর দ্বারা পুনরায় থ্রো করা হবে।
• transform(chunk, controller) : এই মেথডটি তখন কল করা হয় যখন রাইটেবল সাইডে লেখা একটি নতুন চাঙ্ক ট্রান্সফর্ম করার জন্য প্রস্তুত হয়। স্ট্রিম ইমপ্লিমেন্টেশন নিশ্চিত করে যে এই ফাংশনটি শুধুমাত্র পূর্ববর্তী ট্রান্সফর্মগুলো সফল হওয়ার পরেই কল করা হবে, এবং start() সম্পূর্ণ হওয়ার আগে বা flush() কল করার পরে কখনোই নয়। এই ফাংশনটি ট্রান্সফর্ম স্ট্রিমের প্রকৃত ট্রান্সফর্মেশনের কাজটি সম্পাদন করে। এটি controller.enqueue() ব্যবহার করে ফলাফলগুলোকে এনকিউ (enqueue) করতে পারে। এটি রাইটেবল সাইডে লেখা একটিমাত্র চাঙ্কের ফলে রিডেবল সাইডে শূন্য বা একাধিক চাঙ্ক তৈরি করতে পারে, যা নির্ভর করে controller.enqueue() কতবার কল করা হয়েছে তার উপর। যদি ট্রান্সফর্ম করার প্রক্রিয়াটি অ্যাসিঙ্ক্রোনাস হয়, তবে এই ফাংশনটি ট্রান্সফর্মেশনের সাফল্য বা ব্যর্থতা বোঝাতে একটি প্রমিজ (promise) রিটার্ন করতে পারে। একটি রিজেক্টেড প্রমিজ ট্রান্সফর্ম স্ট্রিমের রিডেবল এবং রাইটেবল উভয় দিকেই এরর (error) দেখাবে। যদি কোনো transform() মেথড সরবরাহ করা না হয়, তবে আইডেন্টিটি ট্রান্সফর্ম (identity 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() এর একটি "শাউটিং" সংস্করণ প্রয়োগ করতে পারেন, যা ফেরত আসা `response promise`-টিকে একটি স্ট্রিম হিসাবে গ্রহণ করে এবং খণ্ড খণ্ড করে সমস্ত টেক্সটকে আপারকেস করে। এই পদ্ধতির সুবিধা হলো, আপনাকে পুরো ডকুমেন্টটি ডাউনলোড হওয়ার জন্য অপেক্ষা করতে হবে না, যা বড় ফাইল নিয়ে কাজ করার সময় একটি বিশাল পার্থক্য তৈরি করতে পারে।

function upperCaseStream() {
  return new TransformStream({
    transform(chunk, controller) {
      controller.enqueue(chunk.toUpperCase());
    },
  });
}

function appendToDOMStream(el) {
  return new WritableStream({
    write(chunk) {
      el.append(chunk);
    }
  });
}

fetch('./lorem-ipsum.txt').then((response) =>
  response.body
    .pipeThrough(new TextDecoderStream())
    .pipeThrough(upperCaseStream())
    .pipeTo(appendToDOMStream(document.body))
);

ডেমো

নিচের ডেমোটিতে রিডেবল, রাইটেবল এবং ট্রান্সফর্ম স্ট্রিমের কার্যকারিতা দেখানো হয়েছে। এতে pipeThrough() ও pipeTo() পাইপ চেইনের উদাহরণ এবং tee() ফাংশনের ব্যবহারও অন্তর্ভুক্ত রয়েছে। আপনি চাইলে ডেমোটি নিজস্ব উইন্ডোতে চালাতে পারেন অথবা সোর্স কোড দেখতে পারেন।

ব্রাউজারে দরকারী স্ট্রিমগুলি উপলব্ধ আছে

ব্রাউজারের মধ্যেই বেশ কিছু দরকারি স্ট্রিম তৈরি করা আছে। আপনি সহজেই একটি ব্লব থেকে একটি ReadableStream তৈরি করতে পারেন। Blob ইন্টারফেসের stream() মেথডটি একটি 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() রিকোয়েস্ট স্ট্রিমগুলো হলো বাস্তব ক্ষেত্রে ব্যবহৃত রাইটেবল স্ট্রিমের উদাহরণ।

সিরিয়াল এপিআই পঠনযোগ্য এবং লিখনযোগ্য উভয় প্রকার স্ট্রিম ব্যাপকভাবে ব্যবহার করে।

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

দরকারী সম্পদ

• স্ট্রিম স্পেসিফিকেশন
• সহগামী ডেমো
• স্ট্রিম পলিফিল
• ২০১৬—ওয়েব স্ট্রিমের বছর
• অ্যাসিঙ্ক ইটারেটর এবং জেনারেটর
• স্ট্রিম ভিজ্যুয়ালাইজার

কৃতজ্ঞতা স্বীকার

এই নিবন্ধটি জেক আর্চিবল্ড , ফ্রাঁসোয়া বোফোর্ট , স্যাম ডাটন , ম্যাটিয়াস বুলেন্স , সুরমা , জো মেডলি এবং অ্যাডাম রাইস দ্বারা পর্যালোচিত হয়েছে। জেক আর্চিবল্ডের ব্লগ পোস্টগুলো আমাকে স্ট্রিম বুঝতে অনেক সাহায্য করেছে। কিছু কোড স্যাম্পল গিটহাব ব্যবহারকারী @bellbind- এর অনুসন্ধান থেকে অনুপ্রাণিত এবং এর গদ্যের কিছু অংশ স্ট্রিম সম্পর্কিত এমডিএন ওয়েব ডক্সের ওপর ব্যাপকভাবে নির্ভরশীল। স্ট্রিমস স্ট্যান্ডার্ড -এর লেখকগণ এই স্পেকটি লেখার ক্ষেত্রে এক অসাধারণ কাজ করেছেন।