java类io.reactivex.Observer的实例源码

public Observable<Address> getObservableAddressFromLocation(final double latitude, final double longitude,
                                                            final Context context) {
    return new Observable<Address>() {
        @Override
        protected void subscribeActual(Observer<? super Address> observer) {
            Geocoder geocoder = new Geocoder(context, Locale.getDefault());
            try {
                List<Address> addressList = geocoder.getFromLocation(
                        latitude, longitude, 1);
                if (addressList != null && addressList.size() > 0) {
                    address = addressList.get(0);
                    observer.onNext(address);
                }
            } catch (IOException e) {
                Log.e(TAG, "Unable connect to Geocoder", e);
            }

        }
    }.subscribeOn(Schedulers.newThread())
            .observeOn(AndroidSchedulers.mainThread());
}

@Override
public void getOverseaComingMovieList(String area, int limit, int offset) {
    mManager.getOverseaComingMovie(area, limit, offset)
            .subscribe(new Observer<OverseaComingMovieBean>() {
                @Override
                public void onSubscribe(@NonNull Disposable d) {
                    mView.showLoading();
                }

                @Override
                public void onNext(@NonNull OverseaComingMovieBean data) {
                    mView.addOverseaComingMovieList(data.getData().getComing());
                }

                @Override
                public void onError(@NonNull Throwable e) {
                    mView.showError(ErrorHanding.handleError(e));
                }

                @Override
                public void onComplete() {
                    mView.showContent();
                }
            });
}

public void rx2WithObservable() {
    io.reactivex.Observable.just("test").subscribe(new Observer<String>() {
        @Override
        public void onSubscribe(Disposable d) {

        }

        @Override
        public void onNext(String s) {

        }

        @Override
        public void onError(Throwable e) {

        }

        @Override
        public void onComplete() {

        }
    });
}

@Override
protected void subscribeActual(Observer<? super T> observer) {
    boolean b;

    try {
        b = condition.getAsBoolean();
    } catch (Throwable ex) {
        EmptyDisposable.error(ex, observer);
        return;
    }

    if (b) {
        then.subscribe(observer);
    } else {
        orElse.subscribe(observer);
    }
}

protected <T> Observer<T> failOrCountDownPlainObserver(CountDownLatch latch) {
    return new Observer<T>() {
        @Override
        public void onSubscribe(Disposable d) {

        }

        @Override
        public void onNext(T t) {

        }

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

        @Override
        public void onComplete() {
            latch.countDown();
        }

    };
}

private Observer<ArrayList<Recipe>> networkApiRecipeObserver() {
    return new Observer<ArrayList<Recipe>>() {
        @Override
        public void onSubscribe(Disposable d) {
            mCompositeDisposable.add(d);
        }
        @Override
        public void onNext(ArrayList<Recipe> networkRecipeResult) {
            ArrayList<Recipe> recipeList = new ArrayList<>();
            recipeList.addAll(networkRecipeResult);
            if(mView.isActive()) {
                mView.updateAdapter(recipeList);
            }
        }
        @Override
        public void onError(Throwable e) {
            if(mView.isActive()) {
                mView.displaySnackbarMessage(R.string.network_error_recipes);
            }
        }
        @Override
        public void onComplete() {}
    };
}

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();
                }
            };
        }
    };
}

public static boolean checkWidget(Observer<?> observer, Widget widget) {
    if (null == widget) {
        observer.onError(new NullPointerException("The given widget was null"));
        return false;
    } else if (widget.isDisposed()) {
        observer.onError(new IllegalStateException("The given widget is diposed"));
        return false;
    } else if (!(Thread.currentThread().equals(widget.getDisplay().getThread()))) {
        observer.onError(new IllegalStateException(
                "Expected to be called on the main thread but was " + Thread.currentThread().getName()));
        return false;
    }
    return true;
}

