From 20c39ed540459ce4ca04bc69de6f1ef39d2f8c2b Mon Sep 17 00:00:00 2001
From: lihongjie0209 <lihongjie0209@gmail.com>
Date: Sun, 8 Jun 2025 10:23:07 +0800
Subject: [PATCH] =?UTF-8?q?fix(sse):=20=E4=BC=98=E5=8C=96=20SSE=E8=BF=9E?=
 =?UTF-8?q?=E6=8E=A5=E5=BC=82=E5=B8=B8=E5=A4=84=E7=90=86=E5=92=8C=E5=BF=83?=
 =?UTF-8?q?=E8=B7=B3=E6=9C=BA=E5=88=B6-=20=E5=9C=A8=20GlobalExceptionHandl?=
 =?UTF-8?q?er=20=E4=B8=AD=E5=A2=9E=E5=8A=A0=E5=AF=B9=20IOException=20?=
 =?UTF-8?q?=E7=9A=84=E5=A4=84=E7=90=86=EF=BC=8C=E5=BF=BD=E7=95=A5=E5=B8=B8?=
 =?UTF-8?q?=E8=A7=81=E7=9A=84=E5=AE=A2=E6=88=B7=E7=AB=AF=E6=96=AD=E5=BC=80?=
 =?UTF-8?q?=E5=BC=82=E5=B8=B8=20-=20=E5=9C=A8=20SseService=20=E4=B8=AD?=
 =?UTF-8?q?=E5=A2=9E=E5=8A=A0=E6=B3=A8=E5=86=8C=E6=97=B6=E7=9A=84=E6=97=A5?=
 =?UTF-8?q?=E5=BF=97=E8=AE=B0=E5=BD=95=EF=BC=8C=E8=B0=83=E6=95=B4=E5=BF=83?=
 =?UTF-8?q?=E8=B7=B3=E5=8F=91=E9=80=81=E9=A2=91=E7=8E=87=E4=B8=BA=2030?=
 =?UTF-8?q?=E7=A7=92=E4=B8=80=E6=AC=A1=20-=20=E4=BC=98=E5=8C=96=20SseEmitt?=
 =?UTF-8?q?er=20=E7=9A=84=E9=94=99=E8=AF=AF=E5=A4=84=E7=90=86=E5=92=8C?=
 =?UTF-8?q?=E8=B5=84=E6=BA=90=E9=87=8A=E6=94=BE?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 .../spring/config/GlobalExceptionHandler.java | 24 +++++++++++++++++++
 .../coal/sse/service/SseService.java          | 14 +++++------
 2 files changed, 31 insertions(+), 7 deletions(-)

diff --git a/src/main/java/cn/lihongjie/coal/spring/config/GlobalExceptionHandler.java b/src/main/java/cn/lihongjie/coal/spring/config/GlobalExceptionHandler.java
index fb4b5b2e..cd8dc93a 100644
--- a/src/main/java/cn/lihongjie/coal/spring/config/GlobalExceptionHandler.java
+++ b/src/main/java/cn/lihongjie/coal/spring/config/GlobalExceptionHandler.java
@@ -6,6 +6,7 @@ import cn.lihongjie.coal.exception.BizException;
 import jakarta.servlet.http.HttpServletRequest;
 import jakarta.validation.ConstraintViolation;
 
+import lombok.SneakyThrows;
 import lombok.extern.slf4j.Slf4j;
 
 import org.hibernate.SessionFactory;
@@ -21,6 +22,7 @@ import org.springframework.web.bind.annotation.ExceptionHandler;
 import org.springframework.web.bind.annotation.RestControllerAdvice;
 import org.springframework.web.method.HandlerMethod;
 
+import java.io.IOException;
 import java.util.Arrays;
 import java.util.List;
 import java.util.Map;
@@ -65,6 +67,28 @@ public class GlobalExceptionHandler {
         return R.fail(ex.getCode(), ex.getMessage());
     }
 
+    @SneakyThrows
+    @ExceptionHandler(IOException.class)
+    public void handleIOException(IOException ex, HttpServletRequest request) {
+        boolean isSse = "text/event-stream".equals(request.getHeader("Accept"));
+        String message = ex.getMessage();
+        // 忽略常见的客户端断开异常
+        if ( isSse && message != null && (
+                message.contains("Broken pipe") ||
+                message.contains("Connection reset by peer") ||
+                message.contains("An existing connection was forcibly closed") ||
+                message.contains("你的主机中的软件中止了一个已建立的连接")
+        )) {
+            // 可以选择只记录DEBUG日志，或者直接忽略
+            // log.debug("SSE client closed connection: {}", message);
+            log.info("SSE client closed connection: {}", message);
+            return;
+        }
+        // 如果不是上述异常，可以选择抛出或做其他处理
+        // log.error("IOException not ignored", ex);
+        throw ex; // 或返回错误响应
+    }
+
     private void logEx(Exception ex, HttpServletRequest request, HandlerMethod handlerMethod) {
         if (request.getAttribute("__logged") == null) {
 
diff --git a/src/main/java/cn/lihongjie/coal/sse/service/SseService.java b/src/main/java/cn/lihongjie/coal/sse/service/SseService.java
index 0292d012..26c75da1 100644
--- a/src/main/java/cn/lihongjie/coal/sse/service/SseService.java
+++ b/src/main/java/cn/lihongjie/coal/sse/service/SseService.java
@@ -29,6 +29,8 @@ public class SseService {
 
     public void register(String topicId, SseEmitter emitter) {
 
+        log.info("{} register sse emitter:{}", topicId, emitter);
+
         ScheduledFuture<?> scheduledFuture = taskScheduler.scheduleAtFixedRate(
                 () -> {
                     try {
@@ -42,16 +44,11 @@ public class SseService {
                     } catch (IOException e) {
                         log.error("{} send heartbeat error:{}", topicId, e.getMessage());
 
-                        try{
 
-                            emitter.complete();
-                        }catch (Exception exception){
-                            log.error("{} emitter complete error:{}", topicId, exception.getMessage());
-                        }
 
                     }
                 },
-                Duration.ofMinutes(1));
+                Duration.ofSeconds(30));
 
         RTopic topic = redissonClient.getTopic(topicId);
         int addListener =
@@ -83,12 +80,15 @@ public class SseService {
         emitter.onError(
                 (e) -> {
                     topic.removeListener(addListener);
-                    log.info("{} remove listener:{} onError ", topicId, addListener, e);
+                    log.info("{} remove listener:{} onError {}", topicId, addListener, e.getMessage());
 
                     scheduledFuture.cancel(true);
+
                 });
 
 
+
+
     }
 
     @SneakyThrows