如何在带有SSE(服务器发送事件)的RxJava2 Flowable上处理浏览器关闭情况(操作/事件)

Question

如果消费者关闭浏览器或以某种方式断开与SSE流的连接，我希望得到通知。

例如，当连接从curl开始时，调用doOnSubscribe操作，connectedDeliveryCount增加，但从未减少，尽管我故意用ctrl-c关闭了连接。没有像doOnComplete、doOnTerminate或doOnCancel那样调用任何事件

获取的样本连接

faruk@virtualBox:/usr/lib/jvm$ curl -X GET http://localhost:8080/delivery-stream -v
Note: Unnecessary use of -X or --request, GET is already inferred.
*   Trying 127.0.0.1...
* TCP_NODELAY set
* Connected to localhost (127.0.0.1) port 8080 (#0)
> GET /delivery-stream HTTP/1.1
> Host: localhost:8080
> User-Agent: curl/7.58.0
> Accept: */*
> 
^C
faruk@virtualBox:/usr/lib/jvm$

代码块：

static AtomicInteger connectedDeliveryCount = new AtomicInteger();
@GetMapping(path = "/delivery-stream", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
public Flowable<String> deliveryStream() {
return ReactiveStreamsOrderService.deliverySubjectToFlowable()
.doOnCancel(() -> System.out.println("{doOnCancel}: xxx"))
.doOnComplete(() -> System.out.println("{doOnComplete}: xxx"))
.doOnSubscribe(disposable -> System.out.println("{doOnSubscribe}: xxx"))
.doOnTerminate(() -> System.out.println("{doOnTerminate}: xxx"))
.onErrorReturn(throwable -> "xxx"+throwable.getMessage())
.doOnSubscribe(subscription -> connectedDeliveryCount.incrementAndGet())
.doOnTerminate(() -> connectedDeliveryCount.decrementAndGet())
;
}

Answer 1

如果消费者关闭浏览器或以某种方式断开与SSE流的连接，我希望得到通知。

在套接字上发送内容时，如果客户端已断开连接，则会出现套接字关闭错误。根据我的经验，只有当你尝试使用TCP/IP套接字时，你才会发现这一点。

因此，解决方案是发送定期的"；保持活力"；消息，即使您没有任何信息要发送给它们。例如，每10秒发送一个时间戳或服务器状态字符串。当这触发错误时，您知道客户端在最后10秒内的某个时刻断开了连接。

对不起，我不知道rx-java，但我猜你会在doOnTerminate处理程序中得到错误。