Why LiveDataBus Beats EventBus and RxBus for Android Message Passing

Background

Message passing is a fundamental component of Android. Different pages and components within an app communicate via messages, whether between the four core Android components or between background threads and the UI thread. Over time, developers have used Handlers, BroadcastReceiver, callbacks, and more recently EventBus, RxBus, and LiveData‑based solutions.

EventBus Overview

EventBus is an Android publish/subscribe framework that decouples publishers from subscribers. It can replace Intents, Handlers, BroadcastReceiver, or callbacks for data exchange among Activities, Fragments, Services, and threads. Its main advantages are simplicity and decoupling.

Publish/Subscribe Pattern

The publish/subscribe (pub/sub) model defines a one‑to‑many relationship: a subject notifies all subscribed observers when its state changes.

RxBus Emergence and Principle

RxBus is not a library but a 30‑line file that leverages RxJava’s reactive programming model. RxJava’s Subject (specifically PublishSubject) acts as both observer and observable, enabling simple event emission and subscription.

RxBus Implementations

Example implementations (RxJava 1 and RxJava 2) are shown below.

public final class RxBus {
    private final Subject<Object, Object> bus;
    private RxBus() {
        bus = new SerializedSubject<>(PublishSubject.create());
    }
    public static RxBus getInstance() {
        return SingletonHolder.defaultRxBus;
    }
    public void post(Object o) {
        bus.onNext(o);
    }
    public boolean hasObservable() {
        return bus.hasObservers();
    }
    public <T> Observable<T> toObservable(Class<T> type) {
        return bus.ofType(type);
    }
}

public final class RxBus2 {
    private final Subject<Object> bus;
    private RxBus2() {
        bus = PublishSubject.create().toSerialized();
    }
    public static RxBus2 getInstance() {
        return Holder.BUS;
    }
    public void post(Object obj) {
        bus.onNext(obj);
    }
    public <T> Observable<T> toObservable(Class<T> tClass) {
        return bus.ofType(tClass);
    }
    public Observable<Object> toObservable() {
        return bus;
    }
}

LiveData Introduction and Advantages

LiveData, part of Android Architecture Components, is a lifecycle‑aware observable data holder. It only notifies active observers (STARTED or RESUMED), automatically clears observers when the owner is destroyed, and prevents crashes caused by updates to stopped Activities.

UI stays in sync with real‑time data.

Avoids memory leaks by binding observers to lifecycle.

No crashes when the Activity is in the background.

Eliminates manual lifecycle handling.

Provides instant data refresh on lifecycle changes.

Handles configuration changes gracefully.

Android Architecture Components Overview

The core components are Lifecycle, LiveData, ViewModel, and Room, which together enable clean data‑UI interaction, persistence, and automatic lifecycle management.

Motivation for LiveDataBus

LiveData’s lifecycle awareness makes it an ideal foundation for a communication bus. Compared with EventBus and RxBus, LiveDataBus:

Requires only a single class, reducing APK size.

Depends solely on official Android libraries, improving support.

Automatically handles observer registration/unregistration, eliminating memory‑leak risks.

LiveDataBus Design and Architecture

Key components:

Message : Any Object (e.g., Boolean, String, custom types).

Channel : A LiveData instance identified by a String key.

Bus : A singleton holding a Map<String, MutableLiveData<Object>> of channels.

Subscribe : Retrieve a channel via getChannel(key) and call observe.

Publish : Call setValue or postValue on the channel.

First Implementation (20 lines)

public final class LiveDataBus {
    private final Map<String, MutableLiveData<Object>> bus;
    private LiveDataBus() {
        bus = new HashMap<>();
    }
    public static LiveDataBus get() {
        return SingletonHolder.DEFAULT_BUS;
    }
    public <T> MutableLiveData<T> with(String key, Class<T> type) {
        if (!bus.containsKey(key)) {
            bus.put(key, new BusMutableLiveData<>());
        }
        return (MutableLiveData<T>) bus.get(key);
    }
    public MutableLiveData<Object> with(String key) {
        return with(key, Object.class);
    }
    // ... inner classes omitted for brevity
}

Problem with the First Version

When a new observer registers, it immediately receives the most recent value even if that value was posted before the subscription. This violates typical bus semantics where only events emitted after subscription should be delivered.

Root Cause Analysis

