Clean Architecture & RxJava — changing the repository source on the fly
Recently we at grandcentrix had to implement a different polling logic based on a mutating state: in the default state we were fine to read data from the local cache or query the backend if that cache is empty; in another state we needed to hit the backend for updated data in a regular interval.
This mutating state could be everything in your app: location, network or bluetooth settings, time etc.
While heavily using Clean Architecture and RxJava in the app, we wanted to make the state so irrelevant as possible, not interrupting any subscriptions when the state changes, optimally just start to emit regularly to any existing subscribers. But which part of a Clean Architecture application is responsible for checking and observing that state?
Should it be the presenter that checks the state and executes the use case either once or every x seconds? Pretty fast we said no, the presenter should not know anything about the business logic. Also we needed the logic at different places, so it should be the use case to decide, right?
Well, optimally a use case only transforms data it gets from the data layer resp. the repository, applies some logic on the data, like mapping or filtering, and doesn’t decide how or when to get the data.
Further, when reading the note from Fernando Cejas about the data layer, it sounds like an awesome fit:
The idea behind all this is that the data origin is transparent for the client, which does not care if the data is coming from memory, disk or the cloud, the only truth is that the data will arrive and will be got.
So, the repository is already responsible for deciding what data gets returned from where, why not let it decide how often to return it, too? Combined with RxJava this means that our repository does a new network call every thirty seconds, pushes the new data on the upstream and neither the use case, nor the presenter have to care about the state at all, they just transform resp. display the data they receive. Even better, the repository can react on network or connectivity issues by falling back to the local cache, so no other component has to care about errors (although they could)!
Let’s assume the following repository interface:
Depending on one of two states we want to make the
Observable emit either once or in an interval. Let’s call the two functions that serve the data for either state
specialStateObservable and put them in a new class:
Now, let’s make
Emitter observe any state changes and call a method when the state changes:
Now we would have to subscribe to the specific
Observable depending on the state. The problem is, our repository only has one method to get the data. If a subscriber is already subscribed to
defaultStateObservable() and now we need to switch to
specialStateObservable(), we cannot tell the subscriber to resubscribe.
Subject to the rescue!
We let the subscriber subscribe to a
Subject so we don’t have to interrupt the subscription but can still control which state observable feeds the
Subject! Let’s take a look:
Now we can change the data source on the fly when the state changes while our subscriber doesn’t know anything about it 😎. Each emission of one of the
Observables will be delegated to the
Subject, so our subscriber receives the event.
But subscribing to the
Observables although now one requires the data right now? 🤔 Also we’re leaking the subscription as we never dispose.
Let’s fix this by subscribing and disposing to/from our
Observables depending on if our
Subject has a subscriber or not:
Woah, a lot of changes. Let’s go through it step by step.
When someone subscribes to our
Subject by calling
stateObservable()we call an action based on the subscription and disposal event. If this is the first subscriber to the
Subject we also subscribe to the state
Observable, if the last subscriber disposes from the
Subject we also dispose from the state
Observable, so we’re only subscribed when someone actually requests data.
Also, when the state changes, we now first check if we even have subscribers. If not, we do nothing, as the
onSubscribeConsumer will later trigger the subscription to the internal
Observable. Otherwise, for simplification, we just dispose from the current
Observable and subscribe to the correct
Observable for the new state.
Well, and that’s it already. Now we have a working data stream that changes the emission frequency based on the
state, while any subscribers just stay subscribed.
Let’s put the parts quickly together and use the
Emitter in a repository implementation: