গেট সেট ডার্ট: স্ট্রীম

পূর্ববর্তী একটি পোস্টে আমরা ডার্টের প্রাথমিক অ্যাসিনক্রোনি নিয়ে কথা বলেছি। এবার আমরা ডার্টে অ্যাসিনক্রোনির আরেকটি গুরুত্বপূর্ণ অংশ স্ট্রীম (Stream) নিয়ে কথা বলবো।

Before stream there was iterable…

স্ট্রীম কী সেটা বোঝার আগে আসুন ইটারেবলের সাথে পরিচিত হই। আসলে ইটারেটর আপনি সবসময়ই ব্যবহার করে এসেছেন। যার ভেতর থেকে ক্রমানুসারে বারবার এক-একটা ভ্যালু পাওয়া যায় সেটাই ইটারেবল। যত জায়গায় আপনি for (var x in iterable) এই সিনট্যাক্সে লিখেছেন সবই ইটারেবল। সে অর্থে লিস্ট নিশ্চয়ই ইটারেবল। তবে ইটারেবল মানেই লিস্ট না। ইটারেবল একটা জেনারেটরও হতে পারে। জেনারেটর হচ্ছে এমন একধরনের ফাংশন যা আসলে ইটারেবল রিটার্ন করে। জেনারেটর কীভাবে কাজ করে তা আমরা দেখতে পাবো jpryan.me থেকে নির্লজ্জভাবে কপি করা এই কোড থেকে:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

main() {
  evenNumbersDownFrom(7).forEach(print);
}

// sync* functions return an iterable
Iterable<int> evenNumbersDownFrom(int n) sync* {
  // the body isn't executed until an iterator invokes moveNext()
  int k = n;
  while (k >= 0) {
    if (k % 2 == 0) {
      // 'yield' suspends the function
      yield k;
    }
    k--;
  }

  // when the end of the function is executed,
  // there are no more values in the Iterable, and
  // moveNext() returns false to the caller
  // > 6
  // > 4
  // > 2
  // > 0
}

evenNumbersDownFrom ফাংশনটি একটি ইটারেবল রিটার্ন করে যার ভেতর শুধু ইন্টিজার ভ্যালু পাওয়া যায়। সাধারণ বা সিনক্রোনাস জেনারেটর ডিফাইন করতে sync* কীওয়ার্ডটি ব্যবহার করা হয়। লিস্টের মত জেনারেটরের আগে থেকে ভ্যালু থাকে না। জেনারেটর লেইজি, অর্থাৎ, moveNext কল না হলে জেনারেটর ভ্যালু জেনারেট করে না।

এপর্যন্ত সবই ঠিক থাকে যতক্ষণ সিনক্রোনাস থাকে সব। এবার ভাবতে চেষ্টা করুন এমন একটা ইটারেবল যেটা অ্যাসিনক্রোনাস! রীতিমত ‘খাঁচার ভেতর অচিন পাখি কেমনে আসে যায়’ অবস্থা। এমন ইটারেবলকে আমরা স্ট্রীম বলি। অ্যাসিনক্রোনাস প্রোগ্রামিং কেন কাজের তা আমরা আগের পোস্টটিতে বলেছিলাম। স্ট্রীমও একই কারণে ভালো। বড় ফাইল চাঙ্ক করে ডাউনলোড করতে চান, বা মাউস ইভেন্ট পাঠাতে বা পেয়ে ব্যবহার করতে চান, স্ট্রীমই সবচেয়ে ভালো পদ্ধতি।

স্ট্রীম কীভাবে বানায়?

আসলে উপরের জেনারেটরের কোডটুকু একটু মোডিফাই করলেই অ্যাসিনক্রোনাস জেনারেটর বানানো যায় যেটি স্ট্রীম রিটার্ন করবে। আমরা ডার্ট এর অ্যাসিনক্ প্যাকেজ ইম্পোর্ট করেছি, sync* এর জায়গায় async* ব্যবহার করেছি, এবং রিটার্ন টাইপ Iterable<int> থেকে Stream<int> করেছি।

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

import "dart:async"; // Do not forget to import this

main() {
  evenNumbersDownFrom(7).forEach(print);
}

// async* functions return an stream
Stream<int> evenNumbersDownFrom(int n) async* {
  // the body isn't executed until an iterator invokes moveNext()
  int k = n;
  while (k >= 0) {
    if (k % 2 == 0) {
      // 'yield' suspends the function
      yield k;
    }
    k--;
  }

  // when the end of the function is executed,
  // there are no more values in the Iterable, and
  // moveNext() returns false to the caller
  // > 6
  // > 4
  // > 2
  // > 0
}

