ObservableEchoTransformer.kt

import io.reactivex.Observable
import io.reactivex.ObservableSource
import io.reactivex.ObservableTransformer
import java.util.Optional
import java.util.concurrent.atomic.AtomicReference

/**
 * A transformer that remembers the last value. For observer, you'll get a pair of
 * last value and current value.
 *
 * Note: When there's no last value remembered and initial value is null, observer
 * gets current value along with a null last value.
 */
class ObservableEchoTransformer<T>(
    initialValue: T? = null
) : ObservableTransformer<T, Pair<Optional<T>, T>> {

    private val lastValue = AtomicReference<T?>()

    override fun apply(
        upstream: Observable<T>
    ): ObservableSource<Pair<Optional<T>, T>> {
        return upstream.map { newValue: T ->
            val echoResult = Pair(
                Optional.ofNullable(lastValue.getAndSet(newValue)),
                newValue
            )

            echoResult
        }
    }
}

/**
 * Remember the last value and give you a pair of last value and current value.
 *
 * Note: When there's no last value remembered and initial value is null, observer
 * gets current value along with a null last value.
 */
fun <T> Observable<T>.echo(
    initialValue: T? = null
): Observable<Pair<Optional<T>, T>> {
    return compose(ObservableEchoTransformer(initialValue))
}