Implement your own LiveData | androidinterview.com

Build an observable value holder that only delivers to observers whose screen is at least started, and that forgets an observer the moment its screen is destroyed. Almost everything else in the component falls out of those two sentences.

The part most candidates miss is that the interesting design is two integers. The holder counts its own writes, and every observer remembers the number it last saw. That one pair of counters is why a late observer gets the current value exactly once, why an observer whose screen was stopped catches up the moment it starts again, and why nobody is ever handed the same value twice.

What this really tests

Whether you know what the framework you use every day actually does, and whether you can build against the platform instead of inside it. The lifecycle and the main thread are the only two Android things this depends on, so a strong answer models both as small interfaces of its own and the whole design then compiles and runs on a plain JVM. If you cannot say what happens to a value that is set while the screen is stopped, you have used LiveData without understanding it.

What to clarify first

Ask these before you draw a box. Three of them change the code you write.

Which states count as active. Started and resumed. Say it out loud, because a candidate who says visible has not thought about the difference between a paused screen and a stopped one.
Does a new observer get the current value. Yes, and that replay is the most consequential decision in the whole component. It is exactly what makes it right for state and wrong for one shot events.
Can values be set from a background thread. If yes, you need a second setter with different rules, and you need to say what happens when three of them land before the main thread gets a turn.
Is null a legal value. If yes, then no value yet has to be a different thing from the value is null, which means a sentinel rather than a null check.
Does anything need to combine two sources. If yes, you need a mediator, and map and switchMap come free with it.
Is this a state holder or an event bus. The honest answer is state holder. Agreeing to that early saves you from the pitfall the interviewer is about to walk you into.

Two counters decide every delivery

The naive version keeps a list of observers and calls all of them on every write. It fails on the first real question. A screen registers an observer after the value was already set, so it shows nothing until the next write.

Fix that by replaying the current value to a new observer and you break something else. A screen that goes to the background and comes back would be handed the value it is already showing.

Both problems are one problem. The holder does not know who has seen what. So give it a version that goes up on every write, and give every observer a last seen version that starts below it. A delivery happens only when the observer is behind, and the delivery moves that observer forward. Nothing else in the component needs to remember anything.

Now every case answers itself. A late observer starts behind, so it gets one delivery when it goes active. A stopped observer misses writes, is still behind when its screen starts again, and catches up with exactly one call carrying the newest value rather than a replay of everything it missed. An observer that is already current is skipped. And because the version is on the holder rather than on the value, a write that sets the same object twice still counts as two writes, which is the behaviour people expect.

setValue on the main thread, postValue from anywhere

There are two setters because there are two situations, and the difference is worth being exact about.

setValue must be called on the main thread and it delivers synchronously. By the time it returns, every active observer has run. It throws if you call it from anywhere else, and throwing is the right choice, because delivering to a view from a background thread is a crash you would rather have at the setter than three frames later.

postValue is callable from anywhere. It parks the value and posts one task to the main thread, and that task reads whatever the latest parked value is. So if three values are posted before the main thread gets a turn, one task runs and observers see the third value only. That is coalescing, and it is a feature for a progress percentage and a trap for anything where every value matters, for example a queue of chat messages.

Model the main thread yourself, as an interface with two methods. Am I on it, and please run this on it. Back it with a queue you drain by hand and the whole component is testable with no device, no Looper and no idling resource. The thread that calls drain is the main thread for as long as the drain runs, which is close enough to what a real Looper is.

The classes

Nine types, and each one exists because something would be wrong without it.

LifecycleState is the four states in order, destroyed, created, started, resumed. Ordering is the whole point, because active means at least started and that is one comparison. Destroyed sits at the bottom so a dead screen can never satisfy it.
Lifecycle holds the current state and a list of listeners. Two of its behaviours are load bearing. A new listener is told the current state immediately, which is how a late observer gets its value with no special case in the holder, and moving to destroyed drops every listener.
LifecycleOwner is an activity or a fragment reduced to the one thing we need from it. Taking the owner rather than the activity is what makes this testable.
MainThread is the platform seam. Two methods, and the holder never imports Handler.
Observer is one callback taking one value. It never learns whether it was called because a value arrived or because its screen came back, and that ignorance is the point.
LiveData is the read only holder. It owns the value, the version, the observer map and the active count. Its setters are protected, which is the type discipline the whole component is for.
ObserverWrapper is what the holder actually stores per observer, and it carries the last seen version and the active flag. The observer stays a plain callback and knows none of this. Two subclasses, one bound to a lifecycle and one always active, which is the entire difference between observe and observeForever.
MutableLiveData exists only to widen two methods to public. A view model keeps one of these private and exposes it typed as LiveData, so the UI can watch a value and has no way to set one.
MediatorLiveData is a holder that watches other holders. It subscribes to its sources only while somebody is watching it, and map and switchMap are both a mediator with one source and a rule.

There is no Transformations class in the design as such. In Java the two operators need somewhere to live so they become static methods, and in Kotlin they are extension functions on the holder. The mechanism is identical either way.

The wrapper is the class that makes the component work, because it is the only thing that knows which version an observer has seen and whether its screen is currently allowed to hear anything.

One value, walked through

A view model posts a result from a background thread while the screen is in the background, and the user then comes back to the app. This is the walk to do at a whiteboard, because every mechanism shows up in it.

The background thread calls postValue. Under a small lock the holder sees no value parked, parks this one, and posts a single task to the main thread. A second and third post park their values and post nothing, because a task is already on its way.
The main thread runs the task. It takes the parked value, clears the slot under the same lock, and calls setValue with it. Only the last of the three values ever existed as far as the observers are concerned.
setValue checks it is on the main thread, raises the version by one, stores the value, and starts a dispatch.
The dispatch walks the wrappers. Our observer's screen is stopped, so its active flag is false and it is skipped. Nothing is queued for it and nothing is remembered about it, because its last seen version already says everything.
The user returns. The lifecycle moves to started and tells every listener, including our wrapper.
The wrapper asks whether it should be active, which is one comparison against started. It flips its flag, the holder's active count goes from zero to one, and onActive fires. A holder that owns a location listener starts the listener right here.
Flipping active immediately attempts a delivery for that one wrapper. Its last seen version is behind, so the observer is called once with the current value and its version is moved forward. Had the screen come back without anything having changed, that same check would have found it current and delivered nothing.
Later the screen is destroyed. The wrapper hears it, removes itself from the holder, and unregisters from the lifecycle. The active count drops to zero, onInactive fires, and the location listener stops. Nobody had to write a single line of cleanup in the fragment.

