onNext start another Observable

Question

Is there any cleaner way to do the following?

I have a Android observable that debounces the requests. onNext it calls a second observable.

{// 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 instantiates a second observable:

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

};

It works. But is there a way to "merge" the two streams into one either improving readability or to create a most optimal code?

---

Edit: For sake of completeness and anyone interested in the future I manage to shrink the code into:

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

A couple of things:

I am not sure, why you are wrapping the text change events into a new Observable every time instead of just handing them to your Subject right away:

So, instead of:

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

try this:

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

Of course, then you would have a PublishSubject<String> instead of PublishSubject<Observable<String>>.

But you could skip the onNextSwitch and just debounce your Subject and continue from there.

To further simplify things, you could just use the ViewObservables from the rxjava Android package. I haven't used them lately but it should work like this:

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

A OnTextChangeEvent will then be emitted every time the underlying TextWatcher's afterTextChanged is called (in ButterKnife, by default, onTextChanged is used, so that may be a subtle difference).

In case I missed some important reason for wrapping each text change in an Observable, this can also achieved in a simpler way:

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

Note that I also omitted the subscribeOn(Schedulers.io()) - I have never felt the need to move event listeners from the UI thread to some other thread. Am I missing something here?

Finally, you could just use flatMap to actually execute the search for each search term (I am using my own Observable<OnTextChangeEvent> here):

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

You may need to move your pointObservable to the IO thread if the albumEndpoint doesn't do so already.

I hope this helps!

EDIT: Sorry, I forgot to explain what OnTextChangeEvent is. That's a part of rx-android - you can have a look at the source code here:

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

It is just a simple POJO that holds a reference to the EditText that was changed as well as the current content of the EditText. The latter is what I was using in the flatMap.