Combination Operators

Combination Operators হল RxJS-এর এমন অপারেটরস, যা একাধিক Observables (ডেটা স্ট্রিম) কে একত্রিত করে বা তাদের মধ্যে সম্পর্ক স্থাপন করে। এই অপারেটরগুলির মাধ্যমে আপনি বিভিন্ন সোর্স থেকে আসা ডেটা স্ট্রিমগুলিকে একসাথে ম্যানিপুলেট করতে পারেন এবং একই সময় একাধিক স্ট্রিমের ডেটা নিয়ে কাজ করতে পারেন।

RxJS এ Combination Operators এর মাধ্যমে আপনি একাধিক Observables বা ইভেন্টগুলোর মধ্যে ডেটা শেয়ার করতে পারেন, তাদের সমন্বয় করতে পারেন বা একসাথে প্রসেস করতে পারেন।

কিছু সাধারণ Combination Operators

1. merge()

   merge() অপারেটরটি একাধিক Observables কে একত্রিত করে। এটি সমস্ত স্ট্রিমকে সমান্তরালে মিশিয়ে দেয় এবং যে কোন Observable থেকে যে কোন মান আসলে তা সাবস্ক্রাইবারকে প্রদান করে।

   merge() ব্যবহার করলে একাধিক Observable এর মান একসাথে সাবস্ক্রাইব করা সম্ভব হয় এবং এরা একই সময় চলে, কিন্তু স্ট্রিমগুলোর মধ্যে কোনো নির্দিষ্ট অর্ডার থাকে না।

   উদাহরণ:

   import { of, interval } from 'rxjs';
   import { merge } from 'rxjs/operators';
   
   const observable1 = of('A', 'B', 'C');
   const observable2 = interval(1000); // প্রতি ১ সেকেন্ডে সংখ্যা পাঠাবে
   
   observable1.pipe(
     merge(observable2)
   ).subscribe(console.log);
   

   এখানে, merge() অপারেটরটি observable1 এবং observable2 কে একত্রিত করেছে। observable1 স্ট্রিম থেকে আগেই মানগুলি পাঠানো হবে, আর observable2 প্রতি সেকেন্ডে একটি মান পাঠাবে।

   আউটপুট হতে পারে:

   A
   0
   B
   1
   C
   2
   

2. concat()

   concat() অপারেটরটি একাধিক Observables কে সিরিয়ালি (sequentially) একত্রিত করে। এটি প্রথম Observable এর সমস্ত মান সাবস্ক্রাইব করার পর দ্বিতীয় Observable থেকে ডেটা পাঠানো শুরু করবে এবং এর পরবর্তী পর্যায়েও একইভাবে একে একে মান পাঠাবে। অর্থাৎ, একে একে সবগুলো Observable এর মান সাবস্ক্রাইব করা হয়।

   উদাহরণ:

   import { of, interval } from 'rxjs';
   import { concat } from 'rxjs/operators';
   
   const observable1 = of('A', 'B', 'C');
   const observable2 = of('D', 'E', 'F');
   
   observable1.pipe(
     concat(observable2)
   ).subscribe(console.log);
   

   এখানে, concat() অপারেটরটি observable1 থেকে সমস্ত মান প্রথমে পাঠাবে এবং তারপর observable2 থেকে মান পাঠাবে। আউটপুট হবে:

   A
   B
   C
   D
   E
   F
   

