RxJava as event bus?

Question

I'm start learning RxJava and I like it so far. I have a fragment that communicate with an activity on button click (to replace the current fragment with a new fragment). Google recommends interface for fragments to communicate up to the activity but it's too verbose, I tried to use broadcast receiver which works generally but it had drawbacks.

Since I'm learning RxJava I wonder if it's a good option to communicate from fragments to activities (or fragment to fragment)?. If so, whats the best way to use RxJava for this type of communication?. Do I need to make event bus like this one and if that's the case should I make a single instance of the bus and use it globally (with subjects)?

Answer 1

Yes and it's pretty amazing after you learn how to do it. Consider the following singleton class:

public class UsernameModel {

    private static UsernameModel instance;

    private PublishSubject<String> subject = PublishSubject.create();

    public static UsernameModel instanceOf() {
        if (instance == null) {
            instance = new UsernameModel();
        }
        return instance;
    }

    /**
     * Pass a String down to event listeners.
     */
    public void setString(String string) {
        subject.onNext(string);
    }

    /**
     * Subscribe to this Observable. On event, do something e.g. replace a fragment
     */
    public Observable<String> getStringObservable() {
        return subject;
    }

}

In your Activity be ready to receive events (e.g. have it in the onCreate):

UsernameModel usernameModel = UsernameModel.instanceOf();

//be sure to unsubscribe somewhere when activity is "dying" e.g. onDestroy
subscription = usernameModel.getStringObservable()
        .subscribe(s -> {
            // Do on new string event e.g. replace fragment here
        }, throwable -> {
            // Normally no error will happen here based on this example.
        });

In you Fragment pass down the event when it occurs:

UsernameModel.instanceOf().setString("Nick");

Your activity then will do something.

Tip 1: Change the String with any object type you like.

Tip 2: It works also great if you have Dependency injection.

---

Update: I wrote a more lengthy article

Answer 2

Currently I think my preferred approach to this question is this to:

1.) Instead of one global bus that handles everything throughout the app (and consequently gets quite unwieldy) use "local" buses for clearly defined purposes and only plug them in where you need them.

For example you might have:

• One bus for sending data between your Activitys and your ApiService.
• One bus for communicating between several Fragments in an Activity.
• One bus that sends the currently selected app theme color to all Activitys so that they can tint all icons accordingly.

2.) Use Dagger (or maybe AndroidAnnotations if you prefer that) to make the wiring-everything-together a bit less painful (and to also avoid lots of static instances). This also makes it easier to, e. g. have a single component that deals only with storing and reading the login status in the SharedPreferences - this component could then also be wired directly to your ApiService to provide the session token for all requests.

3.) Feel free to use Subjects internally but "cast" them to Observable before handing them out to the public by calling return subject.asObservable(). This prevents other classes from pushing values into the Subject where they shouldn't be allowed to.

Answer 3

Define events

public class Trigger {

public Trigger() {
}

public static class Increment {
}

public static class Decrement {
}

public static class Reset {
}
}

Event controller

public class RxTrigger {

private PublishSubject<Object> mRxTrigger = PublishSubject.create();

public RxTrigger() {
    // required
}

public void send(Object o) {
    mRxTrigger.onNext(o);
}

public Observable<Object> toObservable() {
    return mRxTrigger;
}
// check for available events
public boolean hasObservers() {
    return mRxTrigger.hasObservers();
}
}

Application.class

public class App extends Application {

private RxTrigger rxTrigger;

public App getApp() {
    return (App) getApplicationContext();
}

@Override
public void onCreate() {
    super.onCreate();
    rxTrigger = new RxTrigger();
}


public RxTrigger reactiveTrigger() {
    return rxTrigger;
}


}

Register event listener wherever required

               MyApplication mApp = (App) getApplicationContext();
               mApp
                    .reactiveTrigger() // singleton object of trigger
                    .toObservable()
                    .subscribeOn(Schedulers.io()) // push to io thread
                    .observeOn(AndroidSchedulers.mainThread()) // listen calls on main thread
                    .subscribe(object -> { //receive events here
                        if (object instanceof Trigger.Increment) {
                            fabCounter.setText(String.valueOf(Integer.parseInt(fabCounter.getText().toString()) + 1));
                        } else if (object instanceof Trigger.Decrement) {
                            if (Integer.parseInt(fabCounter.getText().toString()) != 0)
                                fabCounter.setText(String.valueOf(Integer.parseInt(fabCounter.getText().toString()) - 1));
                        } else if (object instanceof Trigger.Reset) {
                            fabCounter.setText("0");
                        }
                    });

Send/Fire event

 MyApplication mApp = (App) getApplicationContext();
 //increment
 mApp
    .reactiveTrigger()
    .send(new Trigger.Increment());

 //decrement
 mApp
    .reactiveTrigger()
    .send(new Trigger.Decrement());

Full implementation for above library with example -> RxTrigger