java - RxJava 在网络关闭时取消订阅

Question

如果网络请求关闭，我当前正在尝试取消订阅可观察对象，以便不再进一步传输数据。

我的 Observable 是通过以下方式创建的:

Observable.interval(1, TimeUnit.SECONDS)
.map { LocalDateTime.now() }.share()

所以有多个订阅者。但我不知道如果网络关闭如何取消订阅。

我目前正在将带有 vert.x 的服务器发送事件的数据传输到客户端，如下所示:

flow.subscribe({
  response.write("the current time is $it")
}, ::println, {
  response.end()
})

如果我取消客户端的请求，可观察对象将继续“流式传输”数据。

感谢您的帮助

Answer 1

您可以通过调用dispose()来取消订阅订阅者

Disposable disposable = flow.subscribe({
  response.write("the current time is $it")
}, ::println, {
  response.end()
})

disposable.dispose();

更新:自定义可观察

val observable: Observable<LocalDateTime> = Observable.create { emitter ->
    val timer = Timer()
    timer.schedule(timerTask {
        if (emitter.isDisposed) {//<-- cancel emmitting items if disposed
            timer.cancel()
        } else {
            emitter.onNext(LocalDateTime.now())
        }
    }, 0, 1000)

}
disposable = observable.share().subscribe { t ->
    System.out.println(" Hello World! $t");
    disposable?.dispose()//<-- here calling dispose() causes observable to stop emitting items
}