If you enjoy talks from 360 AnDev, please support the conference via Patreon!

The Android development community today has embraced Rx(Java) in all its glory. But as developer understanding of RxJava has matured, they’ve begun to peel back the layers to unleash its true power and potential.

One such power is “Multicasting”, where you get to share work across your app.

Rx itself has a steep learning curve, and the easiest way to grasp the concepts and features of Rx, is through examples.

What we’ll do today is learn by examples!

Introduction

My name is Kaushik, and I work as an Android developer at Instacart. Today I’d to talk to you about Rx, or the active extensions. I will focus on one specific subtopic in Rx Java called multicasting.

Multicasting

If I have this source observable or stream, and I’m casting the events from that source observable to multiple subscribers it’s called multicasting. But why would I even bother multicasting?

As an Android developer, when you set up the source observable, it can be expensive; the most typical example is a network request. If you have a network request that takes time to execute, and I have more than one listener or subscriber that’s interested in the results of that observable results, then I don’t want to re-execute that network request. All I want to do is maybe reuse that result, and multicasting allows that.

There are other cool and powerful things that you can do with multicasting, by virtue of sharing the events across multiple subscribers.

How to Multicast

Primarily, there are two ways to multicast: using subjects and using connectable observables. I’m not going to focus on subjects. I want more info to come on connectable observables because that’s the more idiomatic way, and, it’s also powerful.

Multicasting with ConnectableObservables

Connectable observables are the only observables that permit multicasting. Regular observables don’t permit multicasting.

Consider the following example in Kotlin:

val source: Observable<Long> =
                Observable.interval(1, TimeUnit.SECONDS)

source.subscribe { Log.v(TAG, "subscriber 1 : received event $it") }
source.subscribe { Log.v(TAG, "subscriber 2 : received event $it") }


source.subscribe { Log.v(TAG, "subscriber 1 : received event $it") }
Thread.sleep(1000)
source.subscribe { Log.v(TAG, "subscriber 2 : received event $it") }

val source: Observable<Long> =
                Observable.interval(1, TimeUnit.SECONDS).publish()

source.subscribe { Log.v(TAG, "subscriber 1 : received event $it") }

Thread.sleep(1000)

source.subscribe { Log.v(TAG, "subscriber 2 : received event $it") }

source.connect()

There are two subscribers. The first subscribes to the source, and the minute it subscribes to the source, it prints out the event that it sees at that instant of time. The same happens for the second subscriber.

The next set of events is interesting. You have subscriber one, and subscriber two. But notice that the actual event that’s being seen is different. You see one zero, two one, three two, four three. This is an important property of RxJava. Whenever on a regular observable, whenever you call subscribe, RxJava has immutable behavior, and it recreates the source observable.

Publish is a convenient magical operator that converts an observable to a connectable observable. You put in this operator at the end, and your observable gets changed to a connectable observable. Then, you will have to call connect.

refCount

There is an operator called refCount. It’s called on the source observable, and as a result, I don’t have to call connect. This is the true advantage that you get with refCount.

The concept behind refCount is reference counting in computer science. Think of it as a tally counter. If you have one subscriber, it increments by one. If that person leaves, it decrements. If two come in, then there are two subscribers.

Here, I’m telling Rx any time you see at least one subscriber, start emitting. refCount now is going to start emitting that timer observable.

I called publish and refCount - it automatically understands the number of subscribers and sends the respective events simultaneously. This is how multicasting is done.

autoConnect

If I wanted to have both buttons subscribed, and I wanted the source observable to start emitting only when two subscribers have entered in, how do I do that?

I can use an operator called autoConnect to accomplish this. autoConnect takes N as a parameter, with N standing for the number of subscribers.

I’m going to click subscribe two, and send that event up. It sees there are two subscribers, and autoConnect starts emitting those events.

Unsubscribe from the first one, and it goes all the way to the top. You’re saying autoConnect of two, once autoConnect starts, it doesn’t stop. This is important to remember; it’s a stubborn operator.

Use Cases

I want to run through real world cases where I’ve found multicasting to be useful.

Use case No. 1

With location providers, setting up the source observable is costly. Google is going to get angry if you keep GPS radion on for too long, for example. You want to make sure that that isn’t necessarily held connected for too long.

If you had a background service that had location running, but also have a foreground activity where you want to see the location, you will have two subscribers - the background service and the foreground activity. I would want to multicast this; I wouldn’t want to recreate this.

There’s a library that’s named Rx Location, which sends you the location as an observable. It masks all the different things that you have to do to get your GPS location. What I want to do now is figure out what operator should I use.

Do I want publish or replay? Replay, because it would be nice if I had my last location. If want only most accurate location right now, publish should work because it wouldn’t send you the last event.

Do I want refCount or autoConnect? RefCount, because you don’t want to keep your location provider connected for too long.

Use case No. 2

Share view mode with network request. Here, the share view returns an observable of data. Whatever the API response returns, I want to return our shared stream. This is going to be an observable, and I’m going to store that as a reference. The view model now holds this observable.

I know it has to be a connectable observable. Should I use publish or replay?

Replay, because publish doesn’t replay. Publish doesn’t remember the last event.

There’s one problem with autoConnect: it doesn’t stop on its own. With autoConnect, if your network request completes in time, you’re in the clear.

View models have a helpful call back called onClear. This is where I would store the reference of my disposal from the autoConnect in my view model.

Use case 3

In functional programming, side effects are bad. In Rx Java, doOnNext() is executing side effects. With multicasting, you can easily avoid this.

I’ve split this into three colors (see slides). The top part is my source observable - I’m going to consider doOnNext() as an independent subscriber. The green part, the core one, is my original subscriber. I’m going to tease these three apart to convert that paginator into a connectable observable.

Q & A

Q: I use publish subjects a lot right now and I’ve never used connectable observers before– Kaushik: When should I use connectable observables versus subjects? Typically, when I wanted to use subjects, it’s because I want to sneak in an event. You don’t do that with connectable observables. With connectable observables, you set the source stream up, and then you deal with the subscribers.

Q: If multicasting streams error or terminate, what happens to the subscribers? Kaushik: Think of connectable observables as you would think of regular observables. One big problem with that side effect is, if I did something inside that do on next that errored, that would propagate throughout the stream.

What happens when on errors happen in regular streams? It will terminate. It doesn’t matter if you’ve multitasked or using connectable observable.

There are ways you can handle it: there’s doOnError, onErrorReturn, onErrorResume.

Next Up: RxJava #8 – Rxify: Complex RxJava Operators