Patterns actually used

Observer, obviously, but with a lifetime attached. The interesting half is not the callback list, it is that registration is scoped to something that ends. Say that out loud, because it is the difference between this and a listener list anyone can write.
Template method, for onActive and onInactive. Two empty protected hooks that subclasses fill in. This is what turns the holder into a resource owner. A holder wrapping GPS starts the listener when the first observer becomes active and stops it when the last one goes away, so battery is spent only while somebody is looking. It is the cleanest example on the whole component of a hook earning its keep.
A decorator, in the wrapper. The observer the caller passes in stays a dumb function. Everything about lifetime, activity and versions lives in an object wrapped around it, which is why the same holder can serve a lifecycle bound observer and a forever observer with no branching in the dispatch.

Two things to leave out, and saying so is worth marks.

No thread safe observer list. Registration and delivery are main thread only, and that is enforced rather than assumed. A concurrent collection here would be the wrong answer expensively, because it would hide the actual rule instead of stating it. The only lock in the design guards the one field that genuinely crosses threads, the value parked by postValue.
No generic event bus. It is tempting to let this class carry one shot signals as well as state. It cannot, for the reason in the next section but one, and a design that tries ends up with a flag bolted to the side of it.

The implementation

Both trees are the same design. The holder counts versions, the wrapper decides who is listening, and the lifecycle and the main thread are interfaces we wrote ourselves so the code runs anywhere.

The Kotlin is genuinely different where it should be. The observer is a function type rather than an interface, so a lambda, a method reference and another holder's callback all fit without a wrapper. The wrappers are inner classes with an overridable owner property instead of an instance check. map and switchMap are extension functions on the holder rather than statics on a helper class, so they read as methods and no such class exists. The lifecycle observer is a fun interface, and states are an enum whose ordering does the active check.

Java

com.androidinterview.livedata.core.LiveData.java

package com.androidinterview.livedata.core;

import java.util.ArrayList;
import java.util.LinkedHashMap;
import java.util.Map;

import com.androidinterview.livedata.lifecycle.LifecycleObserver;
import com.androidinterview.livedata.lifecycle.LifecycleOwner;
import com.androidinterview.livedata.lifecycle.LifecycleState;
import com.androidinterview.livedata.thread.MainThread;

// The observable holder. Read only on purpose, because the UI is allowed to
// watch a value and not to set one.
//
// Two counters carry the design. The holder keeps a version that goes up on
// every write, and every observer remembers the version it last saw. A
// delivery happens only when the observer is behind, so a late observer gets
// the current value exactly once, a stopped observer catches up when it starts
// again, and nobody ever sees the same value twice.
public class LiveData<T> {

    static final int START_VERSION = -1;

    // A sentinel, because null is a legal value, so "nothing yet" has to be a
    // different thing from "the value is null".
    private static final Object NOT_SET = new Object();

    private final MainThread mainThread;
    private final Map<Observer<? super T>, ObserverWrapper> observers = new LinkedHashMap<>();
    private final Object postLock = new Object();

    private Object data;
    private Object pending = NOT_SET;
    private int version;
    private int activeCount;
    private boolean dispatching;
    private boolean dispatchInvalidated;

    protected LiveData(MainThread mainThread) {
        this.mainThread = mainThread;
        this.data = NOT_SET;
        this.version = START_VERSION;
    }

    protected LiveData(MainThread mainThread, T initial) {
        this.mainThread = mainThread;
        this.data = initial;
        this.version = START_VERSION + 1;
    }

    @SuppressWarnings("unchecked")
    public T getValue() {
        return data == NOT_SET ? null : (T) data;
    }

    public boolean hasObservers() {
        return !observers.isEmpty();
    }

    public boolean hasActiveObservers() {
        return activeCount > 0;
    }

    // Bind an observer to a screen. The wrapper it registers hears DESTROYED
    // and removes itself, so nothing survives the screen. That is the whole
    // reason this exists instead of a plain listener list.
    public void observe(LifecycleOwner owner, Observer<? super T> observer) {
        assertMainThread("observe");
        if (owner.getLifecycle().currentState() == LifecycleState.DESTROYED) {
            return;
        }
        LifecycleBoundObserver wrapper = new LifecycleBoundObserver(owner, observer);
        ObserverWrapper existing = observers.putIfAbsent(observer, wrapper);
        if (existing != null) {
            if (existing.owner() != owner) {
                throw new IllegalArgumentException("that observer is already bound to another owner");
            }
            return;
        }
        owner.getLifecycle().addObserver(wrapper);
    }

    // No owner, so always active, which is what one holder feeding another
    // needs. The caveat is the important part. Nothing will ever remove this
    // observer for you, so the caller owns a matching removeObserver, and
    // forgetting it leaks the observer and everything it captured.
    public void observeForever(Observer<? super T> observer) {
        assertMainThread("observeForever");
        AlwaysActiveObserver wrapper = new AlwaysActiveObserver(observer);
        ObserverWrapper existing = observers.putIfAbsent(observer, wrapper);
        if (existing != null) {
            if (existing.owner() != null) {
                throw new IllegalArgumentException("that observer is already bound to a lifecycle");
            }
            return;
        }
        wrapper.activeStateChanged(true);
    }

    public void removeObserver(Observer<? super T> observer) {
        assertMainThread("removeObserver");
        ObserverWrapper wrapper = observers.remove(observer);
        if (wrapper == null) {
            return;
        }
        wrapper.detach();
        wrapper.activeStateChanged(false);
    }

    // Main thread only, and it fails loudly. Delivery is synchronous, so by the
    // time this returns every active observer has run.
    protected void setValue(T value) {
        assertMainThread("setValue");
        version++;
        data = value;
        dispatchValue(null);
    }

    // Callable from anywhere, and it coalesces. Post three values before the
    // main thread gets a turn and the queue carries one task that delivers the
    // third. A feature for progress, a trap for anything where every value
    // matters, and the first thing to say about it out loud.
    protected void postValue(T value) {
        boolean shouldPost;
        synchronized (postLock) {
            shouldPost = pending == NOT_SET;
            pending = value;
        }
        if (!shouldPost) {
            return;
        }
        mainThread.post(this::drainPending);
    }

