我是新上rxjava,我想执行一个轮询任务每2秒50次,也是它,如果在任务中的一些条件满足,我将尝试使用可能会终止Observable.interval,但是我发现没有办法来终止它除了扔例外,有没有其他运营商来实现我的目标？BTW此功能用作API以提供可观察对象,因此我无法通过取消描述来控制订户和终止.
Observable.interval(timeout, interval, TimeUnit.SECONDS)
.flatmap(task - > task)
我有一个用例,我想使用协程,但有点混淆如何实现它.
一个ViewModel,它具有一个范围并绑定到UI生命周期并从存储库调用API:
class UserViewModel(): CoroutineScope {
private val job = Job()
override val coroutineContext: CoroutineContext
get() = Dispatchers.Main + job
fun showUser() {
launch {
val user = repo.getUser()
livedata = user
}
}
fun onClean() {
job.cancel()
}
}
存储库使用协同程序来构建网络调用,如下所示:
suspend fun getUser() = GlobalScope { ... }
用例是一旦从ViewModel调用API,就需要始终完全执行存储库功能,因为我们需要捕获来自服务器的所有网络响应.
我怎样才能确保存储库中的协程始终执行,但是一旦视图模型被清除,ViewModel协程将被取消以避免内存泄漏？
我的用例如下：
我创建了一个 Observable 来访问远程服务器以获取一些数据。然而，由于服务器设计不善，可能永远没有响应，也没有例外。为了解决这个问题，我想要一些超时重试机制。
目前我尝试启动一个计时器来停止请求并在其中抛出异常，然后重试直到一定次数的尝试或我的真正超时。我尝试使用mergeWith运算符将请求与Observable.interval映射合并以使用生成错误Observable.error()，但是我无法在订阅者中捕获错误，并且看起来Observable.interval永无止境。
我应该如何处理 RXJAVA 中的任何操作员的这种情况？
我当前的代码如下所示：
Observable.fromEmitter(fetchNetwork->...)
.mergeWith(Observable.interval(...)
.flatmap(n->(observable.error)))
.retryWhen(error->(checkTimeExceed))
.subscribe(handleResult)