Java 类io.reactivex.Observer 实例源码

项目：Nibo 文件：LocationAddress.java

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

项目：ZhaZhaShop 文件：MovieMoreRankPresenter.java

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

项目：rxlint 文件：DanglingSubscriptionTest.java

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() {

        }
    });
}

项目：RxConditions 文件：ObservableIfThen.java

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

项目：vertx-jooq-async 文件：RXVertxDaoTestBase.java

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

    };
}

项目：UdacityBakingAndroid 文件：RecipeListPresenter.java

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

项目：Learning-RxJava 文件：Ch9_7.java

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

项目：RxSWT 文件：Preconditions.java

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

项目：Demos 文件：RxLifecycleActivity.java

@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() {

                }
            });
}

项目：Demos 文件：RxLifecycleNaviActivity.java

@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() {

                }
            });
}

项目：MovingGdufe-Android 文件：JwcApiFactory.java

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

项目：RxJava2-Android-Sample 文件：ReplaySubjectExampleActivity.java

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

项目：NotifyTools 文件：MainActivity.java

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

项目：RxDisposal 文件：SubscriptionDecoratorTest.java

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

    validateDisposableDelegated(testObserver);

    validateOnNext(testObserver);
}

项目：Coin-Tracker 文件：SplashActivity.java

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() {

                }
            });
}

项目：RxJava2-Android-Sample 文件：DeferExampleActivity.java

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

项目：RxSWT 文件：PaintObservable.java

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

项目：RxJava2-Android-Sample 文件：StartExampleActivity.java

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

项目：GitHub 文件：RxBitmapActivity.java

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

项目：RxDataBinding 文件：Preconditions.java

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

项目：GitHub 文件：ConcatExampleActivity.java

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

项目：ConductorMVP 文件：TideDetailsController.java

@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() {

        }
      });
}

项目：Reactive-Programming-With-Java-9 文件：Demo_PublishSubject.java

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

项目：BuildingDebugFeatures 文件：MainActivity.java

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

项目：TakeRest 文件：PhotoViewActivity.java

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() {

                }
            });
}

项目：RxJava2-Android-Sample 文件：TakeUntilExampleActivity.java

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

项目：trust-wallet-android 文件：CoinmarketcapTickerService.java

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

项目：MovingGdufe-Android 文件：OpacApiFactory.java

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

项目：RxSWT 文件：FocusObservable.java

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

}

项目：AppServiceRestFul 文件：UserApi.java

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

项目：RxJava2-Android-Sample 文件：UsingExampleActivity.java

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

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

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

项目：sqlbrite-sqlcipher 文件：BriteDatabase.java

BriteDatabase(Context context,SQLiteOpenHelper helper, @NonNull char[] password, Logger logger, Observable<Set<String>> triggerSource,
    Observer<Set<String>> triggerSink, Scheduler scheduler,
    ObservableTransformer<Query, Query> queryTransformer) {
  SQLiteDatabase.loadLibs(context);
  this.helper = helper;
  this.password = password;
  this.logger = logger;
  this.triggerSource = triggerSource;
  this.triggerSink = triggerSink;
  this.scheduler = scheduler;
  this.queryTransformer = queryTransformer;
}

项目：GitHub 文件：MapExampleActivity.java

private Observer<List<User>> getObserver() {
    return new Observer<List<User>>() {

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

        @Override
        public void onNext(List<User> userList) {
            textView.append(" onNext");
            textView.append(AppConstant.LINE_SEPARATOR);
            for (User user : userList) {
                textView.append(" firstname : " + user.firstname);
                textView.append(AppConstant.LINE_SEPARATOR);
            }
            Log.d(TAG, " onNext : " + userList.size());
        }

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

项目：RxJava2-Android-Sample 文件：CreateExampleActivity.java

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

项目：jobson 文件：JobManagerTest.java

@Test
public void testSubmitJobEventListenersEchoStderrWhenExecutorEchoesStderr() throws InterruptedException {
    final Subject<byte[]> stderr = ReplaySubject.create();
    final byte[] stderrBytes = generateRandomBytes();
    stderr.onNext(stderrBytes);

    final JobExecutor jobExecutor = MockJobExecutor.thatUses(Observable.never(), stderr);
    final JobManager jobManager = createManagerWith(jobExecutor);

    final Semaphore s = new Semaphore(1);
    s.acquire();

    final JobEventListeners listeners = JobEventListeners.createStderrListener(new Observer<byte[]>() {
        @Override
        public void onSubscribe(@NonNull Disposable disposable) {}

        @Override
        public void onNext(@NonNull byte[] bytes) {
            assertThat(bytes).isEqualTo(stderrBytes);
            s.release();
        }

        @Override
        public void onError(@NonNull Throwable throwable) {
            fail("Error from observable");
            s.release();
        }

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

    jobManager.submit(STANDARD_VALID_REQUEST, listeners);

    if (!s.tryAcquire(1, SECONDS)) {
        fail("Timed out before any bytes received");
    }
}

项目：RxSWT 文件：TestObserverUtil.java

@SuppressWarnings({ "unchecked", "rawtypes" })
public static TestObserver<?> getEmptySubscribedTestObserver() {
    Observable<?> empty = Observable.empty();
    TestObserver<?> testObserver = TestObserver.create();
    empty.subscribe((Observer) testObserver);
    return testObserver;
}

项目：RxDisposal 文件：SubscriptionDecoratorTest.java

private void validateOnSubscribe(Observer<? super String> testObserver) throws Exception {
    final Disposable testDisposable = mock(Disposable.class);
    testObserver.onSubscribe(testDisposable);
    verify(mockOnSusbcribe, times(1))
            .accept(argThat(validateDisposeIsDelegated(testDisposable)));

}

项目：AmenEye 文件：HealthFragment.java

@Override
public void onHiddenChanged(boolean hidden) {
    if (hidden && mRefresh.isRefreshing()) { // 隐藏的时候停止 SwipeRefreshLayout 转动
        mRefresh.setRefreshing(false);
    }
    if (!hidden) {
        RxBus.getDefault()
                .toObservableWithCode(RxConstants.BACK_PRESSED_CODE, String.class)
                .subscribeWith(new Observer<String>() {
                    @Override
                    public void onSubscribe(Disposable d) {
                        mDisposable = d;
                    }

                    @Override
                    public void onNext(String value) {
                        if (value.equals(RxConstants.BACK_PRESSED_DATA) && mNewsRecycler != null) {
                            //滚动到顶部
                            mLinearLayoutManager.smoothScrollToPosition(mNewsRecycler, null, 0);
                        }
                    }

                    @Override
                    public void onError(Throwable e) {

                    }

                    @Override
                    public void onComplete() {

                    }
                });
    } else {
        if (mDisposable != null && !mDisposable.isDisposed()) {
            mDisposable.dispose();
        }
    }
}

项目：ITagAntiLost 文件：ToolbarItemClickObservable.java

@Override protected void subscribeActual(Observer<? super MenuItem> observer) {
  if (!ObservableUtils.checkMainThread(observer)) {
    return;
  }
  Listener listener = new Listener(view, observer);
  observer.onSubscribe(listener);
  view.setOnMenuItemClickListener(listener);
}

项目：RxDataBinding 文件：ObservableByteObservable.java

@Override protected void subscribeActual(Observer<? super Byte> observer) {
  if (!checkMainThread(observer)) {
    return;
  }
  Listener listener = new Listener(observableByte, observer);
  observer.onSubscribe(listener);
  observableByte.addOnPropertyChangedCallback(listener.onPropertyChangedCallback);
}