    @SuppressWarnings("unchecked")
    private void drainPending() {
        Object next;
        synchronized (postLock) {
            next = pending;
            pending = NOT_SET;
        }
        setValue((T) next);
    }

    // Called on the first active observer and on losing the last one. A holder
    // that owns something expensive, a location listener or a socket, starts it
    // here and stops it there, so the cost exists only while somebody looks.
    protected void onActive() {
    }

    protected void onInactive() {
    }

    MainThread mainThread() {
        return mainThread;
    }

    int version() {
        return version;
    }

    private void assertMainThread(String operation) {
        if (!mainThread.isMainThread()) {
            throw new IllegalStateException(operation + " must be called from the main thread");
        }
    }

    private void changeActiveCount(int delta) {
        int previous = activeCount;
        activeCount += delta;
        if (previous == 0 && activeCount > 0) {
            onActive();
        }
        if (previous > 0 && activeCount == 0) {
            onInactive();
        }
    }

    // Delivery, with a re-entrancy guard. An observer may set a new value from
    // inside onChanged, and without the guard the two dispatches interleave and
    // observers see values out of order. The flag says a newer value arrived,
    // so abandon this pass and start again.
    private void dispatchValue(ObserverWrapper initiator) {
        if (dispatching) {
            dispatchInvalidated = true;
            return;
        }
        dispatching = true;
        try {
            do {
                dispatchInvalidated = false;
                if (initiator != null) {
                    considerNotify(initiator);
                    initiator = null;
                } else {
                    for (ObserverWrapper wrapper : new ArrayList<>(observers.values())) {
                        considerNotify(wrapper);
                        if (dispatchInvalidated) {
                            break;
                        }
                    }
                }
            } while (dispatchInvalidated);
        } finally {
            dispatching = false;
        }
    }

    // The three questions that decide whether a value reaches an observer, and
    // the order matters. Is it active, is it still allowed to be active, and is
    // it behind. Only the third one moves the version forward.
    @SuppressWarnings("unchecked")
    private void considerNotify(ObserverWrapper wrapper) {
        if (!wrapper.active) {
            return;
        }
        if (!wrapper.shouldBeActive()) {
            wrapper.activeStateChanged(false);
            return;
        }
        if (wrapper.lastVersion >= version) {
            return;
        }
        wrapper.lastVersion = version;
        wrapper.observer.onChanged((T) data);
    }

    // What the holder stores per observer. The observer itself is a plain
    // callback and knows none of this.
    private abstract class ObserverWrapper {

        final Observer<? super T> observer;
        int lastVersion = START_VERSION;
        boolean active;

        ObserverWrapper(Observer<? super T> observer) {
            this.observer = observer;
        }

        abstract boolean shouldBeActive();

        LifecycleOwner owner() {
            return null;
        }

        void detach() {
        }

        // The one place active flips, so the counter and the onActive hook
        // cannot drift apart. Going active tries a delivery straight away,
        // which is how a stopped screen catches up the moment it starts.
        void activeStateChanged(boolean newActive) {
            if (newActive == active) {
                return;
            }
            active = newActive;
            changeActiveCount(active ? 1 : -1);
            if (active) {
                dispatchValue(this);
            }
        }
    }

    private final class AlwaysActiveObserver extends ObserverWrapper {

        AlwaysActiveObserver(Observer<? super T> observer) {
            super(observer);
        }

        @Override
        boolean shouldBeActive() {
            return true;
        }
    }

    // The lifecycle aware half, and it is this small. Active means at least
    // STARTED, DESTROYED means remove yourself, and every state change asks
    // the question again.
    private final class LifecycleBoundObserver extends ObserverWrapper implements LifecycleObserver {

        private final LifecycleOwner owner;

        LifecycleBoundObserver(LifecycleOwner owner, Observer<? super T> observer) {
            super(observer);
            this.owner = owner;
        }

        @Override
        boolean shouldBeActive() {
            return owner.getLifecycle().currentState().isAtLeast(LifecycleState.STARTED);
        }

        @Override
        LifecycleOwner owner() {
            return owner;
        }

        @Override
        void detach() {
            owner.getLifecycle().removeObserver(this);
        }

        @Override
        public void onStateChanged(LifecycleState state) {
            if (state == LifecycleState.DESTROYED) {
                removeObserver(observer);
                return;
            }
            activeStateChanged(shouldBeActive());
        }
    }
}

com.androidinterview.livedata.core.MediatorLiveData.java

package com.androidinterview.livedata.core;

import java.util.ArrayList;
import java.util.LinkedHashMap;
import java.util.Map;

import com.androidinterview.livedata.thread.MainThread;

// A holder that watches other holders. It is the answer to combining two
// sources, and it is also what map and switchMap are built out of.
//
// The trick is in onActive and onInactive. A mediator subscribes to its sources
// only while somebody is watching it, so a chain of five transformations over a
// database query costs nothing when the screen is in the background. Subscribe
// eagerly in addSource instead and you have built a component that keeps every
// upstream source hot forever.
public class MediatorLiveData<T> extends MutableLiveData<T> {

    private final Map<LiveData<?>, Source<?>> sources = new LinkedHashMap<>();

    public MediatorLiveData(MainThread mainThread) {
        super(mainThread);
    }

    public <S> void addSource(LiveData<S> source, Observer<? super S> onChanged) {
        Source<S> plug = new Source<>(source, onChanged);
        Source<?> existing = sources.putIfAbsent(source, plug);
        if (existing != null) {
            if (existing.onChanged != onChanged) {
                throw new IllegalArgumentException("that source is already added with a different observer");
            }
            return;
        }
        if (hasActiveObservers()) {
            plug.plug();
        }
    }

    public void removeSource(LiveData<?> source) {
        Source<?> plug = sources.remove(source);
        if (plug != null) {
            plug.unplug();
        }
    }

    @Override
    protected void onActive() {
        for (Source<?> source : new ArrayList<>(sources.values())) {
            source.plug();
        }
    }

    @Override
    protected void onInactive() {
        for (Source<?> source : new ArrayList<>(sources.values())) {
            source.unplug();
        }
    }

    // One upstream subscription. It keeps a version of its own, because
    // unplugging and plugging back in creates a fresh wrapper on the source
    // whose last seen version starts at nothing. Without this the screen would
    // be handed the same value again every time it came back to the foreground.
    private static final class Source<S> implements Observer<S> {