3. combineLatest()

   combineLatest() অপারেটরটি একাধিক Observable কে একত্রিত করে এবং তাদের মধ্যে সর্বশেষ পাওয়া মানকে সাবস্ক্রাইবারের কাছে পাঠায়। এটি একটি Observable থেকে মান পেলে, অন্যান্য Observable থেকেও সর্বশেষ মান নিয়ে আসে এবং একত্রিত করে নতুন একটি আউটপুট প্রদান করে।

   উদাহরণ:

   import { of, interval } from 'rxjs';
   import { combineLatest } from 'rxjs/operators';
   
   const observable1 = interval(1000); // প্রতি ১ সেকেন্ডে সংখ্যা পাঠাবে
   const observable2 = of('A', 'B', 'C'); // কিছু স্ট্যাটিক মান
   
   observable1.pipe(
     combineLatest(observable2)
   ).subscribe(console.log);
   

   এখানে, combineLatest() প্রতিটি Observable থেকে সর্বশেষ মান নিয়ে আসবে এবং তাদের একত্রিত করবে। আউটপুট হবে:

   [0, 'C']
   [1, 'C']
   [2, 'C']
   [3, 'C']
   

4. zip()

   zip() অপারেটরটি একাধিক Observables কে একত্রিত করে, তবে এটি একসাথে মানগুলো পাঠায়। এটি সোজা ভাষায় বললে, বিভিন্ন Observable থেকে একে একে মান নিয়ে আসে এবং তাদের একটি টিউপল (tuple) হিসাবে পাঠায়। সব Observables এর জন্য সমান সংখ্যা মান আসা পর্যন্ত এটি অপেক্ষা করে।

   উদাহরণ:

   import { of } from 'rxjs';
   import { zip } from 'rxjs/operators';
   
   const observable1 = of('A', 'B', 'C');
   const observable2 = of(1, 2, 3);
   
   observable1.pipe(
     zip(observable2)
   ).subscribe(console.log);
   

   এখানে, zip() অপারেটরটি প্রথম Observable থেকে মান নেয় এবং তারপর একই ইনডেক্সে থাকা দ্বিতীয় Observable এর মান নেয়, এবং একটি জোড়া হিসেবে সাবস্ক্রাইবারকে প্রদান করে। আউটপুট হবে:

   ['A', 1]
   ['B', 2]
   ['C', 3]
   

5. withLatestFrom()

   withLatestFrom() অপারেটরটি একটি Observable এর মানের সাথে অন্য Observable থেকে সর্বশেষ মান যুক্ত করে। এটি প্রধানত দুটি Observable এর মানকে একত্রিত করতে ব্যবহৃত হয় যেখানে একটি Observable অন্যটির মানের সাথে মিলিত হয়।

   উদাহরণ:

   import { of, interval } from 'rxjs';
   import { withLatestFrom } from 'rxjs/operators';
   
   const observable1 = interval(1000); // প্রতি ১ সেকেন্ডে সংখ্যা পাঠাবে
   const observable2 = of('A', 'B', 'C'); // কিছু স্ট্যাটিক মান
   
   observable1.pipe(
     withLatestFrom(observable2)
   ).subscribe(console.log);
   

   এখানে, withLatestFrom() অপারেটরটি observable1 থেকে মান নিয়ে observable2 এর সর্বশেষ মানের সাথে একত্রিত করবে। আউটপুট হতে পারে:

   [0, 'C']
   [1, 'C']
   [2, 'C']
   [3, 'C']
   

সারাংশ

Combination Operators হল RxJS-এর এমন অপারেটরস, যা একাধিক Observables এর ডেটা স্ট্রিমকে একত্রিত করে বা তাদের মধ্যে সম্পর্ক স্থাপন করে। এগুলি যেমন merge, concat, combineLatest, zip, এবং withLatestFrom স্ট্রিমগুলির মধ্যে সমন্বয় সাধন করে এবং একাধিক সোর্স থেকে আসা ডেটা সঠিকভাবে প্রসেস করতে সহায়তা করে। Combination Operators ব্যবহার করে আপনি বিভিন্ন অ্যাসিঙ্ক্রোনাস ডেটার মধ্যে সম্পর্ক স্থাপন করতে পারেন এবং স্ট্রিমের ডেটাকে খুব সহজে ম্যানিপুলেট করতে পারেন।

