Тест PublishProcessor + ConcatMap + RoomDB

Question

У меня проблемы с написанием теста для списка, получающего данные из потока.

мой метод выглядит примерно так:

 override fun getMessages() {

    val d = PublishProcessor.create().onBackpressureDrop()
            .filter { !loading }
            .doOnNext {loading = true}
            .concatMap {
                localMessagesDataStore.getMessages() // Flowable from db
            }
            .observeOn(OBSERVE_SCHEDULER, true)
            .subscribe({
                  SUCCESS
                }, {
                ERROR
            })

    disposables?.add(d)

    fetchNextPage()
}

До сих пор я пробовал множество подходов, но основные части:

• создать PublishProcessor
• добавить concatMap и подписаться
• проверить

Но дело в том, что я всегда получаю ошибки Мокито о неправильном типе возврата или утверждении или о чем-то еще. Я также попробовал с подходом mockito when, но это не сработало.

Вот несколько ссылок на многие из них, которые я попробовал:

https://github.com/akarnokd/RxJava3-preview/blob/master/flowable/src/test/java/io/reactivex/flowable/internal/operators/FlowableConcatTest.java

https://www.javatips.net/api/-master/RxJava-2.x/src/test/java/io/reactivex/internal/operators/flowable/FlowableConcatTest.java

https://proandroiddev.com/testing-concurrency-in-rxjava-831804a9e526

и т.д..

Практически все согласованные результаты в Google содержат ключевые слова из названия проблемы. Любая помощь будет оценена!