        private final LiveData<S> liveData;
        final Observer<? super S> onChanged;
        private int lastVersion = LiveData.START_VERSION;

        Source(LiveData<S> liveData, Observer<? super S> onChanged) {
            this.liveData = liveData;
            this.onChanged = onChanged;
        }

        void plug() {
            liveData.observeForever(this);
        }

        void unplug() {
            liveData.removeObserver(this);
        }

        @Override
        public void onChanged(S value) {
            if (lastVersion == liveData.version()) {
                return;
            }
            lastVersion = liveData.version();
            onChanged.onChanged(value);
        }
    }
}

com.androidinterview.livedata.core.MutableLiveData.java

package com.androidinterview.livedata.core;

import com.androidinterview.livedata.thread.MainThread;

// The writable half, and it exists only to widen two methods from protected to
// public. A view model keeps one of these private and exposes it as a LiveData,
// so the UI can watch the value and has no way to set one. Half the value of
// the whole component is this one line of type discipline.
public class MutableLiveData<T> extends LiveData<T> {

    public MutableLiveData(MainThread mainThread) {
        super(mainThread);
    }

    public MutableLiveData(MainThread mainThread, T initial) {
        super(mainThread, initial);
    }

    @Override
    public void setValue(T value) {
        super.setValue(value);
    }

    @Override
    public void postValue(T value) {
        super.postValue(value);
    }
}

com.androidinterview.livedata.core.Observer.java

package com.androidinterview.livedata.core;

// One callback, one value. The observer never learns whether it was called
// because a value arrived or because its screen came back to the foreground,
// and that ignorance is the whole point of the design.
@FunctionalInterface
public interface Observer<T> {

    void onChanged(T value);
}

com.androidinterview.livedata.core.Transformations.java

package com.androidinterview.livedata.core;

import java.util.function.Function;

// Both operators are a mediator with one source and a rule. Neither needs a
// single new field on the holder, which is the point worth making in an
// interview. Get the mediator right and the operators are four lines each.
public final class Transformations {

    private Transformations() {
    }

    // Same stream, different shape.
    public static <X, Y> LiveData<Y> map(LiveData<X> source, Function<X, Y> mapper) {
        MediatorLiveData<Y> result = new MediatorLiveData<>(source.mainThread());
        result.addSource(source, value -> result.setValue(mapper.apply(value)));
        return result;
    }

    // A new stream per value, and the old one has to go. The removeSource is
    // the whole operator. Skip it and a user who types five characters into a
    // search box ends up with five live queries all writing into one result,
    // and the answer the screen shows is whichever one finished last.
    public static <X, Y> LiveData<Y> switchMap(LiveData<X> source, Function<X, LiveData<Y>> mapper) {
        MediatorLiveData<Y> result = new MediatorLiveData<>(source.mainThread());
        result.addSource(source, new Observer<X>() {

            private LiveData<Y> current;

            @Override
            public void onChanged(X value) {
                LiveData<Y> next = mapper.apply(value);
                if (current == next) {
                    return;
                }
                if (current != null) {
                    result.removeSource(current);
                }
                current = next;
                if (next != null) {
                    result.addSource(next, result::setValue);
                }
            }
        });
        return result;
    }
}

com.androidinterview.livedata.lifecycle.Lifecycle.java

package com.androidinterview.livedata.lifecycle;

import java.util.ArrayList;
import java.util.List;

// The minimum lifecycle that makes the design work. A current state, a list of
// listeners, and a way to move.
//
// Two behaviours here are load bearing and both are copied from the real one.
// A new observer is told the current state immediately, which is how a late
// observer gets the current value without any special case in the holder. And
// moving to DESTROYED drops every observer, so a screen that is gone cannot
// keep a holder alive through a callback.
public final class Lifecycle {

    private final List<LifecycleObserver> observers = new ArrayList<>();
    private LifecycleState state = LifecycleState.CREATED;

    public LifecycleState currentState() {
        return state;
    }

    public void addObserver(LifecycleObserver observer) {
        if (state == LifecycleState.DESTROYED) {
            return;
        }
        observers.add(observer);
        // The immediate callback. Without it, an observer that arrives after
        // the screen is already started would sit inactive until the next
        // state change, and the holder would need a catch up path of its own.
        observer.onStateChanged(state);
    }

    public void removeObserver(LifecycleObserver observer) {
        observers.remove(observer);
    }

    // Iterating a copy, because an observer that hears DESTROYED removes
    // itself while we are still walking the list.
    public void moveTo(LifecycleState next) {
        state = next;
        for (LifecycleObserver observer : new ArrayList<>(observers)) {
            observer.onStateChanged(next);
        }
        if (next == LifecycleState.DESTROYED) {
            observers.clear();
        }
    }
}

com.androidinterview.livedata.lifecycle.LifecycleObserver.java

package com.androidinterview.livedata.lifecycle;

// Something that wants to hear about state changes. One method, because a
// callback per event would mean a new method every time a state is added.
@FunctionalInterface
public interface LifecycleObserver {

    void onStateChanged(LifecycleState state);
}

com.androidinterview.livedata.lifecycle.LifecycleOwner.java

package com.androidinterview.livedata.lifecycle;

// An activity or a fragment, reduced to the only thing our holder needs from
// it. Taking the owner rather than the activity is what keeps this code
// testable, because a test can hand over a plain object with a lifecycle it
// drives by hand.
public interface LifecycleOwner {

    Lifecycle getLifecycle();
}

com.androidinterview.livedata.lifecycle.LifecycleState.java

package com.androidinterview.livedata.lifecycle;

// The four states we need, written in order from dead to fully visible. The
// order is the point, because "is this observer active" is the single question
// isAtLeast answers, and DESTROYED sitting at the bottom means a destroyed
// screen can never be at least STARTED.
//
// The real Lifecycle has INITIALIZED too. It buys nothing here, so it is left
// out rather than copied.
public enum LifecycleState {

    DESTROYED,
    CREATED,
    STARTED,
    RESUMED;

    public boolean isAtLeast(LifecycleState other) {
        return compareTo(other) >= 0;
    }
}

com.androidinterview.livedata.sample.Location.java

package com.androidinterview.livedata.sample;

public record Location(double latitude, double longitude) {}

com.androidinterview.livedata.sample.LocationLiveData.java