RxJS-এর মধ্যে merge(), concat(), এবং combineLatest() অপারেটরগুলো একাধিক Observable-কে একত্রিত করতে ব্যবহৃত হয়। এগুলি মূলত ডেটা স্ট্রিম বা ইভেন্টকে একত্রিত করার জন্য ব্যবহার হয়, তবে প্রত্যেকটি অপারেটরের আচরণ এবং ব্যবহারের ধরন ভিন্ন। এই অপারেটরগুলোর মাধ্যমে আপনি একাধিক অ্যাসিঙ্ক্রোনাস স্ট্রিমের ডেটা সমন্বিত করতে পারেন, এবং তাদের মধ্যে সম্পর্ক বা অর্ডার কিভাবে থাকবে তা নিয়ন্ত্রণ করতে পারেন।

এখানে merge(), concat(), এবং combineLatest() এর বিস্তারিত ব্যবহার ব্যাখ্যা করা হলো।

1. merge() অপারেটর

merge() অপারেটরটি একাধিক Observable কে একত্রিত করে এবং তাদের ডেটাকে একত্রিত করে সাবস্ক্রাইবারদের কাছে পাঠায়। এটি তাদের মধ্যে ডেটার আর্ডার বজায় না রেখে, তাদের যতটুকু দ্রুত সম্ভব ডেটা পাঠিয়ে দেয়। এটি ব্যবহার করা হয় যখন আপনি চান একাধিক স্ট্রিমের ডেটা একই সময়ে প্রসেস করতে।

উদাহরণ:

import { of, merge } from 'rxjs';

// দুটি আলাদা Observable তৈরি করা
const first$ = of(1, 2, 3);
const second$ = of(4, 5, 6);

merge(first$, second$).subscribe(value => {
  console.log(value);  // এটি 1, 2, 3, 4, 5, 6 প্রিন্ট করবে
});

ব্যাখ্যা:
এখানে, merge() অপারেটর দুটি আলাদা Observable (first$ এবং second$) একত্রিত করেছে এবং তাদের ডেটা একই সময়ে সাবস্ক্রাইবারদের কাছে পাঠিয়েছে। ডেটার অর্ডার গুরুত্বপূর্ণ নয়, যেহেতু একসাথে স্ট্রিম করা হচ্ছে।

2. concat() অপারেটর

concat() অপারেটরটি একাধিক Observable কে একত্রিত করে এবং একটির পর একটি ডেটা পাঠায়। এর মানে হল যে প্রথম Observable পুরোপুরি শেষ হওয়ার পর, দ্বিতীয় Observable এর ডেটা পাঠানো শুরু হবে, তারপর তৃতীয়, চতুর্থ, ইত্যাদি। এটি ব্যবহার করা হয় যখন আপনি চান একটি স্ট্রিম পুরোপুরি শেষ হওয়ার পর অন্য স্ট্রিম শুরু হবে।

উদাহরণ:

import { of, concat } from 'rxjs';

// দুটি আলাদা Observable তৈরি করা
const first$ = of(1, 2, 3);
const second$ = of(4, 5, 6);

concat(first$, second$).subscribe(value => {
  console.log(value);  // এটি 1, 2, 3, তারপর 4, 5, 6 প্রিন্ট করবে
});

ব্যাখ্যা:
এখানে, concat() অপারেটরটি প্রথম Observable (first$) এর ডেটা পাঠানোর পর, তার পরে দ্বিতীয় Observable (second$) এর ডেটা পাঠাচ্ছে। এই ক্ষেত্রে, ডেটার অর্ডার বজায় রাখা হয় এবং একটির পর একটি স্ট্রিম পাঠানো হয়।

3. combineLatest() অপারেটর

combineLatest() অপারেটরটি একাধিক Observable কে একত্রিত করে এবং যখনই কোনো Observable এর মান পরিবর্তিত হয়, তখন সব Observable এর সর্বশেষ মানগুলোকে একত্রিত করে সাবস্ক্রাইবারদের কাছে পাঠায়। এটি তখনই ডেটা পাঠায় যখন কোনো একটি Observable নতুন মান প্রদান করে, কিন্তু যতক্ষণ না সমস্ত Observable অন্তত একটি মান প্রদান করে, ততক্ষণ এটি কিছু পাঠায় না।

