Java 类io.reactivex.subscribers.DisposableSubscriber 实例源码

项目：RxBus2 文件：RxBusBuilder.java

public <R> Disposable subscribe(DisposableSubscriber<R> subscriber, FlowableTransformer<T, R> transformer)
{
    Flowable flowable = build(false);
    if (transformer != null)
        flowable = flowable.compose(transformer);

    Subscriber<R> actualSubscriber = subscriber;
    if (mQueuer != null && mQueueSubscriptionSafetyCheckEnabled)
        actualSubscriber = RxBusUtil.wrapSubscriber(subscriber, mQueuer);

    flowable = applySchedular(flowable);
    Disposable disposable = (DisposableSubscriber)flowable.subscribeWith(actualSubscriber);
    if (mBoundObject != null)
        RxDisposableManager.addDisposable(mBoundObject, disposable);
    return disposable;
}

项目：RxBus2 文件：RxBusUtil.java

protected static <T> DisposableSubscriber<T> wrapSubscriber(DisposableSubscriber<T> subscriber, IRxBusQueue isResumedProvider)
{
    return new DisposableSubscriber<T>()
    {
        @Override
        public void onComplete()
        {
            subscriber.onComplete();
        }

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

        @Override
        public void onNext(T t)
        {
            if (RxUtil.safetyQueueCheck(t, isResumedProvider))
                subscriber.onNext(t);
        }
    };
}

项目：KomaMusic 文件：SongsPresenter.java

@Override
public void loadSongs() {
    mDisposables.clear();

    Disposable disposable = mRepository.getAllSongs().subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribeWith(new DisposableSubscriber<List<Song>>() {
                @Override
                public void onError(Throwable throwable) {
                    LogUtils.e(TAG, "onError :" + throwable.toString());
                }

                @Override
                public void onComplete() {
                    LogUtils.i(TAG, "onCompleted");
                }

                @Override
                public void onNext(List<Song> songs) {
                    onLoadSongsFinished(songs);
                }
            });

    mDisposables.add(disposable);
}

项目：KomaMusic 文件：PlaylistsPresenter.java

@Override
public void loadPlaylists() {
    LogUtils.i(TAG, "loadPlaylists");
    mDisposables.clear();

    Disposable disposable = mRepository.getAllPlaylists().subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribeWith(new DisposableSubscriber<List<Playlist>>() {
                @Override
                public void onError(Throwable throwable) {
                    LogUtils.e(TAG, "loadPlaylists Error :" + throwable.toString());
                }

                @Override
                public void onComplete() {

                }

                @Override
                public void onNext(List<Playlist> playlists) {
                    onLoadPlaylistsFinished(playlists);
                }
            });

    mDisposables.add(disposable);
}

项目：rxlint 文件：SubscriberTest.java

public void rx2FlowableWithSubscriber() {
    io.reactivex.Flowable.just("test").subscribeWith(new DisposableSubscriber<String>() {
        @Override
        public void onNext(String s) {

        }

        @Override
        public void onError(Throwable t) {

        }

        @Override
        public void onComplete() {

        }
    });
}

项目：Android-AudioRecorder-App 文件：AudioRecorder.java

private void startRecordThread() {
  audioDataFlowable.subscribeOn(Schedulers.io()).subscribe(recordDataPublishProcessor);
  compositeDisposable.add(recordDataPublishProcessor.onBackpressureBuffer()
      .observeOn(Schedulers.io())
      .subscribeWith(new DisposableSubscriber<byte[]>() {
        @Override public void onNext(byte[] bytes) {
          audioSaveHelper.onDataReady(recordBuffer);
        }

        @Override public void onError(Throwable t) {

        }

        @Override public void onComplete() {
          audioSaveHelper.onRecordingStopped(currentRecordTime);
          synchronized (recorderStateMonitor) {
            recorderState = RECORDER_STATE_IDLE;
            recorderStateMonitor.notifyAll();
          }
        }
      }));
}

项目：Android-AudioRecorder-App 文件：AudioRecordingDbmHandler.java

