-

<dependency>
    <groupId>io.reactivex</groupId>
    <artifactId>rxjava</artifactId>
    <version>${rx.java.version}</version>
</dependency>

این کتابخانه را با Maven اضافه میکنیم

اکنون دو تا چیز را باید مد نظر داشته باشیم  یکی دیتای قابل شنود observable

 و دیگری اشیای شنود کننده یا همون  observer

این اشیا برای شنود باید روی دیتا یک subscribe  انجام بدهند تا قابلیت شنود داشته باشند

در این صورت هر تغییری در دیتا مستقیما درون متد لامبادای سابسکرایپ ظاهر میشود.

در اینجا ما دو گونه متفاوت می توانیم داشته باشیم  یکی پشت سرهم none-blocking  که برای asynchronous استفاده می شود که اکثرا هم هدف همان است فلذا اون یکی رو بیخیال میشم

بعد از جاوا 8 که لامبادا تشریف آورد اینکارها آسون تر شده وقتی سابیکرایپ میکنیم سه تا وضعیت خروجی توسط RX حمایت میشه یکی در حالت آمدن دیتای بعدی ، یکی در حالت ارور و دیگری در حالت خروجی کامل که ما میتونیم داخل همون متد سابسکرایپ کار مورد نظر رو هندل کنیم

String[] li= {"a", "b", "c", "d", "e", "f", "g"};
Observable<String> observable = Observable.from(li);
observable.subscribe(
  i -> result += i,  //OnNext
  Throwable::printStackTrace, //OnError
  () -> result += "_Completed" //OnCompleted
);
assertTrue(result.equals("abcdefg_Completed"));

الان اگر دقت کنیدتوی متد سابسکرایپ در صورت دریافت هر دیتای جدید توی لیست  OnNext رو اینطوری بدست میارید که البته معمولا onComplete بیشتر بدرد می خوره

متناسب با چیزی که نیاز داریم می تونیم از map وfilter , groupBy لامبادا پشت متد استفاده کنیم

تا یادم نرفته اگر لامبادا ساپورت نیستین می توننین مثل روش قدیم داخل سابسکرایپ new کنین

    return new Observer<String>() {
        @Override
        public void onNext(String value) {
           //use value
        }
 
        @Override
        public void onError(Throwable e) {
            System.out.println("error");
        }
 
        @Override
        public void onCompleted() {
            System.out.println("Subscriber1 completed");
        }
    };