উদাহরণ:

import { of, combineLatest } from 'rxjs';

// দুটি Observable তৈরি করা
const first$ = of(1, 2, 3);
const second$ = of('A', 'B', 'C');

combineLatest([first$, second$]).subscribe(([first, second]) => {
  console.log(first, second);  // এটি সর্বশেষ মানের যোড়গুলো প্রিন্ট করবে
});

ব্যাখ্যা:
এখানে, combineLatest() অপারেটরটি first$ এবং second$ দুটি Observable এর সর্বশেষ মান একত্রিত করেছে এবং সাবস্ক্রাইবারকে পাঠিয়েছে। এটি তখনই ডেটা পাঠায় যখন কোনো একটি Observable একটি নতুন মান প্রদান করে, এবং সমস্ত Observable থেকে সর্বশেষ মানের একটি জোড়া পাঠায়।

আউটপুট:

3 C

এখানে, combineLatest() প্রথমে first$ এর সব মান নেয়, তারপর second$ থেকে সর্বশেষ মান নেয় এবং তাদের একত্রিত করে পাঠায়।

সারাংশ

এই অপারেটরগুলি RxJS-এ একাধিক Observable এর ডেটা স্ট্রিমকে একত্রিত করার জন্য ব্যবহৃত হয় এবং ডেটা সংগ্রহের প্রক্রিয়া সহজ করে তোলে।

RxJS একটি শক্তিশালী লাইব্রেরি, যা অ্যাসিঙ্ক্রোনাস ডেটা স্ট্রিম এবং ইভেন্টগুলির উপর কার্যকরীভাবে কাজ করতে সক্ষম। startWith() এবং withLatestFrom() দুটি অপারেটর, যা ডেটা সমন্বয় বা সমন্বিতভাবে একাধিক স্ট্রিম থেকে তথ্য সংগ্রহ এবং ব্যবহারের জন্য ব্যবহৃত হয়। এগুলি বিভিন্ন স্ট্রিমের মধ্যে সম্পর্ক স্থাপন এবং ডেটার মান একত্রিত করতে সাহায্য করে। এই অপারেটরগুলি বিশেষ করে সেই পরিস্থিতিতে কার্যকর যেখানে একাধিক স্ট্রিমের ডেটা একসাথে প্রয়োজন হয়।

startWith() মেথড

startWith() অপারেটরটি ব্যবহৃত হয় যখন আপনি একটি Observable থেকে ডেটা স্ট্রিম প্রাপ্তির আগে একটি ডিফল্ট মান বা শুরু মান (initial value) প্রদান করতে চান। এটি মূলত ডেটার আগেই নির্দিষ্ট একটি মান পাঠানোর জন্য ব্যবহৃত হয়, যাতে সাবস্ক্রিপশন শুরু হওয়ার আগে সেটি উপলব্ধ হয়।

startWith() এর বৈশিষ্ট্য:

• স্ট্রিমের আগে একটি মান প্রদান করা: এটা মূল স্ট্রিমের আগেই ডিফল্ট মান প্রেরণ করে, যাতে আপনি শুরুতে কিছু মান পেতে পারেন।
• অ্যাসিঙ্ক্রোনাস স্ট্রিমের আগে একটি প্রাথমিক মান: এটি বিশেষভাবে ব্যবহারী যখন ডেটা স্ট্রিমটি পরে আসবে এবং আগে কোনো মান প্রয়োজন হবে।

উদাহরণ:

import { of } from 'rxjs';
import { startWith } from 'rxjs/operators';

const observable = of(2, 3, 4).pipe(
  startWith(1)  // স্ট্রিম শুরু হওয়ার আগে ১ পাঠানো হবে
);

