LiveData extensions

CustomLiveDataExtensions.kt

import android.arch.lifecycle.LiveData
import android.arch.lifecycle.MutableLiveData
import android.support.annotation.MainThread
import kotlin.reflect.KClass

/** Only emits [T] when [source] is true ]*/
fun <T : Any> LiveData<T>.filterTrue(source: LiveData<Boolean>): LiveData<T> {
    return this.combineLatest(source) { emission, isStateInitialized ->
        Pair(emission, isStateInitialized)
    }
        .filter { it.second }
        .map { it.first }
}

/**
 * Update the current value of this [MutableLiveData], but only if it differs from its current
 * value. If the current value is equal to [onNext], this function will no-op.
 *
 * If different, the value will be set via a post to the main thread.
 */
fun <T : Any> MutableLiveData<T>.postIfDistinct(onNext: T) {
    if (onNext != this.value) {
        this.postValue(onNext)
    }
}

/**
 * Allow [T] to pass through this filter only if it is an instance of [kClass]
 */
fun <T : Any, U : T> LiveData<T>.filterCast(kClass: KClass<U>): LiveData<U> {
    return this.filter { kClass == it::class }
        .map {
            @Suppress("UNCHECKED_CAST")
            it as U
        }
}

/**
 * Update the current value of this [MutableLiveData], but only if it differs from its current
 * value. If the current value is equal to [onNext], this function will no-op.
 */
@MainThread
fun <T : Any> MutableLiveData<T>.setIfDistinct(onNext: T) {
    if (onNext != this.value) {
        this.value = onNext
    }
}

@MainThread
fun <T : Any> createLiveData(value: T): LiveData<T> {
    val liveData = MutableLiveData<T>()
    liveData.value = value
    return liveData
}

RxLiveDataExtensions.kt

@file:JvmName("RxLiveDataExtensions")

import android.arch.lifecycle.LifecycleOwner
import android.arch.lifecycle.LiveData
import android.arch.lifecycle.MediatorLiveData
import android.arch.lifecycle.Observer
import android.arch.lifecycle.Transformations
import java.io.Closeable

fun <T : Any?> LiveData<T>.filterNullable(predicate: LiveDataFilter<T?>): LiveData<T> {
    val mediator = MediatorLiveData<T>()

    mediator.addSource(this) {
        if (predicate(it)) {
            mediator.postValue(it)
        }
    }

    return mediator
}


fun <T : Any> LiveData<T>.filter(predicate: LiveDataFilter<T>): LiveData<T> {
    val mediator = MediatorLiveData<T>()

    mediator.addSource(this) {
        if (it != null) {
            if (predicate(it)) {
                mediator.postValue(it)
            }
        }
    }

    return mediator
}

fun <T : Any> LiveData<T>.doOnNext(onNext: (T) -> Unit): LiveData<T> {
    val mediator = MediatorLiveData<T>()

    mediator.addSource(this) {
        if (it != null) {
            onNext(it)
            mediator.value = it
        }
    }

    return mediator
}

fun <T : Any, R : Any> LiveData<T>.map(map: LiveDataMap<T, R>): LiveData<R> {
    return Transformations.map(this, map)
}

fun <T : Any, R : Any> LiveData<T>.switchMap(switchMap: LiveDataSwitchMap<T, R>): LiveData<R> {
    return Transformations.switchMap(this, switchMap)
}

fun <A : Any, B : Any, R : Any> LiveData<A>.combineLatest(
    otherSource: LiveData<B>, combineFunction: LiveDataBiFunction<A, B, R>
): LiveData<R> {
    return MediatorLiveData<R>().apply {
        var lastA: A? = null
        var lastB: B? = null

        fun update() {
            val localLastA = lastA
            val localLastB = lastB
            if (localLastA != null && localLastB != null)
                this.value = combineFunction(localLastA, localLastB)
        }

        addSource(this@combineLatest) {
            if (lastA !== it) {
                lastA = it
                update()
            }

        }
        addSource(otherSource) {
            if (lastB !== it) {
                lastB = it
                update()
            }

        }
    }
}

fun <T : Any> LiveData<T>.distinctUntilChanged(): LiveData<T> {
    val mediator = MediatorLiveData<T>()

    var lastEmission: T? = null

    mediator.addSource(
        this,
        object : Observer<T> {
            override fun onChanged(newEmission: T?) {
                // we don't allow null values to pass through
                if (newEmission != null) {
                    if (lastEmission == null) {
                        // if this is the first emission, let it pass
                        lastEmission = newEmission
                        mediator.value = newEmission
                    } else {
                        if (lastEmission != newEmission) {
                            // values are different, send emission
                            lastEmission = newEmission
                            mediator.value = newEmission
                        }
                    }
                }
            }

        }
    )

    return mediator

}

/**
 * Like [LiveData.observeForever], but returns a [Closeable] that you can un-subscribe from.
 */
fun <T : Any> LiveData<T>.subscribe(onNext: (T) -> Unit): Closeable {
    val observer = Observer<T> { onNext(it!!) }

    // no nulls!
    val liveData = this.filterNullable { it != null }

    liveData.observeForever(observer)

    return Closeable { liveData.removeObserver(observer) }
}

/**
 * Like [LiveData.observe], except no nulls can be emitted
 */
fun <T : Any> LiveData<T>.subscribe(owner: LifecycleOwner, onNext: (T) -> Unit) {
    val observer = Observer<T> { onNext(it!!) }

    // no nulls!
    val liveData = this.filterNullable { it != null }

    liveData.observe(owner, observer)
}