LiveData maintains an internal mVersion (initially –1) that increments on each setValue/postValue. Each observer wrapper also has a mLastVersion (initially –1). When a new observer is added, LiveData compares mVersion with mLastVersion; if mVersion > mLastVersion, it dispatches the current value, causing the pre‑subscription delivery.

Solution Strategy

Synchronize the observer’s mLastVersion with LiveData’s mVersion** at registration time. For lifecycle‑aware observe, obtain the internal LifecycleBoundObserver via reflection and set its version. For observeForever, wrap the original observer in a custom ObserverWrapper that filters callbacks originating from observeForever.

Final LiveDataBus Implementation

public final class LiveDataBus {
    private final Map<String, BusMutableLiveData<Object>> bus;
    private LiveDataBus() {
        bus = new HashMap<>();
    }
    public static LiveDataBus get() {
        return SingletonHolder.DEFAULT_BUS;
    }
    public <T> MutableLiveData<T> with(String key, Class<T> type) {
        if (!bus.containsKey(key)) {
            bus.put(key, new BusMutableLiveData<>());
        }
        return (MutableLiveData<T>) bus.get(key);
    }
    public MutableLiveData<Object> with(String key) {
        return with(key, Object.class);
    }
    private static class ObserverWrapper<T> implements Observer<T> {
        private final Observer<T> real;
        ObserverWrapper(Observer<T> real) { this.real = real; }
        @Override public void onChanged(@Nullable T t) {
            if (isCallOnObserve()) return; // ignore callbacks from observeForever
            if (real != null) real.onChanged(t);
        }
        private boolean isCallOnObserve() {
            for (StackTraceElement e : Thread.currentThread().getStackTrace()) {
                if ("android.arch.lifecycle.LiveData".equals(e.getClassName()) &&
                    "observeForever".equals(e.getMethodName())) {
                    return true;
                }
            }
            return false;
        }
    }
    private static class BusMutableLiveData<T> extends MutableLiveData<T> {
        private final Map<Observer, Observer> observerMap = new HashMap<>();
        @Override public void observe(@NonNull LifecycleOwner owner, @NonNull Observer<T> observer) {
            super.observe(owner, observer);
            try { hook(observer); } catch (Exception e) { e.printStackTrace(); }
        }
        @Override public void observeForever(@NonNull Observer<T> observer) {
            if (!observerMap.containsKey(observer)) {
                observerMap.put(observer, new ObserverWrapper<>(observer));
            }
            super.observeForever(observerMap.get(observer));
        }
        @Override public void removeObserver(@NonNull Observer<T> observer) {
            Observer real = observerMap.remove(observer);
            super.removeObserver(real != null ? real : observer);
        }
        private void hook(@NonNull Observer<T> observer) throws Exception {
            // Reflectively set wrapper version to LiveData version
            Class<LiveData> liveDataCls = LiveData.class;
            Field mObservers = liveDataCls.getDeclaredField("mObservers");
            mObservers.setAccessible(true);
            Object observersObj = mObservers.get(this);
            Class<?> observersCls = observersObj.getClass();
            Method get = observersCls.getDeclaredMethod("get", Object.class);
            get.setAccessible(true);
            Object entry = get.invoke(observersObj, observer);
            Object wrapper = ((Map.Entry) entry).getValue();
            Class<?> wrapperCls = wrapper.getClass().getSuperclass();
            Field mLastVersion = wrapperCls.getDeclaredField("mLastVersion");
            mLastVersion.setAccessible(true);
            Field mVersion = liveDataCls.getDeclaredField("mVersion");
            mVersion.setAccessible(true);
            Object version = mVersion.get(this);
            mLastVersion.set(wrapper, version);
        }
    }
    private static class SingletonHolder {
        private static final LiveDataBus DEFAULT_BUS = new LiveDataBus();
    }
}

Usage Example

// Register observer
LiveDataBus.get()
    .with("key_test", Boolean.class)
    .observe(this, new Observer<Boolean>() {
        @Override public void onChanged(@Nullable Boolean b) {
            // handle event
        }
    });

// Publish event
LiveDataBus.get().with("key_test").setValue(true);

Conclusion

LiveDataBus provides a lightweight, lifecycle‑aware communication bus for Android. It solves the pre‑subscription delivery issue of the naïve LiveData implementation, eliminates manual unregistration, reduces APK size, and leverages official Architecture Components for robust, leak‑free event handling.