我在 Kotlin 中有这两个Observable
,其中只是充当计时器,另一个是 HTTP 网络调用响应Observer
。
timerDisposable = Observable.timer(daleyABCControlResetSeconds, TimeUnit.SECONDS, AndroidSchedulers.mainThread())
.doOnNext {
if (getABCUpdate() != null) {
Log.d("ABC", "Media status reset after 3 seconds: ")
updateABCResponse(getABCUpdate())
}
}.subscribe()
disposable = audioApi.setABCUpdate(abcUpdate)
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe({
timerDisposable.dispose()
updateABCResponse(it)
Log.d("ABC", "Media Status updated:")
}, {
Log.d("ABC", "Error updating Media Status: " + it.message)
isABCControlChangeRequested = false
})
我对这种方法不满意,谁能给我正确的方向,以充分发挥rx
的潜力。提前谢谢。
编辑
Observable.combineLatest(Observable.timer(daleyABCControlResetSeconds, TimeUnit.SECONDS, AndroidSchedulers.mainThread())
.doOnNext {
if (getABCUpdate() != null) {
Log.d("ABC", "Media status reset after 3 seconds: ")
updateABCResponse(getABCUpdate())
}
},
audioApi.setABCUpdate(abcUpdate)
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread()),
BiFunction<Long, ABCStatusUpdate, ABCStatusUpdate> { _, abcStatusUpdate ->
abcStatusUpdate
})
.subscribe({
timerDisposable.dispose()
updateABCResponse(abcStatusUpdate)
Log.d("ABC", "Media Status updated:")
}, {
Log.d("ABC", "Error updating Media Status: " + abcStatusUpdate.vol)
isABCControlChangeRequested = false
})
您可以使用 combinelatest、zip 或 merge 进行组合。我认为在您的情况下,组合最新是合适的
Observable.combineLatest(
Observable.timer(daleyABCControlResetSeconds, TimeUnit.SECONDS, AndroidSchedulers.mainThread())
.doOnNext {
if (getABCUpdate() != null) {
Log.d("ABC", "Media status reset after 3 seconds: ")
updateABCResponse(getABCUpdate())
}
},
audioApi.setABCUpdate(abcUpdate)
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread()),
BiFunction<Long, YourApiResponseType, YourApiResponseType> { _, response ->
response})
.subscribe({
timerDisposable.dispose()
updateABCResponse(it)
Log.d("ABC", "Media Status updated:")
}, {
Log.d("ABC", "Error updating Media Status: " + it.message)
isABCControlChangeRequested = false
})
UPD:你可以像这样更改你的代码:
Observable.timer(5, TimeUnit.SECONDS, AndroidSchedulers.mainThread()).startWith(-1L)
.doOnNext {
if (it == -1L) return@doOnNext
//your condition
}