Additional LiveData transformations. Now released as a library: https://github.com/sczerwinski/android-lifecycle

LiveDataTransformations.kt

import androidx.lifecycle.LiveData
import androidx.lifecycle.MediatorLiveData
import androidx.lifecycle.Observer
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.Job
import kotlinx.coroutines.delay
import kotlinx.coroutines.launch
import kotlin.coroutines.CoroutineContext
import kotlin.coroutines.EmptyCoroutineContext

fun <T, R> LiveData<T>.mapNotNull(
    mapFunction: (T) -> R?
): LiveData<R> {
    val result = MediatorLiveData<R>()
    result.addSource(this) { x ->
        val y = mapFunction(x)
        if (y != null) result.value = y
    }
    return result
}

fun <T> LiveData<T>.filter(
    predicate: (T) -> Boolean
): LiveData<T> {
    val result = MediatorLiveData<T>()
    result.addSource(this) { x ->
        if (predicate(x)) result.value = x
    }
    return result
}

fun <T> LiveData<T?>.filterNotNull(): LiveData<T> {
    val result = MediatorLiveData<T>()
    result.addSource(this) { x ->
        if (x != null) result.value = x
    }
    return result
}

inline fun <reified R> LiveData<*>.filterIsInstance(): LiveData<R> {
    val result = MediatorLiveData<R>()
    result.addSource(this) { x ->
        if (x is R) result.value = x
    }
    return result
}

fun <T> LiveData<T>.reduceNotNull(
    reduceFunction: (T, T) -> T
): LiveData<T> {
    val result = MediatorLiveData<T>()
    result.addSource(this) { x ->
        if (x != null) {
            val oldValue = result.value
            result.value =
                if (oldValue == null) x
                else reduceFunction(oldValue, x)
        }
    }
    return result
}

fun <T> LiveData<T?>.reduce(
    reduceFunction: (T?, T?) -> T?
): LiveData<T?> {
    val result = MediatorLiveData<T>()
    result.addSource(this, object : Observer<T?> {

        private var firstTime = true

        override fun onChanged(x: T?) {
            if (firstTime) {
                firstTime = false
                result.value = x
            } else {
                result.value = reduceFunction(result.value, x)
            }
        }
    })
    return result
}

fun <T> LiveData<T>.throttleWithTimeout(
    timeMillis: Long,
    context: CoroutineContext = EmptyCoroutineContext
): LiveData<T> {
    val result = MediatorLiveData<T>()
    result.addSource(this, object : Observer<T?> {

        private var throttleJob: Job? = null

        override fun onChanged(x: T?) {
            throttleJob?.cancel()
            throttleJob = CoroutineScope(context).launch {
                delay(timeMillis)
                result.postValue(x)
            }
        }
    })
    return result
}

LiveDataTransformationsTests.kt

import androidx.arch.core.executor.ArchTaskExecutor
import androidx.arch.core.executor.TaskExecutor
import androidx.lifecycle.MutableLiveData
import androidx.lifecycle.Observer
import io.mockk.impl.annotations.RelaxedMockK
import io.mockk.junit5.MockKExtension
import io.mockk.verifySequence
import kotlinx.coroutines.ExperimentalCoroutinesApi
import kotlinx.coroutines.test.TestCoroutineDispatcher
import org.junit.jupiter.api.AfterEach
import org.junit.jupiter.api.BeforeEach
import org.junit.jupiter.api.DisplayName
import org.junit.jupiter.api.Test
import org.junit.jupiter.api.extension.ExtendWith

@ExtendWith(MockKExtension::class)
class LiveDataTransformationsTests {

    @RelaxedMockK
    lateinit var stringObserver: Observer<String?>

    @RelaxedMockK
    lateinit var intObserver: Observer<Int>

    @BeforeEach
    fun initInstantTaskExecutor() {
        ArchTaskExecutor.getInstance().setDelegate(InstantTaskExecutor())
    }

    @AfterEach
    fun clearInstantTaskExecutor() {
        ArchTaskExecutor.getInstance().setDelegate(null)
    }

    @Test
    @DisplayName(
        value = "GIVEN source LiveData transformed with mapNotNull, " +
                "WHEN posting values to source, " +
                "THEN only non-null transformed values should be observed"
    )
    fun mapNotNull() {
        val source = MutableLiveData<Int>()
        val transformed = source.mapNotNull { number ->
            if (number % 2 == 0) number.toString()
            else null
        }
        transformed.observeForever(stringObserver)

        source.postValue(1)
        source.postValue(2)
        source.postValue(3)
        source.postValue(4)

        verifySequence {
            stringObserver.onChanged("2")
            stringObserver.onChanged("4")
        }
    }