package com.androidinterview.livedata.sample;

import java.util.function.Consumer;

import com.androidinterview.livedata.core.LiveData;
import com.androidinterview.livedata.thread.MainThread;

// Why onActive and onInactive are worth having. A location listener costs
// battery, so it runs while a screen is watching and stops when the last one
// goes away. Rotation is the case that makes this subtle. The old activity is
// destroyed and the new one observes moments later, so a naive design would
// stop the GPS and start it again for no reason. Real code softens that with a
// short delay before releasing, and it is a good thing to mention.
public final class LocationLiveData extends LiveData<Location> {

    private final LocationProvider provider;
    private final Consumer<Location> listener;

    public LocationLiveData(MainThread mainThread, LocationProvider provider) {
        super(mainThread);
        this.provider = provider;
        // postValue, not setValue, because a location callback arrives on
        // whichever thread the provider felt like using.
        this.listener = this::postValue;
    }

    @Override
    protected void onActive() {
        provider.start(listener);
    }

    @Override
    protected void onInactive() {
        provider.stop(listener);
    }
}

com.androidinterview.livedata.sample.LocationProvider.java

package com.androidinterview.livedata.sample;

import java.util.function.Consumer;

// Stands in for the platform location client. It calls back on a thread of its
// own choosing, which is exactly the case postValue exists for.
public interface LocationProvider {

    void start(Consumer<Location> listener);

    void stop(Consumer<Location> listener);
}

com.androidinterview.livedata.sample.SingleLiveEvent.java

package com.androidinterview.livedata.sample;

import java.util.concurrent.atomic.AtomicBoolean;

import com.androidinterview.livedata.core.MutableLiveData;
import com.androidinterview.livedata.core.Observer;
import com.androidinterview.livedata.lifecycle.LifecycleOwner;
import com.androidinterview.livedata.thread.MainThread;

// The famous workaround, written out so you can see why it is a workaround and
// not a fix. The holder replays its current value to whoever starts observing,
// which is right for a name on a screen and wrong for "show a toast", because
// rotation brings a new observer that is behind. The flag breaks the replay.
//
// Two honest flaws. Only one observer can win the flag, so a second observer on
// the same event silently gets nothing, and the wrapping means the caller can
// no longer remove the observer it registered. The bug is that a state holder
// was asked to carry events, which is why one shot signals moved to a channel.
public final class SingleLiveEvent<T> extends MutableLiveData<T> {

    private final AtomicBoolean pending = new AtomicBoolean(false);

    public SingleLiveEvent(MainThread mainThread) {
        super(mainThread);
    }

    @Override
    public void observe(LifecycleOwner owner, Observer<? super T> observer) {
        super.observe(owner, value -> {
            if (pending.compareAndSet(true, false)) {
                observer.onChanged(value);
            }
        });
    }

    @Override
    public void setValue(T value) {
        pending.set(true);
        super.setValue(value);
    }
}

com.androidinterview.livedata.thread.MainThread.java

package com.androidinterview.livedata.thread;

// The main thread, as an interface, so the holder never touches Looper or
// Handler and a test never needs a device.
//
// Two questions is all the holder asks. Am I on it, which guards setValue, and
// please run this on it, which is what postValue uses.
public interface MainThread {

    boolean isMainThread();

    void post(Runnable task);
}

com.androidinterview.livedata.thread.QueuedMainThread.java

package com.androidinterview.livedata.thread;

import java.util.ArrayDeque;
import java.util.Deque;

// A main thread you drain by hand, which is exactly what Looper does in a real
// app and exactly what a test wants.
//
// Work posted from any thread lands on the queue. Whichever thread calls drain
// becomes the main thread for as long as the drain runs, so isMainThread is a
// real answer rather than a hardcoded true.
public final class QueuedMainThread implements MainThread {

    private final Deque<Runnable> queue = new ArrayDeque<>();
    private volatile Thread draining;

    @Override
    public boolean isMainThread() {
        return Thread.currentThread() == draining;
    }

    @Override
    public void post(Runnable task) {
        synchronized (queue) {
            queue.addLast(task);
        }
    }

    // Run everything queued, on this thread, in the order it was posted. A
    // task that posts another task is picked up by the same loop, which is the
    // behaviour a real message queue has.
    public void drain() {
        Thread previous = draining;
        draining = Thread.currentThread();
        try {
            while (true) {
                Runnable task;
                synchronized (queue) {
                    task = queue.pollFirst();
                }
                if (task == null) {
                    return;
                }
                task.run();
            }
        } finally {
            draining = previous;
        }
    }

    // The convenience a caller wants. Run this block as if it were on the main
    // thread, then let everything it posted run too.
    public void runOnMain(Runnable block) {
        post(block);
        drain();
    }
}

Kotlin

com.androidinterview.livedata.core.LiveData.kt

package com.androidinterview.livedata.core

import com.androidinterview.livedata.lifecycle.LifecycleObserver
import com.androidinterview.livedata.lifecycle.LifecycleOwner
import com.androidinterview.livedata.lifecycle.LifecycleState
import com.androidinterview.livedata.thread.MainThread

// An observer is a function, not an interface. There is nothing an interface
// would add here, and a function type means a lambda, a method reference and
// another holder's callback all fit without a wrapper.
typealias Observer<T> = (T) -> Unit

// A sentinel, because null is a perfectly good value to hold, so "no value
// yet" has to be a different thing from "the value is null".
private val NOT_SET = Any()

internal const val START_VERSION = -1

// The observable holder. Read only on purpose, because the UI is allowed to
// watch a value and is not allowed to set one.
//
// Two counters carry the whole design. The holder keeps a version that goes up
// on every write, and every observer remembers the version it last saw. A
// delivery happens only when an observer is behind, which is why a late
// observer gets the current value exactly once, an observer whose screen was
// stopped catches up the moment it starts again, and neither of them ever sees
// the same value twice.
open class LiveData<T> protected constructor(internal val mainThread: MainThread) {

    private val observers = LinkedHashMap<Observer<T>, ObserverWrapper>()
    private val postLock = Any()

    private var data: Any? = NOT_SET
    private var pending: Any? = NOT_SET
    private var activeCount = 0
    private var dispatching = false
    private var invalidated = false

    internal var version = START_VERSION
        private set

    protected constructor(mainThread: MainThread, initial: T) : this(mainThread) {
        data = initial
        version = START_VERSION + 1
    }