আমরা আউটপুটে ফলাফলও একই পাচ্ছি। এরকম সাধারণ ফাংশনের ক্ষেত্রে সেটাই হওয়ার কথা। আমরা স্ট্রীমের ওপর forEach মেথড কল করেছি। forEach, where, first, isEmpty ইত্যাদি যে মেথডগুলো আমরা ইটারেবলের মত স্ট্রীমেও পাচ্ছি। কিন্তু, স্ট্রীমের ওপর ইটারেশন চালানোর জন্য সবচেয়ে সাধারণ উপায় হচ্ছে await for:

1
2
3
4
5
6

main() async {
  var evensDownFromSeven = evenNumbersDownFrom(7);
  await for (var num in evensDownFromSeven){
    print(num);
  }
}

আমরা যে ফাংশনে await for ব্যবহার করছি সেটাকে async হিসেবে ডিক্লিয়ার করছি (এক্ষেত্রে main)। তারপর আমরা সাধারণ ফরলুপের মত await for ব্যবহার করছি। শুধু পার্থক্য হচ্ছে এক্ষেত্রে পরবর্তী ডাটা না আসা পর্যন্ত লুপটি অপেক্ষা করবে।

আগের পোস্টটি যারা পড়েছেন বা Future সম্পর্কে যারা জানেন তাদের নিশ্চয়ই মনে আছে যে Future ব্যবহার করার জন্য async/await বাদেও আছে then। স্ট্রীমের ক্ষেত্রেও তেমন আছে listen।

listen এর ডেফিনিশনটি এমন:

1
2

StreamSubscription<T> listen(void onData(T event),
    {Function onError, void onDone(), bool cancelOnError});

অর্থাৎ listen এর প্যারামিটারগুলোর মধ্যে একটা পজিশনাল। এটি একটি ফাংশন, প্রত্যেকবার ডাটা পেলে সেই ডাটা দিয়ে যা করতে হবে সেটা এই ফাংশনে পাওয়া যাবে। তাছাড়াও অপশনাল কীওয়ার্ড প্যারামিটারে আছে তিনটি ফাংশন যারা এরর ও স্ট্রীম শেষ হলে কী করা হবে সেটি হ্যান্ডেল করবে। উপরের async/await for দিয়ে লেখা কোডটি আমরা এভাবে লিখতে পারি:

1
2
3
4
5
6
7
8
9
10
11

main() async {
  var evensDownFromSeven = evenNumbersDownFrom(7);
  evensDownFromSeven.listen((e) => print(e),
                           onDone: () => print("I have gone all they way down to the 0."));
}

// > 6
// > 4
// > 2
// > 0
// > I have gone all they way down to the 0.

স্ট্রীমের ধরনধারন…

প্রোগ্রামিটিক্যালি পার্থক্য না থাকলেও ব্যবহারের ওপর নির্ভর করলে স্ট্রীম আসলে দুটি ভাগে ভাগ করা যায়। এদুটো আলাদা করতে পারা দরকার প্রোগ্রামিঙের স্বার্থেই।

একটিকে বলে সিঙ্গেল সাবস্ক্রিপশন স্ট্রীম (Single Subscription Stream), অর্থাৎ এমন স্ট্রীম যার ডাটা অ্যাসিনক্রোনাস হলেও ক্রম বজায় রাখে বা একটার সাথে আরেকটা সম্পর্কিত। পুরোটা না পেলে আসলে লাভ নেই। যেমন বড় ফাইল চাঙ্ক করে ডাউনলোড যদি করা হয় তাহলে মাঝখান থেকে লিসেন্ করলে যে ফাইল পাওয়া যাবে তা করাপ্টেড হতে পারে।

আরেকটি হলো ব্রডকাস্ট স্ট্রীম (Broadcast Stream)। এই স্ট্রীমগুলো এমন যে তার ভেতরের ডাটাগুলো একটি আরেকটির ওপর নির্ভরশীল না। যেমন মাউস ইভেন্টের কথাই চিন্তা করুন। মাউসের পজিশন ও ইভেন্টের একটি স্ট্রীম হলে সেটায় আপনি যখনই সাবস্ক্রাইব করেন আর যে আইটেমই পান তা অন্যগুলির ওপর নির্ভরশীল না, স্বাধীনভাবে আপনি ওই আইটেমটি ব্যবহার করতে পারবেন।

ত‌ো…

মোটামুটি এই ছিল স্ট্রীম নিয়ে। এইটি এবং পোস্টের শুরুতে বলা আগের পোস্টটিতে আমরা ডার্টের অ্যাসিনক্রোনির সবচেয়ে প্রাথমিক কিন্তু গুরুত্বপূর্ণ অংশ নিয়ে জানলাম। হ্যাপি কোডিং!