برنامه نویسی واکنشی (Reactive Programming) چیست ؟

یک مثال ساده

Reactive Extensions

RxJava چیست ؟

RxAndroid  چیست ؟

()io:

()mainThread:

()newThread:

()computation

()single:

()immediate:

()trampoline:

()from:

مبانی RxJava : Observable، Observer

نمونه پایه‌ای از RxJava

اضافه کردن وابستگی‌ها (Dependencies)

// RxJava
implementation 'io.reactivex.rxjava2:rxjava:2.1.9'
// RxAndroid
implementation 'io.reactivex.rxjava2:rxandroid:2.0.1'

مراحل اساسی

Observable<String> animalsObservable = Observable.just("Eagle", "Bee", "Lion", "Dog", "Wolf");

• ()onSubscribe : وقتی Observer عضو Observable شود، این متد فراخوانی می‌شود.
• ()onNext : این متد هنگامی که Observable شروع به انتشار اطلاعات می‌کند، فراخوانی می‌شود.
• ()onError : در صورت بروز هرگونه خطا، متد onError () فراخوانی می‌شود.
• ()onComplete : هنگامی که یک Observable کننده انتشار همه موارد را انجام می‌دهد،()onComplete فراخوانی می‌شود.

Observer<String> animalsObserver = getAnimalsObserver();
private Observer<String> getAnimalsObserver() {
        return new Observer<String>() {
            @Override
            public void onSubscribe(Disposable d) {
                Log.d(TAG, "onSubscribe");
            }
            @Override
            public void onNext(String s) {
                Log.d(TAG, "Name: " + s);
            }
            @Override
            public void onError(Throwable e) {
                Log.e(TAG, "onError: " + e.getMessage());
            }
            @Override
            public void onComplete() {
                Log.d(TAG, "All items are emitted!");
            }
        };
    }

• (()SubscribOn (Schedulers.io : این مورد به  Observable می‌گوید که وظیفه خود را بر روی یک                     Background Thread اجرا کند.
• (()observeOn (AndroidSchedulers.mainThread : این مورد به Observer می‌گوید که داده‌های مربوط به Android Ui Thread را دریافت کند تا بتوانید هرگونه کار مرتبط با UI (رابط کاربری) را انجام دهید.

animalsObservable
        .subscribeOn(Schedulers.io())
        .observeOn(AndroidSchedulers.mainThread())
        .subscribe(animalsObserver);

اساس Observable و Observer

import android.support.v7.app.AppCompatActivity;
import android.os.Bundle;
import android.util.Log;
import info.androidhive.rxandroidexamples.R;
import io.reactivex.Observable;
import io.reactivex.Observer;
import io.reactivex.android.schedulers.AndroidSchedulers;
import io.reactivex.disposables.Disposable;
import io.reactivex.schedulers.Schedulers;
public class Example1Activity extends AppCompatActivity {
    /**
     * Basic Observable, Observer, Subscriber example
     * Observable emits list of animal names
     */
    private static final String TAG = Example1Activity.class.getSimpleName();
    @Override
    protected void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        setContentView(R.layout.activity_example1);
        // observable
        Observable<String> animalsObservable = getAnimalsObservable();
        // observer
        Observer<String> animalsObserver = getAnimalsObserver();
        // observer subscribing to observable
        animalsObservable
                .observeOn(Schedulers.io())
                .subscribeOn(AndroidSchedulers.mainThread())
                .subscribe(animalsObserver);
    }
    private Observer<String> getAnimalsObserver() {
        return new Observer<String>() {
            @Override
            public void onSubscribe(Disposable d) {
                Log.d(TAG, "onSubscribe");
            }
            @Override
            public void onNext(String s) {
                Log.d(TAG, "Name: " + s);
            }
            @Override
            public void onError(Throwable e) {
                Log.e(TAG, "onError: " + e.getMessage());
            }
            @Override
            public void onComplete() {
                Log.d(TAG, "All items are emitted!");
            }
        };
    }
    private Observable<String> getAnimalsObservable() {
        return Observable.just("Eagle", "Bee", "Lion", "Dog", "Wolf");
    }
}

معرفی Disposable

import android.os.Bundle;
import android.support.v7.app.AppCompatActivity;
import android.util.Log;
import info.androidhive.rxandroidexamples.R;
import io.reactivex.Observable;
import io.reactivex.Observer;
import io.reactivex.android.schedulers.AndroidSchedulers;
import io.reactivex.disposables.CompositeDisposable;
import io.reactivex.disposables.Disposable;
import io.reactivex.observers.DisposableObserver;
import io.reactivex.schedulers.Schedulers;
public class Example2Activity extends AppCompatActivity {
    /**
     * Basic Observable, Observer, Subscriber example
     * Observable emits list of animal names
     * You can see Disposable introduced in this example
     */
    private static final String TAG = Example2Activity.class.getSimpleName();
    private Disposable disposable;
    @Override
    protected void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        setContentView(R.layout.activity_example2);
        // observable
        Observable<String> animalsObservable = getAnimalsObservable();
        // observer
        Observer<String> animalsObserver = getAnimalsObserver();
        // observer subscribing to observable
        animalsObservable
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribeWith(animalsObserver);
    }
    private Observer<String> getAnimalsObserver() {
        return new Observer<String>() {
            @Override
            public void onSubscribe(Disposable d) {
                Log.d(TAG, "onSubscribe");
                disposable = d;
            }
            @Override
            public void onNext(String s) {
                Log.d(TAG, "Name: " + s);
            }
            @Override
            public void onError(Throwable e) {
                Log.e(TAG, "onError: " + e.getMessage());
            }
            @Override
            public void onComplete() {
                Log.d(TAG, "All items are emitted!");
            }
        };
    }
    private Observable<String> getAnimalsObservable() {
        return Observable.just("Eagle", "Bee", "Lion", "Dog", "Wolf");
    }
    @Override
    protected void onDestroy() {
        super.onDestroy();
        // don't send events once the activity is destroyed
        disposable.dispose();
    }
}

عملگرهای RxJava Operators) RxJava) 

•  Introduction
• Map
• Buffer
• Debounce
• ThrottleFirst
• FlatMap
• ConcatMap
• SwitchMap

Map

Buffer

Debounce

ThrottleFirst

FlatMap

ConcatMap

SwitchMap

Observers چندگانه و CompositDisposable

CompositeDisposable 

• AnimalObserver : از عملگر ()filter برای فیلتر کردن نام حیوانات با حرف "b" استفاده می‌شود.
• AnimalObserverAllCaps : عملگر ()filter برای فیلتر کردن نام حیوانات با شروع حرف "c" استفاده می‌شود. از نقشه عملگر map() بعدی برای تبدیل نام هر حیوانی به حروف بزرگ استفاده می‌شود. استفاده از چندین اپراتور روی یک observer  مفرد، زنجیره عملگرها نام (chaining of operators) دارد.

import android.os.Bundle;
import android.support.v7.app.AppCompatActivity;
import android.util.Log;
import info.androidhive.rxandroidexamples.R;
import io.reactivex.Observable;
import io.reactivex.android.schedulers.AndroidSchedulers;
import io.reactivex.disposables.CompositeDisposable;
import io.reactivex.functions.Function;
import io.reactivex.functions.Predicate;
import io.reactivex.observers.DisposableObserver;
import io.reactivex.schedulers.Schedulers;
public class Example4Activity extends AppCompatActivity {
    /**
     * Basic Observable, Observer, Subscriber example
     * Observable emits list of animal names
     * You can see filter() operator is used to filter out the
     * animal names that starts with letter `b`
     */
    private static final String TAG = Example4Activity.class.getSimpleName();
    private CompositeDisposable compositeDisposable = new CompositeDisposable();
    @Override
    protected void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        setContentView(R.layout.activity_example4);
        Observable<String> animalsObservable = getAnimalsObservable();
        DisposableObserver<String> animalsObserver = getAnimalsObserver();
        DisposableObserver<String> animalsObserverAllCaps = getAnimalsAllCapsObserver();
        /**
         * filter() is used to filter out the animal names starting with `b`
         * */
        compositeDisposable.add(
                animalsObservable
                        .subscribeOn(Schedulers.io())
                        .observeOn(AndroidSchedulers.mainThread())
                        .filter(new Predicate<String>() {
                            @Override
                            public boolean test(String s) throws Exception {
                                return s.toLowerCase().startsWith("b");
                            }
                        })
                        .subscribeWith(animalsObserver));
        /**
         * filter() is used to filter out the animal names starting with 'c'
         * map() is used to transform all the characters to UPPER case
         * */
        compositeDisposable.add(
                animalsObservable
                        .subscribeOn(Schedulers.io())
                        .observeOn(AndroidSchedulers.mainThread())
                        .filter(new Predicate<String>() {
                            @Override
                            public boolean test(String s) throws Exception {
                                return s.toLowerCase().startsWith("c");
                            }
                        })
                        .map(new Function<String, String>() {
                            @Override
                            public String apply(String s) throws Exception {
                                return s.toUpperCase();
                            }
                        })
                        .subscribeWith(animalsObserverAllCaps));
    }
    private DisposableObserver<String> getAnimalsObserver() {
        return new DisposableObserver<String>() {
            @Override
            public void onNext(String s) {
                Log.d(TAG, "Name: " + s);
            }
            @Override
            public void onError(Throwable e) {
                Log.e(TAG, "onError: " + e.getMessage());
            }
            @Override
            public void onComplete() {
                Log.d(TAG, "All items are emitted!");
            }
        };
    }
    private DisposableObserver<String> getAnimalsAllCapsObserver() {
        return new DisposableObserver<String>() {
            @Override
            public void onNext(String s) {
                Log.d(TAG, "Name: " + s);
            }
            @Override
            public void onError(Throwable e) {
                Log.e(TAG, "onError: " + e.getMessage());
            }
            @Override
            public void onComplete() {
                Log.d(TAG, "All items are emitted!");
            }
        };
    }
    private Observable<String> getAnimalsObservable() {
        return Observable.fromArray(
                "Ant", "Ape",
                "Bat", "Bee", "Bear", "Butterfly",
                "Cat", "Crab", "Cod",
                "Dog", "Dove",
                "Fox", "Frog");
    }
    @Override
    protected void onDestroy() {
        super.onDestroy();
        // don't send events once the activity is destroyed
        compositeDisposable.clear();
    }
}

Flowables

• Flowable چیست؟
• Backpressure
• Backpressure Strategies
• Flowables and Observables

1. Observable
2. Flowable

1. Observables are not backpressure-aware 
2. Flowables are backpressure-aware 

backpressure