    @Suppress("UNCHECKED_CAST")
    val value: T?
        get() = if (data === NOT_SET) null else data as T?

    val hasObservers: Boolean get() = observers.isNotEmpty()

    val hasActiveObservers: Boolean get() = activeCount > 0

    // Bind an observer to a screen. The holder never keeps that observer alive
    // past the screen, because the wrapper it registers hears DESTROYED and
    // removes itself. That one line is the reason this exists instead of a
    // plain listener list.
    open fun observe(owner: LifecycleOwner, observer: Observer<T>) {
        assertMainThread("observe")
        if (owner.lifecycle.currentState == LifecycleState.DESTROYED) return
        val wrapper = LifecycleBoundObserver(owner, observer)
        val existing = observers.putIfAbsent(observer, wrapper)
        if (existing != null) {
            require(existing.owner === owner) { "that observer is already bound to another owner" }
            return
        }
        owner.lifecycle.addObserver(wrapper)
    }

    // No owner, so always active. Useful when the watcher is not a screen, for
    // example one holder feeding another. The caveat is the important part.
    // Nothing will ever remove this observer for you, so the caller owns a
    // matching removeObserver, and forgetting it leaks the observer and
    // everything the lambda captured.
    fun observeForever(observer: Observer<T>) {
        assertMainThread("observeForever")
        val wrapper = AlwaysActiveObserver(observer)
        val existing = observers.putIfAbsent(observer, wrapper)
        if (existing != null) {
            require(existing.owner == null) { "that observer is already bound to a lifecycle" }
            return
        }
        wrapper.activeStateChanged(true)
    }

    fun removeObserver(observer: Observer<T>) {
        assertMainThread("removeObserver")
        val wrapper = observers.remove(observer) ?: return
        wrapper.detach()
        wrapper.activeStateChanged(false)
    }

    // Main thread only, and it fails loudly rather than quietly. Delivery is
    // synchronous, so by the time this returns every active observer has run.
    protected open fun setValue(value: T) {
        assertMainThread("setValue")
        version++
        data = value
        dispatchValue(null)
    }

    // Callable from anywhere, and it coalesces. Post three values before the
    // main thread gets a turn and the queue carries one task that delivers the
    // third. A feature for progress, a trap for anything where every value
    // matters, and the first thing to say about it out loud.
    protected open fun postValue(value: T) {
        val shouldPost = synchronized(postLock) {
            val first = pending === NOT_SET
            pending = value
            first
        }
        if (!shouldPost) return
        mainThread.post {
            @Suppress("UNCHECKED_CAST")
            val next = synchronized(postLock) { pending.also { pending = NOT_SET } } as T
            setValue(next)
        }
    }

    // Called when the holder gains its first active observer and when it loses
    // its last. A holder that owns something expensive, a location listener or
    // a socket, starts it here and stops it there, so the cost exists only
    // while somebody is looking.
    protected open fun onActive() = Unit

    protected open fun onInactive() = Unit

    private fun assertMainThread(operation: String) =
        check(mainThread.isMainThread) { "$operation must be called from the main thread" }

    private fun changeActiveCount(delta: Int) {
        val previous = activeCount
        activeCount += delta
        if (previous == 0 && activeCount > 0) onActive()
        if (previous > 0 && activeCount == 0) onInactive()
    }

    // Delivery, with a re-entrancy guard. An observer may set a new value from
    // inside its own callback, and without the guard the two dispatches
    // interleave and observers see values out of order. The flag says a newer
    // value arrived, so abandon this pass and start again.
    private fun dispatchValue(initiator: ObserverWrapper?) {
        if (dispatching) {
            invalidated = true
            return
        }
        dispatching = true
        var target = initiator
        try {
            do {
                invalidated = false
                if (target != null) {
                    considerNotify(target)
                    target = null
                } else {
                    for (wrapper in observers.values.toList()) {
                        considerNotify(wrapper)
                        if (invalidated) break
                    }
                }
            } while (invalidated)
        } finally {
            dispatching = false
        }
    }

    // The three questions that decide whether a value reaches an observer, and
    // the order matters. Is it active, is it still allowed to be active, and is
    // it behind. Only the third moves its version forward.
    @Suppress("UNCHECKED_CAST")
    private fun considerNotify(wrapper: ObserverWrapper) {
        if (!wrapper.active) return
        if (!wrapper.shouldBeActive()) {
            wrapper.activeStateChanged(false)
            return
        }
        if (wrapper.lastVersion >= version) return
        wrapper.lastVersion = version
        wrapper.observer(data as T)
    }

    // What the holder stores per observer. The observer itself is a plain
    // function and knows none of this.
    private abstract inner class ObserverWrapper(val observer: Observer<T>) {

        var lastVersion = START_VERSION
        var active = false

        open val owner: LifecycleOwner? get() = null

        abstract fun shouldBeActive(): Boolean

        open fun detach() = Unit

        // The one place active flips, so the counter and the onActive hook can
        // never drift apart. Going active tries a delivery straight away, which
        // is how a stopped screen catches up the moment it starts.
        fun activeStateChanged(newActive: Boolean) {
            if (newActive == active) return
            active = newActive
            changeActiveCount(if (active) 1 else -1)
            if (active) dispatchValue(this)
        }
    }

    private inner class AlwaysActiveObserver(observer: Observer<T>) : ObserverWrapper(observer) {
        override fun shouldBeActive() = true
    }

    // The lifecycle aware half of the component, and it is this small. Active
    // means at least STARTED, DESTROYED means remove yourself, and every state
    // change asks the question again.
    private inner class LifecycleBoundObserver(
        override val owner: LifecycleOwner,
        observer: Observer<T>,
    ) : ObserverWrapper(observer), LifecycleObserver {

        override fun shouldBeActive() =
            owner.lifecycle.currentState.isAtLeast(LifecycleState.STARTED)

        override fun detach() = owner.lifecycle.removeObserver(this)

        override fun onStateChanged(state: LifecycleState) {
            if (state == LifecycleState.DESTROYED) removeObserver(observer)
            else activeStateChanged(shouldBeActive())
        }
    }
}

// The writable half, and it exists only to widen two methods. A view model
// keeps one of these private and exposes it as a LiveData, so the UI can watch
// the value and has no way to set one. Half the value of the component is that
// single line of type discipline.
open class MutableLiveData<T> : LiveData<T> {

