如何解决如何 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)
}
版权声明:本文内容由互联网用户自发贡献,该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容, 请发送邮件至 dio@foxmail.com 举报,一经查实,本站将立刻删除。