RxJava 在网络关闭时取消订阅答案

【问题标题】：RxJava unsubscribe on Network closeRxJava 在网络关闭时取消订阅
【发布时间】：2019-07-12 11:58:01
【问题描述】：

如果网络请求已关闭，我目前正在尝试取消订阅 observable，以不再进一步流式传输数据。

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

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

所以有多个订阅者。但是如果网络关闭了，我不知道如何退订。

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

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

如果我取消来自客户端的请求，observable 将继续“流式传输”数据。

感谢您的帮助

【问题讨论】：

标签： java networking stream rx-java vert.x

【解决方案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
}

【讨论】：

这会在请求结束之前调用 dispose，不是吗？
当你想取消请求并取消订阅 observable 时，你必须调用 dispose()
您可以创建自定义的 observable 并自己处理所有问题，我更新了答案
我做错了。这有效：response.closeHandler{ disposal.dispose()}

【解决方案2】：

您可以使用takeUntil 运算符和响应closeHandler：

router.get("/test").handler(ctx -> {
    ctx.response().setChunked(true);
    Observable.interval(1, TimeUnit.SECONDS)
            .share()
            .doFinally(() -> System.out.println("Disposed"))
            .takeUntil(Observable.create(e -> ctx.response().closeHandler(event -> e.onComplete())))
            .subscribe(l -> ctx.response().write(l.toString()));
});

【讨论】：

vertx 在取消来自客户端的请求时不会调用关闭处理程序 :-(