• Hot sources: به این عنوان به شکل یک فشار نسبی فکر کنید. Observables بدون در نظر گرفتن اینکه آیا آن‌ها می‌توانند به کار خود ادامه دهند یا خیر، هدایت اشیا را به سمت Observers ادامه می‌دهند. در اصل Observables اشیاء را منتشر می‌کنند و این امر به عهده Observers است که به کار خود ادامه دهند. هنگامی که Observables نتواند اشیای منتشر شده را نگهداری کند، بایدبافر شوند یا به طریقی آن‌ها handle شوند. در ادامه در مورد برخی استراتژی‌های بافر صحبت خواهیم کرد.
• Cold sources: این مورد برخلاف Hot sources است. می‌توان گفت که اشیاء توسط Observables به صورت lazily منتشر می‌شوند. به معنی اینکه: Observables هنگامی که Observerها اشیاء را می‌خواهند و با یک نرخ مناسب برای Observer، اشیاء را منتشر می‌کنند. اشیاء منتشر شده توسط Observables  نیازی به بافر ندارند زیرا تمام فرایند به‌طور کلی در اختیار Observables است. به آن به عنوان یک رابطه کششی نگاه کنید.Observerها وقتی اشیاء یا آیتم‌ها را می‌خواهند می‌توانند از  Observables آن‌ها را دریافت کنند.

Backpressure Strategies

استراتژی 1 ()onBackpressureBuffer :

Flowable.range(0, 1000000)
    .onBackpressureBuffer()
    .observeOn(Schedulers.computation())
    .subscribe(new FlowableSubscriber<Integer>() {
        @Override
        public void onSubscribe(Subscription s) {
        }
        @Override
        public void onNext(Integer integer) {
            Log.d(TAG, "onNext: " + integer);
        }
        @Override
        public void onError(Throwable t) {
            Log.e(TAG, "onError: ", t);
        }
        @Override
        public void onComplete() {
        }
    });

Observable<Integer> observable = Observable
        .just(1, 2, 3, 4, 5);
Flowable<Integer> flowable = observable.toFlowable(BackpressureStrategy.BUFFER);

Flowable to Observable: 

Observable<Integer> observable = Observable
        .just(1, 2, 3, 4, 5);
Flowable<Integer> flowable = observable.toFlowable(BackpressureStrategy.BUFFER);
Observable<Integer> backToObservable = flowable.toObservable(

ساخت عملگرهای  (Operators : Create، Just، Range، Repeat)

• مقدمه
• چرا باید این اپراتورها برای شما مهم باشد
• ()create
• ()just
• ()range
• ()repeat

مقدمه

1. ()create
2. ()just
3. ()range
4. ()repeat

چرا باید این اپراتورها برای شما مهم باشند

• ایجاد یک Observable از یک single object
• ایجاد یک Observable از لیستی از اشیاء

ایجاد یک Observable از یک single object

1. یک شیء را برای تبدیل شدن به یک Observable تعریف کنید.
2. یک Observable ایجاد کنید.
3. Observable را Subscribe کرده و شیء منتشر شده را دریافت کنید.

// Instantiate the object to become an Observable
final Task task = new Task("Walk the dog", false, 4);
// Create the Observable
Observable<Task> singleTaskObservable = Observable
        .create(new ObservableOnSubscribe<Task>() {
            @Override
            public void subscribe(ObservableEmitter<Task> emitter) throws Exception {
                if(!emitter.isDisposed()){
                    emitter.onNext(task);
                    emitter.onComplete();
                }
            }
        })
        .subscribeOn(Schedulers.io())
        .observeOn(AndroidSchedulers.mainThread());
// Subscribe to the Observable and get the emitted object
singleTaskObservable.subscribe(new Observer<Task>() {
    @Override
    public void onSubscribe(Disposable d) {
    }
    @Override
    public void onNext(Task task) {
        Log.d(TAG, "onNext: single task: " + task.getDescription());
    }
    @Override
    public void onError(Throwable e) {
    }
    @Override
    public void onComplete() {
    }
});

ایجاد یک Observable از لیستی از اشیاء

1. Observable را ایجاد کنید.
2. در داخل متد Subscribe از طریق لیست Taskهای تکرار شده می‌توانید ()onNext را فراخوانی کنید.
3. پس از اتمام حلقه (Loop)، می‌توانید متد ()onComplete را فراخوانی کنید.
4. Observable را Subscribe کرده و شیء منتشر شده را دریافت کنید.

// Create the Observable
Observable<Task> taskListObservable = Observable
    .create(new ObservableOnSubscribe<Task>() {
        @Override
        public void subscribe(ObservableEmitter<Task> emitter) throws Exception {
            // Inside the subscribe method iterate through the list of tasks and call onNext(task)
            for(Task task: DataSource.createTasksList()){
                if(!emitter.isDisposed()){
                    emitter.onNext(task);
                }
            }
            // Once the loop is complete, call the onComplete() method
            if(!emitter.isDisposed()){
                emitter.onComplete();
            }
        }
    })
    .subscribeOn(Schedulers.io())
    .observeOn(AndroidSchedulers.mainThread());
// Subscribe to the Observable and get the emitted objects
taskListObservable.subscribe(new Observer<Task>() {
@Override
public void onSubscribe(Disposable d) {
}
@Override
public void onNext(Task task) {
    Log.d(TAG, "onNext: task list: " + task.getDescription());
}
@Override
public void onError(Throwable e) {
}
@Override
public void onComplete() {
}
});

()just

Observable.just("first", "second", "third", "fourth", "fifth", "sixth",
          "seventh", "eighth", "ninth", "tenth")
          .subscribeOn(Schedulers.io()) // What thread to do the work on
          .observeOn(AndroidSchedulers.mainThread()) // What thread to observe the results on
          .subscribe(new Observer<String>() { // view the results by creating a new observer
              @Override
              public void onSubscribe(Disposable d) {
                  Log.d(TAG, "onSubscribe: called");
              }
              @Override
              public void onNext(String s) {
                  Log.d(TAG, "onNext: " + s);
              }
              @Override
              public void onError(Throwable e) {
                  Log.e(TAG, "onError: ", e);
              }
              @Override
              public void onComplete() {
                  Log.d(TAG, "onComplete: done...");
              }
          });

()range

Observable.range(0,11)
        .observeOn(Schedulers.io())
        .subscribeOn(AndroidSchedulers.mainThread())
        .subscribe(new Observer<Integer>() {
            @Override
            public void onSubscribe(Disposable d) {
            }
            @Override
            public void onNext(Integer integer) {
                Log.d(TAG, "onNext: " + integer);
            }
            @Override
            public void onError(Throwable e) {
            }
            @Override
            public void onComplete() {
            }
        });

()repeat

Observable.range(0,3)
        .repeat(2)
        .observeOn(Schedulers.io())
        .subscribeOn(AndroidSchedulers.mainThread())
        .subscribe(new Observer<Integer>() {
            @Override
            public void onSubscribe(Disposable d) {
            }
            @Override
            public void onNext(Integer integer) {
                Log.d(TAG, "onNext: " + integer);
            }
            @Override
            public void onError(Throwable e) {
            }
            @Override
            public void onComplete() {
            }
        });

final Handler handler = new Handler();
final Runnable runnable = new Runnable() {
    int elapsedTime = 0;
    @Override
    public void run() {
        if(elapsedTime >= 5){ // if greater than 5 seconds
            handler.removeCallbacks(this);
        }
        else{
            elapsedTime = elapsedTime + 1;
            handler.postDelayed(this, 1000);
            Log.d(TAG, "run: " + elapsedTime);
        }
    }
};
handler.postDelayed(runnable, 1000);

// emit an observable every time interval
Observable<Long> intervalObservable = Observable
        .interval(1, TimeUnit.SECONDS)
        .subscribeOn(Schedulers.io())
        .takeWhile(new Predicate<Long>() { // stop the process if more than 5 seconds passes
            @Override
            public boolean test(Long aLong) throws Exception {
                return aLong <= 5;
            }
        })
        .observeOn(AndroidSchedulers.mainThread());

intervalObservable.subscribe(new Observer<Long>() {
    @Override
    public void onSubscribe(Disposable d) {
    }
    @Override
    public void onNext(Long aLong) {
        Log.d(TAG, "onNext: interval: " + aLong);
    }
    @Override
    public void onError(Throwable e) {
    }
    @Override
    public void onComplete() {
    }
});

()timer

// emit single observable after a given delay
Observable<Long> timeObservable = Observable
        .timer(3, TimeUnit.SECONDS)
        .subscribeOn(Schedulers.io())
        .observeOn(AndroidSchedulers.mainThread());

timeObservable.subscribe(new Observer<Long>() {
    long time = 0; // variable for demonstating how much time has passed
    @Override
    public void onSubscribe(Disposable d) {
        time = System.currentTimeMillis() / 1000;
    }
    @Override
    public void onNext(Long aLong) {
        Log.d(TAG, "onNext: " + ((System.currentTimeMillis() / 1000) - time) + " seconds have elapsed." );
    }
    @Override
    public void onError(Throwable e) {
    }
    @Override
    public void onComplete() {
    }
});

ساخت عملگر‌های (fromArray, fromIterable, fromCallable)

• مقدمه
• ()fromArray
• ()fromIterable
• ()fromCallable

public class Task {
    private String description;
    private boolean isComplete;
    private int priority;
    public Task(String description, boolean isComplete, int priority) {
        this.description = description;
        this.isComplete = isComplete;
        this.priority = priority;
    }
    // getter and setters ....
}

()fromarray

Task[] list = new Task[5];
list[0] = (new Task("Take out the trash", true, 3));
list[1] = (new Task("Walk the dog", false, 2));
list[2] = (new Task("Make my bed", true, 1));
list[3] = (new Task("Unload the dishwasher", false, 0));
list[4] = (new Task("Make dinner", true, 5));
Observable<Task> taskObservable = Observable
        .fromArray(list)
        .subscribeOn(Schedulers.io())
        .observeOn(AndroidSchedulers.mainThread());
taskObservable.subscribe(new Observer<Task>() {
    @Override
    public void onSubscribe(Disposable d) {
    }
    @Override
    public void onNext(Task task) {
        Log.d(TAG, "onNext: : " + task.getDescription());
    }
    @Override
    public void onError(Throwable e) {
    }
    @Override
    public void onComplete() {
    }
});)

List<Task> taskList = new ArrayList<>();
taskList.add(new Task("Take out the trash", true, 3));
taskList.add(new Task("Walk the dog", false, 2));
taskList.add(new Task("Make my bed", true, 1));
taskList.add(new Task("Unload the dishwasher", false, 0));
taskList.add(new Task("Make dinner", true, 5));
Observable<Task> taskObservable = Observable
        .fromIterable(taskList)
        .subscribeOn(Schedulers.io())
        .observeOn(AndroidSchedulers.mainThread());
taskObservable.subscribe(new Observer<Task>() {
    @Override
    public void onSubscribe(Disposable d) {
    }
    @Override
    public void onNext(Task task) {
        Log.d(TAG, "onNext: : " + task.getDescription());
    }
    @Override
    public void onError(Throwable e) {
    }
    @Override
    public void onComplete() {
    }
});

()fromCallable

Output: T

1. برای اجرای متد بر روی background thread می‌توانید از یک callable استفاده کنید.
2. سپس نتایج را به Main Thread برگردانید.

// create Observable (method will not execute yet)
Observable<Task> callable = Observable
        .fromCallable(new Callable<Task>() {
            @Override
            public Task call() throws Exception {
                return MyDatabase.getTask();
            }
        })
        .subscribeOn(Schedulers.io())
        .observeOn(AndroidSchedulers.mainThread());
