How to make an Event Bus with RxJava and RxAndroid

If you’ve ever needed to communicate between different parts of your application, it can be painful. To alleviate this, you can use an event bus like Otto. But, with the addition of RxJava and RxAndroid to the Android ecosystem you don’t need to rely on Otto anymore. Otto is actually deprecated in favour of these newer libraries, since making your own event bus with them is actually quite easy.

I came up with my own solution that works well for my purposes. You can use it as is if you want, or tweak it to fit your needs.

First attempt

If you just want to pass arbitrary data around your app this is all you need:

Using a PublishSubject means that a new subscriber will only receive events emitted after they subscribed. It will not repeat old events to new subscribers.

This simple solution has a few problems though. You can’t pick what kind of events your subscriber will receive, it’s going to see everything. The only way to ensure you get the data you want, is to define a new class for each event, such as XYZDataDownloadedEvent or PurpleButtonClickedEvent. Then you have to do an instanceof check and cast it. Personally, I don’t like having to create a new class for each type of event I want to broadcast.

Also, this solution can cause memory leaks if you don’t unsubscribe from each subscription. Ideally I want to be able to publish an event and subscribe to updates for that event. I don’t want to have to manage subscriptions in each place I subscribe.

Second attempt

To address these issues, my next iteration looked like this:

I added a CompositeDisposable to manage all of the subscriptions for a given object (typically an Activity or Fragment). I also added a map to keep track of the different types of subjects. That way, I can easily keep track of all of the subscriptions and subjects.

Also, to simplify management of unsubscribing and keeping a reference to those subscriptions, I created a BaseActivity and BaseFragment.

Calling RxBus.unregister(this) in onDestroy() is all that’s required for cleanup. If there are subscriptions associated to that object they will be unsubscribed and removed. I also wrote comments to make it clear that if you subscribe, you need to call unregister. In case you aren’t using a base class that handles it or subscribe to the bus from somewhere else.

Whenever I create a new subject, it’s set to be subscribed on the main thread. For my purposes, all of the events being posted will trigger UI updates. You could always extend it to allow for subscribing on different threads if you want. The current implementation makes it simple and covers the majority of use cases.

Final implementation

The last change I made is to how you define what subject you are subscribed to. Initially you could just pass in a String key, defined wherever you like. I like having all these keys organized in one place though. Also, I wanted to limit the events you could subscribe and publish to. So, I changed the String parameter to an int, and created a set of integer constants with the IntDef annotation. You can see my completed RxBus.java class below:

Potential issues

While working on this, I made a list of problems I had with the implementation. Some of which I believe can be addressed, others I’m not sure.

I’m still passing around objects which have to be cast to the correct type. I’m not sure if there’s a way around this, because the subject publishes Objects. So, the subscriber will only receive Objects.

You can pass in any object to associate your subscription with, so there’s no guarantee that you’ve actually unsubscribed. I tried to address this with my comments, saying that you must call unregister. But there’s no guarantee that it gets called, which will cause memory leaks.

The BaseActivity and BaseFragment unregister from the bus in onDestroy(). This means that if you start a new activity, the old activity will still be subscribed. So if you publish an event that the previous activity is subscribed to, it may end up causing your app to crash with java.lang.IllegalStateException: Can not perform this action after onSaveInstanceState. I didn’t want to call unregister in onStop() because if you go back to the previous activity, it won’t be subscribed anymore. If you are careful with how you manage your subjects, this won't be an issue. Ideally the subscriptions would pause and resume with the lifecycle, and finally get destroyed with the lifecycle.

Lastly, I’m using static members instead of the singleton pattern. Technically I believe that using the singleton pattern is more memory efficient. Since it will only create the class when it needs to. But, in my case since I’m using RxBus in onCreate() for most of my activities, it won’t really save anything. Plus, the amount of memory it uses is negligible. Some people also think that static variables are evil.

Final thoughts

This solution isn’t perfect, but I felt it was a good compromise between complexity and ease of use.

Pierce Zaifman

Android app developer. Hates wasting time and takes satisfaction in improving efficiency in all aspects of life. Also enjoys helping people solve their problems.

London, Ontario, Canada

Subscribe to Pierce Zaifman

Get the latest posts delivered right to your inbox.

or subscribe via RSS.