二通りの方法を示します。

1. LiveData を BroadcastChannel に変換する

いきなりテストです。

class MyTest {
    private val dispatcher = TestCoroutineDispatcher()

    @[Rule JvmField]
    val rule = InstantTaskExecutorRule()

    @Before
    fun before() {
        Dispatchers.setMain(dispatcher)
    }

    @After
    fun after() {
        Dispatchers.resetMain()
        dispatcher.cleanupTestCoroutines()
    }

    @Test
    fun test() = dispatcher.runBlockingTest {
        // ソースとなる LiveData
        val source = object : MutableLiveData<Int>(0) {
            override fun onActive() = println("onActive")
            override fun onInactive() = println("onInactive")
        }

        // BroadcastChannel に変換
        val broadcastChannel = source.asFlow().broadcastIn(this)

        // Flow に変換 (都度 broadcastChannel.openSubscription() しても良い)
        val flow = broadcastChannel.asFlow()

        val job1 = launch { flow.collect { println("coroutine1: $it") } }

        source.value = 1

        job1.cancel()

        source.value = 2

        val job2 = launch { flow.collect { println("coroutine2: $it") } }

        source.value = 3

        job2.cancel()

        broadcastChannel.cancel()
    }
}

実行結果は以下です。

onActive
coroutine1: 0
coroutine1: 1
coroutine2: 3
onInactive

coroutine1 では初期値 0 および 1 を受信していますが、coroutine2 では 3 だけを受信しており、直前の値 2 を受信していないことがわかります。

利点

特別な仕組みを作りこむ必要がない

欠点

おまじないっぽい？
多くの関数が @ExperimentalCoroutinesApi だったり @FlowPreview だったりする

2. LiveData.getVersion() を使う

androidx.lifecycle パッケージ配下に以下の関数を用意します。

package androidx.lifecycle

internal fun LiveData<*>.version() = version

次に、任意のパッケージ配下に以下の関数を用意します。

@MainThread
fun <T> LiveData<T>.observeChangesForever(action: (T) -> Unit): Observer<T> =
    object : Observer<T> {
        private val version = version()
        override fun onChanged(t: T) {
            if (version() > version) action(t)
        }
    }.also { observeForever(it) }

省略していますが、LifecycleOwner を受け取る関数も同じように作ることができます。

テストは以下です。

class MyTest {
    @[Rule JvmField]
    val instantTaskExecutorRule: InstantTaskExecutorRule = InstantTaskExecutorRule()

    @Test
    fun test() {
        // ソースとなる LiveData
        val source = object : MutableLiveData<Int>(0) {
            override fun onActive() = println("onActive")
            override fun onInactive() = println("onInactive")
        }

        val observer1 = source.observeChangesForever { println("observer1: $it") }

        source.value = 1

        source.removeObserver(observer1)

        source.value = 2

        val observer2 = source.observeChangesForever { println("observer2: $it") }

        source.value = 3

        source.removeObserver(observer2)
    }
}

実行結果は以下です。

onActive
observer1: 1
onInactive
onActive
observer2: 3
onInactive

observer1・observer2 共に直前の値 (0 および 2) を受信していないことがわかります。

利点

わかりやすい (かも)

欠点

作りこみが必要
LiveData.getValue() はパッケージプライベートなので、今後動かなくなるかも

二通りの方法を示しましたが両者は挙動が異なるので、使い方には注意が必要です。

以上です。