/**
 * This function creates a [LiveData] of a [Pair] of the two types provided. The resulting LiveData is updated whenever either input LiveData updates and both LiveData have updated at least once before.
 *
 * If the zip of A and B is C, and A and B are updated in this pattern: `AABA`, C would be updated twice (once with the second A value and first B value, and once with the third A value and first B value).
 *
 * @param a the first LiveData
 * @param b the second LiveData
 */
fun <A, B> zipLiveData(a: LiveData<A>, b: LiveData<B>): LiveData<Pair<A, B>> {
    return MediatorLiveData<Pair<A, B>>().apply {
        var lastA: A? = null
        var lastB: B? = null

        fun update() {
            val localLastA = lastA
            val localLastB = lastB
            if (localLastA != null && localLastB != null)
                this.value = Pair(localLastA, localLastB)
        }

        addSource(a) {
            lastA = it
            update()
        }
        addSource(b) {
            lastB = it
            update()
        }
    }
}

typealias LiveDataObserver<T> = (T) -> Unit

        /** return `true` if this emission should pass through the filter */
typealias LiveDataFilter<T> = (T) -> Boolean

typealias LiveDataMap<T, R> = (T) -> R

typealias LiveDataSwitchMap<T, R> = (T) -> LiveData<R>

typealias LiveDataBiFunction<T, U, R> = (T, U) -> R

typealias LiveDataAction<T> = (T) -> Unit


fun <T : Any> LiveData<Optional<T>>.filterSome(): LiveData<T> {
    return filter { it is Some }
        .map { (it as Optional<T>).toNullable()!! }
}

fun <T : Any> LiveData<Optional<T>>.filterNone(): LiveData<Unit> {
    return filter { it == None }
        .map { Unit }
}

RxLiveDataExtensionsKtTest.kt

import android.arch.core.executor.testing.InstantTaskExecutorRule
import android.arch.lifecycle.MutableLiveData
import com.nhaarman.mockitokotlin2.mock
import com.nhaarman.mockitokotlin2.times
import com.nhaarman.mockitokotlin2.verify
import com.nhaarman.mockitokotlin2.verifyNoMoreInteractions
import org.junit.Rule
import org.junit.Test
import org.junit.rules.TestRule


/**
 * Test for [RxLiveDataExtensions]
 */
class RxLiveDataExtensionsTest {

    @JvmField
    @Rule
    var rule: TestRule = InstantTaskExecutorRule()

    /**
     * Tests [doOnNext]
     */
    @Test
    fun operator_doOnNext() {
        val mockObserver = mock<(String) -> Unit> { }

        val source = MutableLiveData<String>()

        source.doOnNext(mockObserver)
            .subscribe { }

        source.value = "1"
        source.value = "2"
        source.value = null // null should be ignored ignored
        source.value = "3"

        verify(mockObserver).invoke("1")
        verify(mockObserver).invoke("2")
        verify(mockObserver).invoke("3")

        verifyNoMoreInteractions(mockObserver)
    }

    @Test
    fun operator_subscribe() {
        val mockObserver = mock<(String) -> Unit> { }

        val source = MutableLiveData<String>()

        source.subscribe(mockObserver)

        source.value = "1"
        source.value = "2"
        source.value = null // null ignored
        source.value = "3"

        verify(mockObserver).invoke("1")
        verify(mockObserver).invoke("2")
        verify(mockObserver).invoke("3")
        verifyNoMoreInteractions(mockObserver)
    }

    @Test
    fun operator_filter() {
        val mockObserver = mock<(Int) -> Unit> { }

        val source = MutableLiveData<Int>()

        source.filter { it != 2 }
            .subscribe(mockObserver)

        source.value = 1
        source.value = 2 // 2 will be filtered out
        source.value = null // null should be ignored
        source.value = 3

        verify(mockObserver).invoke(1)
        verify(mockObserver).invoke(3)
        verifyNoMoreInteractions(mockObserver)
    }

    @Test
    fun operator_combineLatest() {
        val mockObserver = mock<(Pair<String, Int>) -> Unit> { }

        val sourceNames = MutableLiveData<String>()
        val sourceAges = MutableLiveData<Int>()

        sourceNames.combineLatest(sourceAges) { name, age -> Pair(name, age) }
            .subscribe(mockObserver)

        sourceNames.value = "Zak"

        sourceNames.value = "Grace"
        sourceAges.value = 24

        sourceNames.value = "Kelly"
        sourceAges.value = 25

        sourceNames.value = "Jack"
        sourceAges.value = 27
        sourceAges.value = 28 // happy birthday

        verify(mockObserver, times(1)).invoke(Pair("Grace", 24))
        verify(mockObserver, times(1)).invoke(Pair("Kelly", 24))
        verify(mockObserver, times(1)).invoke(Pair("Kelly", 25))
        verify(mockObserver, times(1)).invoke(Pair("Jack", 25))
        verify(mockObserver, times(1)).invoke(Pair("Jack", 27))
        verify(mockObserver, times(1)).invoke(Pair("Jack", 28))
        verifyNoMoreInteractions(mockObserver)
    }

    @Test
    fun operator_distinctUntilChanged() {
        val mockObserver = mock<(Int) -> Unit> { }

        val source = MutableLiveData<Int>()

        source.distinctUntilChanged()
            .subscribe(mockObserver)

        source.value = 1
        source.value = 2
        source.value = 2
        source.value = 2
        source.value = 3

        verify(mockObserver, times(1)).invoke(1)
        verify(mockObserver, times(1)).invoke(2)
        verify(mockObserver, times(1)).invoke(3)
        verifyNoMoreInteractions(mockObserver)
    }
}