@Override
protected void onStart() {
    super.onStart();
    Observable.interval(1, TimeUnit.SECONDS)
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .compose(provider.<Long>bindToLifecycle())
            .subscribe(new Observer<Long>() {
                @Override
                public void onSubscribe(@NonNull Disposable d) {

                }

                @Override
                public void onNext(@NonNull Long aLong) {
                    Log.i("接收数据", String.valueOf(aLong));
                }

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

                }

                @Override
                public void onComplete() {

                }
            });
}

@Override
protected void onStart() {
    super.onStart();
    Observable.interval(1, TimeUnit.SECONDS)
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .compose(provider.<Long>bindToLifecycle())
            .subscribe(new Observer<Long>() {
                @Override
                public void onSubscribe(@NonNull Disposable d) {

                }

                @Override
                public void onNext(@NonNull Long aLong) {
                    Log.i("接收数据", String.valueOf(aLong));
                }

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

                }

                @Override
                public void onComplete() {

                }
            });
}

public void getCet(String zkzh,String name, Observer<Cet> sub ) {
    ApiUtils.getApi(AppConfig.jwPwd).create(JwcApi.class).getCet(zkzh,name)
    .map(new HttpResultFunc<Cet>())
    .observeOn(AndroidSchedulers.mainThread())
    .subscribeOn(Schedulers.io())
    .subscribe(sub);
}