// method will be executed since now something has subscribed
callable.subscribe(new Observer<Task>() {
    @Override
    public void onSubscribe(Disposable d) {
    }
    @Override
    public void onNext(Task task) {
        Log.d(TAG, "onNext: : " + task.getDescription());
    }
    @Override
    public void onError(Throwable e) {
    }
    @Override
    public void onComplete() {
    }
});

ساخت عملگر (fromFuture)

• مقدمه
• ()fromFuture
• مثال

مقدمه

()fromFuture

implementation "com.squareup.retrofit2:adapter-rxjava2:2.5.0"

def retrofitVersion = "2.5.0"
def rxjava_version = '2.2.7'
def rxandroid_version = '2.1.1'
def lifecycle_version = "1.1.1"
// ViewModel and LiveData
implementation "android.arch.lifecycle:extensions:$lifecycle_version"
// Retrofit
implementation "com.squareup.retrofit2:retrofit:$retrofitVersion"
implementation "com.squareup.retrofit2:converter-gson:$retrofitVersion"
// RxJava
implementation "io.reactivex.rxjava2:rxjava:$rxjava_version"
// RxAndroid
implementation "io.reactivex.rxjava2:rxandroid:$rxandroid_version"
// RxJava Call Adapter
implementation "com.squareup.retrofit2:adapter-rxjava2:2.5.0"

public interface RequestApi {
    @GET("todos/1")
    Observable<ResponseBody> makeObservableQuery();
}

.addCallAdapterFactory(RxJava2CallAdapterFactory.create())

public class ServiceGenerator {
    public static final String BASE_URL = "https://jsonplaceholder.typicode.com";
    private static Retrofit.Builder retrofitBuilder =
            new Retrofit.Builder()
                    .baseUrl(BASE_URL)
                    .addCallAdapterFactory(RxJava2CallAdapterFactory.create())
                    .addConverterFactory(GsonConverterFactory.create());
    private static Retrofit retrofit = retrofitBuilder.build();
    private static RequestApi requestApi = retrofit.create(RequestApi.class);
    public static RequestApi getRequestApi(){
        return requestApi;
    }
}

public class Repository {
    private static Repository instance;
    public static Repository getInstance(){
        if(instance == null){
            instance = new Repository();
        }
        return instance;
    }
    public Future<Observable<ResponseBody>> makeFutureQuery(){
        final ExecutorService executor = Executors.newSingleThreadExecutor();
        final Callable<Observable<ResponseBody>> myNetworkCallable = new Callable<Observable<ResponseBody>>() {
            @Override
            public Observable<ResponseBody> call() throws Exception {
                return ServiceGenerator.getRequestApi().makeObservableQuery();
            }
        };
        final Future<Observable<ResponseBody>> futureObservable = new Future<Observable<ResponseBody>>(){
            @Override
            public boolean cancel(boolean mayInterruptIfRunning) {
                if(mayInterruptIfRunning){
                    executor.shutdown();
                }
                return false;
            }
            @Override
            public boolean isCancelled() {
                return executor.isShutdown();
            }
            @Override
            public boolean isDone() {
                return executor.isTerminated();
            }
            @Override
            public Observable<ResponseBody> get() throws ExecutionException, InterruptedException {
                return executor.submit(myNetworkCallable).get();
            }
            @Override
            public Observable<ResponseBody> get(long timeout, TimeUnit unit) throws ExecutionException, InterruptedException, TimeoutException {
                return executor.submit(myNetworkCallable).get(timeout, unit);
            }
        };
        return futureObservable;
    }
}

public class MainViewModel extends ViewModel {
    private Repository repository;
    public MainViewModel() {
        repository = Repository.getInstance();
    }
    public Future<Observable<ResponseBody>> makeFutureQuery(){
        return repository.makeFutureQuery();
    }
}

MainViewModel viewModel = ViewModelProviders.of(this).get(MainViewModel.class);
try {
    viewModel.makeFutureQuery().get()
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(new Observer<ResponseBody>() {
                @Override
                public void onSubscribe(Disposable d) {
                    Log.d(TAG, "onSubscribe: called.");
                }
                @Override
                public void onNext(ResponseBody responseBody) {
                    Log.d(TAG, "onNext: got the response from server!");
                    try {
                        Log.d(TAG, "onNext: " + responseBody.string());
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                }
                @Override
                public void onError(Throwable e) {
                    Log.e(TAG, "onError: ", e);
                }
                @Override
                public void onComplete() {
                    Log.d(TAG, "onComplete: called.");
                }
            });
} catch (ExecutionException e) {
    e.printStackTrace();
} catch (InterruptedException e) {
    e.printStackTrace();
}

ساخت عملگر (fromPublisher)

• مقدمه
• ()fromPublisher
• مثال

مقدمه

()fromPublisher

مثالی از FromPublisher:

<uses-permission android:name="android.permission.INTERNET"/>

implementation "com.squareup.retrofit2:adapter-rxjava2:2.9.0"

implementation "android.arch.lifecycle:reactivestreams:1.1.1"

def retrofitVersion = "2.5.0"
def rxjava_version = '2.2.7'
def rxandroid_version = '2.1.1'
def lifecycle_version = "1.1.1"
// ViewModel and LiveData
implementation "android.arch.lifecycle:extensions:$lifecycle_version"
// Retrofit
implementation "com.squareup.retrofit2:retrofit:$retrofitVersion"
implementation "com.squareup.retrofit2:converter-gson:$retrofitVersion"
// RxJava
implementation "io.reactivex.rxjava2:rxjava:$rxjava_version"
// RxAndroid
implementation "io.reactivex.rxjava2:rxandroid:$rxandroid_version"
// RxJava Call Adapter (Call object to Observable)
implementation "com.squareup.retrofit2:adapter-rxjava2:2.5.0"
// Convert Observable to LiveData
implementation "android.arch.lifecycle:reactivestreams:1.1.1"

RequestApi :

public interface RequestApi {
    @GET("todos/1")
    Flowable<ResponseBody> makeQuery();
}

ServiceGenerator:

.addCallAdapterFactory(RxJava2CallAdapterFactory.create())

public class ServiceGenerator {
    public static final String BASE_URL = "https://jsonplaceholder.typicode.com";
    private static Retrofit.Builder retrofitBuilder =
            new Retrofit.Builder()
                    .baseUrl(BASE_URL)
                    .addCallAdapterFactory(RxJava2CallAdapterFactory.create())
                    .addConverterFactory(GsonConverterFactory.create());
    private static Retrofit retrofit = retrofitBuilder.build();
    private static RequestApi requestApi = retrofit.create(RequestApi.class);
    public static RequestApi getRequestApi(){
        return requestApi;
    }
}

Repository:

public class Repository {
    private static Repository instance;
    public static Repository getInstance(){
        if(instance == null){
            instance = new Repository();
        }
        return instance;
    }
    public LiveData<ResponseBody> makeReactiveQuery(){
        return LiveDataReactiveStreams.fromPublisher(ServiceGenerator.getRequestApi()
                .makeQuery()
                .subscribeOn(Schedulers.io()));
    }
}

ViewModel:

public class MainViewModel extends ViewModel {
    private Repository repository;
    public MainViewModel() {
        repository = Repository.getInstance();
    }
    public LiveData<ResponseBody> makeQuery(){
        return repository.makeReactiveQuery();
    }
}

MainActivity:

عملگرهای Filter : Filter

مثال 1: فیلتر کردن  یک فیلد String در یک لیست

1. ما یک Observable از لیستی از Task Objects را ایجاد می‌کنیم.
2.  Tasks را برای توضیحات خاص فیلتر می‌کنیم.
3. سپس  Subscribe کردن و منتشر کردن کارهایی (Task) که از  Test عبور می‌کنند.

Observable<Task> taskObservable = Observable
            .fromIterable(DataSource.createTasksList())
            .filter(new Predicate<Task>() {
                @Override
                public boolean test(Task task) throws Exception {
                    if(task.getDescription().equals("Walk the dog")){
                        return true;
                    }
                    return false;
                }
            })
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread());
taskObservable.subscribe(new Observer<Task>() {
            @Override
            public void onSubscribe(Disposable d) {
            }
            @Override
            public void onNext(Task task) {
                Log.d(TAG, "onNext: This task matches the description: " + task.getDescription());
            }
            @Override
            public void onError(Throwable e) {
            }
            @Override
            public void onComplete() {
            }
});

public static List<Task> createTasksList(){
        List<Task> tasks = new ArrayList<>();
        tasks.add(new Task("Take out the trash", true, 3));
        tasks.add(new Task("Walk the dog", false, 2));
        tasks.add(new Task("Make my bed", true, 1));
        tasks.add(new Task("Unload the dishwasher", false, 0));
        tasks.add(new Task("Make dinner", true, 5));
        return tasks;
    }

مثال 2 : فیلتر کردن  یک فیلد Boolean در یک لیست

1. ما یک Observable از لیستی از Task Objects را ایجاد می‌کنیم.
2. Tasksرا در قسمت ()isComplete فیلتر می‌کنیم.
3. سپس  Subscribe کردن و منتشر کردن کارهایی (Task) که از  Test عبور می‌کنند.

Observable<Task> taskObservable = Observable
            .fromIterable(DataSource.createTasksList())
            .filter(new Predicate<Task>() {
                @Override
                public boolean test(Task task) throws Exception {
                    return task.isComplete();
                }
            })
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread());
taskObservable.subscribe(new Observer<Task>() {
            @Override
            public void onSubscribe(Disposable d) {
            }
            @Override
            public void onNext(Task task) {
                Log.d(TAG, "onNext: This is a completed task: " + task.getDescription());
            }
            @Override
            public void onError(Throwable e) {
            }
            @Override
            public void onComplete() {
            }
        });

public static List<Task> createTasksList(){
        List<Task> tasks = new ArrayList<>();
        tasks.add(new Task("Take out the trash", true, 3));
        tasks.add(new Task("Walk the dog", false, 2));
        tasks.add(new Task("Make my bed", true, 1));
        tasks.add(new Task("Unload the dishwasher", false, 0));
        tasks.add(new Task("Make dinner", true, 5));
        return tasks;
    }

عملگرهای Distinct : Filter

• مقدمه
• اسفاده نادرست از ()Distinct
• استفاده درست از ()Distinct

مقدمه

استفاده نادرست از ()Distinct

1. ما یک Observable از لیستی از Task Objects را ایجاد می‌کنیم.
2. اپراتور  ()Distinct را اعمال کرده و یک custom function را پاس می‌دهیم.

public static List<Task> createTasksList(){
    List<Task> tasks = new ArrayList<>();
    tasks.add(new Task("Take out the trash", true, 3));
    tasks.add(new Task("Walk the dog", false, 2));
    tasks.add(new Task("Make my bed", true, 1));
    tasks.add(new Task("Unload the dishwasher", false, 0));
    tasks.add(new Task("Make dinner", true, 5));
    tasks.add(new Task("Make dinner", true, 5)); // duplicate for testing the distinct operator
    return tasks;
}

