https://speakerdeck.com/hkusu/extension-functions-for-livedata で説明したコードですが https://gist.github.com/hkusu/6bdc35e75614df559135174535e3c3ce の方が新しいです

LiveDataExt.kt

package io.github.hkusu.sample.ext

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

inline fun <T, U> LiveData<T>.mapNotNull(
    crossinline block: (T) -> U? // null または戻り値なしの return で更新をキャンセルできるように
): LiveData<U> {
    val result = MediatorLiveData<U>()
    result.addSource(this) { t ->
        block.invoke(t)?.takeUnless { it is Unit }?.let { result.value = it }
    }
    return result
}

// 現状では inline fun とすると落ちる https://github.com/Kotlin/kotlinx.coroutines/issues/1150
fun <T, U> LiveData<T>.coMap(
    coroutineScope: CoroutineScope, // 通常は ViewModel の viewModelScope を渡す
    coroutineContext: CoroutineContext = coroutineScope.coroutineContext, // context を切り替えたい場合に利用(例: Dispatchers.IO)。block 中で withContext で切り替えても可
    block: suspend (T) -> U // liveData-ktx の map に合わせて T は NotNull にしているが null が渡されうる
): LiveData<U> {
    val result = MediatorLiveData<U>()
    result.addSource(this) {
        coroutineScope.launch(coroutineContext) {
            result.postValue(block.invoke(it)) // mainスレッド以外からも更新できるように
        }
    }
    return result
}
// この拡張関数を利用しない場合、
// val hogeLiveData = fugalLveData.switchMap {
//     liveData(viewModelScope.coroutineContext) {
//         emit("hoge")
//     }
// }
// のようにも書ける

fun <T, U> LiveData<T>.coMapNotNull(
    coroutineScope: CoroutineScope,
    coroutineContext: CoroutineContext = coroutineScope.coroutineContext,
    block: suspend (T) -> U?
): LiveData<U> {
    val result = MediatorLiveData<U>()
    result.addSource(this) { t ->
        coroutineScope.launch(coroutineContext) {
            block.invoke(t)?.takeUnless { it is Unit }?.let { result.postValue(it) }
        }
    }
    return result
}

inline fun <T1, T2, U> merge(
    liveData1: LiveData<T1>,
    liveData2: LiveData<T2>,
    crossinline block: (T1?, T2?) -> U?
): LiveData<U> {
    val result = MediatorLiveData<U>()
    result.addSource(liveData1) { t1 ->
        block.invoke(t1, liveData2.value)?.takeUnless { it is Unit }?.let { result.value = it }
    }
    result.addSource(liveData2) { t2 ->
        block.invoke(liveData1.value, t2)?.takeUnless { it is Unit }?.let { result.value = it }
    }
    return result
}

inline fun <T1, T2, T3, U> merge(
    liveData1: LiveData<T1>,
    liveData2: LiveData<T2>,
    liveData3: LiveData<T3>,
    crossinline block: (T1?, T2?, T3?) -> U?
): LiveData<U> {
    val result = MediatorLiveData<U>()
    result.addSource(liveData1) { t1 ->
        block.invoke(t1, liveData2.value, liveData3.value)?.takeUnless { it is Unit }?.let { result.value = it }
    }
    result.addSource(liveData2) { t2 ->
        block.invoke(liveData1.value, t2, liveData3.value)?.takeUnless { it is Unit }?.let { result.value = it }
    }
    result.addSource(liveData3) { t3 ->
        block.invoke(liveData1.value, liveData2.value, t3)?.takeUnless { it is Unit }?.let { result.value = it }
    }
    return result
}

fun <T1, T2, U> coMerge(
    liveData1: LiveData<T1>,
    liveData2: LiveData<T2>,
    coroutineScope: CoroutineScope,
    coroutineContext: CoroutineContext = coroutineScope.coroutineContext,
    block: suspend (T1?, T2?) -> U?
): LiveData<U> {
    val result = MediatorLiveData<U>()
    result.addSource(liveData1) { t1 ->
        coroutineScope.launch(coroutineContext) {
            block.invoke(t1, liveData2.value)?.takeUnless { it is Unit }?.let { result.postValue(it) }
        }
    }
    result.addSource(liveData2) { t2 ->
        coroutineScope.launch(coroutineContext) {
            block.invoke(liveData1.value, t2)?.takeUnless { it is Unit }?.let { result.postValue(it) }
        }
    }
    return result
}

fun <T1, T2, T3, U> coMerge(
    liveData1: LiveData<T1>,
    liveData2: LiveData<T2>,
    liveData3: LiveData<T3>,
    coroutineScope: CoroutineScope,
    coroutineContext: CoroutineContext = coroutineScope.coroutineContext,
    block: suspend (T1?, T2?, T3?) -> U?
): LiveData<U> {
    val result = MediatorLiveData<U>()
    result.addSource(liveData1) { t1 ->
        coroutineScope.launch(coroutineContext) {
            block.invoke(t1, liveData2.value, liveData3.value)?.takeUnless { it is Unit }?.let { result.postValue(it) }
        }
    }
    result.addSource(liveData2) { t2 ->
        coroutineScope.launch(coroutineContext) {
            block.invoke(liveData1.value, t2, liveData3.value)?.takeUnless { it is Unit }?.let { result.postValue(it) }
        }
    }
    result.addSource(liveData3) { t3 ->
        coroutineScope.launch(coroutineContext) {
            block.invoke(liveData1.value, liveData2.value, t3)?.takeUnless { it is Unit }?.let { result.postValue(it) }
        }
    }
    return result
}

inline fun <T> LiveData<T>.filter(
    crossinline block: (T) -> Boolean
): LiveData<T> {
    val result = MediatorLiveData<T>()
    result.addSource(this) {
        if (block.invoke(it)) {
            result.value = it
        }
    }
    return result
}

fun <T> LiveData<T>.coFilter(
    coroutineScope: CoroutineScope,
    coroutineContext: CoroutineContext = coroutineScope.coroutineContext,
    block: suspend (T) -> Boolean
): LiveData<T> {
    val result = MediatorLiveData<T>()
    result.addSource(this) {
        coroutineScope.launch(coroutineContext) {
            if (block.invoke(it)) {
                result.postValue(it)
            }
        }
    }
    return result
}

class Observe<T>(private val liveData: LiveData<T>) {

    private val observer = Observer<T> {}

    fun start() {
        liveData.observeForever(observer)
    }

    fun cancel() {
        liveData.removeObserver(observer)
    }
}

fun <T> LiveData<T>.observe(): Observe<T> = Observe(this).apply { start() }