RXAndroid Part 3 PublishSubject

By | April 14, 2017

In the previous series of tutorial on RxAndroid part 1 and 2 we have seen about basics of RxAndroid like creating observable , observer and subscription and how to apply operators to observable for manipulating data emitted by observable . In this tutorial we will see one of the interesting concept of Rxjava called Subject.

Subjects are special objects that are both an Observable and an Observer. you can think of it like a pipe You can put things into one end of the Subject and it will come out the other.

There are several types of Subjects in this tutorial we will see one of the simplest subject : a PublishSubject. With a PublishSubject, as soon as you put something in one end of the pipe it immediately comes out the other.



Lets understand the concept by a sample example increment counter on button click .

1.  Create and Subscribe 

Inside the onResume Create and Subscribe for publishSubject .

This is like subscribing to one end of the pipe to listen when there is data emission on another end .

private PublishSubject<Integer> mCounterEmitter;
private Subscription mSubscription;
private int mCounter = 0;

mCounterEmitter = PublishSubject.create();

mSubscription = mCounterEmitter.subscribe(
        new Observer<Integer>() {
            public void onCompleted() {

            public void onError(Throwable e) {

            public void onNext(Integer integer) {

2. Publish (onNext)

Set the onClickListener for button view , inside the onClick method increment the mCounter variable and call onNext method by passing mCounter variable as a paramter to it .

This is like emitting data from one of end of the pipe .

button.setOnClickListener(new View.OnClickListener() {
    public void onClick(View v) {

3. UnSubscribe

if (mSubscription != null && !mSubscription.isUnsubscribed()) {



Now lets make an sample example where user search for countries by name using pubjectSubject .

The following example is similar to Filtering RecyclerView Using SearchView but the different is there we are searching for data which is stored  in local variable and here in this example we make network request for user search query and show the result  .

1. Create and Subscribe

Inside the onResume create and subscribe for publishSubject. as you can notice we are calling debounce() on publishsubject object . So what is debounce ? Every Single time the user adds or remove character in SearchView . It is not good to send out a request to the server on every single keystroke .

What is Debounce ?

The debounce operator emits only those items from the source Observable that are not followed by another item within a specified duration.



private PublishSubject<String> mSearchResultsSubject;
private Subscription mSearchSubscription;

mSearchResultsSubject = PublishSubject.create();
mSearchSubscription = mSearchResultsSubject
        .debounce(400, TimeUnit.MILLISECONDS)
        .map(new Func1<String, List<String>>() {
            public List<String> call(String data) {
                return mRestClient.searchForCity(data);

        .subscribe(new Observer<List<String>>() {
            public void onCompleted() {


            public void onError(Throwable e) {


            public void onNext(List<String> cities) {

This is what debounce() allows us to do. It tells mSearchResultsSubject to only emit the last value that came into it after nothing new has come into the mSearchResultsSubject for 400 milliseconds .

2. Publish Text Change Listener

Adding TextChangeL Listener to SearchView , Inside the onQueryTextChange call the onNext method using publishSubject object by passing string (new text) .

public boolean onQueryTextChange(String newText) {
    return true;

3. Handle Search Result

This method is used called inside the onNext method for updating UI (recylcerView with filter list).

private void handleSearchResults(List<String> cities) {
    if (cities.isEmpty()) {
    } else {

4. UnSubscribe

Finally inside the onDestroy method call UnSubscribe on subscription object.

if (mSearchSubscription != null && !mSearchSubscription.isUnsubscribed()) {