Java 类io.reactivex.observers.DisposableObserver 实例源码

public static <T> ObservableOperator<List<T>, T> myToList() {
        return observer -> new DisposableObserver<T>() {
            ArrayList<T> list = new ArrayList<>();

            @Override
            public void onNext(T value) {
//add to List, but don't pass anything downstream
                list.add(value);
            }

            @Override
            public void onError(Throwable t) {
                observer.onError(t);
            }

            @Override
            public void onComplete() {
                observer.onNext(list); //push List downstream
                observer.onComplete();
            }
        };
    }

private DisposableObserver<Long> getObserver() {
    return new DisposableObserver<Long>() {

        @Override
        public void onNext(Long value) {
            textView.append(" onNext : value : " + value);
            textView.append(AppConstant.LINE_SEPARATOR);
            Log.d(TAG, " onNext : value : " + value);
        }

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

        @Override
        public void onComplete() {
            textView.append(" onComplete");
            textView.append(AppConstant.LINE_SEPARATOR);
            Log.d(TAG, " onComplete");
        }
    };
}

void onRunSchedulerExampleButtonClicked() {
    disposables.add(sampleObservable()
        // Run on a background thread
        .subscribeOn(Schedulers.io())
        // Be notified on the main thread
        .observeOn(AndroidSchedulers.mainThread())
        .subscribeWith(new DisposableObserver<String>() {
            @Override public void onComplete() {
                Log.d(TAG, "onComplete()");
            }

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

            @Override public void onNext(String string) {
                Log.d(TAG, "onNext(" + string + ")");
            }
        }));
}