@Override public void startDbmThread() {
  disposableSubscriber = audioRecorder.getAudioDataFlowable()
      .onBackpressureLatest()
      .observeOn(Schedulers.newThread())
      .subscribeWith(new DisposableSubscriber<byte[]>() {
        @Override public void onNext(byte[] moreData) {
          if (moreData != null && moreData.length > 0) {
            onDataReceived(moreData);
          }
        }

        @Override public void onError(Throwable t) {
          Log.i("Visual Error ", t.getMessage() + " ");
        }

        @Override public void onComplete() {
          Log.i("Visualise ", "complete");
        }
      });
}

项目：commitstrip-reader 文件：RandomStripPresenter.java

@Override
public void fetchStrip(Integer numberOfStripPerPage, int page) {

    mSubscriptions.add(
            fetchRandomStrip(numberOfStripPerPage)
                    .subscribeWith(new DisposableSubscriber<StripDto>() {
                        @Override public void onNext(StripDto strip) {
                            mAlreadyDisplayedId.add(strip.getId());
                            mListCurrentDisplayStrip.add(strip);
                        }

                        @Override public void onError(Throwable e) {}

                        @Override public void onComplete() {
                            List<StripWithImageDto> displayStrips = new ArrayList<>();
                            for (int i = 0; i < mListCurrentDisplayStrip.size(); i++) {
                                displayStrips.add(
                                        convertStripDtoToStripWithImageDto(mListCurrentDisplayStrip.get(i)));
                            }

                            mListStripView.addMoreStrips(displayStrips);
                            mListCurrentDisplayStrip.clear();
                        }
                    }));
}

项目：apollo-android 文件：GitHuntEntryDetailActivity.java

