Introduction
Threading and lifecycle are one of the top concerns when developing applications for the Android platform. UI has to be interacted with on a dedicated thread (main thread) but in order to keep the UI responsible to user input and rendering, blocking or CPU intensive calculations should be kept off it.
In addition, views can get destroyed and recreated in a way that is outside of a given application's control unlike a desktop
Swing application. This means background tasks must be stopped and listeners removed to prevent leaking references to now logically dead objects.
RxJava and
RxAndroid can help with threading concerns and there are other libraries that tap into the various lifecycle events; in general, this means someone will call
dispose() on a particular flow or
clear() on a
CompositeDisposable to mass-cancel multiples of them.
Having a rich set of transformative and coordinating operators along with support for normal values, errors and finite sequences may be overwhelming compared to a classical
Listener-based API. Google's
LiveData is one of such classical
Listener style APIs but unlike Swing's
ActionListener for example, there are explicit requirements that interaction with the
LiveData object itself happens on the main thread and signals will be dispatched from the main thread to
Observers to it.
LiveData API
Unfortunately, I wasn't able to locate a public repository for the LiveData sources and had to rely on the sources downloaded from Google's
Maven repository:
compile "android.arch.lifecycle:reactivestreams:1+"
There is an interoperation library associated with LiveData that allows presenting and consuming events from any Reactive-Streams
Publisher. This will transitively import the actual LiveData library. Note that LiveData is currently considered
beta and may change arbitrarily before release. That said, I don't think the core structure and premise will actually change.
The main consumer type is the
android.arch.lifecycle.Observer with its single
onChanged(T) method. Given an arbitrary
LiveData instance, one can use
observe(LifecycleOwner, Observer) or
observeForever(Observer) methods to attach the listener to a source. The
onChanged(T) will be called on the
main thread. Cancelling an observation requires one to call
removeObserver() for which one has to remember both the original
LiveData object and the
Observer instance itself. All the methods return
void. The
LifecycleOwner will be listened for destroy events, at which point the
Observers are removed from the
LiveData instance that depended on that specific source. This is similar to
RxBinding's
bindToLifecycle() approach composed onto a flow.
The
LiveData abstract class doesn't feature any transformative or coordinating operators on itself, but the library features an utility class,
Transformations, currently with a
map and
switchMap operations. Both will execute the transformations and emissions on the main thread, but otherwise they match the behavior of the same RxJava operators.
In theory, one could interpret
LiveData as a cold source, but its usage patterns point to a hot source, just like a
BehaviorSubject in RxJava;
LiveData remembers the last value and emits it first to new consumers. To signal to consumers,
LiveData has protected methods
setValue and
postValue.
setValue requires the value to be set on the main thread and
postValue will schedule a task to the main thread to call
setValue there. To manually invoke these methods, a
MutableLiveData object is available where the two methods have been made public.
Possible shortcomings
The main shortcoming I see with LiveData is the ubiquitous "return to main thread" mentality. For example, querying data from a
LiveData-enabled datasource and preprocessing it should happen in a background thread as well before the aggregated data is sent back to the main thread for displaying them. Unfortunately, the datasource will notify the preprocessing
Observer on the main thread for each resulting item, which
Observer then has to schedule tasks on a background thread for all those items. The main thread is involved too early in such flows which wastes time for the main thread.
A second shortcoming could be the lack of terminal event notifications included in the
Observer API. Of course, one can define an union-class with the value, error and complete subtypes, just like RxJava's Notification object, to work around this problem. Needless to say, such wrapper will cost allocation and indirection.
Lastly,
addListener/
removeListener pairs compose relatively poorly and requires the operator to remember both the original
LiveData and the
Observer that was used when the link was established. An
onSubscribe(Disposable) style cancellation support worked out much nicer in RxJava 2. Of course, having a similar method in
Observer would break its lambda-friendly Single Abstract Method structure.
Reactive-Streams interoperation
Consuming a LiveData as Publisher
RxJava has infrastructure support to work with such classical
Listener-style sources. Specifically, the
create() operator on
Observable and
Flowable show such an example usage in its JavaDoc. Consequently, talking to
LiveData is a straightforward routine:
LiveData<String> source = new MutableLiveData<String>();
Flowable<String> f = Flowable.<String>create(emitter -> {
Observer<String> o = v -> {
emitter.onNext(v);
};
source.observeForever(o);
emitter.setCancellable(() ->
AndroidSchedulers.mainThread()
.scheduleDirect(() ->
mld.removeObserver(o)
)
);
}, BackpressureStrategy.LATEST);
The Reactive-Streams bridge in
LiveDataReactiveStreams, of course, has no access to
Flowable and has to work out the
Subscription management manually. Unfortunately, the
toPublisher() implementation is wrong. If there was a public repository, I could contribute a correct one (without using RxJava of course).
Let's see the implementation piece by piece to demonstrate how
not to write Publishers. (The file is marked as Apache 2.0 open source, should be fine to quote it here, right?)
public static <T> Publisher<T> toPublisher(
final LifecycleOwner lifecycle, final LiveData<T> liveData) {
return new Publisher<T>() {
boolean mObserving;
boolean mCanceled;
long mRequested;
@Nullable
T mLatest;
@Override
public void subscribe(final Subscriber<T> subscriber) {
// continued below
}
};
}
The first red flag comes from the mutable, non thread-safe state of the anonymous
Publisher. Hot sources may have mutable fields but those are protected with proper synchronization primitives (see
PublishProcessor for example). However, these fields should be part of the state of the
Subscription that is sent to a
Subscriber as
Publishers are expected to be consumed by any number of
Subscribers and interaction with one
Subscriber should not affect the interactions with the others.
Of course, a
Publisher can chose to only service a single
Subscriber (as with
UnicastProcessor), but for that, there is no field to let the
subscribe() know there is already someone consuming the
Publisher.
Next, the
Observer instance is defined with a backpressure strategy of keeping the latest if the downstream is not ready.
final Observer observer = new Observer() {
@Override
public void onChanged(@Nullable T t) {
if (mCanceled) {
return;
}
if (mRequested > 0) {
mLatest = null;
subscriber.onNext(t);
if (mRequested != Long.MAX_VALUE) {
mRequested--;
}
} else {
mLatest = t;
}
}
};
Apart from the wrong shared state of the
mCanceled,
mRequested and
mLatest, the code relies on the fact that both the
onChanged invocation and the
request() call happens on the main thread (see below). Having a per
Subscriber atomic
mRequested instead would save a trip to the main thread for an increment.
The next segment deals with the
request() call from the downstream.
subscriber.onSubscribe(new Subscription() {
@Override
public void request(final long n) {
if (n < 0 || mCanceled) {
return;
}
ArchTaskExecutor.getInstance().executeOnMainThread(new Runnable() {
@Override
public void run() {
if (mCanceled) {
return;
}
// Prevent overflowage.
mRequested = mRequested + n >= mRequested
? mRequested + n : Long.MAX_VALUE;
if (!mObserving) {
mObserving = true;
liveData.observe(lifecycle, observer);
} else if (mLatest != null) {
observer.onChanged(mLatest);
mLatest = null;
}
}
});
}
The first (smaller) problem is that non-positive requests should be rewarded with an
onError(IllegalArgumentException) properly serialized with any
onNext calls. Here, it is ignored. In addition,
request() may be called from any thread and such call may not see
mCanceled properly. The
mObserving adds an odd delay to start observing the actual
LiveData object. (Since most sequences will issue a
request() almost immediately, I'm not sure what benefit this can give, however, there were voices pretty keen on such behavior recently who couldn't understand that to control the first invocation of
request(), one would have to consume a
Publisher directly with a
Subscriber and not have any intermediate operators between the two.)
Lastly, let's see
cancel():
@Override
public void cancel() {
if (mCanceled) {
return;
}
ArchTaskExecutor.getInstance().executeOnMainThread(new Runnable() {
@Override
public void run() {
if (mCanceled) {
return;
}
if (mObserving) {
liveData.removeObserver(observer);
mObserving = false;
}
mLatest = null;
mCanceled = true;
}
});
}
Since
removeObserver has to be called from the main thread (which by itself could be inconvenient), a task is scheduled which will also update
mCanceled to true. Since
cancel() can be invoked from any thread, a non-volatile
mCanceled may not be visible. In addition, since the
mCanceled is set to
true on the main thread, there could be a significant delay between a
cancel() and its effects, letting
onChanged() calls still through. Setting a volatile
mCanceled true right in the
cancel() method could have the stopping effects much earlier.
(In fact, this was a problem with our early
unsubscribeOn() where the Reactive-Streams TCK would complain about still receiving too events after it has cancelled the flow. The solution was to set a volatile cancelled flag that would be read by
onNext and drop events until the
upstream.cancel() got executed asynchronously.)
It is relatively simple to fix the mistakes here: combine
Observer and
Subscription into one class and move the
mXXX fields into this class.
Exposing a Publisher as LiveData
The unfortunate lack of terminal events in
LiveData forces us to deal with only
onNext calls. There is no way to tell an
Observer that no more events will come (unless agreeing upon an event container type) and then get rid of them en-masse (there is no
clearObservers()).
With RxJava, this could be done relatively easily:
MutableLiveData<String> mld2 = new MutableLiveData<>();
Disposable d = Flowable.range(1, 5).map(Object::toString)
.subscribe(mld2::postValue);
Of course, one has to manage the
Disposable as usual and possibly decide, when to
subscribe() to a cold source with respect to available
Observers of the (
Mutable)
LiveData. A
refCount-like behavior is possible but for that, one has to implement a custom
LiveData object.
Apparently, there exist events on the
LiveData class,
onActive and
onInactive, that will be called when the first
Observer arrives and the last
Observer leaves respectively. The
ReactiveStreamsLiveData.fromPublisher() uses them in its implementation:
private static class PublisherLiveData<T> extends LiveData<T> {
private WeakReference<Subscription> mSubscriptionRef;
private final Publisher mPublisher;
private final Object mLock = new Object();
PublisherLiveData(@NonNull final Publisher publisher) {
mPublisher = publisher;
}
// the rest is below
}
The
mSubscriptionRef holds the current active connection's
Subscription and the
mLock protects it so that a concurrent call to
onSubscribe may not clash with the deactivation of the
LiveData object. As we'll see in a short, this is not enough because the race can leave the upstream's
Subscription from an old activation not cancelled and thus still posting to the
LiveData object.
The
onActivate event should subscribe to the source and relay its
onNext events:
@Override
protected void onActive() {
super.onActive();
mPublisher.subscribe(new Subscriber<T>() {
@Override
public void onSubscribe(Subscription s) {
// Don't worry about backpressure. If the stream is too noisy then
// backpressure can be handled upstream.
synchronized (mLock) {
s.request(Long.MAX_VALUE);
mSubscriptionRef = new WeakReference<>(s);
}
}
@Override
public void onNext(final T t) {
postValue(t);
}
@Override
public void onError(Throwable t) {
synchronized (mLock) {
mSubscriptionRef = null;
}
// Errors should be handled upstream, so propagate as a crash.
throw new RuntimeException(t);
}
@Override
public void onComplete() {
synchronized (mLock) {
mSubscriptionRef = null;
}
}
});
}
Here, apart from the
mSubscriptionRef management, the only small mistake is the call to
request() from within the lock itself. A bigger mistake is that
onError should not throw.
The
onInactive should cancel any existing
Subscription if the
Subscription representing that connection has been received:
@Override
protected void onInactive() {
super.onInactive();
synchronized (mLock) {
WeakReference<Subscription> subscriptionRef = mSubscriptionRef;
if (subscriptionRef != null) {
Subscription subscription = subscriptionRef.get();
if (subscription != null) {
subscription.cancel();
}
mSubscriptionRef = null;
}
}
}
Unfortunately, the call to
onSubscribe is not guaranteed to happen on the same thread the
Publisher.subscribe() is invoked. Even though RxJava's
Flowable almost always does so, other Reactive-Streams implementations are free to delay the call to
onSubscribe and issue it from any thread. (We deal with such situations in
Flowable via the
deferred requesting/cancellation mechanism.)
Consider the following scenario. We know
onActive and
onInactive execute on the main thread thus they are serialized in respect to each other. Let's assume a
Publisher will signal
onSubscribe in a delayed manner when subscribed to. The main thread executes
onActive that calls
subscribe() on the
Publisher. Then the main thread executes
onInactive() after. Since the
Publisher hasn't called
onSubscribe yet, there is no
mSubscriptionRef and thus there is nothing to cancel. The
onSubscribe() is then invoked by the
Publisher and it starts streaming events (to nobody practically). Now the main thread executes
onActive again, triggering yet another subscription to the
Publisher. After
onSubscribe() is called again, the previous
Subscription is overwritten and there are now twice as many
onNext events posted to the
LiveData (these could be duplicates in case the
Publisher is hot/multicasting, or arbitrary items from various stages of a cold
Publisher). Since the previous
Subscription is lost, it is not possible to stop the first subscription other than letting it terminate on its own pace.
One possible solution is to host the
Subscriber in an
AtomicReference inside the custom
LiveData, plus have that
Subscriber store the incoming
Subscription in an
AtomicReference too. Therefore, an
onInactive can get rid of the previous
Subscriber and at the same time, instruct it to cancel the
Subscription incoming through
onSubscribe without requesting from it:
static final class PublisherLiveData<T> extends LiveData<T> {
final Publisher<T> mPublisher;
final AtomicReference<SubscriberLiveData> mSubscriber;
PublisherLiveData(@NonNull final Publisher<T> publisher) {
mPublisher = publisher;
mSubscriber = new AtomicReference<>();
}
@Override
protected void onActive() {
super.onActive();
SubscriberLiveData s = new SubscriberLiveData();
mSubscriber.set(s);
mPublisher.subscribe(s);
}
@Override
protected void onInactive() {
super.onInactive();
SubscriberLiveData s = mSubscriber.getAndSet(null);
if (s != null) {
s.cancel();
}
}
final class SubscriberLiveData
extends AtomicReference<Subscription>
implements Subscriber<T>, Subscription {
@Override
public void onSubscribe(Subscription s) {
if (compareAndSet(null, s)) {
s.request(Long.MAX_VALUE);
} else {
s.cancel();
}
}
@Override
public void onNext(T item) {
postValue(item);
}
@Override
public void onError(Throwable ex) {
lazySet(this);
mSubscriber.compareAndSet(this, null);
ex.printStackTrace();
}
@Override
public void onComplete() {
lazySet(this);
mSubscriber.compareAndSet(this, null);
}
@Override
public void cancel() {
Subscription s = getAndSet(this);
if (s != null && s != this) {
s.cancel();
}
}
@Override
public void request(long n) {
// never called
}
}
}
Although
onActive and
onInactive would update
mSubscriber from the main thread, the atomics is preferable since we want to set the current
Subscriber to
null once the
Publisher terminated, thus playing nice with the GC. (For comparison, the
refCount in RxJava is more involved because connection and disconnection may be triggered from any thread at any time.)
Conclusion
The
LiveData class can be considered a classical (gen 0) reactive-push component which has practically one purpose on Android: notify its
Observers of data (changes) on the
main thread. If the typical thread crossing in the app is mainly from a background thread to the main thread,
LiveData may be just enough to suit one's needs. It's one step above
Agera after all since
LiveData actually transmits data to be acted upon, not just "something happened" signals. However, if possibly failing and finite sources are present, which also have to be coordinated, transformed and kept on background thread(s) as long as possible, I believe
RxJava is a better choice.
Nevertheless, there are standard interoperation bridges provided with
LiveData which would allow it to work with or pose as a Reactive-Streams
Publisher, but unfortunately, the implementation of this bridge has concurrency flaws that could be easily fixed by a PR, if there was an open repository for the library (because you'd want to post unit tests as well).