Observable<Task> taskObservable = Observable
        .fromIterable(DataSource.createTasksList())
        .distinct(new Function<Task, Task>() { // <--- WRONG 
            @Override
            public Task apply(Task task) throws Exception { 
                return task;
            }
        })
        .subscribeOn(Schedulers.io())
        .observeOn(AndroidSchedulers.mainThread());
taskObservable.subscribe(new Observer<Task>() {
    @Override
    public void onSubscribe(Disposable d) {
    }
    @Override
    public void onNext(Task task) {
        Log.d(TAG, "onNext: " + task.getDescription());
    }
    @Override
    public void onError(Throwable e) {
    }
    @Override
    public void onComplete() {
    }
});

استفاده درست از ()Distinct

1. ما یک Observable از لیستی از Task Objects را ایجاد می‌کنیم.
2. اعمال اپراتور ()Distinct و پاس دادن یک
3. اپراتور  ()Distinct را اعمال کرده و یک custom function را پاس می‌دهیم.
4. تفاوت در نوع بازگشت عملکرد است. توجه کنید که ما یک رشته را بر می‌گردانیم و در آن رشته چک کردن را انجام می‌دهیم.

public static List<Task> createTasksList(){
    List<Task> tasks = new ArrayList<>();
    tasks.add(new Task("Take out the trash", true, 3));
    tasks.add(new Task("Walk the dog", false, 2));
    tasks.add(new Task("Make my bed", true, 1));
    tasks.add(new Task("Unload the dishwasher", false, 0));
    tasks.add(new Task("Make dinner", true, 5));
    tasks.add(new Task("Make dinner", true, 5)); // duplicate for testing the distinct operator
    return tasks;
}

Observable<Task> taskObservable = Observable
        .fromIterable(DataSource.createTasksList())
        .distinct(new Function<Task, String>() { // <--- CORRECT 
            @Override
            public String apply(Task task) throws Exception {
                return task.getDescription();
            }
        })
        .subscribeOn(Schedulers.io())
        .observeOn(AndroidSchedulers.mainThread());
taskObservable.subscribe(new Observer<Task>() {
    @Override
    public void onSubscribe(Disposable d) {
    }
    @Override
    public void onNext(Task task) {
        Log.d(TAG, "onNext: " + task.getDescription());
    }
    @Override
    public void onError(Throwable e) {
    }
    @Override
    public void onComplete() {
    }
});

عملگرهای Take : Filter و TakeWhile

• مقدمه
• ()Take
• ()TakeWhile

مقدمه

()take

public class DataSource {
    public static List<Task> createTasksList(){
        List<Task> tasks = new ArrayList<>();
        tasks.add(new Task("Take out the trash", true, 3));
        tasks.add(new Task("Walk the dog", false, 2));
        tasks.add(new Task("Make my bed", true, 1));
        tasks.add(new Task("Unload the dishwasher", false, 0));
        tasks.add(new Task("Make dinner", true, 5));
        return tasks;
    }
}

Observable<Task> taskObservable = Observable
        .fromIterable(DataSource.createTasksList())
        .take(3)
        .subscribeOn(Schedulers.io())
        .observeOn(AndroidSchedulers.mainThread());
taskObservable.subscribe(new Observer<Task>() {
    @Override
    public void onSubscribe(Disposable d) {
    }
    @Override
    public void onNext(Task task) {
        Log.d(TAG, "onNext: " + task.getDescription());
    }
    @Override
    public void onError(Throwable e) {
    }
    @Override
    public void onComplete() {
    }
});

public class DataSource {
    public static List<Task> createTasksList(){
        List<Task> tasks = new ArrayList<>();
        tasks.add(new Task("Take out the trash", true, 3));
        tasks.add(new Task("Walk the dog", false, 2));
        tasks.add(new Task("Make my bed", true, 1));
        tasks.add(new Task("Unload the dishwasher", false, 0));
        tasks.add(new Task("Make dinner", true, 5));
        return tasks;
    }
}

Observable<Task> taskObservable = Observable
        .fromIterable(DataSource.createTasksList())
        .takeWhile(new Predicate<Task>() {
            @Override
            public boolean test(Task task) throws Exception {
                return task.isComplete();
            }
        })
        .subscribeOn(Schedulers.io())
        .observeOn(AndroidSchedulers.mainThread());
taskObservable.subscribe(new Observer<Task>() {
    @Override
    public void onSubscribe(Disposable d) {
    }
    @Override
    public void onNext(Task task) {
        Log.d(TAG, "onNext: " + task.getDescription());
    }
    @Override
    public void onError(Throwable e) {
    }
    @Override
    public void onComplete() {
    }
});

• Introduction
• ()Map
• ()Buffer
• ()Debounce
• ()ThrottleFirst
• ()FlatMap
• ()ConcatMap
• ()SwitchMap

مقدمه(Introduction)

کار یک اپراتور به شرح زیر است : اپراتورها مواردی را که توسط یک Observable منتشر می‌شود را تبدیل می‌کنند. در زیر برخی از متداول‌ترین اپراتورهای دگرگونی وجود دارد که فرض می‌کنیم شما از آن‌ها استفاده می‌کنید.

Map

یک تابع را برای هر مورد منتشر شده اعمال می‌کند. با استفاده از تابعی از آن، هر مورد منتشر شده را تغییر می‌دهد. (نظم ارسال حفظ می‌شود.) نمودار Map :

Buffer

به‌طور دوره ای items را از یک Observable به صورت bundles جمع می‌کند و به جای اینکه یکبار items را منتشر کند، bundles را منتشر می‌کند. (نظم ارسال حفظ می‌شود) نمودار Buffer:

Debounce

اگر یک زمان خاص گذشته باشد، یک Item را از یک Observable منتشر می‌کند بدون آنکه یک Item دیگر را منتشر کند. این عملگر برای کلیک دکمه (button clicks) بسیار عالی است. اگر کاربر بارها و بارها یک دکمه را کلیک کند، دیگر نیازی به اجرای چندین بار کار نیست. می‌توانیم از اپراتور ()Debounce استفاده کنیم تا با معرفی یک بازه‌ی زمانی مجاز بین کلیک، کلیک‌های خود را کنترل کنیم. اگر مدت زمانی سپری نشده باشد، می‌توانیم از هر دو متد جلوگیری کنیم. (نظم ارسال حفظ می‌شود.) نمودار Debounce:

ThrottleFirst

موارد منتشر شده توسط منبع Observable را که در یک بازه‌ی زمانی قرار دارند، فیلتر می‌کند.(نظم ارسال حفظ می‌شود.) نمودار ThrottleFirst:

FlatMap

موارد منتشر شده توسط یک Observable را به Observables تبدیل می‌کند، و سپس انتشار از آن را به یک Observable Single قسمت می‌کند. اگر با LiveData آشنا باشید، MediatorLiveData می‌تواند کاری بسیار مشابه انجام دهد. در مورد ()FlatMap در ادامه بیشتر صحبت می‌کنیم. (نظم ارسال حفظ نمی‌شود.) نمودار FlatMap:

ConcatMap

موارد منتشر شده توسط یک Observable را به Observables تبدیل می‌کند. این در اصل همان مورد ()FlatMap است، اما نظم ارسال حفظ می‌شود. اما از آنجا که ()ConcatMap باید منتظر بماند تا هر یک از Observable کار خود را انجام دهند پس از نظر فنی غیر همزمان نیست. (نظم ارسال حفظ می‌شود.) نمودار ConcatMap:

SwitchMap

()SwitchMap آیتم‌های منتشر شده توسط یک Observable را به یک Observable تبدیل می‌کند درست مثل ()ConcatMap  و ()FlatMap. تفاوت این است که به محض مشترک شدن یک Observer جدید، Observer  قبلی را لغو می‌کند. ()SwitchMap  محدودیتی را حل می‌کند که ()ConcatMap و ()FlatMap هم دارند. (نظم ارسال حفظ می‌شود) نمودار SwitchMap:

عملگرهای تبدیل کننده : Map

یک تابع را برای هر مورد منتشر شده اعمال می‌کند. با استفاده از تابعی از آن، هر مورد منتشر شده را تغییر می‌دهد. (نظم ارسال حفظ می‌شود.) نمودار Map :

برای دو مثال زیر به این دو کلاس مراجعه می‌کنیم:

Task.java (کلاس Task)

public class Task {
    private String description;
    private boolean isComplete;
    private int priority;
    public Task(String description, boolean isComplete, int priority) {
        this.description = description;
        this.isComplete = isComplete;
        this.priority = priority;
    }
    // getter and setters ....
}

DataSource.java (کلاس DataSource)

public class DataSource {
    public static List<Task> createTasksList(){
        List<Task> tasks = new ArrayList<>();
        tasks.add(new Task("Take out the trash", true, 3));
        tasks.add(new Task("Walk the dog", false, 2));
        tasks.add(new Task("Make my bed", true, 1));
        tasks.add(new Task("Unload the dishwasher", false, 0));
        tasks.add(new Task("Make dinner", true, 5));
        return tasks;
    }
}

