android - onNext 启动另一个 Observable

Question

有没有更简洁的方法来执行以下操作？

我有一个 Android observable 可以消除请求。 onNext 它调用第二个可观察对象。

{// when creating the android activity
searchTextEmitterSubject = PublishSubject.create();
subscription = AndroidObservable.bindActivity(this, Observable.switchOnNext(searchTextEmitterSubject))
                .debounce(100, TimeUnit.MILLISECONDS, Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Observer<String>() {
                    @Override
                    public void onCompleted() {
                    }

                    @Override
                    public void onError(Throwable e) {
                    }

                    @Override
                    public void onNext(String s) {
                        Log.d("a",s);
                        pa.doSearch(s);
                    }
                });
}

@OnTextChanged(R.id.ed)
public void onTextEntered(CharSequence charsEntered) {
    searchTextEmitterSubject.onNext(getASearchObservableFor(charsEntered.toString()));
}


private Observable<String> getASearchObservableFor(final String searchText) {
    return Observable.create( (Subscriber<? super String> subscriber) ->
            subscriber.onNext(searchText)).subscribeOn(Schedulers.io());
}

doSearch 实例化第二个可观察对象:

public void doSearch(String string) {
    AlbumEndpoint albumEndpoint = API.getRestAdapter().create(AlbumEndpoint.class);
    Observable<Point> observable = albumEndpoint.searchPoint(string);
    mAdapterDataSubscription = AndroidObservable.bindActivity((Activity) getContext(), observable)
            .subscribe(mAdapterDataObserver);

}

private Observer<Point> mAdapterDataObserver = new Observer<Point>() {

    @Override
    public void onCompleted() {
        mAdapterDataSubscription.unsubscribe();
    }

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

    @Override
    public void onNext(Point point) {
        List<Point.Result> results = point.getResults();
        mData.addAll(results);
        notifyDataSetChanged();
    }

};

它有效。但是有没有一种方法可以将两个流“合并”为一个流，以提高可读性或创建最佳代码？

---

编辑:为了完整性和任何对 future 感兴趣的人，我设法将代码压缩为:

Observable<EditText> searchTextObservable = ViewObservable.text(ed);
searchTextObservable.debounce(100, TimeUnit.MILLISECONDS)
        .flatMap(editText -> {
            String string = editText.getText().toString();
            AlbumEndpoint albumEndpoint = getRestAdapter().create(AlbumEndpoint.class);
            return albumEndpoint.searchPoint(string);
        })
        .observeOn(AndroidSchedulers.mainThread())
        .subscribe(mAdapterDataObserver);

Answer 1

一些事情:

我不确定，为什么你每次都将文本更改事件包装到一个新的 Observable 中，而不是立即将它们交给你的主题:

所以，而不是:

@OnTextChanged(R.id.ed)
public void onTextEntered(CharSequence charsEntered) {
    searchTextEmitterSubject.onNext(getASearchObservableFor(charsEntered.toString()));
}

试试这个:

@OnTextChanged(R.id.ed)
public void onTextEntered(CharSequence charsEntered) {
    searchTextSubject.onNext(charsEntered.toString());
}

当然，那么你会有一个PublishSubject<String>而不是 PublishSubject<Observable<String>> .

但是你可以跳过 onNextSwitch只是debounce您的主题并从那里继续。

为了进一步简化事情，您可以只使用 rxjava Android 包中的 ViewObservables。我最近没有使用它们，但它应该像这样工作:

Observable<OnTextChangeEvent> searchTextObservable = ViewObservable.text(tvSearch);

A OnTextChangeEvent每次底层 TextWatcher 的 afterTextChanged 时都会发出被调用(在 ButterKnife 中，默认使用 onTextChanged，所以这可能是一个细微的差别)。

如果我错过了将每个文本更改包装在 Observable 中的一些重要原因，这也可以通过更简单的方式实现:

@OnTextChanged(R.id.ed)
public void onTextEntered(CharSequence charsEntered) {
    searchTextEmitterSubject.onNext(
        Observable.just(charsEntered.toString())
    );
}

请注意，我还省略了 subscribeOn(Schedulers.io()) - 我从未觉得有必要将事件监听器从 UI 线程移动到其他线程。我在这里遗漏了什么吗？

最后，您可以使用 flatMap实际执行每个搜索词的搜索(我在这里使用我自己的 Observable<OnTextChangeEvent>):

searchTextObservable
.debounce(100, TimeUnit.MILLISECONDS)
.flatMap(new Func1<OnTextChangeEvent, Observable<Point>>() {

    @Override
    public Observable<Point> call(OnTextChangeEvent event) {
        final String string = event.text.toString();
        AlbumEndpoint albumEndpoint = API.getRestAdapter().create(AlbumEndpoint.class);
        Observable<Point> pointObservable = albumEndpoint.searchPoint(string);
        return AndroidObservable.bindActivity((Activity) getContext(), pointObservable);
    }

})
.observeOn(AndroidSchedulers.mainThread())
.subscribe(mAdapterDataObserver);

如果 albumEndpoint 尚未这样做，您可能需要将 pointObservable 移动到 IO 线程。

希望对您有所帮助!

编辑:抱歉，我忘了解释什么 OnTextChangeEvent是。这是 rx-android 的一部分——您可以在此处查看源代码:

https://github.com/ReactiveX/RxAndroid/blob/0.x/src/main/java/rx/android/events/OnTextChangeEvent.java

它只是一个简单的 POJO，它包含对已更改的 EditText 的引用以及 EditText 的当前内容。后者是我在 flatMap 中使用的.