observable.subscribe(console.log);

এখানে, startWith(1) অপারেটরটি ব্যবহার করে Observable ১ মান পাঠাবে, তারপর ২, ৩, ৪ মানগুলি পাঠানো হবে। আউটপুট হবে:

1
2
3
4

withLatestFrom() মেথড

withLatestFrom() অপারেটরটি একটি রিয়্যাক্টিভ ডেটা সমন্বয়ের জন্য ব্যবহৃত হয়, যেখানে এটি দুটি Observable এর মধ্যে ডেটা সম্পর্ক স্থাপন করতে সাহায্য করে। এটি মূলত প্রথম Observable থেকে ডেটা গ্রহণ করে এবং তারপরে দ্বিতীয় Observable এর সর্বশেষ মান গ্রহণ করে। এটি বিশেষভাবে তখন ব্যবহার করা হয় যখন একাধিক স্ট্রিম থেকে সর্বশেষ মানগুলির সাথে কাজ করা প্রয়োজন।

withLatestFrom() এর বৈশিষ্ট্য:

• একাধিক স্ট্রিমের সাথে সম্পর্ক স্থাপন: এটি প্রথম Observable এর প্রতিটি মানের সাথে দ্বিতীয় Observable এর সর্বশেষ মান সংযুক্ত করে।
• ডেটার সমন্বয়: একাধিক Observable এর মানগুলির মধ্যে সম্পর্ক স্থাপন করতে সাহায্য করে, যেখানে একটি স্ট্রিমের মান অন্যটির সাথে সম্পর্কিত হয়।

উদাহরণ:

import { interval } from 'rxjs';
import { map, withLatestFrom } from 'rxjs/operators';

const observable1 = interval(1000); // প্রতি সেকেন্ডে একটি মান
const observable2 = interval(1500); // প্রতি ১.৫ সেকেন্ডে একটি মান

observable1.pipe(
  withLatestFrom(observable2)  // observable2 থেকে সর্বশেষ মান গ্রহণ করবে
).subscribe(([val1, val2]) => {
  console.log(`observable1: ${val1}, observable2: ${val2}`);
});

এখানে, observable1 প্রতি সেকেন্ডে একটি নতুন মান তৈরি করছে এবং observable2 প্রতি ১.৫ সেকেন্ডে একটি মান তৈরি করছে। withLatestFrom() ব্যবহার করে observable1 থেকে প্রতিটি মানের সাথে observable2 এর সর্বশেষ মান যুক্ত করা হচ্ছে।

আউটপুট (প্রথম ৫টি):

observable1: 0, observable2: 0
observable1: 1, observable2: 1
observable1: 2, observable2: 1
observable1: 3, observable2: 2
observable1: 4, observable2: 2

এখানে দেখা যাচ্ছে, observable1 এর মান প্রতি সেকেন্ডে পরিবর্তিত হলেও, observable2 থেকে সর্বশেষ মানটি সবসময় যুক্ত হচ্ছে।

startWith() এবং withLatestFrom() এর সমন্বয়

আপনি যদি startWith() এবং withLatestFrom() মেথড দুটি একসাথে ব্যবহার করতে চান, তখন আপনি প্রথম Observable থেকে একটি ডিফল্ট মান গ্রহণ করতে পারেন এবং পরবর্তী Observable থেকে সর্বশেষ মান নেওয়ার মাধ্যমে তাদের সমন্বয় করতে পারবেন।

উদাহরণ:

import { interval, of } from 'rxjs';
import { startWith, withLatestFrom, map } from 'rxjs/operators';

const observable1 = interval(1000); // প্রতি সেকেন্ডে একটি মান
const observable2 = of('A', 'B', 'C'); // একটি সিঙ্ক্রোনাস স্ট্রিম