مثال : (Mapping (Task -> String

در این مثال ما یک  custom map function  ایجاد می‌کنیم که فیلد توضیحات را از Task Objects استخراج می‌کند و فقط آن پارامتر را منتشر می‌کند. تابع:

Function<Task, String> extractDescriptionFunction = new Function<Task, String>() {
    @Override
    public String apply(Task task) throws Exception {
        Log.d(TAG, "apply: doing work on thread: " + Thread.currentThread().getName());
        return task.getDescription();
    }
};

Observable<String> extractDescriptionObservable = Observable
        .fromIterable(DataSource.createTasksList())
        .subscribeOn(Schedulers.io())
        .map(extractDescriptionFunction)
        .observeOn(AndroidSchedulers.mainThread());
extractDescriptionObservable.subscribe(new Observer<String>() {
    @Override
    public void onSubscribe(Disposable d) {
    }
    @Override
    public void onNext(String s) {
        Log.d(TAG, "onNext: extracted description: " + s);
    }
    @Override
    public void onError(Throwable e) {
    }
    @Override
    public void onComplete() {
    }
});

مثال : (Mapping (Task -> Updated Task

Function<Task, Task> completeTaskFunction = new Function<Task, Task>() {
    @Override
    public Task apply(Task task) throws Exception {
        Log.d(TAG, "apply: doing work on thread: " + Thread.currentThread().getName());
        task.setComplete(true);
        return task;
    }
};

Observable<Task> completeTaskObservable = Observable
        .fromIterable(DataSource.createTasksList())
        .subscribeOn(Schedulers.io())
        .map(completeTaskFunction)
        .observeOn(AndroidSchedulers.mainThread());
completeTaskObservable.subscribe(new Observer<Task>() {
    @Override
    public void onSubscribe(Disposable d) {
    }
    @Override
    public void onNext(Task task) {
        Log.d(TAG, "onNext: is this task complete? " + task.isComplete());
    }
    @Override
    public void onError(Throwable e) {
    }
    @Override
    public void onComplete() {
    }
});

مثال : Order of Emitted Objects

// Create an Observable using the fromIterable operator
Observable<Task> mappedObservable = Observable
        .fromIterable(DataSource.createTasksList())
        .subscribeOn(Schedulers.io())
        .map(new Function<Task, Task>() {
            @Override
            public Task apply(Task task) throws Exception {
                task.setComplete(true);
                return task;
            }
        })
        .observeOn(AndroidSchedulers.mainThread());
// subscribe to the Observable and view the emitted results
mappedObservable.subscribe(new Observer<Task>() {
    @Override
    public void onSubscribe(Disposable d) {
    }
    @Override
    public void onNext(Task task) {
        Log.d(TAG, "onNext: mapped: " + task.getDescription());
    }
    @Override
    public void onError(Throwable e) {
    }
    @Override
    public void onComplete() {
    }
});

عملگرهای تبدیل کننده: Buffer

public class Task {
    private String description;
    private boolean isComplete;
    private int priority;
    public Task(String description, boolean isComplete, int priority) {
        this.description = description;
        this.isComplete = isComplete;
        this.priority = priority;
    }
    // getter and setters ....
}

public class DataSource {
    public static List<Task> createTasksList(){
        List<Task> tasks = new ArrayList<>();
        tasks.add(new Task("Take out the trash", true, 3));
        tasks.add(new Task("Walk the dog", false, 2));
        tasks.add(new Task("Make my bed", true, 1));
        tasks.add(new Task("Unload the dishwasher", false, 0));
        tasks.add(new Task("Make dinner", true, 5));
        return tasks;
    }
}

یک مثال ساده:

1. Observable را ایجاد می‌کنیم.
2. اپراتور ()Buffer را به کار می‌بریم.
3. Subscribe کردن و مشاهده کردن نتیجه منتشر شده در log.

// Create an Observable
Observable<Task> taskObservable = Observable
        .fromIterable(DataSource.createTasksList())
        .subscribeOn(Schedulers.io());
taskObservable
        .buffer(2) // Apply the Buffer() operator
        .observeOn(AndroidSchedulers.mainThread())
        .subscribe(new Observer<List<Task>>() { // Subscribe and view the emitted results
            @Override
            public void onSubscribe(Disposable d) {
            }
            @Override
            public void onNext(List<Task> tasks) {
                Log.d(TAG, "onNext: bundle results: -------------------");
                for(Task task: tasks){
                    Log.d(TAG, "onNext: " + task.getDescription());
                }
            }
            @Override
            public void onError(Throwable e) {
            }
            @Override
            public void onComplete() {
            }
        });

مثال : Tracking UI Interactions

def rxbinding_version = "4.0.0"
// Rx Binding Library
implementation "com.jakewharton.rxbinding4:rxbinding:$rxbinding_version"

// global disposables object
CompositeDisposable disposables = new CompositeDisposable();
// detect clicks to a button
RxView.clicks(findViewById(R.id.button))
.map(new Function<Unit, Integer>() { // convert the detected clicks to an integer
    @Override
    public Integer apply(Unit unit) throws Exception {
        return 1;
    }
})
.buffer(4, TimeUnit.SECONDS) // capture all the clicks during a 4 second interval
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Observer<List<Integer>>() {
    @Override
    public void onSubscribe(Disposable d) {
        disposables.add(d); // add to disposables to you can clear in onDestroy
    }
    @Override
    public void onNext(List<Integer> integers) {
        Log.d(TAG, "onNext: You clicked " + integers.size() + " times in 4 seconds!");
    }
    @Override
    public void onError(Throwable e) {
    }
    @Override
    public void onComplete() {
    }
});
// make sure to clear disposables when the activity is destroyed
@Override
protected void onDestroy() {
    super.onDestroy();
    disposables.clear();
}

عملگرهای تبدیل کننده : Debounce

مثال : محدود کردن درخواست‌‌های سرور

def rxbinding_version = "4.0.0"
// Rx Binding Library
implementation "com.jakewharton.rxbinding4:rxbinding:$rxbinding_version"

<?xml version="1.0" encoding="utf-8"?>
<androidx.constraintlayout.widget.ConstraintLayout
    xmlns:android="http://schemas.android.com/apk/res/android"
    xmlns:app="http://schemas.android.com/apk/res-auto"
    xmlns:tools="http://schemas.android.com/tools"
    android:layout_width="match_parent"
    android:layout_height="match_parent"
    tools:context=".MainActivity">
    <androidx.appcompat.widget.SearchView
        android:layout_width="match_parent"
        android:layout_height="?attr/actionBarSize"
        app:layout_constraintTop_toTopOf="parent"
        app:layout_constraintRight_toRightOf="parent"
        app:layout_constraintLeft_toLeftOf="parent"
        android:id="@+id/search_view">
    </androidx.appcompat.widget.SearchView>
  </androidx.constraintlayout.widget.ConstraintLayout>

• ساخت Observable
• گوش دادن به ورودی متن در SearchView.
• فرستادن Query به emitter.
• استفاده از اپراتور ()Debounce، برای محدود کردن درخواست ها.
• Subscirbe کردن Observer
• ارسال درخواست به سرور
• پاک کردن disposable در ()onDestroy.

public class MainActivity extends AppCompatActivity {
    private static final String TAG = "MainActivity";
    //ui
    private SearchView searchView;
    // vars
    private CompositeDisposable disposables = new CompositeDisposable();
    private long timeSinceLastRequest; // for log printouts only. Not part of logic.
    @Override
    protected void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        setContentView(R.layout.activity_main);
        searchView = findViewById(R.id.search_view);
        timeSinceLastRequest = System.currentTimeMillis();
        // create the Observable
        Observable<String> observableQueryText = Observable
                .create(new ObservableOnSubscribe<String>() {
                    @Override
                    public void subscribe(final ObservableEmitter<String> emitter) throws Exception {
                        // Listen for text input into the SearchView
                        searchView.setOnQueryTextListener(new SearchView.OnQueryTextListener() {
                            @Override
                            public boolean onQueryTextSubmit(String query) {
                                return false;
                            }
                            @Override
                            public boolean onQueryTextChange(final String newText) {
                                if(!emitter.isDisposed()){
                                    emitter.onNext(newText); // Pass the query to the emitter
                                }
                                return false;
                            }
                        });
                    }
                })
                .debounce(500, TimeUnit.MILLISECONDS) // Apply Debounce() operator to limit requests
                .subscribeOn(Schedulers.io());
        // Subscribe an Observer
        observableQueryText.subscribe(new Observer<String>() {
            @Override
            public void onSubscribe(Disposable d) {
                disposables.add(d);
            }
            @Override
            public void onNext(String s) {
                Log.d(TAG, "onNext: time  since last request: " + (System.currentTimeMillis() - timeSinceLastRequest));
                Log.d(TAG, "onNext: search query: " + s);
                timeSinceLastRequest = System.currentTimeMillis();
                // method for sending a request to the server
                sendRequestToServer(s);
            }
            @Override
            public void onError(Throwable e) {
            }
            @Override
            public void onComplete() {
            }
        });
    }
    // Fake method for sending a request to the server
    private void sendRequestToServer(String query){
        // do nothing
    }
    @Override
    protected void onDestroy() {
        super.onDestroy();
        disposables.clear(); // clear disposables
    }

عملگرهای تبدیل کننده : ThrottleFirst

مثال : اسپم شدن Button را محدود کنید

def rxbinding_version = "4.0.0"
// Rx Binding Library
implementation "com.jakewharton.rxbinding4:rxbinding:$rxbinding_version"

<?xml version="1.0" encoding="utf-8"?>
<androidx.constraintlayout.widget.ConstraintLayout
    xmlns:android="http://schemas.android.com/apk/res/android"
    xmlns:app="http://schemas.android.com/apk/res-auto"
    xmlns:tools="http://schemas.android.com/tools"
    android:layout_width="match_parent"
    android:layout_height="match_parent"
    tools:context=".MainActivity">
    <Button
        android:layout_width="match_parent"
        android:layout_height="wrap_content"
        app:layout_constraintTop_toTopOf="parent"
        app:layout_constraintBottom_toBottomOf="parent"
        app:layout_constraintRight_toRightOf="parent"
        app:layout_constraintLeft_toLeftOf="parent"
        android:id="@+id/button"
        android:text="button"/>
  </androidx.constraintlayout.widget.ConstraintLayout>

•  یک Click Listener برای Button با RxBinding Library ایجاد کنید.
• با کنترل کردن کلیک پش از گذشت 500 میلی ثانیه می‌توانید یک کلیک جدید ثبت کنید.
• هنگام ثبت یک کلیک متد هایی را اجرا کنید.
•  Observable را Dispose کنید.

public class MainActivity extends AppCompatActivity {
    private static final String TAG = "MainActivity";
    //ui
    private Button button;
    // vars
    private CompositeDisposable disposables = new CompositeDisposable();
    private long timeSinceLastRequest;
    @Override
    protected void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        setContentView(R.layout.activity_main);
        button = findViewById(R.id.button);
        timeSinceLastRequest = System.currentTimeMillis();
        // Set a click listener to the button with RxBinding Library
        RxView.clicks(button)
            .throttleFirst(500, TimeUnit.MILLISECONDS) // Throttle the clicks so 500 ms must pass before registering a new click
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(new Observer<Unit>() {
                @Override
                public void onSubscribe(Disposable d) {
                    disposables.add(d);
                }
                @Override
                public void onNext(Unit unit) {
                    Log.d(TAG, "onNext: time since last clicked: " + (System.currentTimeMillis() - timeSinceLastRequest));
                    someMethod(); // Execute some method when a click is registered
                }
                @Override
                public void onError(Throwable e) {
                }
                @Override
                public void onComplete() {
                }
            });
    }
    private void someMethod(){
        timeSinceLastRequest = System.currentTimeMillis();
        // do something
    }
    @Override
    protected void onDestroy() {
        super.onDestroy();
        disposables.clear(); // Dispose observable
    }
}

عملگرهای تبدیل کننده : FlatMap

• ساخت Observables از اشیاء منتشر شده توسط Observables دیگر.
• ترکیب کردن منبع چندین Observables به Single Obsevables (این چیزی است که به عنوان "flattening" شناخته می‌شود).

apply plugin: 'com.android.application'
android {
    compileSdkVersion 28
    defaultConfig {
        applicationId "com.codingwithmitch.rxjavaflatmapexample"
        minSdkVersion 21
        targetSdkVersion 28
        versionCode 1
        versionName "1.0"
        testInstrumentationRunner "androidx.test.runner.AndroidJUnitRunner"
    }
    buildTypes {
        release {
            minifyEnabled false
            proguardFiles getDefaultProguardFile('proguard-android-optimize.txt'), 'proguard-rules.pro'
        }
    }
}
dependencies {
    def retrofitVersion = "2.5.0"
    def rxjava_version = '2.2.7'
    def rxandroid_version = '2.1.1'
    def recyclerview_version = "1.0.0"
    implementation fileTree(dir: 'libs', include: ['*.jar'])
    implementation 'androidx.appcompat:appcompat:1.0.0-beta01'
    implementation 'androidx.constraintlayout:constraintlayout:1.1.2'
    testImplementation 'junit:junit:4.12'
    androidTestImplementation 'androidx.test:runner:1.1.0-alpha4'
    androidTestImplementation 'androidx.test.espresso:espresso-core:3.1.0-alpha4'
    // Retrofit
    implementation "com.squareup.retrofit2:retrofit:$retrofitVersion"
    implementation "com.squareup.retrofit2:converter-gson:$retrofitVersion"
    // RxJava
    implementation "io.reactivex.rxjava2:rxjava:$rxjava_version"
    // RxJava Call Adapter
    implementation "com.squareup.retrofit2:adapter-rxjava2:2.5.0"
    // RxAndroid
    implementation "io.reactivex.rxjava2:rxandroid:$rxandroid_version"
    // Recyclerview
    implementation "androidx.recyclerview:recyclerview:$recyclerview_version"
}

<?xml version="1.0" encoding="utf-8"?>
<manifest xmlns:android="http://schemas.android.com/apk/res/android"
    package="com.codingwithmitch.rxjavaflatmapexample">
    <uses-permission android:name="android.permission.INTERNET"/>
    <application
        android:allowBackup="true"
        android:icon="@mipmap/ic_launcher"
        android:label="@string/app_name"
        android:roundIcon="@mipmap/ic_launcher_round"
        android:supportsRtl="true"
        android:theme="@style/AppTheme">
        <activity android:name=".MainActivity">
            <intent-filter>
                <action android:name="android.intent.action.MAIN" />
                <category android:name="android.intent.category.LAUNCHER" />
            </intent-filter>
        </activity>
    </application>
</manifest>

<?xml version="1.0" encoding="utf-8"?>
<androidx.constraintlayout.widget.ConstraintLayout xmlns:android="http://schemas.android.com/apk/res/android"
    xmlns:app="http://schemas.android.com/apk/res-auto"
    xmlns:tools="http://schemas.android.com/tools"
    android:layout_width="match_parent"
    android:layout_height="match_parent"
    tools:context=".MainActivity">
    <androidx.recyclerview.widget.RecyclerView
        android:layout_width="match_parent"
        android:layout_height="match_parent"
        android:id="@+id/recycler_view"
        android:orientation="vertical">
    </androidx.recyclerview.widget.RecyclerView>
</androidx.constraintlayout.widget.ConstraintLayout>

//This is the layout for the RecyclerView list-items
<?xml version="1.0" encoding="utf-8"?>
<LinearLayout xmlns:android="http://schemas.android.com/apk/res/android"
    android:orientation="horizontal"
    android:layout_width="match_parent"
    android:layout_height="wrap_content"
    android:weightSum="100"
    android:padding="20dp">
    <RelativeLayout
        android:layout_width="0dp"
        android:layout_height="wrap_content"
        android:layout_weight="80"
        android:layout_gravity="center_vertical" >
        <TextView
            android:layout_width="match_parent"
            android:layout_height="wrap_content"
            android:id="@+id/title"
            android:text="this is a title"
            android:textColor="#000"
            android:textSize="17sp"
            />
    </RelativeLayout>
    <RelativeLayout
        android:layout_width="0dp"
        android:layout_height="wrap_content"
        android:layout_weight="20"
        android:layout_gravity="center_vertical">
        <TextView
            android:layout_width="wrap_content"
            android:layout_height="wrap_content"
            android:id="@+id/num_comments"
            android:textSize="14sp"
            android:layout_centerInParent="true"/>
        <ProgressBar
            android:layout_width="20dp"
            android:layout_height="20dp"
            android:id="@+id/progress_bar"
            style="@style/Widget.AppCompat.ProgressBar"
            android:layout_centerInParent="true"
            android:visibility="gone"/>
    </RelativeLayout>
</LinearLayout>

public class Post {
    @SerializedName("userId")
    @Expose()
    private int userId;
    @SerializedName("id")
    @Expose()
    private int id;
    @SerializedName("title")
    @Expose()
    private String title;
    @SerializedName("body")
    @Expose()
    private String body;
    private List<Comment> comments;
    public Post(int userId, int id, String title, String body, List<Comment> comments) {
        this.userId = userId;
        this.id = id;
        this.title = title;
        this.body = body;
        this.comments = comments;
    }
    public int getUserId() {
        return userId;
    }
    public void setUserId(int userId) {
        this.userId = userId;
    }
    public int getId() {
        return id;
    }
    public void setId(int id) {
        this.id = id;
    }
    public String getTitle() {
        return title;
    }
    public void setTitle(String title) {
        this.title = title;
    }
    public String getBody() {
        return body;
    }
    public void setBody(String body) {
        this.body = body;
    }
    public List<Comment> getComments() {
        return comments;
    }
    public void setComments(List<Comment> comments) {
        this.comments = comments;
    }
    @Override
    public String toString() {
        return "Post{" +
                "userId=" + userId +
                ", id=" + id +
                ", title='" + title + '\'' +
                ", body='" + body + '\'' +
                '}';
    }
}

public class Comment {
    @Expose
    @SerializedName("postId")
    private int postId;
    @Expose
    @SerializedName("id")
    private int id;
    @Expose
    @SerializedName("name")
    private String name;
    @Expose
    @SerializedName("email")
    private String email;
    @Expose
    @SerializedName("body")
    private String body;
    public Comment(int postId, int id, String name, String email, String body) {
        this.postId = postId;
        this.id = id;
        this.name = name;
        this.email = email;
        this.body = body;
    }
    public int getPostId() {
        return postId;
    }
    public void setPostId(int postId) {
        this.postId = postId;
    }
    public int getId() {
        return id;
    }
    public void setId(int id) {
        this.id = id;
    }
    public String getName() {
        return name;
    }
    public void setName(String name) {
        this.name = name;
    }
    public String getEmail() {
        return email;
    }
    public void setEmail(String email) {
        this.email = email;
    }
    public String getBody() {
        return body;
    }
    public void setBody(String body) {
        this.body = body;
    }
}

RequestApi.java

• ()getPosts لیست پست‌ها را بازیابی می‌کند.
• ()getComments لیستی از نظرات را برای یک پست خاص بازیابی می‌کند.

public interface RequestApi {
    @GET("posts")
    Observable<List<Post>> getPosts();
    @GET("posts/{id}/comments")
    Observable<List<Comment>> getComments(
            @Path("id") int id
    );
}

ServiceGenerator.java

public class ServiceGenerator {
    public static final String BASE_URL = "https://jsonplaceholder.typicode.com";
    private static Retrofit.Builder retrofitBuilder =
            new Retrofit.Builder()
                    .baseUrl(BASE_URL)
                    .addCallAdapterFactory(RxJava2CallAdapterFactory.create())
                    .addConverterFactory(GsonConverterFactory.create());
    private static Retrofit retrofit = retrofitBuilder.build();
    private static RequestApi requestApi = retrofit.create(RequestApi.class);
    public static RequestApi getRequestApi(){
        return requestApi;
    }
}

RecyclerAdapter.java

public class RecyclerAdapter extends RecyclerView.Adapter<RecyclerAdapter.MyViewHolder> {
    private static final String TAG = "RecyclerAdapter";
    private List<Post> posts = new ArrayList<>();
    @NonNull
    @Override
    public MyViewHolder onCreateViewHolder(@NonNull ViewGroup parent, int viewType) {
        View view = LayoutInflater.from(parent.getContext()).inflate(R.layout.layout_post_list_item, null, false);
        return new MyViewHolder(view);
    }
    @Override
    public void onBindViewHolder(@NonNull MyViewHolder holder, int position) {
        holder.bind(posts.get(position));
    }
    @Override
    public int getItemCount() {
        return posts.size();
    }
    public void setPosts(List<Post> posts){
        this.posts = posts;
        notifyDataSetChanged();
    }
    public void updatePost(Post post){
        posts.set(posts.indexOf(post), post);
        notifyItemChanged(posts.indexOf(post));
    }
    public List<Post> getPosts(){
        return posts;
    }
    public class MyViewHolder extends RecyclerView.ViewHolder{
        TextView title, numComments;
        ProgressBar progressBar;
        public MyViewHolder(@NonNull View itemView) {
            super(itemView);
            title = itemView.findViewById(R.id.title);
            numComments = itemView.findViewById(R.id.num_comments);
            progressBar = itemView.findViewById(R.id.progress_bar);
        }
        public void bind(Post post){
            title.setText(post.getTitle());
            if(post.getComments() == null){
                showProgressBar(true);
                numComments.setText("");
            }
            else{
                showProgressBar(false);
                numComments.setText(String.valueOf(post.getComments().size()));
            }
        }
        private void showProgressBar(boolean showProgressBar){
            if(showProgressBar) {
                progressBar.setVisibility(View.VISIBLE);
            }
            else{
                progressBar.setVisibility(View.GONE);
            }
        }
    }
}

MainActivity.java

public class MainActivity extends AppCompatActivity {
    private static final String TAG = "MainActivity";
    //ui
    private RecyclerView recyclerView;
    // vars
    private CompositeDisposable disposables = new CompositeDisposable();
    private RecyclerAdapter adapter;
    @Override
    protected void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        setContentView(R.layout.activity_main);
        recyclerView = findViewById(R.id.recycler_view);
        initRecyclerView();
        getPostsObservable()
                .subscribeOn(Schedulers.io())
                .flatMap(new Function<Post, ObservableSource<Post>>() {
                    @Override
                    public ObservableSource<Post> apply(Post post) throws Exception {
                        return getCommentsObservable(post);
                    }
                })
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Observer<Post>() {
                    @Override
                    public void onSubscribe(Disposable d) {
                        disposables.add(d);
                    }
                    @Override
                    public void onNext(Post post) {
                        updatePost(post);
                    }
                    @Override
                    public void onError(Throwable e) {
                        Log.e(TAG, "onError: ", e);
                    }
                    @Override
                    public void onComplete() {
                    }
                });
    }
    private Observable<Post> getPostsObservable(){
        return ServiceGenerator.getRequestApi()
                .getPosts()
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .flatMap(new Function<List<Post>, ObservableSource<Post>>() {
                    @Override
                    public ObservableSource<Post> apply(final List<Post> posts) throws Exception {
                        adapter.setPosts(posts);
                        return Observable.fromIterable(posts)
                                .subscribeOn(Schedulers.io());
                    }
                });
    }
    private void updatePost(final Post p){
        Observable
                .fromIterable(adapter.getPosts())
                .filter(new Predicate<Post>() {
                    @Override
                    public boolean test(Post post) throws Exception {
                        return post.getId() == p.getId();
                    }
                })
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Observer<Post>() {
                    @Override
                    public void onSubscribe(Disposable d) {
                        disposables.add(d);
                    }
                    @Override
                    public void onNext(Post post) {
                        Log.d(TAG, "onNext: updating post: " + post.getId() + ", thread: " + Thread.currentThread().getName());
                        adapter.updatePost(post);
                    }
                    @Override
                    public void onError(Throwable e) {
                        Log.e(TAG, "onError: ", e);
                    }
                    @Override
                    public void onComplete() {
                    }
                });
    }
    private Observable<Post> getCommentsObservable(final Post post){
        return ServiceGenerator.getRequestApi()
                .getComments(post.getId())
                .map(new Function<List<Comment>, Post>() {
                    @Override
                    public Post apply(List<Comment> comments) throws Exception {
                        int delay = ((new Random()).nextInt(5) + 1) * 1000; // sleep thread for x ms
                        Thread.sleep(delay);
                        Log.d(TAG, "apply: sleeping thread " + Thread.currentThread().getName() + " for " + String.valueOf(delay)+ "ms");
                        post.setComments(comments);
                        return post;
                    }
                })
                .subscribeOn(Schedulers.io());
    }
    private void initRecyclerView(){
        adapter = new RecyclerAdapter();
        recyclerView.setLayoutManager(new LinearLayoutManager(this));
        recyclerView.setAdapter(adapter);
    }
    @Override
    protected void onDestroy() {
        super.onDestroy();
        disposables.clear();
    }
}