private Observer<Integer> getSecondObserver() {
    return new Observer<Integer>() {

        @Override
        public void onSubscribe(Disposable d) {
            textView.append(" Second onSubscribe : isDisposed :" + d.isDisposed());
            Log.d(TAG, " Second onSubscribe : " + d.isDisposed());
            textView.append(AppConstant.LINE_SEPARATOR);
        }

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

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

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

public Observer<List<Net.BingWallBean>> bingObserver() {
    Observer<List<Net.BingWallBean>> observer = new Observer<List<Net.BingWallBean>>() {
        Disposable dd;

        @Override
        public void onSubscribe(Disposable d) {
            //                d.dispose();//断开订阅关系；
            //                d.isDisposed();//判断是否还有订阅关系
            //                dd = d;
            Log.d(TAG, "onSubscribe: ");
        }

        @Override
        public void onNext(List<Net.BingWallBean> ll) {
            Log.d(TAG, "onNext: " + ll.size());
            bingAdapter=new BingAdapter(ll);
            rv.setAdapter(bingAdapter);
            bingAdapter.notifyDataSetChanged();
        }

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

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

@Test
public void testWrap1Arg() throws Exception {
    Observer<? super String> testObserver = createLambdaObserver(1);

    validateDisposableDelegated(testObserver);

    validateOnNext(testObserver);
}

private void getAllCoins() {
    if (CoinHelper.getInstance().getAllCachedCoinTags().size() > 0) return;

    service.getAllCoins()
            .subscribeOn(Injection.provideSchedulerProvider().io())
            .observeOn(Injection.provideSchedulerProvider().computation())
            .subscribe(new Observer<CoinListResponse>() {
                @Override
                public void onSubscribe(@NonNull Disposable d) {

                }

                @Override
                public void onNext(@NonNull CoinListResponse coinListResponse) {
                    Logger.i("getAllCoins done");
                    CoinHelper coinHelper = CoinHelper.getInstance();
                    coinHelper.setContext(SplashActivity.this);
                    coinHelper.updateAllCachedCoins(coinListResponse.getData(), true);
                }

                @Override
                public void onError(@NonNull Throwable e) {
                    Logger.i("getAllCoins error");
                    e.printStackTrace();
                }

                @Override
                public void onComplete() {

                }
            });
}

private Observer<String> getObserver() {
    return new Observer<String>() {

        @Override
        public void onSubscribe(Disposable d) {
            Log.d(TAG, " onSubscribe : " + d.isDisposed());
        }

        @Override
        public void onNext(String 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");
        }
    };
}

@Override
protected void subscribeActual(Observer<? super PaintEvent> observer) {
    if (!Preconditions.checkWidget(observer, control)) {
        return;
    }
    RxPaintListener listener = new RxPaintListener(control, observer);
    observer.onSubscribe(listener);
    control.addDisposeListener(e -> listener.dispose());
    control.addPaintListener(listener);
}

private Observer<String> getObserver() {
    return new Observer<String>() {

        @Override
        public void onSubscribe(Disposable d) {
            Log.d(TAG, " onSubscribe : " + d.isDisposed());
        }

        @Override
        public void onNext(String 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");
        }
    };
}

@OnClick(R.id.requestImage)
public void requestImage(View view) {
    ServerApi.getBitmap("aaa", "bbb")//
            .subscribeOn(Schedulers.io())//
            .doOnSubscribe(new Consumer<Disposable>() {
                @Override
                public void accept(@NonNull Disposable disposable) throws Exception {
                    showLoading();
                }
            })//
            .observeOn(AndroidSchedulers.mainThread())//
            .subscribe(new Observer<Response<Bitmap>>() {
                @Override
                public void onSubscribe(@NonNull Disposable d) {
                    addDisposable(d);
                }

                @Override
                public void onNext(@NonNull Response<Bitmap> response) {
                    handleResponse(response);
                    imageView.setImageBitmap(response.body());
                }

                @Override
                public void onError(@NonNull Throwable e) {
                    e.printStackTrace();            //请求失败
                    handleError(null);
                    showToast("请求失败");
                }

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

public static boolean checkMainThread(Observer<?> observer) {
  if (Looper.myLooper() != Looper.getMainLooper()) {
    observer.onError(new IllegalStateException(
        "Expected to be called on the main thread but was " + Thread.currentThread().getName()));
    return false;
  }
  return true;
}

private Observer<String> getObserver() {
    return new Observer<String>() {

        @Override
        public void onSubscribe(Disposable d) {
            Log.d(TAG, " onSubscribe : " + d.isDisposed());
        }

        @Override
        public void onNext(String 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");
        }
    };
}

@Override protected void attachView(final TideDetailsView view) {
  super.attachView(view);
  DemoApplication.app(getApplicationContext()).injector().inject(this);
  view.setTitle(tideLocationName);
  noaaApi.getTideInfo(noaaApiId)
      .observeOn(AndroidSchedulers.mainThread())
      .compose(this.<TideInfo>bindToLifecycle())
      .subscribe(new Observer<TideInfo>() {
        @Override public void onSubscribe(@NonNull Disposable d) {

        }

        @Override public void onNext(@NonNull TideInfo tideInfo) {
          if (tideInfo.getData() != null && !tideInfo.getData().isEmpty()) {
            List<Observation> observations = tideInfo.getData();
            BigDecimal highestMeasuredTideHeight =
                Collections.max(filterOutNullMeasurements(observations), OBSERVATION_COMPARATOR)
                    .getVerifiedWaterLevel();
            BigDecimal lowestMeasuredTideHeight =
                Collections.min(filterOutNullMeasurements(observations), OBSERVATION_COMPARATOR)
                    .getVerifiedWaterLevel();
            BigDecimal latestMeasuredTideHeight =
                observations.get(observations.size() - 1).getVerifiedWaterLevel();
            view.setTideHeights(latestMeasuredTideHeight, lowestMeasuredTideHeight,
                highestMeasuredTideHeight);
          }
        }

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

        @Override public void onComplete() {

        }
      });
}

public static void main(String[] args) {
    // TODO Auto-generated method stub
    Observer<Long> observer=new Observer<Long>() {

        @Override
        public void onComplete() {
            // TODO Auto-generated method stub
            System.out.println("It's Done");

        }

        @Override
        public void onError(Throwable throwable) {
            // TODO Auto-generated method stub
            throwable.printStackTrace();

        }

        @Override
        public void onNext(Long value) {
            // TODO Auto-generated method stub
            System.out.println(":"+value);
        }

        @Override
        public void onSubscribe(Disposable disposable) {
            // TODO Auto-generated method stub
            System.out.println("onSubscribe");

        }
    };

    PublishSubject< Long> publishSubject=PublishSubject.create();
    publishSubject.onNext(1L);
    publishSubject.onNext(2l);
    publishSubject.subscribe(observer);
    publishSubject.onNext(10l);
    publishSubject.onNext(20l);
    publishSubject.onComplete();
}

private void search(final String username) {
    gitHubService.user(username).subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread()).subscribe(new Observer<GitHubUser>() {
        @Override
        public void onSubscribe(Disposable d) {
            disposable = d;
        }

        @Override
        public void onNext(GitHubUser gitHubUser) {
            Timber.v(gitHubUser.avatar_url);
            Intent intent = new Intent(MainActivity.this, UserActivity.class);
            intent.putExtras(UserActivity.BundleBuilder.createWithUser(gitHubUser));
            startActivity(intent);
            usernameEditText.setText("");
        }

        @Override
        public void onError(Throwable e) {
            Toast.makeText(MainActivity.this, "Could not find user '" + username + "'", Toast.LENGTH_SHORT).show();
        }

        @Override
        public void onComplete() {
            Timber.v("onComplete");
        }
    });
}

private void save(){
    GetFrescoBitmap.saveImage(PhotoViewActivity.this, url)
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(new Observer<Uri>() {
                @Override
                public void onSubscribe(Disposable d) {

                }

                @Override
                public void onNext(Uri value) {
                    if (value != null) {
                        ToastUtil.makeTextShort(PhotoViewActivity.this, getString(R.string.save_success) + value.toString());
                    }
                }

                @Override
                public void onError(Throwable e) {
                    ToastUtil.makeTextShort(PhotoViewActivity.this, getString(R.string.save_error) +
                            e.getMessage());
                }

                @Override
                public void onComplete() {

                }
            });
}

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

        @Override
        public void onSubscribe(Disposable d) {
            Log.d(TAG, " onSubscribe : " + d.isDisposed());
        }

        @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");
        }
    };
}

@Override
public Observer<? super Response<T>> apply(Observer<? super T> observer) throws Exception {
    return new DisposableObserver<Response<T>>() {
        @Override
        public void onNext(Response<T> response) {
            if (isDisposed()) {
                return;
            }
            observer.onNext(response.body());
            observer.onComplete();
        }

        @Override
        public void onError(Throwable e) {
            if (!isDisposed()) {
                observer.onError(e);
            }
        }

        @Override
        public void onComplete() {
            if (!isDisposed()) {
                observer.onComplete();
            }
        }
    };
}

public void  getRenewBookVerifyCode(Observer<StrObjectResponse> sub ) {
    ApiUtils.getApi(AppConfig.idsPwd).create(OpacApi.class).getRenewBookVerifyCode()
            .map(new HttpResultFunc<StrObjectResponse>())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribeOn(Schedulers.io())
            .subscribe(sub);
}

@Override
protected void subscribeActual(Observer<? super FocusEvent> observer) {
    if (!Preconditions.checkWidget(observer, control)) {
        return;
    }
    AbstractFocusListener listener = getFocusListener(control, observer);
    observer.onSubscribe(listener);
    control.addDisposeListener(e -> listener.dispose());
    control.addFocusListener(listener);

}

public void register(String phone, String password, String username, int sex, int age, Observer<UserBean> subscriber)
{
    UserBean userBean = new UserBean();
    userBean.setPhone(phone);
    userBean.setPassword(password);
    userBean.setUsername(username);
    userBean.setSex(sex);
    userBean.setAge(age);
    Observable observable = userService.register(userBean)
            .map(new HttpResultFunc<UserBean>());

    toSubscribe(observable, subscriber);
}