private DisposableObserver<String> getDisposableObserver() {

        return new DisposableObserver<String>() {

            @Override
            public void onNext(String response) {
                Log.d(TAG, "onResponse response : " + response);
            }

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

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

private DisposableObserver<String> getDisposableObserver() {

        return new DisposableObserver<String>() {

            @Override
            public void onNext(String response) {
                Log.d(TAG, "onResponse response : " + response);
            }

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

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

private void changeCycle(Cycle cycle) {
    if (addSubscriptionAdaptor != null) {
        addSubscriptionAdaptor.clearData();
    }
    if (disposableUpdates != null) {
        disposableUpdates.dispose();
    }
    disposableUpdates = subscribeToSubscriptionUpdates.execute(
            new DisposableObserver<UserSubscriptionDto>() {
                @Override public void onNext(@NonNull UserSubscriptionDto subscriptionDto) {
                    showUsersCollectionInView(subscriptionDto);
                }

                @Override public void onError(@NonNull Throwable throwable) {

                }

                @Override public void onComplete() {

                }
            }, cycle == null ? Params.forCaseAll() : Params.forCase(cycle));
}

private void getCompanies() {
    Disposable disposable = companiesRepository
            .getCompanies()
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribeWith(new DisposableObserver<List<Company>>() {
                @Override
                public void onNext(List<Company> value) {
                    view.showCompanies(value);
                }

                @Override
                public void onError(Throwable e) {

                }

                @Override
                public void onComplete() {

                }
            });
    compositeDisposable.add(disposable);
}

private void startLocationScan() {

        RxLocation rxLocation = new RxLocation(this);

        LocationRequest locationRequest = LocationRequest.create()
                .setPriority(LocationRequest.PRIORITY_HIGH_ACCURACY)
                .setInterval(TimeUnit.SECONDS.toMillis(5));

        rxLocationObserver = rxLocation.location()
                .updates(locationRequest)
                .subscribeOn(Schedulers.io())
                .flatMap(location -> rxLocation.geocoding().fromLocation(location).toObservable())
                .observeOn(Schedulers.io())
                .subscribeWith(new DisposableObserver<Address>() {
                    @Override public void onNext(Address address) {

                        boolean isLocationsEnabled = App.INSTANCE.getSharedPreferences().isLocationsEnabled();
                        if (isLocationsEnabled) {
                            mOWDevice.setGpsLocation(address);
                        } else if (rxLocationObserver != null) {
                            rxLocationObserver.dispose();
                        }
                    }

                    @Override public void onError(Throwable e) {
                        Log.e(TAG, "onError: error retreiving location", e);
                    }

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

private void disposeOnResume() {
    Disposable d = Observable
            .interval(1, TimeUnit.SECONDS)
            .subscribeWith(new DisposableObserver<Long>() {
                @Override
                public void onNext(Long value) {
                    Log.d(TAG, "Timer A:" + value.toString() + " Seconds");
                }

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

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

    /*
     * Dispose observer on ActivityEvent.RESUME
     */
    dispose(d, ActivityEvent.RESUME);
}

private void disposeOnDestroy() {
    Disposable d = Observable
            .interval(1, TimeUnit.SECONDS)
            .subscribeWith(new DisposableObserver<Long>() {
                @Override
                public void onNext(Long value) {
                    Log.d(TAG, "Timer B:" + value.toString() + " Seconds");
                }

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

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

    /*
     * Dispose observer
     */
    dispose(d);
}

private void checkIfPageIsBookmarked(Integer... pages) {
  compositeDisposable.add(bookmarkModel.getIsBookmarkedObservable(pages)
      .observeOn(AndroidSchedulers.mainThread())
      .subscribeWith(new DisposableObserver<Pair<Integer, Boolean>>() {

        @Override
        public void onNext(Pair<Integer, Boolean> result) {
          bookmarksCache.put(result.first, result.second);
        }

        @Override
        public void onError(Throwable e) {
        }

        @Override
        public void onComplete() {
          supportInvalidateOptionsMenu();
        }
      }));
}

@Override
public void bind(PagerActivity what) {
  minimumPage = Constants.NO_PAGE;
  maximumPage = Constants.NO_PAGE;
  lastPage = Constants.NO_PAGE;

  disposable = what.getViewPagerObservable()
      .subscribeWith(new DisposableObserver<Integer>() {
        @Override
        public void onNext(Integer value) {
          onPageChanged(value);
        }

        @Override
        public void onError(Throwable e) {
        }

        @Override
        public void onComplete() {
        }
      });
}

void onRunSchedulerExampleButtonClicked() {
    disposables.add(sampleObservable()
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribeWith(new DisposableObserver<String>() {
                @Override
                public void onComplete() {
                    Log.d(TAG, "onComplete()");
                }

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

                @Override
                public void onNext(String string) {
                    Log.d(TAG, "onNext(" + string + ")");
                }
            }));
}

private void getAyahCoordinates(Integer... pages) {
  compositeDisposable.add(
      Observable.fromArray(pages)
          .flatMap(coordinatesModel::getAyahCoordinates)
          .observeOn(AndroidSchedulers.mainThread())
          .subscribeWith(new DisposableObserver<Pair<Integer, Map<String, List<AyahBounds>>>>() {
            @Override
            public void onNext(Pair<Integer, Map<String, List<AyahBounds>>> coordinates) {
              if (screen != null) {
                screen.setAyahCoordinatesData(coordinates.first, coordinates.second);
              }
            }

            @Override
            public void onError(Throwable e) {
            }

            @Override
            public void onComplete() {
              if (quranSettings.shouldHighlightBookmarks()) {
                getBookmarkedAyahs(pages);
              }
            }
          })
  );
}

private void getUserProfile() {
    manage(getUserProfile.execute(new DisposableObserver<UserProfile>() {
        @Override public void onNext(@NonNull UserProfile userProfile) {
            userProfileView.setName(userProfile.getUserFullName());
            userProfileView.setEmail(userProfile.getUserEmail());
            maxSubs = userProfile.getSubAvailable();
            updateCounts();
        }

        @Override public void onError(@NonNull Throwable e) {

        }

        @Override public void onComplete() {

        }
    }, null));
}

public void initialize(SubscriptionType subscriptionType) {
    manage(subscribeToSubscriptionUpdates.execute(new DisposableObserver<SubscriptionDto>() {
        @Override public void onNext(@NonNull SubscriptionDto subscriptionDto) {
            showUsersCollectionInView(subscriptionDto);
        }

        @Override public void onError(@NonNull Throwable throwable) {

        }

        @Override public void onComplete() {

        }
    }, SubscribeToSubscriptionUpdates.Params.forCase(subscriptionType)));
    this.hideViewRetry();
    this.showViewLoading();
}

@Override
public void getRecommendedWares() {
    addDisposabe(mDataManager.getData(new DisposableObserver<HomeIndex>() {
        @Override
        public void onNext(HomeIndex homeIndex) {
            mHomeView.setRecommendedWares(homeIndex);
        }

        @Override
        public void onError(Throwable e) {

        }

        @Override
        public void onComplete() {

        }
    },HomeIndex.class, "recommend.txt"));
}

@Override
public void getMoreRecommendedWares() {
    addDisposabe(mDataManager.getData(new DisposableObserver<HomeIndex>() {
        @Override
        public void onNext(HomeIndex homeIndex) {
            mHomeView.setMoreRecommendedWares(homeIndex);
        }

        @Override
        public void onError(Throwable e) {

        }

        @Override
        public void onComplete() {

        }
    },HomeIndex.class, "recommended.txt"));
}

@Override
public void getMoreFindData() {
    addDisposabe(mDataManager.getData(new DisposableObserver<FindsBean>() {
        @Override
        public void onNext(FindsBean findsBean) {
            mFindView.setMoreData(findsBean);
        }

        @Override
        public void onError(Throwable e) {

        }

        @Override
        public void onComplete() {

        }
    },FindsBean.class, "find.txt"));
}

/**
 * 接受消息
 *
 * @param tag      标志
 * @param callBack 回调
 */
public <T> DisposableObserver registerNoThread(@NonNull final Object tag,
                                               @NonNull final RxBusCallBack<T> callBack) {
    RxBusEvent rxBusEvent = rxBusEventArrayMap.get(tag);
    if (RxUtils.isEmpty(rxBusEvent)) {
        rxBusEvent = new RxBusEvent();
        rxBusEvent.subject = PublishSubject.create().toSerialized();
        rxBusEvent.disposable =
                rxBusEvent.subject
                        .ofType(callBack.busOfType())
                        .subscribeWith(new RxBusObserver<T>() {
                            @Override
                            public void onError(@io.reactivex.annotations.NonNull Throwable e) {
                                super.onError(e);
                                callBack.onBusError(e);
                            }

                            @Override
                            public void onNext(@io.reactivex.annotations.NonNull T t) {
                                super.onNext(t);
                            }
                        });
        rxBusEventArrayMap.put(tag, rxBusEvent);
    }
    return rxBusEvent.disposable;
}

@Test
public void viewsAreCalledBeforeLifecycleIsReadyWithoutLifecycleAwareRx() throws Exception {
    // Lifecycle is "active" once it is STARTED, it's not ready yet at INITIALIZED or CREATED.
    lifecycleOwner.handleLifecycleEvent(Lifecycle.Event.ON_CREATE);

    Observable.interval(1, TimeUnit.MILLISECONDS)
        .subscribeWith(new DisposableObserver<Long>() {
            @Override
            public void onNext(final Long value) {
                LifecycleTest.this.methodOnViewCalled = true;
            }

            @Override
            public void onError(final Throwable e) {
            }

            @Override
            public void onComplete() {
            }
        });

    // Need to wait to give it time to potentially fail
    TimeUnit.MILLISECONDS.sleep(100);
    assertEquals(true, methodOnViewCalled);
}

public void trainNewModel() {
    Register register = new Register(repository,
            threadExecutor.getOriginScheduler(),
            threadExecutor.getPostScheduler());
    view.onRegisterStart();
    register.execute(new DisposableObserver<Integer>() {
        @Override
        public void onNext(@NonNull Integer modelNumber) {
            view.onRegisterDone();
            train(modelNumber);
        }

        @Override
        public void onError(@NonNull Throwable e) {
            view.onRegisterDone();
            train(-1);
        }

        @Override
        public void onComplete() {
        }
    });

}

public void rx2subscriptionWithoutSavingReturnSubscribeWith() {
    io.reactivex.Observable.just("Test").subscribeWith(new DisposableObserver<String>() {
        @Override
        public void onNext(String s) {

        }

        @Override
        public void onError(Throwable e) {

        }

        @Override
        public void onComplete() {

        }
    });
}

public void rx2ObservableWithSubscriber() {
    io.reactivex.Observable.just("test").subscribeWith(new DisposableObserver<String>() {
        @Override
        public void onNext(String s) {

        }

        @Override
        public void onError(Throwable e) {

        }

        @Override
        public void onComplete() {

        }
    });
}

private void addCheckPointMarkers() {
  getCheckPointDataSource().getAllCheckPoints()
      .toObservable()
      .doOnNext(checkPoints -> allCheckPoints = (ArrayList<CheckPoint>) checkPoints)
      .flatMap(Observable::fromIterable)
      .subscribeOn(Schedulers.io())
      .observeOn(AndroidSchedulers.mainThread())
      .subscribeWith(new DisposableObserver<CheckPoint>() {
        @Override public void onNext(CheckPoint checkPoint) {
          getView().addMarkerOnMap(checkPoint);
        }

        @Override public void onError(Throwable e) {
          getView().showError(e.getLocalizedMessage());
        }

        @Override public void onComplete() {
          getView().notifyListAdapter();
        }
      });
}

public void setHouseId(Long houseId) {
    this.houseId = houseId;
    getHouseDetailUseCase.execute(new DisposableObserver<HouseEntity>() {
        @Override
        public void onNext(HouseEntity value) {
            initViewModel(value);
        }

        @Override
        public void onError(Throwable e) {

        }

        @Override
        public void onComplete() {

        }
    }, GetHouseDetail.Params.setHouse(houseId));
}

public void start() {
    mGoogleApiClientRepository.connect();
    mGoogleApiClientRepository.subscribeToThingsMessage(new DisposableObserver<ThingsMessage>() {
        @Override
        public void onNext(ThingsMessage message) {
            handleThingsMessage(message);
        }

        @Override
        public void onError(Throwable e) {
            Timber.e(e);
        }

        @Override
        public void onComplete() {

        }
    });
}

private void initializeSession() {
    manage(getLoginStatus.execute(new DisposableObserver<GetLoginStatus.LoginStatusType>() {
        @Override public void onNext(@io.reactivex.annotations.NonNull
                GetLoginStatus.LoginStatusType loginStatusType) {
            if (loginStatusType == GetLoginStatus.LoginStatusType.ACTIVE) {
                loginFlowListener.openMainActivity();
            } else {
                Log.d(TAG, "onAuthStateChanged:signed_out");
            }
        }

        @Override public void onError(@io.reactivex.annotations.NonNull Throwable e) {

        }

        @Override public void onComplete() {

        }
    }, null));
}

private DisposableObserver<Long> getObserver() {
    return new DisposableObserver<Long>() {

        @Override
        public void onNext(Long value) {
            textView.append(" onNext : value : " + value);
            textView.append(AppConstant.LINE_SEPARATOR);
            Log.d(TAG, " onNext : value : " + value);
        }

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

        @Override
        public void onComplete() {
            textView.append(" onComplete");
            textView.append(AppConstant.LINE_SEPARATOR);
            Log.d(TAG, " onComplete");
        }
    };
}

private void initializeIndividualObservers(Cycle cycle,
        final BreakDownListener breakDownListener) {
    manage(subscriptionBreakdownUpdates.execute(
            new DisposableObserver<SubscriptionBreakdownUpdates.BreakdownDto>() {
                @Override public void onNext(
                        @NonNull SubscriptionBreakdownUpdates.BreakdownDto breakdownDto) {
                    breakDownListener.update(breakdownDto.getData());
                }

                @Override public void onError(@NonNull Throwable e) {

                }

                @Override public void onComplete() {

                }
            }, SubscriptionBreakdownUpdates.Params.forCase(cycle)));
}

@Override
public void onResume() {
    super.onResume();
    mCompositeDisposable.add(Observable.fromCallable(() -> mApp.getDBAccessHelper().getAllArtist())
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread()).subscribeWith(new DisposableObserver<ArrayList<Artist>>() {
                @Override
                public void onNext(ArrayList<Artist> hashMaps) {
                    mArtistList = hashMaps;
                    mAdapter.updateData(mArtistList);
                    mAdapter.notifyDataSetChanged();
                }

                @Override
                public void onError(Throwable e) {
                    Logger.log("" + e.getCause());
                }

                @Override
                public void onComplete() {

                }
            })
    );
}

@Override
public void onResume() {
    super.onResume();
    mCompositeDisposable
            .add(Observable.fromCallable(() -> mApp.getDBAccessHelper().getAllGenres())
                    .subscribeOn(Schedulers.io())
                    .observeOn(AndroidSchedulers.mainThread())
                    .subscribeWith(new DisposableObserver<ArrayList<Genre>>() {
                        @Override
                        public void onNext(ArrayList<Genre> hashMaps) {
                            mGenres = hashMaps;
                            mAdapter.updateData(hashMaps);
                            mAdapter.notifyDataSetChanged();
                        }

                        @Override
                        public void onError(Throwable e) {

                        }

                        @Override
                        public void onComplete() {

                        }
                    }));
}

@Override public void initialize() {
    final Handler finalHandler = new Handler();
    this.manage(
            getLoginStatus.execute(new DisposableObserver<GetLoginStatus.LoginStatusType>() {
                @Override public void onNext(
                        @NonNull final GetLoginStatus.LoginStatusType loginStatusType) {
                    Runnable runnable = () -> {
                        if (loginStatusType == GetLoginStatus.LoginStatusType.ACTIVE) {
                            splashScreenFlowListener.openMainScreen();
                        } else {
                            splashScreenFlowListener.openLoginScreen();
                        }
                    };
                    finalHandler.postDelayed(runnable, 2000);
                }

                @Override public void onError(@NonNull Throwable e) {
                    splashScreenView.showError(e.getMessage());
                }

                @Override public void onComplete() {

                }
            }, null));
}

private void loadData() {
    mCompositeDisposable.add(Observable.fromCallable(() -> CursorHelper.getTracksForSelection("SONGS", ""))
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribeWith(new DisposableObserver<ArrayList<Song>>() {
                @Override
                public void onComplete() {

                }

                @Override
                public void onError(Throwable e) {

                }

                @Override
                public void onNext(ArrayList<Song> data) {
                    mSongList.clear();
                    mSongList.addAll(data);
                    mAdapter.update(mSongList);
                }
            }));
}

public void getMovieDetails(long movieId) {
    isMovieLoading.set(true);
    errorViewShowing.set(false);
    mMoviesRepository.getMovieDetails(movieId)
            .observeOn(AndroidSchedulers.mainThread())
            .subscribeOn(Schedulers.io())
            .subscribeWith(new DisposableObserver<Movie>() {
                @Override
                public void onNext(Movie value) {
                    setMovie(value);
                }

                @Override
                public void onError(Throwable e) {
                    errorViewShowing.set(true);
                    isMovieLoading.set(false);
                }

                @Override
                public void onComplete() {
                    isMovieLoading.set(false);
                    errorViewShowing.set(false);
                }
            });
}

private void getUserSubscriptionCount() {
    manage(subscribeToUserSubscriptionCountUpdates.execute(new DisposableObserver<Integer>() {
        @Override public void onNext(@io.reactivex.annotations.NonNull Integer integer) {
            currentSubs = integer;
            updateCounts();
        }

        @Override public void onError(@io.reactivex.annotations.NonNull Throwable e) {

        }

        @Override public void onComplete() {

        }
    }, null));
}

public static <T> ObservableOperator<T, T> doOnEmpty(Action
                                                             action) {
    return new ObservableOperator<T, T>() {
        @Override
        public Observer<? super T> apply(Observer<? super T>
                                                 observer) throws Exception {
            return new DisposableObserver<T>() {
                boolean isEmpty = true;

                @Override
                public void onNext(T value) {
                    isEmpty = false;
                    observer.onNext(value);
                }

                @Override
                public void onError(Throwable t) {
                    observer.onError(t);
                }

                @Override
                public void onComplete() {
                    if (isEmpty) {
                        try {
                            action.run();
                        } catch (Exception e) {
                            onError(e);
                            return;
                        }
                    }
                    observer.onComplete();
                }
            };
        }
    };
}

void doSomeWork() {
    disposables.add(sampleObservable()
            // Run on a background thread
            .subscribeOn(Schedulers.io())
            // Be notified on the main thread
            .observeOn(AndroidSchedulers.mainThread())
            .subscribeWith(new DisposableObserver<String>() {
                @Override
                public void onComplete() {
                    textView.append(" onComplete");
                    textView.append(AppConstant.LINE_SEPARATOR);
                    Log.d(TAG, " onComplete");
                }

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

                @Override
                public void onNext(String value) {
                    textView.append(" onNext : value : " + value);
                    textView.append(AppConstant.LINE_SEPARATOR);
                    Log.d(TAG, " onNext value : " + value);
                }
            }));
}

@Override
protected <T> void execute(BaseCallback<T> callback) {
    DisposableObserver disposableObserver = new ApiCallbackSubscriber(callback);
    if (super.tag != null) {
        ApiManager.get().add(super.tag, disposableObserver);
    }
    if (isLocalCache) {
        this.cacheExecute(getSubType(callback)).subscribe(disposableObserver);
    } else {
        this.execute(getType(callback)).subscribe(disposableObserver);
    }
}

@Test
@SuppressWarnings("unchecked")
public void testUserListPresenterInitialize() {
  given(mockUserListView.context()).willReturn(mockContext);

  userListPresenter.initialize();

  verify(mockUserListView).hideRetry();
  verify(mockUserListView).showLoading();
  verify(mockGetUserList).execute(any(DisposableObserver.class), any(Void.class));
}