عملگرهای تبدیل کننده : ConcatMap

مثالی از ConcatMap:

MainActivity.java

package com.codingwithmitch.rxjavaflatmapexample;
import androidx.appcompat.app.AppCompatActivity;
import androidx.recyclerview.widget.LinearLayoutManager;
import androidx.recyclerview.widget.RecyclerView;
import io.reactivex.Observable;
import io.reactivex.ObservableSource;
import io.reactivex.Observer;
import io.reactivex.android.schedulers.AndroidSchedulers;
import io.reactivex.disposables.CompositeDisposable;
import io.reactivex.disposables.Disposable;
import io.reactivex.functions.Function;
import io.reactivex.functions.Predicate;
import io.reactivex.schedulers.Schedulers;
import android.os.Bundle;
import android.util.Log;
import com.codingwithmitch.rxjavaflatmapexample.models.Comment;
import com.codingwithmitch.rxjavaflatmapexample.models.Post;
import com.codingwithmitch.rxjavaflatmapexample.requests.ServiceGenerator;
import java.util.List;
import java.util.Random;
public class MainActivity extends AppCompatActivity {
    private static final String TAG = "MainActivity";
    //ui
    private RecyclerView recyclerView;
    // vars
    private CompositeDisposable disposables = new CompositeDisposable();
    private RecyclerAdapter adapter;
    @Override
    protected void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        setContentView(R.layout.activity_main);
        recyclerView = findViewById(R.id.recycler_view);
        initRecyclerView();
        getPostsObservable()
                .subscribeOn(Schedulers.io())
                .concatMap(new Function<Post, ObservableSource<Post>>() {
                    @Override
                    public ObservableSource<Post> apply(Post post) throws Exception {
                        return getCommentsObservable(post);
                    }
                })
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Observer<Post>() {
                    @Override
                    public void onSubscribe(Disposable d) {
                        disposables.add(d);
                    }
                    @Override
                    public void onNext(Post post) {
                        updatePost(post);
                    }
                    @Override
                    public void onError(Throwable e) {
                        Log.e(TAG, "onError: ", e);
                    }
                    @Override
                    public void onComplete() {
                    }
                });
    }
    private Observable<Post> getPostsObservable(){
        return ServiceGenerator.getRequestApi()
                .getPosts()
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .flatMap(new Function<List<Post>, ObservableSource<Post>>() {
                    @Override
                    public ObservableSource<Post> apply(final List<Post> posts) throws Exception {
                        adapter.setPosts(posts);
                        return Observable.fromIterable(posts)
                                .subscribeOn(Schedulers.io());
                    }
                });
    }
    private void updatePost(Post post){
        adapter.updatePost(post);
    }
    private Observable<Post> getCommentsObservable(final Post post){
        return ServiceGenerator.getRequestApi()
                .getComments(post.getId())
                .map(new Function<List<Comment>, Post>() {
                    @Override
                    public Post apply(List<Comment> comments) throws Exception {
                        int delay = ((new Random()).nextInt(5) + 1) * 1000; // sleep thread for x ms
                        Thread.sleep(delay);
                        Log.d(TAG, "apply: sleeping thread " + Thread.currentThread().getName() + " for " + String.valueOf(delay)+ "ms");
                        post.setComments(comments);
                        return post;
                    }
                })
                .subscribeOn(Schedulers.io());
    }
    private void initRecyclerView(){
        adapter = new RecyclerAdapter();
        recyclerView.setLayoutManager(new LinearLayoutManager(this));
        recyclerView.setAdapter(adapter);
    }
    @Override
    protected void onDestroy() {
        super.onDestroy();
        disposables.clear();
    }
}

