如何使用SSE在RxJava2 Flowable上处理浏览器关闭案例动作/事件服务器发送事件

问题描述

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

作为示例，当连接以curl开始时，将调用doOnSubscribe操作，并且connectedDeliveryCount会增加，但不会减少，尽管我故意使用ctrl-c关闭了连接。没有一个事件像doOnComplete或doOnTerminate或doOnCancel

那样被调用

与get的示例连接

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())
                ;
    }

解决方法

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

在套接字上发送内容时，如果客户端断开连接，则会收到套接字关闭错误。以我的经验，直到尝试使用TCP / IP套接字，您才发现这一点。

因此，解决方案是沿套接字发送常规的“保持活动”消息，即使您没有任何要发送给他们的信息也是如此。例如，每10秒钟发送一个时间戳或服务器状态字符串。当这触发错误时，您知道客户端在最近10秒钟的某个时间断开了连接。

对不起，我不认识rx-java，但是我的猜测是您会在doOnTerminate处理程序中收到错误消息。