آموزش عملگر ها یا اپراتور های Distinct ، Reduce و Count در RxJava

 Observable numbersObservable = Observable.just(10,10, 15, 20, 100, 200, 100, 300, 20, 100);
    numbersObservable
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .distinct()
            .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() {

                }
            });

 Observable numbersObservable = Observable.just(1,2,3,4,5);  
  numbersObservable
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .count()
            .subscribeWith(new SingleObserver<Long>() {
                @Override
                public void onSubscribe(Disposable d) {

                }

                @Override
                public void onSuccess(Long count) {
                    Log.d("count is :" + count);
                }

                @Override
                public void onError(Throwable e) {

                }
            });

  Observable
         .range(1, 10)
         .reduce(new BiFunction() {
             @Override
             public Integer apply(Integer number, Integer sum) throws Exception {
                 return sum + number;
             }
         })
         .subscribe(new MaybeObserver() {
             @Override
             public void onSubscribe(Disposable d) {
                 disposable = d;
             }
       @Override
        public void onSuccess(Integer integer) {
            Log.e(TAG, "Sum of numbers from 1 - 10 is: " + integer);
        }

        @Override
        public void onError(Throwable e) {
            Log.e(TAG, "onError: " + e.getMessage());
        }

        @Override
        public void onComplete() {
            Log.e(TAG, "onComplete");
        }
    });