嗨,我有一个 rxjava 平面图,我想在其中调用协程用例 onStandUseCase,这是一个 api 调用
最初的用例也是基于 rxjava 的,它曾经返回 Observable<GenericResponse>
它工作得很好
现在我将使用更改为基于协程,它只返回 GenericResponse
请问如何修改平面图以与协程用例一起正常工作
subscriptions += view.startFuellingObservable
.onBackpressureLatest()
.doOnNext { view.showLoader(false) }
.flatMap {
if (!hasOpenInopIncidents()) {
//THIS IS WHERE THE ERROR IS IT RETURNS GENERICRESPONSE
onStandUseCase(OnStandUseCase.Params("1", "2", TimestampedAction("1", "2", DateTime.now()))) {
}
} else {
val incidentOpenResponse = GenericResponse(false)
incidentOpenResponse.error = OPEN_INCIDENTS
Observable.just(incidentOpenResponse)
}
}
.subscribe(
{ handleStartFuellingClicked(view, it) },
{ onStartFuellingError(view) }
)
OnStandUseCase.ktclass OnStandUseCase @Inject constructor(
private val orderRepository: OrderRepository,
private val serviceOrderTypeProvider: ServiceOrderTypeProvider
) : UseCaseCoroutine<GenericResponse, OnStandUseCase.Params>() {
override suspend fun run(params: Params) = orderRepository.notifyOnStand(
serviceOrderTypeProvider.apiPathFor(params.serviceType),
params.id,
params.action
)
data class Params(val serviceType: String, val id: String, val action: TimestampedAction)
}
用例协程abstract class UseCaseCoroutine<out Type, in Params> where Type : Any {
abstract suspend fun run(params: Params): Type
operator fun invoke(params: Params, onResult: (type: Type) -> Unit = {}) {
val job = GlobalScope.async(Dispatchers.IO) { run(params) }
GlobalScope.launch(Dispatchers.Main) { onResult(job.await()) }
}
}
startFuellingObservable 是val startFuellingObservable: Observable<Void>
这是错误的图像有关如何解决此问题的任何建议,请
提前致谢
R
最佳答案
有integration library连接 RxJava 和 Kotlin 协程。rxSingle
可用于将挂起函数转换为 Single
. OP 想要一个 Observable
,所以我们可以调用toObservable()
为转换。
.flatMap {
if (!hasOpenInopIncidents()) {
rxSingle {
callYourSuspendFunction()
}.toObservable()
} else {
val incidentOpenResponse = GenericResponse(false)
incidentOpenResponse.error = OPEN_INCIDENTS
Observable.just(incidentOpenResponse)
}
}
请注意 Observable
两个分支中的 s 只包含一个元素。我们可以通过使用 Observable#concatMapSingle 使这一事实更加明显。 ..concatMapSingle {
if (!hasOpenInopIncidents()) {
rxSingle { callYourSuspendFunction() }
} else {
val incidentOpenResponse = GenericResponse(false)
incidentOpenResponse.error = OPEN_INCIDENTS
Single.just(incidentOpenResponse)
}
}
关于android - 如何从 rxjava flatmap 调用协程用例,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/65616836/