如何 doOnNext 只消耗 PublishSubject 的前 n 项？

问题描述

我有一个PublishSubject：

val myPublishSubject = remember {
    PublishSubject.create<Long>().apply {
        doOnNext {
            Logger.debug(TAG) { "Got new Long $it" }
        }
            .takeuntil(someObservable)
            .subscribe()
    }
}

我有一个服务，它在另一个函数的某个地方获取一个长值流：

while(notFinished){
    val newLong = getSomeLong()
    myPublishSubject.onNext(newLong)
}

在这里，上面的 doOnNext{} 是永远记录的。例如，我如何只允许前 20 个 Long 值？ .take(20) 无效！

解决方法

take(20) 应该可以工作，这个测试通过了：

@Test
fun consumeOnlyFirst20ValueTest() {
    val ps = PublishSubject.create<Long>()

    CoroutineScope(Dispatchers.Default).launch {
        for (i in 1L..1000L) {
            ps.onNext(i)
            delay(50)
        }
    }

    var i = 0
    ps.take(20)
            .doOnNext {
                i++
            }.subscribeOn(Schedulers.io())
            .subscribe()

    Thread.sleep(6000)
    assertEquals(20,i)
}