observable1.pipe(
  startWith(0), // স্ট্রিমের আগে ০ পাঠাবে
  withLatestFrom(observable2)  // observable2 থেকে সর্বশেষ মান গ্রহণ করবে
).subscribe(([val1, val2]) => {
  console.log(`observable1: ${val1}, observable2: ${val2}`);
});

এখানে, observable1 এর সেকেন্ডে একটি নতুন মান তৈরি হচ্ছে এবং observable2 সিঙ্ক্রোনাসভাবে কিছু মান প্রদান করছে। প্রথমে observable1 এর আগে startWith(0) ব্যবহার করে 0 পাঠানো হচ্ছে এবং পরে observable2 থেকে সর্বশেষ মান নেয়া হচ্ছে।

আউটপুট হবে:

observable1: 0, observable2: A
observable1: 1, observable2: B
observable1: 2, observable2: C

সারাংশ

• startWith(): এটি একটি প্রাথমিক মান প্রদান করতে ব্যবহৃত হয় যা একটি Observable স্ট্রিমের আগে পাঠানো হয়। এটি অ্যাসিঙ্ক্রোনাস স্ট্রিমে আগে কিছু মান বা স্টেট পাঠানোর জন্য উপযোগী।
• withLatestFrom(): এটি দুটি Observable এর ডেটা সমন্বয় করার জন্য ব্যবহৃত হয়, যেখানে একটি Observable এর মানের সাথে অপর Observable এর সর্বশেষ মান যুক্ত করা হয়।

এগুলি একসাথে ব্যবহার করলে আপনি একাধিক স্ট্রিমের ডেটার মধ্যে সম্পর্ক স্থাপন এবং তাদের সংমিশ্রণ করতে পারবেন।

RxJS-এ forkJoin() এবং zip() হল দুইটি অপারেটর যা একাধিক Observable এর মানগুলোকে একত্রিত করতে ব্যবহৃত হয়। তারা আপনাকে একাধিক স্ট্রিম (Observables) থেকে মান সংগ্রহ করতে এবং সেগুলোকে একত্রিত করে একটি নির্দিষ্ট আউটপুট প্রদান করতে সহায়তা করে। তবে, তাদের আচরণে কিছু পার্থক্য রয়েছে এবং এটি নির্ভর করে আপনি কোন পরিস্থিতিতে কোন অপারেটরটি ব্যবহার করবেন।

1. forkJoin()

forkJoin() অপারেটরটি একাধিক Observable এর মানগুলিকে একত্রিত করে, তবে এটি শুধুমাত্র তখনই কাজ করবে যখন সমস্ত Observables complete হয়ে যাবে। এটি সকল Observable এর শেষ মান গ্রহণ করবে এবং সেই মানগুলিকে একটি অ্যারেতে বা অবজেক্টে একত্রিত করে একটি একক আউটপুট প্রদান করবে। যদি কোন Observable ভুল (error) হয়, তবে forkJoin() সেই Observable এর মান গ্রহণ করবে না এবং পুরো স্ট্রিমটি error হবে।

বৈশিষ্ট্য:

• Complete হওয়া আবশ্যক: সমস্ত Observable এর complete হওয়ার পরই forkJoin() সেই সমস্ত Observable এর শেষ মানগুলো একত্রিত করবে।
• একক মান: প্রতিটি Observable এর শেষ মান সংগ্রহ করা হয় এবং তা একসাথে পাঠানো হয়।

উদাহরণ:

import { forkJoin, of, timer } from 'rxjs';
import { delay } from 'rxjs/operators';

// তিনটি Observable তৈরি
const observable1 = of('First Value').pipe(delay(1000));
const observable2 = of('Second Value').pipe(delay(2000));
const observable3 = of('Third Value').pipe(delay(3000));

// forkJoin ব্যবহার করে একত্র করা
forkJoin([observable1, observable2, observable3]).subscribe({
  next: (values) => console.log('All values:', values),
  complete: () => console.log('All observables completed!')
});

আউটপুট:

All values: [ 'First Value', 'Second Value', 'Third Value' ]
All observables completed!

