连接两个LiveData，一开始订阅第一个LiveData，在一定条件下，改为订阅第二个LiveData

ConcatLiveData

import android.arch.lifecycle.LiveData;
import android.arch.lifecycle.MediatorLiveData;
import android.arch.lifecycle.Observer;
import android.support.annotation.NonNull;
import android.support.annotation.Nullable;

/**
 *
 * 每次InputSource更新后。
 * <ol>
 *      <li>结果交给InputObserver处理</li>
 *      <li>回调getOutputSource</li>
 *      <li>当getOutputSource返回不为null时, 改为订阅它返回的LiveData</li>
 * </ol>
 */
public abstract class ConcatLiveData<I, O> {

    private final MediatorLiveData<O> mResult = new MediatorLiveData<>();

    public ConcatLiveData() {
        init();
    }

    private void init() {
        LiveData<I> inputSource = getInputSource();
        Observer<I> inputObserver = getInputObserver();

        mResult.addSource(inputSource, input -> {

            inputObserver.onChanged(input);

            LiveData<O> outputSource = getOutputSource();
            if (outputSource != null) {
                mResult.removeSource(inputSource);
                mResult.addSource(outputSource, mResult::setValue);
            }
        });
    }

    public LiveData<O> asLiveData() {
        return mResult;
    }

    @NonNull
    protected abstract LiveData<I> getInputSource();

    @NonNull
    protected abstract Observer<I> getInputObserver();

    @Nullable
    protected abstract LiveData<O> getOutputSource();

}

ConcatLiveDataTest

import android.arch.core.executor.testing.InstantTaskExecutorRule;
import android.arch.lifecycle.Lifecycle;
import android.arch.lifecycle.LifecycleOwner;
import android.arch.lifecycle.LifecycleRegistry;
import android.arch.lifecycle.LiveData;
import android.arch.lifecycle.MutableLiveData;
import android.arch.lifecycle.Observer;
import android.support.annotation.NonNull;
import android.support.annotation.Nullable;

import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.TestRule;

import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.only;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;

public class ConcatLiveDataTest {

    private static final float PRICE = 5.0f;

    @Rule
    public TestRule rule = new InstantTaskExecutorRule();


    private static LifecycleOwner mockLifecycleOwner() {
        LifecycleOwner owner = mock(LifecycleOwner.class);
        LifecycleRegistry lifecycle = new LifecycleRegistry(owner);
        lifecycle.handleLifecycleEvent(Lifecycle.Event.ON_RESUME);
        when(owner.getLifecycle()).thenReturn(lifecycle);
        return owner;
    }

    @SuppressWarnings("unchecked")
    private static <T> Observer<T> mockObserver() {
        return (Observer<T>) mock(Observer.class);
    }

    @Test
    public void onGetNextSource() {
        LifecycleOwner owner = mockLifecycleOwner();
        Observer<Float> settleObserver = mockObserver();
        Observer<String> billObserver = mockObserver();

        MutableLiveData<Integer> buy = new MutableLiveData<>();
        LiveData<Float> settleUntilBuyNumberLargerThen2 = new ConcatLiveData<Integer, Float>() {

            private int mBuyNumber = 0;

            @NonNull
            @Override
            protected LiveData<Integer> getInputSource() {
                return buy;
            }

            @NonNull
            @Override
            protected Observer<Integer> getInputObserver() {
                return buyNumber -> mBuyNumber = buyNumber == null ? 0 : buyNumber;
            }

            @Nullable
            @Override
            protected LiveData<Float> getOutputSource() {
                if (mBuyNumber > 2) {
                    return settle(mBuyNumber, PRICE);
                } else {
                    return null;
                }
            }
        }.asLiveData();
        LiveData<String> printBill = new ConcatLiveData<Float, String>() {

            private float mPrice;

            @NonNull
            @Override
            protected LiveData<Float> getInputSource() {
                return settleUntilBuyNumberLargerThen2;
            }

            @NonNull
            @Override
            protected Observer<Float> getInputObserver() {
                return price -> {
                    if (price != null) {
                        mPrice = price;
                    }
                };
            }

            @Override
            protected LiveData<String> getOutputSource() {
                return printBill(mPrice);
            }
        }.asLiveData();

        settleUntilBuyNumberLargerThen2.observe(owner, settleObserver);
        printBill.observe(owner, billObserver);

        buy.setValue(1);
        verify(settleObserver, never()).onChanged(any());
        verify(billObserver, never()).onChanged(any());
        buy.setValue(2);
        verify(settleObserver, never()).onChanged(any());
        verify(billObserver, never()).onChanged(any());
        buy.setValue(3);
        verify(settleObserver, only()).onChanged(3 * PRICE);
        verify(billObserver, only()).onChanged("$" + 3 * PRICE);
    }

    /**
     * 结账
     */
    private LiveData<Float> settle(int number, float price) {
        return new LiveData<Float>() {
            {
                setValue(number * price);
            }
        };
    }

    /**
     * 打印价格单
     */
    private LiveData<String> printBill(float price) {
        return new LiveData<String>() {
            {
                setValue("$" + price);
            }
        };
    }
}