Observables কে Combine করার জন্য বিভিন্ন Operators

RxJava-তে Observables-কে combine বা একত্রিত করার জন্য বিভিন্ন operators ব্যবহার করা হয়। এই operators বিভিন্ন data streams একত্রিত করে বা তাদের সম্পর্কিত ডেটা প্রসেস করতে সাহায্য করে।

Observables Combine করার জন্য গুরুত্বপূর্ণ Operators

1. merge()

• ব্যবহার:
  • একাধিক Observables-এর ডেটা একত্রিত করে একটি single Observable তৈরি করে।
  • Observables-এর ডেটা sequential বা parallel যে কোনোভাবে emit হতে পারে।
• উদাহরণ:

import io.reactivex.Observable;

public class MergeExample {
    public static void main(String[] args) {
        Observable<String> observable1 = Observable.just("A", "B", "C");
        Observable<String> observable2 = Observable.just("1", "2", "3");

        Observable.merge(observable1, observable2)
            .subscribe(item -> System.out.println("Received: " + item));
    }
}

আউটপুট:

Received: A  
Received: B  
Received: C  
Received: 1  
Received: 2  
Received: 3  

2. zip()

• ব্যবহার:
  • দুটি বা আরও বেশি Observables থেকে corresponding items একত্রিত করে একটি নতুন ডেটা তৈরি করে।
  • এটি index-এর ভিত্তিতে Observables-এর ডেটা মিলিত করে।
• উদাহরণ:

import io.reactivex.Observable;

public class ZipExample {
    public static void main(String[] args) {
        Observable<String> observable1 = Observable.just("A", "B", "C");
        Observable<String> observable2 = Observable.just("1", "2", "3");

        Observable.zip(observable1, observable2, (item1, item2) -> item1 + item2)
            .subscribe(item -> System.out.println("Received: " + item));
    }
}

আউটপুট:

Received: A1  
Received: B2  
Received: C3  

3. concat()

• ব্যবহার:
  • Observables-এর ডেটা sequentially emit করে।
  • একটি Observable শেষ হওয়ার পর আরেকটি Observable এর ডেটা emit শুরু হয়।
• উদাহরণ:

import io.reactivex.Observable;

public class ConcatExample {
    public static void main(String[] args) {
        Observable<String> observable1 = Observable.just("A", "B", "C");
        Observable<String> observable2 = Observable.just("1", "2", "3");

        Observable.concat(observable1, observable2)
            .subscribe(item -> System.out.println("Received: " + item));
    }
}

আউটপুট:

Received: A  
Received: B  
Received: C  
Received: 1  
Received: 2  
Received: 3  

4. combineLatest()

• ব্যবহার:
  • সর্বশেষ emit হওয়া ডেটাগুলো একত্রিত করে একটি নতুন Observable তৈরি করে।
  • Observables-এর ডেটা স্ট্রিম asynchronous হলে এটি উপকারী।
• উদাহরণ:

import io.reactivex.Observable;
import java.util.concurrent.TimeUnit;

public class CombineLatestExample {
    public static void main(String[] args) throws InterruptedException {
        Observable<String> observable1 = Observable.interval(1, TimeUnit.SECONDS)
            .map(tick -> "A" + tick);
        Observable<String> observable2 = Observable.interval(2, TimeUnit.SECONDS)
            .map(tick -> "B" + tick);

        Observable.combineLatest(observable1, observable2, (item1, item2) -> item1 + "-" + item2)
            .subscribe(item -> System.out.println("Received: " + item));

        Thread.sleep(5000); // Sleep to allow the example to run
    }
}

আউটপুট (সম্ভাব্য):

Received: A0-B0  
Received: A1-B0  
Received: A1-B1  
Received: A2-B1  
Received: A3-B1  

5. flatMap()

• ব্যবহার:
  • একটি Observable-এর প্রতিটি ডেটা আইটেম থেকে নতুন Observables তৈরি করে এবং সেগুলো merge করে।
  • এটি asynchronous ডেটা স্ট্রিমের জন্য উপযুক্ত।
• উদাহরণ:

import io.reactivex.Observable;

public class FlatMapExample {
    public static void main(String[] args) {
        Observable<String> observable = Observable.just("A", "B", "C");

        observable
            .flatMap(item -> Observable.just(item + "1", item + "2"))
            .subscribe(result -> System.out.println("Received: " + result));
    }
}

আউটপুট:

Received: A1  
Received: A2  
Received: B1  
Received: B2  
Received: C1  
Received: C2  

6. switchMap()

• ব্যবহার:
  • একটি Observable-এর প্রতিটি আইটেম থেকে নতুন Observable তৈরি করে, কিন্তু পুরনো Observable-এর ডেটা discard করে।
  • সাধারণত asynchronous calls-এর জন্য ব্যবহৃত হয়।
• উদাহরণ:

import io.reactivex.Observable;

public class SwitchMapExample {
    public static void main(String[] args) throws InterruptedException {
        Observable<String> observable = Observable.just("A", "B", "C");

        observable
            .switchMap(item -> Observable.just(item + "1").delay(1, TimeUnit.SECONDS))
            .subscribe(result -> System.out.println("Received: " + result));

        Thread.sleep(4000); // Sleep to allow the example to run
    }
}

আউটপুট (সম্ভাব্য):

Received: C1

উপসংহার

Observables-কে combine করার জন্য বিভিন্ন operators ব্যবহার করা হয়, এবং প্রতিটি operator-এর নিজস্ব ব্যবহারক্ষেত্র রয়েছে।

• merge(): Parallel streams combine করার জন্য।
• concat(): Sequential streams combine করার জন্য।
• zip(): Corresponding items combine করার জন্য।
• combineLatest(): সর্বশেষ emitted ডেটা combine করার জন্য।
• flatMap(): Nested asynchronous operations-এর জন্য।
• switchMap(): নতুন Observable-এ switch করার জন্য।

যথাযথ operator বেছে নেওয়া আপনার ডেটা স্ট্রিমের প্রয়োজনের উপর নির্ভর করে।