    constructor(mainThread: MainThread) : super(mainThread)

    constructor(mainThread: MainThread, initial: T) : super(mainThread, initial)

    public override fun setValue(value: T) = super.setValue(value)

    public override fun postValue(value: T) = super.postValue(value)
}

com.androidinterview.livedata.core.MediatorLiveData.kt

package com.androidinterview.livedata.core

import com.androidinterview.livedata.thread.MainThread

// A holder that watches other holders. It answers "combine these two sources",
// and it is also what map and switchMap are built out of.
//
// The trick is in onActive and onInactive. A mediator subscribes to its sources
// only while somebody is watching it, so a chain of five transformations over a
// database query costs nothing while the screen is in the background. Subscribe
// eagerly in addSource instead and you have built a component that keeps every
// upstream source hot forever.
open class MediatorLiveData<T>(mainThread: MainThread) : MutableLiveData<T>(mainThread) {

    private val sources = LinkedHashMap<LiveData<*>, Source<*>>()

    fun <S> addSource(source: LiveData<S>, onChanged: Observer<S>) {
        val plug = Source(source, onChanged)
        val existing = sources.putIfAbsent(source, plug)
        if (existing != null) {
            require(existing.onChanged === onChanged) {
                "that source is already added with a different observer"
            }
            return
        }
        if (hasActiveObservers) plug.plug()
    }

    fun removeSource(source: LiveData<*>) {
        sources.remove(source)?.unplug()
    }

    override fun onActive() = sources.values.toList().forEach { it.plug() }

    override fun onInactive() = sources.values.toList().forEach { it.unplug() }

    // One upstream subscription. It keeps a version of its own, because
    // unplugging and plugging back in creates a fresh wrapper on the source
    // whose last seen version starts at nothing. Without this, the screen would
    // be handed the same value again every time it returned to the foreground.
    private class Source<S>(private val liveData: LiveData<S>, val onChanged: Observer<S>) {

        private var lastVersion = START_VERSION

        private val relay: Observer<S> = { value ->
            if (lastVersion != liveData.version) {
                lastVersion = liveData.version
                onChanged(value)
            }
        }

        fun plug() = liveData.observeForever(relay)

        fun unplug() = liveData.removeObserver(relay)
    }
}

// Both operators are a mediator with one source and a rule, and neither needs
// a single new field on the holder. That is the point worth making out loud.
// Get the mediator right and the operators are three lines each.
//
// Java gathers these as static methods on a Transformations class because it
// has nowhere else to put them. Kotlin has extension functions, so map reads
// as a method on the holder and no helper class exists at all.
fun <X, Y> LiveData<X>.map(transform: (X) -> Y): LiveData<Y> =
    MediatorLiveData<Y>(mainThread).apply {
        addSource(this@map) { setValue(transform(it)) }
    }

// A new stream per value, and the old one has to go. The removeSource is the
// whole operator. Skip it and a user who types five characters into a search
// box ends up with five live queries writing into one result, and the screen
// shows whichever one finished last.
fun <X, Y> LiveData<X>.switchMap(transform: (X) -> LiveData<Y>?): LiveData<Y> =
    MediatorLiveData<Y>(mainThread).apply {
        var current: LiveData<Y>? = null
        addSource(this@switchMap) { x ->
            val next = transform(x)
            if (next !== current) {
                current?.let { removeSource(it) }
                current = next
                next?.let { source -> addSource(source) { setValue(it) } }
            }
        }
    }

com.androidinterview.livedata.lifecycle.Lifecycle.kt

package com.androidinterview.livedata.lifecycle

// The four states we need, in order from dead to fully visible. The order is
// the point, because "is this observer active" is the one question isAtLeast
// answers, and DESTROYED at the bottom means a destroyed screen can never be
// at least STARTED.
//
// The real Lifecycle has INITIALIZED as well. It buys nothing here, so it is
// left out rather than copied.
enum class LifecycleState {
    DESTROYED,
    CREATED,
    STARTED,
    RESUMED;

    fun isAtLeast(other: LifecycleState) = ordinal >= other.ordinal
}

// One callback for every state change. A method per event would mean a new
// method every time a state is added.
fun interface LifecycleObserver {
    fun onStateChanged(state: LifecycleState)
}

// An activity or a fragment, reduced to the only thing the holder needs. A
// test hands over a plain object with a lifecycle it drives by hand.
interface LifecycleOwner {
    val lifecycle: Lifecycle
}

// The minimum lifecycle that makes the design work. A current state, a list of
// listeners, and a way to move between states.
//
// Two behaviours are load bearing and both come from the real one. A new
// observer is told the current state straight away, which is how a late
// observer gets the current value with no special case in the holder. And
// moving to DESTROYED drops every observer, so a screen that is gone cannot
// keep a holder alive through a callback.
class Lifecycle {

    private val observers = mutableListOf<LifecycleObserver>()

    var currentState: LifecycleState = LifecycleState.CREATED
        private set

    fun addObserver(observer: LifecycleObserver) {
        if (currentState == LifecycleState.DESTROYED) return
        observers += observer
        observer.onStateChanged(currentState)
    }

    fun removeObserver(observer: LifecycleObserver) {
        observers -= observer
    }

    // Walking a copy, because an observer that hears DESTROYED removes itself
    // while we are still going through the list.
    fun moveTo(next: LifecycleState) {
        currentState = next
        observers.toList().forEach { it.onStateChanged(next) }
        if (next == LifecycleState.DESTROYED) observers.clear()
    }
}

com.androidinterview.livedata.sample.LocationLiveData.kt

package com.androidinterview.livedata.sample

import com.androidinterview.livedata.core.LiveData
import com.androidinterview.livedata.thread.MainThread

data class Location(val latitude: Double, val longitude: Double)

// Stands in for the platform location client. It calls back on a thread of its
// own choosing, which is the exact case postValue exists for.
interface LocationProvider {
    fun start(listener: (Location) -> Unit)
    fun stop(listener: (Location) -> Unit)
}

