java - RxJava 中断 onNext

Question

我有一个发出长值的可观察值。一旦我收到一个>0的长值，那么我只需要中断发出并点击onComplete。有没有比我所做的黑客方法更好的方法来做到这一点

long id = -1l;
 methodA()
    .subscribeOn(Schedulers.io())
    .flatMapObservable(list -> Observable.from(list))
    .flatMap(eachObject -> methodB(eachObject))
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe(new Subscriber<Long>() {
            @Override
            public void onCompleted() {
                //some callbacks
            }
            @Override
           public void onError(Throwable e) {
                //just say error
           }
           @Override
           public void onNext(Long mId) {
                   if (id < 0) {
                    id = mId;
                   }
           }
  });

Answer 1

使用takeWhile

返回一个 Observable，只要指定条件为 true，它就会发出源 Observable 发出的项目。

Parameters:

predicate a function that evaluates an item emitted by the source Observable and returns a Boolean

Returns:

an Observable that emits the items from the source Observable so long as each item satisfies the condition defined by predicate

public Observable<T> takeWhile(final Func1<T, Boolean> predicate)

methodA()
    .subscribeOn(Schedulers.io())
    .flatMapObservable(Observable::from)
    .flatMap(this::methodB)
    .observeOn(AndroidSchedulers.mainThread())
    .takeWhile(id -> id <= 0)
    .subscribe(...);