এখানে, forkJoin() সমস্ত Observable এর শেষ মানগুলিকে একত্রিত করেছে এবং একটি অ্যারে হিসেবে সাবস্ক্রাইবারকে পাঠিয়েছে। সমস্ত Observable এর complete হওয়া পর্যন্ত এটি অপেক্ষা করেছে।

2. zip()

zip() অপারেটরটি একাধিক Observable এর মানগুলোকে সমান্তরালভাবে একত্রিত করে। এটি সমস্ত Observable এর মান একত্র করে একটি নতুন অবজেক্ট বা অ্যারে তৈরি করে, তবে এটি প্রত্যেক Observable থেকে একটি করে মান একত্রিত করে পাঠায়। এটি সমস্ত Observable থেকে একটি করে মান গ্রহণ করবে এবং তারপর সেই মানগুলোকে একটি নতুন অ্যারে বা অবজেক্টে একত্রিত করবে।

বৈশিষ্ট্য:

• একযোগে মান গ্রহণ: একাধিক Observable থেকে একটি করে মান সংগ্রহ করা হয়।
• একই সংখ্যক মান: zip() তখনই একটি আউটপুট প্রদান করবে যখন সমস্ত Observable থেকে একটি করে মান পাওয়া যাবে। যদি কোনো Observable মান না পাঠায়, তবে zip() কোনো আউটপুট প্রদান করবে না।

উদাহরণ:

import { zip, of, timer } from 'rxjs';
import { delay } from 'rxjs/operators';

// তিনটি Observable তৈরি
const observable1 = of('First Value').pipe(delay(1000));
const observable2 = of('Second Value').pipe(delay(2000));
const observable3 = of('Third Value').pipe(delay(3000));

// zip ব্যবহার করে একত্র করা
zip(observable1, observable2, observable3).subscribe({
  next: (values) => console.log('Zipped values:', values),
  complete: () => console.log('All observables zipped!')
});

আউটপুট:

Zipped values: [ 'First Value', 'Second Value', 'Third Value' ]
All observables zipped!

এখানে, zip() প্রত্যেক Observable থেকে একসাথে একটি মান গ্রহণ করেছে এবং একটি অ্যারে তৈরি করেছে। সব Observable থেকে একটি মান প্রাপ্ত হওয়া পর্যন্ত এটি অপেক্ষা করেছে।

forkJoin() এবং zip() এর মধ্যে পার্থক্য

সারাংশ

• forkJoin() সব Observable এর শেষ মান একত্রিত করে এবং তখনই আউটপুট দেয় যখন সমস্ত Observable সম্পূর্ণ (complete) হয়ে যায়। এটি এক ধরনের নিশ্চিতভাবে একত্রিত করা প্রক্রিয়া, যেখানে সমস্ত স্ট্রিমের শেষ মান কেবলমাত্র ফেরত দেয়া হয়।
• zip() একযোগভাবে মান সংগ্রহ করে এবং প্রত্যেক Observable থেকে একটি করে মান নেয়। এটি সমান্তরালভাবে একাধিক Observable এর মান একত্র করে, এবং যখন সমস্ত Observable থেকে একটি করে মান পাওয়া যায়, তখন এটি আউটপুট প্রদান করে।

এই দুটি অপারেটরের মাধ্যমে আপনি একাধিক Observable এর মানগুলোকে একত্রিত করে প্রয়োজন অনুযায়ী ব্যবহার করতে পারেন।

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 এর মান সংযুক্ত করে।

এই অপারেটরগুলোর মাধ্যমে আপনি ডাইনামিক ডেটা স্ট্রিমের সাথে আরও সহজে কাজ করতে পারেন এবং বিভিন্ন ধরনের অ্যাসিঙ্ক্রোনাস বা ইভেন্ট-ভিত্তিক ডেটা স্ট্রিমের সাথে সমন্বয় করতে পারেন।