عملگرهای تبدیل کننده : SwitchMap

حالت 1:

حالت 2:

مثالی از ()SwitchMap:

• مرحله‌ی 1:
  لیستی از پست‌های وبلاگ را از آدرس اینترنتی (URL) بازیابی کنید: /jsonplaceholder.typicode.com/posts
• مرحله‌ی 2: هنگامی که یک مورد لیست کلیک می‌شود، درخواست دیگری به json placeholder api ارسال می‌شود، اما این بار برای یک پست خاص. بنابراین آدرس اینترنتی چیزی شبیه به این خواهد بود: /jsonplaceholder.typicode.com/posts/5 با این کار پست با id = 5 بازیابی می‌شود.
• مرحله‌ی 3: قبل از اینکه درخواست انجام شود‌، ما شبیه سازی می‌کنیم که شبکه ای با سرعت عملکرد پایین یا کند چگونه باشد. ما درخواست را مجبور می‌کنیم قبل از اجرا 3000 ms منتظر بماند. progress bar بروزرسانی می‌شود که زمان سپری شده از کلیک یک آیتم لیست را نشان می‌دهد. سرعت آهسته‌ی شبکه با استفاده از اپراتور ()Interval، اپراتور ()TakeWhile و ()Filter شبیه سازی می‌شود.
• مرحله 4: اگر کاربر قبل از اجرای درخواست شبکه، مورد جدیدی از لیست را کلیک کند، کل فرایند دوباره شروع می‌شود. اپراتور ()SwitchMap  از این امر مراقبت می‌کند.
• مرحله 5: در صورت تکمیل فرآیند (پس از گذشت 3000 میلی ثانیه بدون کلیک بر روی گزینه ای جدید)، درخواست انجام می‌شود و کاربر به یک Activity جدید که عنوان پست را نشان می‌دهد، هدایت می‌شوید.

build.gradle

apply plugin: 'com.android.application'
android {
    compileSdkVersion 28
    defaultConfig {
        applicationId "com.codingwithmitch.rxjavaflatmapexample"
        minSdkVersion 21
        targetSdkVersion 28
        versionCode 1
        versionName "1.0"
        testInstrumentationRunner "androidx.test.runner.AndroidJUnitRunner"
    }
    buildTypes {
        release {
            minifyEnabled false
            proguardFiles getDefaultProguardFile('proguard-android-optimize.txt'), 'proguard-rules.pro'
        }
    }
}
dependencies {
    def retrofitVersion = "2.5.0"
    def rxjava_version = '2.2.7'
    def rxandroid_version = '2.1.1'
    def recyclerview_version = "1.0.0"
    implementation fileTree(dir: 'libs', include: ['*.jar'])
    implementation 'androidx.appcompat:appcompat:1.0.0-beta01'
    implementation 'androidx.constraintlayout:constraintlayout:1.1.2'
    testImplementation 'junit:junit:4.12'
    androidTestImplementation 'androidx.test:runner:1.1.0-alpha4'
    androidTestImplementation 'androidx.test.espresso:espresso-core:3.1.0-alpha4'
    // Retrofit
    implementation "com.squareup.retrofit2:retrofit:$retrofitVersion"
    implementation "com.squareup.retrofit2:converter-gson:$retrofitVersion"
    // RxJava
    implementation "io.reactivex.rxjava2:rxjava:$rxjava_version"
    // RxJava Call Adapter
    implementation "com.squareup.retrofit2:adapter-rxjava2:2.5.0"
    // RxAndroid
    implementation "io.reactivex.rxjava2:rxandroid:$rxandroid_version"
    // Recyclerview
    implementation "androidx.recyclerview:recyclerview:$recyclerview_version"
}

AndroidManifest.xml

<?xml version="1.0" encoding="utf-8"?>
<manifest xmlns:android="http://schemas.android.com/apk/res/android"
    package="com.codingwithmitch.rxjavaflatmapexample">
    <uses-permission android:name="android.permission.INTERNET"/>
    <application
        android:allowBackup="true"
        android:icon="@mipmap/ic_launcher"
        android:label="@string/app_name"
        android:roundIcon="@mipmap/ic_launcher_round"
        android:supportsRtl="true"
        android:theme="@style/AppTheme">
        <activity android:name=".MainActivity">
            <intent-filter>
                <action android:name="android.intent.action.MAIN" />
                <category android:name="android.intent.category.LAUNCHER" />
            </intent-filter>
        </activity>
        <activity android:name=".ViewPostActivity" />
    </application>
</manifest>

activity_main.xml

<?xml version="1.0" encoding="utf-8"?>
<LinearLayout
    xmlns:android="http://schemas.android.com/apk/res/android"
    xmlns:app="http://schemas.android.com/apk/res-auto"
    xmlns:tools="http://schemas.android.com/tools"
    android:orientation="vertical"
    android:layout_width="match_parent"
    android:layout_height="match_parent"
    tools:context=".MainActivity">
    <ProgressBar
        android:id="@+id/progress_bar"
        style="@style/Widget.AppCompat.ProgressBar.Horizontal"
        android:layout_width="match_parent"
        android:layout_height="wrap_content"
        android:layout_marginTop="16dp"
        android:padding="10dp"
        />
    <androidx.recyclerview.widget.RecyclerView
        android:id="@+id/recycler_view"
        android:layout_width="match_parent"
        android:layout_height="match_parent"
        android:orientation="vertical">
    </androidx.recyclerview.widget.RecyclerView>
</LinearLayout>

layout_post_list_item.xml

//This is the layout for the RecyclerView list-items.
<?xml version="1.0" encoding="utf-8"?>
<LinearLayout xmlns:android="http://schemas.android.com/apk/res/android"
    android:orientation="horizontal"
    android:layout_width="match_parent"
    android:layout_height="wrap_content"
    android:padding="20dp">
    <TextView
        android:layout_width="match_parent"
        android:layout_height="wrap_content"
        android:id="@+id/title"
        android:text="this is a title"
        android:textColor="#000"
        android:textSize="17sp"
        />
</LinearLayout>

activity_view_post.xml

<?xml version="1.0" encoding="utf-8"?>
<androidx.constraintlayout.widget.ConstraintLayout
    xmlns:android="http://schemas.android.com/apk/res/android"
    xmlns:app="http://schemas.android.com/apk/res-auto"
    xmlns:tools="http://schemas.android.com/tools"
    android:layout_width="match_parent"
    android:layout_height="match_parent"
    tools:context=".MainActivity"
    android:padding="15dp">
    <TextView
        android:layout_width="wrap_content"
        android:layout_height="wrap_content"
        android:id="@+id/text"
        app:layout_constraintTop_toTopOf="parent"
        app:layout_constraintRight_toRightOf="parent"
        app:layout_constraintLeft_toLeftOf="parent"
        app:layout_constraintBottom_toBottomOf="parent"/>
</androidx.constraintlayout.widget.ConstraintLayout>

Post.java

