1. Overview
The default behavior of multiple subscribers isn’t always desirable. In this article, we’ll cover how to change this behavior and handle multiple subscribers in a proper way.
But first, let’s have a look at the default behavior of multiple subscribers.
2. Default Behaviour
Let’s say we have the following Observable:
private static Observable getObservable() {
return Observable.create(subscriber -> {
subscriber.onNext(gettingValue(1));
subscriber.onNext(gettingValue(2));
subscriber.add(Subscriptions.create(() -> {
LOGGER.info("Clear resources");
}));
});
}
This emits two elements as soon as the Subscribers subscribes.
In our example we have two Subscribers:
LOGGER.info("Subscribing");
Subscription s1 = obs.subscribe(i -> LOGGER.info("subscriber#1 is printing " + i));
Subscription s2 = obs.subscribe(i -> LOGGER.info("subscriber#2 is printing " + i));
s1.unsubscribe();
s2.unsubscribe();
Imagine that getting each element is a costly operation – it may include, for example, an intensive computation or opening an URL-connection.
To keep things simple we’ll just return a number:
private static Integer gettingValue(int i) {
LOGGER.info("Getting " + i);
return i;
}
Here is the output:
Subscribing
Getting 1
subscriber#1 is printing 1
Getting 2
subscriber#1 is printing 2
Getting 1
subscriber#2 is printing 1
Getting 2
subscriber#2 is printing 2
Clear resources
Clear resources
As we can see getting each element as well as clearing the resources is performed twice by default – once for each Subscriber. This isn’t what we want. The ConnectableObservable class helps to fix the problem.
3. ConnectableObservable
The ConnectableObservable class allows to share the subscription with multiple subscribers and not to perform the underlying operations several times.
But first, let’s create a ConnectableObservable.
3.1. publish()
publish() method is what creates a ConnectableObservable from an Observable:
ConnectableObservable obs = Observable.create(subscriber -> {
subscriber.onNext(gettingValue(1));
subscriber.onNext(gettingValue(2));
subscriber.add(Subscriptions.create(() -> {
LOGGER.info("Clear resources");
}));
}).publish();
But for now, it does nothing. What makes it work is the connect() method.
3.2. connect()
*Until ConnectableObservable‘s connect() method isn’t called Observable‘s onSubcribe() callback isn’t triggered* even if there are some subscribers.
Let’s demonstrate this:
LOGGER.info("Subscribing");
obs.subscribe(i -> LOGGER.info("subscriber #1 is printing " + i));
obs.subscribe(i -> LOGGER.info("subscriber #2 is printing " + i));
Thread.sleep(1000);
LOGGER.info("Connecting");
Subscription s = obs.connect();
s.unsubscribe();
We subscribe and then wait for a second before connecting. The output is:
Subscribing
Connecting
Getting 1
subscriber #1 is printing 1
subscriber #2 is printing 1
Getting 2
subscriber #1 is printing 2
subscriber #2 is printing 2
Clear resources
As we can see:
- Getting elements occurs only once as we wanted
- Clearing resources occur only once as well
- Getting elements starts a second after the subscribing.
- Subscribing doesn’t trigger emitting of elements anymore. Only connect() does this
This delay can be beneficial – sometimes we need to give all the subscribers the same sequence of elements even if one of them subscribes earlier than another.
3.3. The Consistent View of the Observables – connect() After subscribe()
This use case can’t be demonstrated on our previous Observable as it runs cold and both subscribers get the whole sequence of elements anyway.
Imagine, instead, that an element emitting doesn’t depend on the moment of the subscription, events emitted on mouse clicks, for example. Now also imagine that a second Subscriber subscribes a second after the first.
The first Subscriber will get all the elements emitted during this example, whereas the second Subscriber will only receive some elements.
On the other hand, using the connect() method in the right place can give both subscribers the same view on the Observable sequence.
Example of Hot Observable
Let’s create a hot Observable. It will be emitting elements on mouse clicks on JFrame.
Each element will be the x-coordinate of the click:
private static Observable getObservable() {
return Observable.create(subscriber -> {
frame.addMouseListener(new MouseAdapter() {
@Override
public void mouseClicked(MouseEvent e) {
subscriber.onNext(e.getX());
}
});
subscriber.add(Subscriptions.create(() {
LOGGER.info("Clear resources");
for (MouseListener listener : frame.getListeners(MouseListener.class)) {
frame.removeMouseListener(listener);
}
}));
});
}
The Default Behavior of Hot Observable
Now if we subscribe two Subscribers one after another with a second interval, run the program and start clicking, we’ll see that the first Subscriber will get more elements:
public static void defaultBehaviour() throws InterruptedException {
Observable obs = getObservable();
LOGGER.info("subscribing #1");
Subscription subscription1 = obs.subscribe((i) ->
LOGGER.info("subscriber#1 is printing x-coordinate " + i));
Thread.sleep(1000);
LOGGER.info("subscribing #2");
Subscription subscription2 = obs.subscribe((i) ->
LOGGER.info("subscriber#2 is printing x-coordinate " + i));
Thread.sleep(1000);
LOGGER.info("unsubscribe#1");
subscription1.unsubscribe();
Thread.sleep(1000);
LOGGER.info("unsubscribe#2");
subscription2.unsubscribe();
}
subscribing #1
subscriber#1 is printing x-coordinate 280
subscriber#1 is printing x-coordinate 242
subscribing #2
subscriber#1 is printing x-coordinate 343
subscriber#2 is printing x-coordinate 343
unsubscribe#1
clearing resources
unsubscribe#2
clearing resources
connect()* After *subscribe()
To make both subscribers get the same sequence we’ll convert this Observable to the ConnectableObservable and call connect() after the subscription both Subscribers:
public static void subscribeBeforeConnect() throws InterruptedException {
ConnectableObservable obs = getObservable().publish();
LOGGER.info("subscribing #1");
Subscription subscription1 = obs.subscribe(
i -> LOGGER.info("subscriber#1 is printing x-coordinate " + i));
Thread.sleep(1000);
LOGGER.info("subscribing #2");
Subscription subscription2 = obs.subscribe(
i -> LOGGER.info("subscriber#2 is printing x-coordinate " + i));
Thread.sleep(1000);
LOGGER.info("connecting:");
Subscription s = obs.connect();
Thread.sleep(1000);
LOGGER.info("unsubscribe connected");
s.unsubscribe();
}
Now they’ll get the same sequence:
subscribing #1
subscribing #2
connecting:
subscriber#1 is printing x-coordinate 317
subscriber#2 is printing x-coordinate 317
subscriber#1 is printing x-coordinate 364
subscriber#2 is printing x-coordinate 364
unsubscribe connected
clearing resources
So the point is to wait for the moment when all subscribers are ready and then call connect().
In a Spring application, we may subscribe all of the components during the application startup for example and call connect() in onApplicationEvent().
But let’s return to our example; note that all the clicks before the connect() method are missed. If we don’t want to miss elements but on the contrary process them we can put connect() earlier in the code and force the Observable to produce events in the absence of any Subscriber.
3.4. Forcing Subscription in the Absence of Any Subscriber – connect() Before subscribe()
To demonstrate this let’s correct our example:
public static void connectBeforeSubscribe() throws InterruptedException {
ConnectableObservable obs = getObservable()
.doOnNext(x -> LOGGER.info("saving " + x)).publish();
LOGGER.info("connecting:");
Subscription s = obs.connect();
Thread.sleep(1000);
LOGGER.info("subscribing #1");
obs.subscribe((i) -> LOGGER.info("subscriber#1 is printing x-coordinate " + i));
Thread.sleep(1000);
LOGGER.info("subscribing #2");
obs.subscribe((i) -> LOGGER.info("subscriber#2 is printing x-coordinate " + i));
Thread.sleep(1000);
s.unsubscribe();
}
The steps are relatively simple:
- First, we connect
- Then we wait for one second and subscribe the first Subscriber
- Finally, we wait for another second and subscribe the second Subscriber
Note that we’ve added doOnNext() operator. Here we could store elements in the database for example but in our code, we just print “saving… “.
If we launch the code and begin clicking we’ll see that the elements are emitted and processed immediately after the connect() call:
connecting:
saving 306
saving 248
subscribing #1
saving 377
subscriber#1 is printing x-coordinate 377
saving 295
subscriber#1 is printing x-coordinate 295
saving 206
subscriber#1 is printing x-coordinate 206
subscribing #2
saving 347
subscriber#1 is printing x-coordinate 347
subscriber#2 is printing x-coordinate 347
clearing resources
If there were no subscribers, the elements would still be processed.
So the connect() method starts emitting and processing elements regardless of whether someone is subscribed as if there was an artificial Subscriber with an empty action which consumed the elements.
And if some real Subscribers subscribe, this artificial mediator just propagates elements to them.
To unsubscribe the artificial Subscriber we perform:
s.unsubscribe();
Where:
Subscription s = obs.connect();
3.5. autoConnect()
This method implies that connect() isn’t called before or after subscriptions but automatically when the first Subscriber subscribes.
Using this method, we can’t call connect() ourselves as the returned object is a usual Observable which doesn’t have this method but uses an underlying ConnectableObservable:
public static void autoConnectAndSubscribe() throws InterruptedException {
Observable obs = getObservable()
.doOnNext(x -> LOGGER.info("saving " + x)).publish().autoConnect();
LOGGER.info("autoconnect()");
Thread.sleep(1000);
LOGGER.info("subscribing #1");
Subscription s1 = obs.subscribe((i) ->
LOGGER.info("subscriber#1 is printing x-coordinate " + i));
Thread.sleep(1000);
LOGGER.info("subscribing #2");
Subscription s2 = obs.subscribe((i) ->
LOGGER.info("subscriber#2 is printing x-coordinate " + i));
Thread.sleep(1000);
LOGGER.info("unsubscribe 1");
s1.unsubscribe();
Thread.sleep(1000);
LOGGER.info("unsubscribe 2");
s2.unsubscribe();
}
Note that we can’t also unsubscribe the artificial Subscriber. We can unsubscribe all the real Subscribers but the artificial Subscriber will still process the events.
To understand this let’s look at what is happening at the end after the last subscriber has unsubscribed:
subscribing #1
saving 296
subscriber#1 is printing x-coordinate 296
saving 329
subscriber#1 is printing x-coordinate 329
subscribing #2
saving 226
subscriber#1 is printing x-coordinate 226
subscriber#2 is printing x-coordinate 226
unsubscribe 1
saving 268
subscriber#2 is printing x-coordinate 268
saving 234
subscriber#2 is printing x-coordinate 234
unsubscribe 2
saving 278
saving 268
As we can see clearing resources doesn’t happen and saving elements with doOnNext() continues after the second unsubscribing. This means that the artificial Subscriber doesn’t unsubscribe but continues to consume elements.
3.6. refCount()
refCount() is similar to autoConnect() in that connecting also happens automatically as soon as the first Subscriber subscribes.
Unlike autoconnect() disconnecting also happens automatically when the last Subscriber unsubscribes:
public static void refCountAndSubscribe() throws InterruptedException {
Observable obs = getObservable()
.doOnNext(x -> LOGGER.info("saving " + x)).publish().refCount();
LOGGER.info("refcount()");
Thread.sleep(1000);
LOGGER.info("subscribing #1");
Subscription subscription1 = obs.subscribe(
i -> LOGGER.info("subscriber#1 is printing x-coordinate " + i));
Thread.sleep(1000);
LOGGER.info("subscribing #2");
Subscription subscription2 = obs.subscribe(
i -> LOGGER.info("subscriber#2 is printing x-coordinate " + i));
Thread.sleep(1000);
LOGGER.info("unsubscribe#1");
subscription1.unsubscribe();
Thread.sleep(1000);
LOGGER.info("unsubscribe#2");
subscription2.unsubscribe();
}
refcount()
subscribing #1
saving 265
subscriber#1 is printing x-coordinate 265
saving 338
subscriber#1 is printing x-coordinate 338
subscribing #2
saving 203
subscriber#1 is printing x-coordinate 203
subscriber#2 is printing x-coordinate 203
unsubscribe#1
saving 294
subscriber#2 is printing x-coordinate 294
unsubscribe#2
clearing resources
4. Conclusion
The ConnectableObservable class helps to handle multiple subscribers with little effort.
Its methods look similar but change the subscribers’ behavior greatly due to implementation subtleties meaning even the order of the methods matters.
The full source code for all the examples used in this article can be found in the GitHub project.