如何在 Kotlin 中将以下 Observables 合并为一个?
How to combine following Observables into one in Kotlin?
我在 Kotlin 中有这两个 Observable
s,其中一个只是用作计时器,另一个是 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 进行合并。我认为在你的情况下 combinelatest 是合适的
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
})
更新:
您可以这样更改代码:
Observable.timer(5, TimeUnit.SECONDS, AndroidSchedulers.mainThread()).startWith(-1L)
.doOnNext {
if (it == -1L) return@doOnNext
//your condition
}
我在 Kotlin 中有这两个 Observable
s,其中一个只是用作计时器,另一个是 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 进行合并。我认为在你的情况下 combinelatest 是合适的
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
})
更新: 您可以这样更改代码:
Observable.timer(5, TimeUnit.SECONDS, AndroidSchedulers.mainThread()).startWith(-1L)
.doOnNext {
if (it == -1L) return@doOnNext
//your condition
}