public class Post implements Parcelable {
    @SerializedName("userId")
    @Expose()
    private int userId;
    @SerializedName("id")
    @Expose()
    private int id;
    @SerializedName("title")
    @Expose()
    private String title;
    @SerializedName("body")
    @Expose()
    private String body;
    private List<Comment> comments;
    public Post(int userId, int id, String title, String body, List<Comment> comments) {
        this.userId = userId;
        this.id = id;
        this.title = title;
        this.body = body;
        this.comments = comments;
    }
    protected Post(Parcel in) {
        userId = in.readInt();
        id = in.readInt();
        title = in.readString();
        body = in.readString();
    }
    public static final Creator<Post> CREATOR = new Creator<Post>() {
        @Override
        public Post createFromParcel(Parcel in) {
            return new Post(in);
        }
        @Override
        public Post[] newArray(int size) {
            return new Post[size];
        }
    };
    public int getUserId() {
        return userId;
    }
    public void setUserId(int userId) {
        this.userId = userId;
    }
    public int getId() {
        return id;
    }
    public void setId(int id) {
        this.id = id;
    }
    public String getTitle() {
        return title;
    }
    public void setTitle(String title) {
        this.title = title;
    }
    public String getBody() {
        return body;
    }
    public void setBody(String body) {
        this.body = body;
    }
    public List<Comment> getComments() {
        return comments;
    }
    public void setComments(List<Comment> comments) {
        this.comments = comments;
    }
    @Override
    public String toString() {
        return "Post{" +
                "userId=" + userId +
                ", id=" + id +
                ", title='" + title + '\'' +
                ", body='" + body + '\'' +
                '}';
    }
    @Override
    public int describeContents() {
        return 0;
    }
    @Override
    public void writeToParcel(Parcel dest, int flags) {
        dest.writeInt(userId);
        dest.writeInt(id);
        dest.writeString(title);
        dest.writeString(body);
    }
}

RequestApi.java

• ()getPosts لیست پست‌ها را بازیابی می‌کند.
• ()getComments لیستی از نظرات را برای یک پست خاص بازیابی می‌کند.

public interface RequestApi {
    @GET("posts")
    Observable<List<Post>> getPosts();
    @GET("posts/{id}")
    Observable<Post> getPost(
            @Path("id") int id
    );
}

ServiceGenerator.java

public class ServiceGenerator {
    public static final String BASE_URL = "https://jsonplaceholder.typicode.com";
    private static Retrofit.Builder retrofitBuilder =
            new Retrofit.Builder()
                    .baseUrl(BASE_URL)
                    .addCallAdapterFactory(RxJava2CallAdapterFactory.create())
                    .addConverterFactory(GsonConverterFactory.create());
    private static Retrofit retrofit = retrofitBuilder.build();
    private static RequestApi requestApi = retrofit.create(RequestApi.class);
    public static RequestApi getRequestApi(){
        return requestApi;
    }
}

RecyclerAdapter.java

public class RecyclerAdapter extends RecyclerView.Adapter<RecyclerAdapter.MyViewHolder> {
    private static final String TAG = "RecyclerAdapter";
    private List<Post> posts = new ArrayList<>();
    private OnPostClickListener onPostClickListener;
    public RecyclerAdapter(OnPostClickListener onPostClickListener) {
        this.onPostClickListener = onPostClickListener;
    }
    @NonNull
    @Override
    public MyViewHolder onCreateViewHolder(@NonNull ViewGroup parent, int viewType) {
        View view = LayoutInflater.from(parent.getContext()).inflate(R.layout.layout_post_list_item, null, false);
        return new MyViewHolder(view, onPostClickListener);
    }
    @Override
    public void onBindViewHolder(@NonNull MyViewHolder holder, int position) {
        holder.bind(posts.get(position));
    }
    @Override
    public int getItemCount() {
        return posts.size();
    }
    public void setPosts(List<Post> posts){
        this.posts = posts;
        notifyDataSetChanged();
    }
    public void updatePost(Post post){
        posts.set(posts.indexOf(post), post);
        notifyItemChanged(posts.indexOf(post));
    }
    public List<Post> getPosts(){
        return posts;
    }
    public class MyViewHolder extends RecyclerView.ViewHolder implements View.OnClickListener {
        OnPostClickListener onPostClickListener;
        TextView title;
        public MyViewHolder(@NonNull View itemView, OnPostClickListener onPostClickListener) {
            super(itemView);
            title = itemView.findViewById(R.id.title);
            this.onPostClickListener = onPostClickListener;
            itemView.setOnClickListener(this);
        }
        public void bind(Post post){
            title.setText(post.getTitle());
        }
        @Override
        public void onClick(View v) {
            onPostClickListener.onPostClick(getAdapterPosition());
        }
    }
    public interface OnPostClickListener{
        void onPostClick(int position);
    }
}

ViewPostActivity.java

public class ViewPostActivity extends AppCompatActivity {
    private static final String TAG = "ViewPostActivity";
    private TextView text;
    @Override
    protected void onCreate(@Nullable Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        setContentView(R.layout.activity_view_post);
        text = findViewById(R.id.text);
        getIncomingIntent();
    }
    private void getIncomingIntent(){
        if(getIntent().hasExtra("post")){
            Post post = getIntent().getParcelableExtra("post");
            text.setText(post.getTitle());
        }
    }
}

MainActivity.java

1. ما از یک شیء PublishSubject استفاده می‌کنیم تا تشخیص دهیم که یک Post Object از لیست انتخاب شده است. وقتی تنظیم شود، فرایند ایجاد Observable را شروع می‌کند. می‌توانید متد Override شده onPostClick را مشاهده کنید.
2. سپس ما از عملگر SwitchMap استفاده می‌کنیم و یک <Observable<Post را از یک Post Object ایجاد می‌کنیم. این تضمین می‌کند که فقط یک Observer در هر زمان معین Observable را Subscribe می‌کند. بنابراین اگر پست دیگری انتخاب شود، از این رو  Observer قبلی از بین می‌رود، بنابراین این فرآیند را دوباره تنظیم می‌کند.
3. ()takeWhile() ،Interval و ()Filter برای شبیه سازی یک اتصال ضعیف به شبکه هستند. آن‌ها progress bar را تنظیم می‌کنند و زمان سپری شده را ردیابی می‌کنند.
4. اگر زمان سپری شود، اپراتور ()FlapMap کار را با تبدیل یک Long Object (حاصل عملگر Interval) به یک  Observable<Post> Object پایان می‌دهد.
5. و در آخر، Object Post به Observer منتقل می‌شود.
6. بنابراین هر بار که یک پست جدید از لیست انتخاب می‌شود، فرآیند به دلیل وجود عملگر ()SwitchMap دوباره شروع می‌شود.

public class MainActivity extends AppCompatActivity implements RecyclerAdapter.OnPostClickListener {
    private static final String TAG = "MainActivity";
    //ui
    private RecyclerView recyclerView;
    private ProgressBar progressBar;
    // vars
    private CompositeDisposable disposables = new CompositeDisposable();
    private RecyclerAdapter adapter;
    private PublishSubject<Post> publishSubject = PublishSubject.create(); // for selecting a post
    private static final int PERIOD = 100;
    @Override
    protected void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        setContentView(R.layout.activity_main);
        recyclerView = findViewById(R.id.recycler_view);
        progressBar = findViewById(R.id.progress_bar);
        initRecyclerView();
        retrievePosts();
    }
    private void initSwitchMapDemo(){
        publishSubject
                // apply switchmap operator so only one Observable can be used at a time.
                // it clears the previous one
                .switchMap(new Function<Post, ObservableSource<Post>>() {
                    @Override
                    public ObservableSource<Post> apply(final Post post) throws Exception {
                        return Observable
                                // simulate slow network speed with interval + takeWhile + filter operators
                                .interval(PERIOD, TimeUnit.MILLISECONDS)
                                .subscribeOn(AndroidSchedulers.mainThread())
                                .takeWhile(new Predicate<Long>() { // stop the process if more than 5 seconds passes
                                    @Override
                                    public boolean test(Long aLong) throws Exception {
                                        Log.d(TAG, "test: " + Thread.currentThread().getName() + ", " + aLong);
                                        progressBar.setMax(3000 - PERIOD);
                                        progressBar.setProgress(Integer.parseInt(String.valueOf((aLong * PERIOD) + PERIOD)));
                                        return aLong <= (3000 / PERIOD);
                                    }
                                })
                                .filter(new Predicate<Long>() {
                                    @Override
                                    public boolean test(Long aLong) throws Exception {
                                        return aLong >= (3000 / PERIOD);
                                    }
                                })
                                // flatmap to convert Long from the interval operator into a Observable<Post>
                                .subscribeOn(Schedulers.io())
                                .flatMap(new Function<Long, ObservableSource<Post>>() {
                                    @Override
                                    public ObservableSource<Post> apply(Long aLong) throws Exception {
                                        return ServiceGenerator.getRequestApi()
                                                .getPost(post.getId());
                                    }
                                });
                    }
                })
                .subscribe(new Observer<Post>() {
                    @Override
                    public void onSubscribe(Disposable d) {
                        disposables.add(d);
                    }
                    @Override
                    public void onNext(Post post) {
                        Log.d(TAG, "onNext: done.");
                        navViewPostActivity(post);
                    }
                    @Override
                    public void onError(Throwable e) {
                        Log.e(TAG, "onError: ", e);
                    }
                    @Override
                    public void onComplete() {
                    }
                });
    }
    private void retrievePosts(){
        ServiceGenerator.getRequestApi()
                .getPosts()
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
        .subscribe(new Observer<List<Post>>() {
            @Override
            public void onSubscribe(Disposable d) {
                disposables.add(d);
            }
            @Override
            public void onNext(List<Post> posts) {
                adapter.setPosts(posts);
            }
            @Override
            public void onError(Throwable e) {
                Log.e(TAG, "onError: ", e);
            }
            @Override
            public void onComplete() {
            }
        });
    }
    @Override
    protected void onResume() {
        super.onResume();
        progressBar.setProgress(0);
        initSwitchMapDemo();
    }
    private void initRecyclerView(){
        adapter = new RecyclerAdapter(this);
        recyclerView.setLayoutManager(new LinearLayoutManager(this));
        recyclerView.setAdapter(adapter);
    }
    private void navViewPostActivity(Post post){
        Intent intent = new Intent(this, ViewPostActivity.class);
        intent.putExtra("post", post);
        startActivity(intent);
    }
    @Override
    protected void onPause() {
        Log.d(TAG, "onPause: called.");
        disposables.clear();
        super.onPause();
    }
    @Override
    public void onPostClick(final int position) {
        Log.d(TAG, "onPostClick: clicked.");
        // submit the selected post object to be queried
        publishSubject.onNext(adapter.getPosts().get(position));
    }
}

منابع این مقاله:

اگر به یادگیری بیشتر در زمینه‌ی برنامه نویسی اندروید علاقه داری، با شرکت در دوره‌ی برنامه نویسی اندروید در کمتر از یکسال به یک توسعه‌دهنده اندروید همه فن حریف تبدیل می‌شوی که آماده‌ی استخدام، دریافت پروژه و حتی پیاده‌سازی اپلیکیشن خودت هستی.