    @Test
    @DisplayName(
        value = "GIVEN source LiveData with filter, " +
                "WHEN posting values to source, " +
                "THEN only values meeting the predicate should be observed"
    )
    fun filter() {
        val source = MutableLiveData<Int>()
        val filtered = source.filter { number -> number % 2 == 0 }
        filtered.observeForever(intObserver)

        source.postValue(1)
        source.postValue(2)
        source.postValue(3)
        source.postValue(4)

        verifySequence {
            intObserver.onChanged(2)
            intObserver.onChanged(4)
        }
    }

    @Test
    @DisplayName(
        value = "GIVEN source LiveData with filterNotNull, " +
                "WHEN posting values to source, " +
                "THEN only non-null values should be observed"
    )
    fun filterNotNull() {
        val source = MutableLiveData<Int?>()
        val filtered = source.filterNotNull()
        filtered.observeForever(intObserver)

        source.postValue(1)
        source.postValue(2)
        source.postValue(null)
        source.postValue(3)

        verifySequence {
            intObserver.onChanged(1)
            intObserver.onChanged(2)
            intObserver.onChanged(3)
        }
    }

    @Test
    @DisplayName(
        value = "GIVEN source LiveData with filterIsInstance, " +
                "WHEN posting values to source, " +
                "THEN only values of given type should be observed"
    )
    fun filterIsInstance() {
        val source = MutableLiveData<Any?>()
        val filtered = source.filterIsInstance<Int>()
        filtered.observeForever(intObserver)

        source.postValue(1)
        source.postValue("text")
        source.postValue(2)
        source.postValue(null)
        source.postValue(3)
        source.postValue("4")

        verifySequence {
            intObserver.onChanged(1)
            intObserver.onChanged(2)
            intObserver.onChanged(3)
        }
    }

    @Test
    @DisplayName(
        value = "GIVEN source LiveData with reduce, " +
                "WHEN posting values to source, " +
                "THEN reduced value should be observed after each emitted item"
    )
    fun reduce() {
        val source = MutableLiveData<String>()
        val reduced = source.reduce { a, b -> "$a, $b" }
        reduced.observeForever(stringObserver)

        source.postValue("first")
        source.postValue("second")
        source.postValue(null)
        source.postValue("third")

        verifySequence {
            stringObserver.onChanged("first")
            stringObserver.onChanged("first, second")
            stringObserver.onChanged("first, second, null")
            stringObserver.onChanged("first, second, null, third")
        }
    }

    @Test
    @DisplayName(
        value = "GIVEN source LiveData with reduceNotNull, " +
                "WHEN posting values to source, " +
                "THEN reduced value should be observed after each emitted non-null item"
    )
    fun reduceNotNull() {
        val source = MutableLiveData<Int>()
        val reduced = source.reduceNotNull { a, b -> a + b }
        reduced.observeForever(intObserver)

        source.postValue(1)
        source.postValue(2)
        source.postValue(null)
        source.postValue(3)
        source.postValue(4)

        verifySequence {
            intObserver.onChanged(1)
            intObserver.onChanged(3)
            intObserver.onChanged(6)
            intObserver.onChanged(10)
        }
    }

    @Test
    @DisplayName(
        value = "GIVEN source LiveData with throttleWithTimeout, " +
                "WHEN posting multiple values to source, " +
                "THEN only the latest values after timeout should be observed"
    )
    @ExperimentalCoroutinesApi
    fun throttleWithTimeout() {
        val dispatcher = TestCoroutineDispatcher()

        val source = MutableLiveData<Int>()
        val throttled = source.throttleWithTimeout(timeMillis = 9_000L, context = dispatcher)
        throttled.observeForever(intObserver)

        source.postValue(1)
        dispatcher.advanceTimeBy(delayTimeMillis = 8_000L)
        source.postValue(2)
        dispatcher.advanceTimeBy(delayTimeMillis = 8_000L)
        source.postValue(3)
        dispatcher.advanceTimeBy(delayTimeMillis = 8_000L)
        source.postValue(4)
        dispatcher.advanceTimeBy(delayTimeMillis = 10_000L)
        source.postValue(5)
        dispatcher.advanceTimeBy(delayTimeMillis = 10_000L)
        source.postValue(6)
        dispatcher.advanceTimeBy(delayTimeMillis = 8_000L)
        source.postValue(7)
        dispatcher.advanceTimeBy(delayTimeMillis = 10_000L)

        verifySequence {
            intObserver.onChanged(4)
            intObserver.onChanged(5)
            intObserver.onChanged(7)
        }
    }

    private class InstantTaskExecutor : TaskExecutor() {
        override fun executeOnDiskIO(runnable: Runnable) = runnable.run()
        override fun postToMainThread(runnable: Runnable) = runnable.run()
        override fun isMainThread(): Boolean = true
    }
}