// Why onActive and onInactive are worth having. A location listener costs
// battery, so it runs while a screen is watching and stops when the last one
// goes away. Rotation is the case that makes this subtle. The old activity is
// destroyed and the new one observes moments later, so a naive design stops the
// GPS and starts it again for nothing. Real code softens that with a short
// delay before releasing, and that is a good thing to mention.
class LocationLiveData(
    mainThread: MainThread,
    private val provider: LocationProvider,
) : LiveData<Location>(mainThread) {

    // postValue, not setValue, because the callback arrives on whichever
    // thread the provider felt like using.
    private val listener: (Location) -> Unit = { postValue(it) }

    override fun onActive() = provider.start(listener)

    override fun onInactive() = provider.stop(listener)
}

com.androidinterview.livedata.sample.SingleLiveEvent.kt

package com.androidinterview.livedata.sample

import com.androidinterview.livedata.core.MutableLiveData
import com.androidinterview.livedata.core.Observer
import com.androidinterview.livedata.lifecycle.LifecycleOwner
import com.androidinterview.livedata.thread.MainThread
import java.util.concurrent.atomic.AtomicBoolean

// The famous workaround, written out so you can see why it is a workaround and
// not a fix. The holder replays its current value to whoever starts observing,
// which is right for a name on a screen and wrong for "show a toast", because
// rotation brings a new observer that is behind. The flag breaks the replay.
//
// Two honest flaws. Only one observer can win the flag, so a second observer on
// the same event silently gets nothing, and the wrapping means the caller can
// no longer remove the function it registered. The bug is that a state holder
// was asked to carry events, which is why one shot signals moved to a channel.
class SingleLiveEvent<T>(mainThread: MainThread) : MutableLiveData<T>(mainThread) {

    private val pending = AtomicBoolean(false)

    override fun observe(owner: LifecycleOwner, observer: Observer<T>) {
        super.observe(owner) { value ->
            if (pending.compareAndSet(true, false)) observer(value)
        }
    }

    override fun setValue(value: T) {
        pending.set(true)
        super.setValue(value)
    }
}

com.androidinterview.livedata.thread.MainThread.kt

package com.androidinterview.livedata.thread

// The main thread as an interface, so the holder never touches Looper or
// Handler and a test never needs a device. Two questions is all the holder
// asks. Am I on it, which guards setValue, and please run this, which is what
// postValue uses.
interface MainThread {
    val isMainThread: Boolean
    fun post(task: () -> Unit)
}

// A main thread you drain by hand, which is what Looper does in an app and
// what a test wants. Work posted from any thread lands on the queue, and
// whichever thread calls drain is the main thread for as long as that runs, so
// isMainThread is a real answer rather than a hardcoded true.
class QueuedMainThread : MainThread {

    private val queue = ArrayDeque<() -> Unit>()

    @Volatile
    private var draining: Thread? = null

    override val isMainThread: Boolean
        get() = Thread.currentThread() === draining

    override fun post(task: () -> Unit) {
        synchronized(queue) { queue.addLast(task) }
    }

    // Run everything queued, on this thread, in the order it was posted. A task
    // that posts another task is picked up by the same loop, which is how a
    // real message queue behaves.
    fun drain() {
        val previous = draining
        draining = Thread.currentThread()
        try {
            while (true) {
                val task = synchronized(queue) { queue.removeFirstOrNull() } ?: return
                task()
            }
        } finally {
            draining = previous
        }
    }

    // Run this block as if it were on the main thread, then let everything it
    // posted run as well.
    fun runOnMain(block: () -> Unit) {
        post(block)
        drain()
    }
}

The single event problem, and where this sits now

The replay that makes this component right is also the thing that breaks it. The holder is a state container, so it hands its current value to whoever starts observing. That is correct for a name on a screen. It is wrong for show a toast, because the user rotates the device, a new observer arrives behind the version, and the toast fires again.

SingleLiveEvent was the well known patch. A flag is raised on set and cleared by the first delivery, so the replay finds nothing to replay. It works and it has two honest flaws. Only one observer can win the flag, so a second observer on the same event silently gets nothing, and the wrapping means the caller can no longer remove the observer it registered. The other common shape, an event class with a get content if not handled method, has the same problem in different clothes.

Neither is a fix, because the bug is that a state holder was asked to carry events. That is what pushed one shot signals to a channel or a shared flow, where a consumer and a replay policy are things you state rather than things you patch around.

Where LiveData sits today is a fair question and it deserves a straight answer. StateFlow does the same job with better tools. It is a hot state holder with a replay of one, it composes with the rest of coroutines, it is not tied to Android so it works in shared code, and back pressure and cancellation are already solved. What it does not have is lifecycle awareness, which is why collection is wrapped in a repeat on lifecycle block on the UI side. LiveData is not deprecated and there are hundreds of thousands of screens using it happily. New Kotlin code reaches for StateFlow for state and a channel for events, and the reason to still be able to build this from memory is that it is the clearest small example of lifecycle scoped observation there is.

Concurrency and edge cases

The only genuinely shared field is the parked value. Observers are registered, removed and notified on the main thread and nowhere else, so the map and the counters need no protection. postValue is the one door open to other threads, so the park and the take happen under one small lock and nothing else does.

Re-entrant writes. An observer is allowed to set a new value from inside its own callback. Without a guard the two dispatches interleave and observers see values out of order. The dispatch keeps a flag saying a newer value arrived, abandons the current pass and starts again, so everyone lands on the newest value and nobody sees a stale one after a fresh one.

observeForever, and its caveat. No owner means always active, which is what one holder feeding another needs. Nothing will ever remove that observer for you. The caller owns a matching remove, and forgetting it leaks the observer and everything the lambda captured, which is usually the fragment. Never call it from a view.

Re-subscribing a mediator without duplicating. When a mediator goes inactive it unplugs from its sources, and plugging back in creates a fresh wrapper whose last seen version starts at nothing. So each source keeps a version of its own and skips a value it has already forwarded. Without that, every trip to the background and back would push the same value through the chain again.

Other cases worth a sentence each.

Observing an already destroyed owner. Returns silently rather than registering something that can never fire.
The same observer registered twice. Ignored the second time, and registering one observer against two different owners throws, because that is always a bug.
Null as a value. Legal, which is why no value yet is a sentinel object and not a null check.
Rotation and an expensive source. The old screen is destroyed before the new one observes, so a naive holder stops the GPS and starts it again a moment later. Real code delays the release slightly. Mention it, because it shows you have run this on a device.
switchMap and a fast typist. Five keystrokes mean five queries. The removeSource on the previous source is the entire operator, and without it the screen shows whichever query happened to finish last.