private void subscribeRepoCommentAdded() {
  ApolloSubscriptionCall<RepoCommentAddedSubscription.Data> subscriptionCall = application.apolloClient()
      .subscribe(new RepoCommentAddedSubscription(repoFullName));

  disposables.add(Rx2Apollo.from(subscriptionCall)
      .subscribeOn(Schedulers.io())
      .observeOn(AndroidSchedulers.mainThread())
      .subscribeWith(
          new DisposableSubscriber<Response<RepoCommentAddedSubscription.Data>>() {
            @Override public void onNext(Response<RepoCommentAddedSubscription.Data> response) {
              commentsListViewAdapter.addItem(response.data().commentAdded().content());
            }

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

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

项目：RxJava-Android-Samples 文件：RotationPersist2Fragment.java

@Override
public void setStream(Flowable<Integer> intStream) {
  DisposableSubscriber<Integer> d =
      new DisposableSubscriber<Integer>() {
        @Override
        public void onNext(Integer integer) {
          _log(String.format("Worker frag spits out - %d", integer));
        }

        @Override
        public void onError(Throwable e) {
          Timber.e(e, "Error in worker demo frag observable");
          _log("Dang! something went wrong.");
        }

        @Override
        public void onComplete() {
          _log("Observable is complete");
        }
      };

  intStream.doOnSubscribe(subscription -> _log("Subscribing to intsObservable")).subscribe(d);

  _disposables.add(d);
}

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

public static <T> FlowableOperator<T, T> doOnEmpty(Action
                                                           action) {
    return new FlowableOperator<T, T>() {
        @Override
        public Subscriber<? super T> apply(Subscriber<? super
                T> subscriber) throws Exception {
            return new DisposableSubscriber<T>() {
                boolean isEmpty = true;

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

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

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

项目：store2realm 文件：MainPresenter.java

public void loadPosts(boolean filterAndSortPosts) {
    SortingMode sortingMode = null;
    Filter filter = null;
    if(filterAndSortPosts){
        sortingMode = new SortingMode("userId", SortType.ASCENDING);
        filter = new Filter("userId", 5);
    }

    Disposable d = postService.getAll(filter, sortingMode)
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribeWith(new DisposableSubscriber<Optional<List<Post>>>(){
                @Override
                public void onNext(Optional<List<Post>> items) {
                    mainView.updatePosts(items.get());
                }

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

                @Override
                public void onComplete() {

                }
            });

    disposable.add(d);
}

项目：CleanArchitechture 文件：UseCase.java

/**
 * Executes the current use case.
 *
 * @param subscriber {@link DisposableSubscriber} which will be listening to the observable build
 * by {@link #buildUseCaseFlowable(Params)} ()} method.
 * @param params Parameters (Optional) used to build/execute this use case.
 */
public void execute(DisposableSubscriber<T> subscriber, Params params) {
  Preconditions.checkNotNull(subscriber);
  if (subscriber.isDisposed())
    subscriber.dispose();
  final Flowable<T> observable = this.buildUseCaseFlowable(params)
      .subscribeOn(schedulerProvider.io())
      .observeOn(schedulerProvider.ui());
  addDisposable(observable.subscribeWith(subscriber));
}

项目：KomaMusic 文件：AlbumsPresenter.java

@Override
public void loadAlbums() {
    LogUtils.i(TAG, "loadAlbums");

    mDisposables.clear();

    Disposable disposable = mRepository.getAllAlbums().subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribeWith(new DisposableSubscriber<List<Album>>() {
                @Override
                public void onError(Throwable throwable) {
                    LogUtils.e(TAG, "loadAlbums onError : " + throwable.toString());
                }

                @Override
                public void onComplete() {

                }

                @Override
                public void onNext(List<Album> albumList) {
                    onLoadSongsFinished(albumList);
                }
            });

    mDisposables.add(disposable);
}

项目：KomaMusic 文件：MyFavoritePresenter.java

@Override
public void loadMyFavoriteSongs() {
    if (mDisposables != null) {
        mDisposables.clear();
    }

    Disposable disposable = mRepository.getMyFavoriteSongs().subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribeWith(new DisposableSubscriber<List<Song>>() {
                @Override
                public void onError(Throwable e) {
                    LogUtils.e(TAG, "loadMyFavoriteSongs error : " + e.toString());
                }

                @Override
                public void onComplete() {

                }

                @Override
                public void onNext(List<Song> songs) {
                    onLoadFinished(songs);
                }
            });

    mDisposables.add(disposable);
}

项目：KomaMusic 文件：RecentlyAddedPresenter.java

@Override
public void loadRecentlyAddedSongs() {
    if (mDisposables != null) {
        mDisposables.clear();
    }

    Disposable disposable = mRepository.getRecentlyAddedSongs().subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribeWith(new DisposableSubscriber<List<Song>>() {
                @Override
                public void onError(Throwable e) {
                    LogUtils.e(TAG, "loadRecentlyAddedSongs onError : " + e.toString());
                }

                @Override
                public void onComplete() {

                }

                @Override
                public void onNext(List<Song> songs) {
                    onLoadSongsFinished(songs);
                }
            });

    mDisposables.add(disposable);
}

项目：KomaMusic 文件：RecentlyPlayPresenter.java

@Override
public void loadRecentlyPlayedSongs() {
    if (mDisposables != null) {
        mDisposables.clear();
    }

    Disposable disposable = mRepository.getRecentlyPlayedSongs().subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribeWith(new DisposableSubscriber<List<Song>>() {
                @Override
                public void onError(Throwable e) {
                    LogUtils.e(TAG, "loadRecentlyPlayedSongs error : " + e.toString());
                }

                @Override
                public void onComplete() {

                }

                @Override
                public void onNext(List<Song> songs) {
                    onLoadPlayedSongsFinished(songs);
                }
            });

    mDisposables.add(disposable);
}

项目：KomaMusic 文件：ArtistsPresenter.java

@Override
public void loadArtists() {
    LogUtils.i(TAG, "loadArtists");

    mDisposables.clear();

    Disposable subscription = mRepository.getAllArtists().subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribeWith(new DisposableSubscriber<List<Artist>>() {
                @Override
                public void onError(Throwable throwable) {
                    LogUtils.e(TAG, "loadArtists onError : " + throwable.toString());
                }

                @Override
                public void onComplete() {

                }

                @Override
                public void onNext(List<Artist> artists) {
                    onLoadArtistsFinished(artists);
                }
            });

    mDisposables.add(subscription);
}

项目：KomaMusic 文件：PlayQueuePresenter.java

@Override
public void loadPlayQueue() {
    LogUtils.i(TAG, "loadPlayQueue");

    if (mDisposables != null) {
        mDisposables.clear();
    }

    Disposable disposable = mRepository.getQueueSongs().subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribeWith(new DisposableSubscriber<List<Song>>() {
                @Override
                public void onNext(List<Song> songs) {
                    onLoadPlayQueueFinished(songs);
                }

                @Override
                public void onError(Throwable t) {

                }

                @Override
                public void onComplete() {

                }
            });

    mDisposables.add(disposable);
}

项目：KomaMusic 文件：ArtistDetailPresenter.java

@Override
public void loadArtistAlbums(long artistId) {
    LogUtils.i(TAG, "loadArtistSongs atistId : " + artistId);

    if (mDisposables != null) {
        mDisposables.clear();
    }

    if (mView != null) {
        Disposable disposable = mRepository.getArtistAlbums(mView.getArtistId())
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribeWith(new DisposableSubscriber<List<Album>>() {
                    @Override
                    public void onError(Throwable throwable) {
                        LogUtils.e(TAG, "loadArtistAlbums onError : " + throwable.toString());
                    }

                    @Override
                    public void onComplete() {

                    }

                    @Override
                    public void onNext(List<Album> albums) {
                        if (mView != null) {
                            mView.showArtistAlbums(albums);
                        }
                    }
                });

        mDisposables.add(disposable);
    }
}

项目：KomaMusic 文件：AlbumDetailPresenter.java

@Override
public void loadAlbumSongs(long albumID) {
    LogUtils.i(TAG, "loadAlbumSongs albumId : " + albumID);

    if (mDisposables != null) {
        mDisposables.clear();
    }

    if (mView != null) {
        Disposable disposable = mRepository.getAlbumSongs(mView.getAlbumId())
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribeWith(new DisposableSubscriber<List<Song>>() {
                    @Override
                    public void onError(Throwable throwable) {
                        LogUtils.e(TAG, "loadAlbumSongs onError : " + throwable.toString());
                    }

                    @Override
                    public void onComplete() {

                    }

                    @Override
                    public void onNext(List<Song> songs) {
                        if (mView != null) {
                            mView.showAlbumSongs(songs);
                        }
                    }
                });

        mDisposables.add(disposable);
    }
}

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

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


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

        }

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

        }

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

        @Override
        protected void onStart() {
            // TODO Auto-generated method stub
            request(Long.MAX_VALUE);
        }
    };

    Flowable.rangeLong(5, 4).subscribe(disposableSubscriber);
    disposableSubscriber.dispose();

}

项目：rxjava2 文件：FlowableRange.java

public static void main(String[] args) {

       subscriber = new DisposableSubscriber<Integer>() {

            public void onStart() {
                request(5);

              while (true){     // Emulate some processing
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                request(1);
              }
            }

            public void onNext(Integer t) {
                System.out.println("processing "+ t);

                if (t==8) {
                    subscriber.dispose();
                }
            }

            public void onError(Throwable thr) {
                System.err.println("In onError(): " + thr.getMessage());
            }

            public void onComplete() {
                System.out.println("Done");
            }
        };

        Flowable.range(1, 10)
                .delay(1, TimeUnit.SECONDS)
                .subscribe(subscriber);
    }

项目：android-architecture-boilerplate 文件：UseCase.java

/**
 * Executes the current use case.
 *
 * @param observer {@link DisposableObserver} which will be listening to the observable build
 *                 by {@link #buildUseCaseObservable(P)} ()} method.
 * @param params   Parameters (Optional) used to build/execute this use case.
 */
public void execute(DisposableSubscriber<T> observer, P params) {
    Preconditions.checkNotNull(observer);
    final Flowable<T> observable = this.buildUseCaseObservable(params)
            .subscribeOn(Schedulers.from(mThreadExecutor))
            .observeOn(mPostExecutionThread.getScheduler());
    addDisposable(observable.subscribeWith(observer));
}

项目：RxAndroid-Examples 文件：MainActivity.java

@OnClick(R.id.btnInterval)
public void startTimerInterval(View view) {
    if (subscriberInterval != null && !subscriberInterval.isDisposed()) {
        subscriberInterval.dispose();
    }
    subscriberInterval = new DisposableSubscriber<Long>() {
        @Override
        public void onNext(Long aLong) {
            addLogMessage("Timer interval:  " + TimeUtil.getCurrentTime());
        }

        @Override
        public void onError(Throwable t) {
            addLogMessage("ERROR Timer interval: " + t.getMessage());
        }

        @Override
        public void onComplete() {
            addLogMessage("Timer interval completed !!");
        }
    };
    addLogMessage("START 2s timer interval...");
    Flowable.interval(POLL_INTERVAL, TimeUnit.SECONDS)
            .observeOn(AndroidSchedulers.mainThread())
            .subscribeOn(Schedulers.io())
            .subscribe(subscriberInterval);
}

项目：RxAndroid-Examples 文件：MainActivity.java

@OnClick(R.id.btnDelayInterval)
public void startDelayTimer(View view) {
    if (subscriberDelayInterval != null && !subscriberDelayInterval.isDisposed()) {
        subscriberDelayInterval.dispose();
    }
    subscriberDelayInterval = new DisposableSubscriber<Long>() {
        @Override
        public void onNext(Long aLong) {
            addLogMessage("Delay Timer interval: " + TimeUtil.getCurrentTime());
        }

        @Override
        public void onError(Throwable t) {
            addLogMessage("ERROR delay timer: " + t.getMessage());
        }

        @Override
        public void onComplete() {
            addLogMessage("Delay timer completed !!");
        }
    };

    addLogMessage("START timer interval after " + DELAY_TIME + "s !!");

    Flowable.interval(DELAY_TIME, POLL_INTERVAL, TimeUnit.SECONDS)
            .observeOn(AndroidSchedulers.mainThread())
            .subscribeOn(Schedulers.io())
            .subscribe(subscriberDelayInterval);
}

项目：DuckDuckDefine-Android 文件：SearchPresenter.java

/**
 * Perform search based on param
 *
 * @param term
 */
public void performSearch(String term) {
    Preconditions.checkNotNull(disposables);

    disposables.add(
        dataManager.getSearchResults(term)
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .doOnSubscribe(subscription -> getMvpView().showSearchInProgress())
                .subscribeWith(new DisposableSubscriber<SearchDefinition>() {
                    @Override
                    public void onNext(SearchDefinition searchDefinition) {
                        getMvpView().showSearchDefinition(term, searchDefinition);
                    }

                    @Override
                    public void onError(Throwable t) {
                        handleRequestError(t, "performSearch()");
                    }

                    @Override
                    public void onComplete() {
                        // Not used
                    }
                })
    );
}

项目：Inshorts 文件：ListActivityViewModel.java

/**
 * This method subscribes to the Flowable to listen to the items emitted.
 *
 * @param subscriber To listen to items emitted
 * @param sortOption Order in which the list should be fetched
 */
private void fetchArticleList(DisposableSubscriber<List<Article>> subscriber,
                              SortOption sortOption) {

    getFlowableBySortOption(sortOption)
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(subscriber);

    mCompositeDisposable.add(subscriber);
}

项目：Android-CleanArchitecture-Java 文件：GetWikiEntryUseCase.java

@Override
public void execute(Input input, DisposableSubscriber<WikiEntry> subscriber) {

    Flowable.just(input.title)
            .flatMap(title -> repo.getWikiEntry(title))
            .subscribeOn(Schedulers.newThread())
            .observeOn(input.observerOnScheduler)
            .subscribe(subscriber);

    Log.d(TAG, "called subscribe on getWikiEntry flowable");

    disposables.add(subscriber);
}

项目：PosTrainer 文件：AlarmDetailPresenter.java

public void getReminder(){
    compositeDisposable.add(
            getAlarm.runUseCase(view.getAlarmId())
                    .subscribeOn(schedulerProvider.io())
                    .observeOn(schedulerProvider.ui())
                    .subscribeWith(
                            new DisposableSubscriber<Alarm>() {
                                @Override
                                public void onNext(Alarm alarm) {
                                    view.setAlarmTitle(alarm.getAlarmTitle());
                                    view.setVibrateOnly(alarm.isVibrateOnly());
                                    view.setRenewAutomatically(alarm.isRenewAutomatically());
                                    view.setPickerTime(alarm.getHourOfDay(), alarm.getMinute());
                                    view.setCurrentAlarmState(alarm.isActive());
                                }

                                @Override
                                public void onError(Throwable e) {
                                    view.makeToast(R.string.error_invalid_alarm_id);
                                    view.startAlarmListActivity();
                                }

                                @Override
                                public void onComplete() {

                                }
                            })
    );
}

项目：PosTrainer 文件：AlarmListPresenter.java

/**
 * Checks Repository for any existing reminders.
 * returns one of:
 * List of 1-5 Reminders : Display Reminders to User
 * Nothing : Display create RealmAlarm Prompt to User
 * error : Display database error
 */
private void getReminders() {
    compositeDisposable.add(
            getAlarmList.runUseCase()
                    .subscribeOn(schedulerProvider.io())
                    .observeOn(schedulerProvider.ui())
                    .subscribeWith(
                            new DisposableSubscriber<List<Alarm>>() {
                                @Override
                                public void onNext(List<Alarm> alarms) {
                                    view.setAlarmListData(alarms);
                                }

                                @Override
                                public void onError(Throwable e) {
                                    view.makeToast(R.string.error_database_connection_failure);

                                }

                                @Override
                                public void onComplete() {
                                    view.setNoAlarmListDataFound();
                                }
                            }
                    )
    );


}

项目：rxtools 文件：SubjectMapTest.java

@Test
public void testQueryBatchOfKeys()
{
    final AtomicInteger counter = new AtomicInteger(0);

    for (int i = 0; i < loop; ++i) {
        final int index = i;

        keys[i] = "run-" + i;

        subscribe(source.get(keys[i]), new DisposableSubscriber<Integer>() {
            @Override
            public void onComplete()
            {
                fail("Unexpected completion on observable");
            }

            @Override
            public void onError(Throwable e)
            {
                fail("Unexpected error on observable");
            }

            @Override
            public void onNext(Integer value)
            {
                assertEquals(index, value % 10);

                counter.incrementAndGet();
            }
        });
    }

    for (int i = 0; i < runs; ++i) {
        source.onNext(keys[i % 10], i);
    }

    assertEquals(runs, counter.get());
}

项目：IncDec 文件：IncDecCircular.java

/** Rx Java **/

    private void startIncrementObserver(){

        _Incrementsubscriber=new DisposableSubscriber<Long>() {

            @Override
            public void onNext(Long aLong) {
                if(isLeftButtonLongPressed||isRightButtonLongPressed) {
                    IncrementAction();
                }
                else
                    _Incrementsubscriber.dispose();

            }

            @Override
            public void onError(Throwable t) {
                Log.i(TAG,t.getMessage());
            }

            @Override
            public void onComplete() {

            }
        };

        Flowable.interval(0,seconds,TimeUnit.MILLISECONDS)
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(_Incrementsubscriber);

    }

项目：IncDec 文件：IncDecCircular.java

private void startDecrementObserver(){

        _Decrementsubscriber=new DisposableSubscriber<Long>() {

            @Override
            public void onNext(Long aLong) {
                if(isLeftButtonLongPressed||isRightButtonLongPressed) {
                    DecrementAction();
                }
                else
                    _Decrementsubscriber.dispose();

            }

            @Override
            public void onError(Throwable t) {
                Log.i(TAG,t.getMessage());
            }

            @Override
            public void onComplete() {

            }
        };

        Flowable.interval(0,seconds,TimeUnit.MILLISECONDS)
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(_Decrementsubscriber);


    }

项目：IncDec 文件：IncDecImageButton.java

/** Rx Java **/

    private void startIncrementObserver(){

        _Incrementsubscriber=new DisposableSubscriber<Long>() {

            @Override
            public void onNext(Long aLong) {
                if(isLeftButtonLongPressed||isRightButtonLongPressed) {
                    IncrementAction();
                }
                else
                    _Incrementsubscriber.dispose();

            }

            @Override
            public void onError(Throwable t) {
                Log.i(TAG,t.getMessage());
            }

            @Override
            public void onComplete() {

            }
        };

        Flowable.interval(0,seconds, TimeUnit.MILLISECONDS)
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(_Incrementsubscriber);

    }

项目：IncDec 文件：IncDecImageButton.java

private void startDecrementObserver(){

        _Decrementsubscriber=new DisposableSubscriber<Long>() {

            @Override
            public void onNext(Long aLong) {
                if(isLeftButtonLongPressed||isRightButtonLongPressed) {
                    DecrementAction();
                }
                else
                    _Decrementsubscriber.dispose();

            }

            @Override
            public void onError(Throwable t) {
                Log.i(TAG,t.getMessage());
            }

            @Override
            public void onComplete() {

            }
        };

        Flowable.interval(0,seconds,TimeUnit.MILLISECONDS)
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(_Decrementsubscriber);


    }

项目：commitstrip-reader 文件：ListStripPresenter.java

@Override
public void fetchStrip(Integer numberOfStripPerPage, int page) {
    if (Configuration.OFFLINE_MODE || !CheckInternetConnection.isOnline()) {

        mListStripView.disableRefreshStrip();
    }

    mSubscriptions.add(mStripRepository
            .fetchListStrip(numberOfStripPerPage, page, false)
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribeWith(new DisposableSubscriber<StripDto>() {
                @Override
                public void onNext(StripDto strip) {
                    mListCurrentDisplayStrip.add(strip);
                }

                @Override
                public void onError(Throwable e) {}

                @Override
                public void onComplete() {
                    List<StripWithImageDto> displayStrips = new ArrayList<>();
                    for (int i = 0; i < mListCurrentDisplayStrip.size(); i++) {
                        displayStrips.add(
                                convertStripDtoToStripWithImageDto(mListCurrentDisplayStrip.get(i)));
                    }

                    mListStripView.addMoreStrips(displayStrips);
                    mListCurrentDisplayStrip.clear();
                }
            }));
}

项目：commitstrip-reader 文件：RandomStripPresenter.java

@Override
public void refreshStrip() {
    // Save all item in strips flux
    mListStripCurrentUpdate = new ArrayList<>();

    mSubscriptions.add(
            fetchRandomStrip(mListStripView.getNumberStripPerPage())
                    .subscribeWith(new DisposableSubscriber<StripDto>() {
                        @Override public void onNext(StripDto strip) {
                            mListStripCurrentUpdate.add(strip);
                        }

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

                        @Override public void onComplete() {
                            if (mListStripCurrentUpdate.size() >= 0) {

                                mListStripView.clearStripDisplayed();

                                List<StripWithImageDto> displayNewStrip = new ArrayList<>();
                                for (int i = 0; i < mListStripCurrentUpdate.size(); i++) {
                                    displayNewStrip.add(
                                            convertStripDtoToStripWithImageDto(mListStripCurrentUpdate.get(i)));
                                }

                                mListStripView.addMoreStripsFromTheStart(displayNewStrip);
                            }

                            mListStripCurrentUpdate.clear();
                            mListStripView.cancelRefreshStrip();
                        }
                    }));
}

项目：darealfragmentation 文件：SensorRx.java

public DisposableSubscriber<String> obtainSubscriber() {
  return new DisposableSubscriber<String>() {
    @Override public void onComplete() {
      System.out.println("onComplete\n");
    }

    @Override public void onError(Throwable e) {
      System.out.println("onError\n");
    }

    @Override public void onNext(String value) {
      System.out.println("onNext : value : " + value + "\n");
    }
  };
}

项目：RxJava-Android-Samples 文件：RotationPersist1Fragment.java

@Override
public void observeResults(Flowable<Integer> intsFlowable) {

  DisposableSubscriber<Integer> d =
      new DisposableSubscriber<Integer>() {
        @Override
        public void onNext(Integer integer) {
          _log(String.format("Worker frag spits out - %d", integer));
        }

        @Override
        public void onError(Throwable e) {
          Timber.e(e, "Error in worker demo frag observable");
          _log("Dang! something went wrong.");
        }

        @Override
        public void onComplete() {
          _log("Observable is complete");
        }
      };

  intsFlowable
      .doOnSubscribe(
          subscription -> {
            _log("Subscribing to intsObservable");
          })
      .subscribe(d);

  _disposables.add(d);
}