Dynamic Data Streams বা ডাইনামিক ডেটা স্ট্রিম এমন স্ট্রিম যার মধ্যে ডেটার পরিবর্তন ঘটতে থাকে এবং এই ডেটা প্রাপ্তির সময় বা শর্তও পরিবর্তিত হতে পারে। RxJS-এ একাধিক স্ট্রিম একত্রিত (combine) করতে এবং তাদের মধ্যে সম্পর্ক স্থাপন করার জন্য কিছু শক্তিশালী combination techniques রয়েছে। এই টেকনিকগুলির মাধ্যমে আপনি বিভিন্ন স্ট্রিমকে সমন্বিত করতে পারেন এবং তাদের উপর বিভিন্ন কার্যকরী অপারেশন প্রয়োগ করতে পারেন।
এখানে combination techniques বা স্ট্রিম কম্বাইনিং অপারেটরস নিয়ে আলোচনা করা হবে:
1. combineLatest()
combineLatest() অপারেটরটি একাধিক Observable থেকে সর্বশেষ মানগুলো নিয়ে একটি নতুন Observable তৈরি করে। এটি সব স্ট্রিমের সর্বশেষ মানগুলো নিয়ে একটি নতুন মান তৈরি করবে এবং সেই নতুন মানটি সাবস্ক্রাইবারে পাঠাবে। combineLatest() প্রতিটি Observable-এর সর্বশেষ মান নেওয়ার জন্য অপেক্ষা করে এবং সেই মানগুলিকে একটি একক আউটপুটে একত্রিত করে।
বৈশিষ্ট্য:
- সব সাবস্ক্রাইবারের সর্বশেষ মানের সমন্বয়।
- একাধিক Observable কে একত্রিত করে একটি নতুন Observable তৈরি করে।
উদাহরণ:
import { combineLatest, of } from 'rxjs';
const observable1 = of('A', 'B');
const observable2 = of(1, 2, 3);
combineLatest([observable1, observable2]).subscribe(([val1, val2]) => {
console.log(`Combined values: ${val1}, ${val2}`);
});
আউটপুট:
Combined values: B, 3
এখানে, combineLatest() দুইটি স্ট্রিমের সর্বশেষ মান একত্রিত করে পাঠাচ্ছে, যেমন B এবং 3।
2. merge()
merge() অপারেটরটি একাধিক Observable কে একত্রিত করে এবং তাদের মান সমান্তরালভাবে (concurrently) সাবস্ক্রাইবে পাঠায়। এটি বিভিন্ন Observable এর মধ্যে সমস্ত ডেটা একযোগে গ্রহণ করে এবং একে একে পাঠাতে থাকে।
বৈশিষ্ট্য:
- সমান্তরালভাবে স্ট্রিম পরিচালনা।
- একাধিক Observable এর মান সমানভাবে গ্রহণ করে।
উদাহরণ:
import { merge, of, interval } from 'rxjs';
import { take } from 'rxjs/operators';
const observable1 = of('A', 'B');
const observable2 = interval(1000).pipe(take(3));
merge(observable1, observable2).subscribe(val => {
console.log(val);
});
আউটপুট:
A
B
0
1
2
এখানে, merge() দুটি স্ট্রিমের মান একযোগে সাবস্ক্রাইব করে, যেটি একে একে প্রদর্শিত হচ্ছে।
3. concat()
concat() অপারেটরটি একাধিক Observable কে সিরিয়ালি একত্রিত করে। এটি প্রথম Observable এর ডেটা শেষ হওয়ার পর পরবর্তী Observable এর ডেটা সাবস্ক্রাইব শুরু করে। এখানে, স্ট্রিমগুলো একে একে সম্পন্ন হয় এবং একটির পর একটি নতুন স্ট্রিম আসে।
বৈশিষ্ট্য:
- সিরিয়াল প্রক্রিয়া। প্রথম Observable সম্পূর্ণ না হওয়া পর্যন্ত পরবর্তী Observable শুরু হয় না।
- একাধিক স্ট্রিম একত্রিত করা হয় নির্দিষ্ট ক্রমে।
উদাহরণ:
import { concat, of, interval } from 'rxjs';
import { take } from 'rxjs/operators';
const observable1 = of('A', 'B');
const observable2 = interval(1000).pipe(take(3));
concat(observable1, observable2).subscribe(val => {
console.log(val);
});
আউটপুট:
A
B
0
1
2
এখানে, concat() প্রথমে observable1 সম্পন্ন করেছে, তারপর observable2 শুরু হয়েছে।
4. zip()
zip() অপারেটরটি একাধিক Observable-এর মানকে একত্রিত করে এবং সেগুলির মানকে জোড়া (pair) আকারে পাঠায়। এটি একটি "zip" ফাংশনের মতো কাজ করে যেখানে প্রতিটি Observable-এর মান একে অপরের সাথে মিলিয়ে পাঠানো হয়।
বৈশিষ্ট্য:
- যুগল (pairing) ডেটা।
- একে অপরের সাথে মিলিয়ে (zip) মান পাঠানো হয়।
উদাহরণ:
import { zip, of } from 'rxjs';
const observable1 = of('A', 'B', 'C');
const observable2 = of(1, 2, 3);
zip(observable1, observable2).subscribe(([val1, val2]) => {
console.log(`Zipped values: ${val1}, ${val2}`);
});
আউটপুট:
Zipped values: A, 1
Zipped values: B, 2
Zipped values: C, 3
এখানে, zip() দুটি Observable এর মান একে অপরের সাথে যুগল (pair) করে পাঠাচ্ছে।
5. forkJoin()
forkJoin() অপারেটরটি একাধিক Observable থেকে সর্বশেষ মানগুলো নিয়ে একটি একক Observable তৈরি করে। এটি একধরনের "wait until all observables complete" অপারেটর। প্রতিটি Observable সম্পূর্ণ হলে তাদের সর্বশেষ মানগুলো একটি অ্যারে আকারে একত্রিত করে।
বৈশিষ্ট্য:
- সবগুলো Observable সম্পূর্ণ না হওয়া পর্যন্ত অপেক্ষা করা।
- শেষ মান গুলি একত্রিত করে একটি অ্যারেতে পাঠানো হয়।
উদাহরণ:
import { forkJoin, of } from 'rxjs';
const observable1 = of('A');
const observable2 = of(1);
const observable3 = of(true);
forkJoin([observable1, observable2, observable3]).subscribe(([val1, val2, val3]) => {
console.log(`ForkJoin values: ${val1}, ${val2}, ${val3}`);
});
আউটপুট:
ForkJoin values: A, 1, true
এখানে, forkJoin() সমস্ত Observable সম্পূর্ণ হলে তাদের সর্বশেষ মানগুলো একত্রিত করে পাঠিয়েছে।
6. withLatestFrom()
withLatestFrom() অপারেটরটি একটি Observable এর মানের সাথে অন্য একটি Observable-এর সর্বশেষ মানকে সংযুক্ত করে। এটি দুটি Observable থেকে ডেটা একত্রিত করে এবং প্রথম Observable এর ডেটাকে সেকেন্ড Observable-এর সর্বশেষ মানের সাথে পাঠায়।
বৈশিষ্ট্য:
- অন্য Observable এর সর্বশেষ মান ব্যবহার করে বর্তমান Observable এর মান প্রক্রিয়া করা।
উদাহরণ:
import { interval } from 'rxjs';
import { withLatestFrom, map } from 'rxjs/operators';
const source = interval(1000); // প্রতি সেকেন্ডে একটি মান দিবে
const latest = interval(2000); // প্রতি ২ সেকেন্ডে একটি মান দিবে
source.pipe(
withLatestFrom(latest),
map(([sourceVal, latestVal]) => `${sourceVal} with latest ${latestVal}`)
).subscribe(console.log);
আউটপুট:
0 with latest 0
1 with latest 0
2 with latest 1
3 with latest 1
4 with latest 2
এখানে, withLatestFrom() প্রথম Observable থেকে মান নেয়ার পাশাপাশি অন্য Observable এর সর্বশেষ মান নিয়ে একত্রিত করছে।
সারাংশ
RxJS-এ Combination Techniques বা combining operators ব্যবহার করে আপনি বিভিন্ন স্ট্রিম একত্রিত করতে পারেন। এই অপারেটরগুলো আপনাকে একাধিক স্ট্রিমের মধ্যে সম্পর্ক স্থাপন করতে সাহায্য করে, যেমন:
- combineLatest(): সব Observable-এর সর্বশেষ মান একত্রিত করে।
- merge(): সব Observable কে একযোগে (concurrently) সাবস্ক্রাইব করে।
- concat(): Observable গুলো সিরিয়াল অর্ডারে একত্রিত করে।
- zip(): Observable গুলোর মান জোড়া (pair) আকারে পাঠায়।
- forkJoin(): সব Observable সম্পূর্ণ হওয়ার পর তাদের সর্বশেষ মান একত্রিত করে।
- withLatestFrom(): অন্য Observable এর সর্বশেষ মান নিয়ে মূল Observable এর মান সংযুক্ত করে।
এই অপারেটরগুলোর মাধ্যমে আপনি ডাইনামিক ডেটা স্ট্রিমের সাথে আরও সহজে কাজ করতে পারেন এবং বিভিন্ন ধরনের অ্যাসিঙ্ক্রোনাস বা ইভেন্ট-ভিত্তিক ডেটা স্ট্রিমের সাথে সমন্বয় করতে পারেন।
Read more
