RxValue.Java

package **; // <-- add own package name

import android.support.annotation.Nullable;

import rx.Observable;
import rx.subjects.PublishSubject;
import rx.subjects.Subject;

/** Generic implementation of value/property that triggers subscribers on value change. */
public class RxValue<T> {
    /** Observable and an Observer of the value. */
    private final Subject<T, T> subject = PublishSubject.create();
    /** Last known value. */
    private T lastValue;
    /** Is value in empty state or not. */
    private boolean isEmpty;

    private RxValue() {
        lastValue = null;
        isEmpty = true;
    }

    /** Hidden constructor. */
    private RxValue(final T initialValue) {
        lastValue = initialValue;
    }

    /** Static method for simplified instance creation. Less verbose syntax. */
    public static <T> RxValue<T> of(T initialValue) {
        return new RxValue<>(initialValue);
    }

    /** Create rx value that does is empty from the beginning. */
    public static <T> RxValue<T> empty() {
        return new RxValue<>();
    }

    /** Get the current value. */
    @Nullable
    public T value() {
        return lastValue;
    }

    /** Set the instance to a new value. */
    public T set(final T value) {
        if (isEmpty || value != lastValue) {
            isEmpty = false;
            lastValue = value;

            if (subject.hasObservers()) {
                subject.onNext(lastValue);
            }
        }

        return value();
    }

    /** Get instance of observable for triggering. */
    public Observable<T> asObservable() {
        if (isEmpty) {
            return subject;
        }

        return Observable.merge(Observable.just(lastValue), subject);
    }
}

Unit Tests

package **; // <-- add own package name

import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.JUnit4;
import org.mockito.ArgumentCaptor;
import org.mockito.Captor;
import org.mockito.Mock;
import org.mockito.Mockito;

import rx.functions.Action1;

import static org.assertj.core.api.Assertions.*;
import static org.mockito.Mockito.*;

/** Unit tests of the {@link RxValue} class. */
@RunWith(JUnit4.class)
public class RxValueTest extends AncestorTest /* <-- can be skipped */ {

    //region Mocks
    @Mock /* package */ Action1<Boolean> subscriber;
    @Captor /* package */ ArgumentCaptor<Boolean> catcher;
    //endregion

    //region Tests
    @Test
    public void testInitializeValues() throws Exception {
        final RxValue<Boolean> underTheTest_False = RxValue.of(false);
        assertThat(underTheTest_False.value()).isFalse();

        final RxValue<Boolean> underTheTest_True = RxValue.of(true);
        assertThat(underTheTest_True.value()).isTrue();
    }

    @Test
    public void testSubscriberCallWithCurrentValue() throws Exception {
        final RxValue<Boolean> underTheTest = RxValue.of(false);
        underTheTest.asObservable().subscribe(subscriber);

        // verify that after subscription we do not change the value
        assertThat(underTheTest.value()).isFalse();

        // on attaching we receive the current value
        verify(subscriber).call(anyBoolean());
    }

    @Test
    public void testOnValueChanged() throws Exception {
        final RxValue<Boolean> underTheTest = RxValue.of(false);
        underTheTest.asObservable().subscribe(subscriber);
        underTheTest.set(true);

        // verify number of calls, 1 - on subscribe, 2 - on value change
        verify(subscriber, times(2)).call(anyBoolean());
    }

    @Test
    public void testOnValueChangedSameValue() throws Exception {
        final RxValue<Boolean> underTheTest = RxValue.of(false);
        underTheTest.asObservable().subscribe(subscriber);

        underTheTest.set(true);

        // do second call, due to no value change we should not receive any updates
        underTheTest.set(true);

        // verify number of calls, 1 - on subscribe, 2 - on value change
        verify(subscriber, times(2)).call(anyBoolean());
    }

    @Test
    @SuppressWarnings("unchecked")
    public void testOnValueChangedParams() throws Exception {
        final RxValue<Boolean> underTheTest = RxValue.of(false);
        underTheTest.asObservable().subscribe(subscriber);

        // reset mock class for next call
        Mockito.reset(subscriber);

        underTheTest.set(true);

        // verify the call value
        verify(subscriber).call(catcher.capture());
        assertThat(catcher.getValue()).isTrue();
    }
    //endregion
}