From da5c5c1e0917d4763c113d179de51075f95e59f3 Mon Sep 17 00:00:00 2001 From: weicw1996 Date: Fri, 23 Oct 2026 12:57:03 +0800 Subject: [PATCH] =?UTF-8?q?fix(ai):=20SSE=20=E5=B8=A7=E4=BC=98=E5=85=88?= =?UTF-8?q?=E6=8A=95=E7=BB=99=E5=9C=A8=E9=80=94=E6=B5=81=E5=BC=8F=E8=AF=B7?= =?UTF-8?q?=E6=B1=82=EF=BC=8C=E4=BF=AE=E3=80=8C=E7=94=9F=E6=88=90=E5=90=8E?= =?UTF-8?q?=E6=8C=89=E9=92=AE=E6=B0=B8=E4=B9=85=E4=B8=8D=E5=8F=AF=E7=82=B9?= =?UTF-8?q?=E3=80=8D?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit WebSocketServer.send 原本 WS 优先：只要同一账号还有 WS 会话（问答助手页会建 连接），文稿工作台那条 /stream 的帧就全被投到 WS 上，SSE 一帧收不到——正文只能 等末帧，收尾的 __END__ 永远不来，前端 loading 不复位，生成按钮卡死。 改为 SSE 优先：ChatStreamRegistry 是按请求登记的，登记期间发出的帧都属于那条 请求；没有在途请求时 emit 返回 false，照旧走 WS。握手回执「连接成功」改由 register() 直接写给该会话，避免被派发到别的请求。 测试：补两条回归（有在途 SSE 时即使 WS 在线也投 SSE；没有在途 SSE 时仍走 WS）， ai 包 158 条全绿。 --- .../gxwebsoft/ai/config/WebSocketServer.java | 22 +++++-- .../ai/config/ChatStreamRegistryTest.java | 66 +++++++++++++++++++ 2 files changed, 82 insertions(+), 6 deletions(-) diff --git a/src/main/java/com/gxwebsoft/ai/config/WebSocketServer.java b/src/main/java/com/gxwebsoft/ai/config/WebSocketServer.java index 3293edc..934c7c0 100644 --- a/src/main/java/com/gxwebsoft/ai/config/WebSocketServer.java +++ b/src/main/java/com/gxwebsoft/ai/config/WebSocketServer.java @@ -58,7 +58,9 @@ public class WebSocketServer { logger.info("websocket connected, userId={}", userId); try { - send(userId, "连接成功"); + // 握手回应属于这条连接本身，不能走 {@link #send}：那条路会优先投给 + // 在途的 HTTP 流，可能把「连接成功」送到别的请求上，端上就等不到握手回执。 + session.getBasicRemote().sendText("连接成功"); } catch (IOException e) { logger.warn("websocket 握手回应发送失败, userId={}", userId, e); } @@ -124,17 +126,25 @@ public class WebSocketServer { /** * 向指定用户推送一条文本；用户不在线时静默丢弃。 * - *

优先走 WebSocket；没有截活会话时退到 HTTP 流式通道 - * （{@link ChatStreamRegistry}）。生产网关在 {@code /api} 前丢 Upgrade 头， - * WS 握手根本进不来，端上因此一直在用 HTTP 流那条路。 + *

先投在途的 HTTP 流通道（SSE），没有才退到 WebSocket。 + * + *

HTTP 流通道是按请求登记的：登记期间发出的每一帧，都属于那条正在跑的 + * {@code /stream} 请求。反过来按「这个用户有没有 WS 会话」优先就会出事——同一个账号 + * 在问答助手开着（WS 会话在 {@code SESSIONS} 里）时，再到文稿工作台生成，文稿页 + * 那条 SSE 一帧都收不到：正文只能等末帧，收尾的 {@code __END__} 永远不来， + * 生成按钮就卡在加载态、点不动（见 issue 20）。 + * + *

没有在途请求时 {@link ChatStreamRegistry#emit} 返回 {@code false}，照旧走 WS。 + * 生产网关在 {@code /api} 前丢 Upgrade 头、WS 握手进不来，端上用的正是 SSE 那条路。 */ static void send(String userId, String message) throws IOException { + if (ChatStreamRegistry.emit(userId, message)) { + return; + } Session target = SESSIONS.get(userId); if (target != null) { target.getBasicRemote().sendText(message); - return; } - ChatStreamRegistry.emit(userId, message); } diff --git a/src/test/java/com/gxwebsoft/ai/config/ChatStreamRegistryTest.java b/src/test/java/com/gxwebsoft/ai/config/ChatStreamRegistryTest.java index ad9e17f..9b132c7 100644 --- a/src/test/java/com/gxwebsoft/ai/config/ChatStreamRegistryTest.java +++ b/src/test/java/com/gxwebsoft/ai/config/ChatStreamRegistryTest.java @@ -2,7 +2,10 @@ package com.gxwebsoft.ai.config; import org.junit.jupiter.api.Test; +import javax.websocket.RemoteEndpoint; +import javax.websocket.Session; import java.io.IOException; +import java.lang.reflect.Proxy; import java.util.ArrayList; import java.util.Collections; import java.util.List; @@ -97,4 +100,67 @@ class ChatStreamRegistryTest { assertEquals(Collections.singletonList("{\"answer\":\"fallback\"}"), received); } + + /** + * 回归（issue 20）：同一账号开着问答助手（WS 会话在线）时，文稿工作台的 + * {@code /stream} 请求也必须收到自己的帧。原先按「先看 WS 会话」派发， + * 这些帧全被投到 WS 上，SSE 拿不到收尾的 {@code __END__}，生成按钮卡在加载态点不动。 + */ + @Test + void 有在途流式通道时即使WebSocket在线也投给流式通道() throws IOException { + String user = newUser(); + List received = new ArrayList<>(); + List wsOutbox = new ArrayList<>(); + ChatStreamRegistry.bind(user, received::add); + + Session session = fakeSession(wsOutbox); + WebSocketServer.register(session, user); + try { + WebSocketServer.send(user, "{\"answer\":\"stream\"}"); + } finally { + WebSocketServer.unregister(user, session); + } + + assertEquals(Collections.singletonList("{\"answer\":\"stream\"}"), received, + "在途的流式请求必须收到自己的帧，否则 __END__ 永远来不了"); + assertEquals(Collections.singletonList("连接成功"), wsOutbox, + "握手回执归这条连接，业务帧不该再派给 WS"); + } + + /** 没有在途请求时照旧走 WebSocket，别为了修 SSE 把 WS 通道弄哑。 */ + @Test + void 没有在途流式通道时仍走WebSocket() throws IOException { + String user = newUser(); + List wsOutbox = new ArrayList<>(); + Session session = fakeSession(wsOutbox); + WebSocketServer.register(session, user); + try { + WebSocketServer.send(user, "{\"answer\":\"socket\"}"); + } finally { + WebSocketServer.unregister(user, session); + } + + assertEquals( + java.util.Arrays.asList("连接成功", "{\"answer\":\"socket\"}"), + wsOutbox, + "没有在途流式请求时，帧必须照旧走 WS"); + } + + /** 只关心 sendText 的去向，其余方法一律返回 null。 */ + private static Session fakeSession(List outbox) { + return (Session) Proxy.newProxyInstance(Session.class.getClassLoader(), + new Class[]{Session.class}, (proxy, method, args) -> { + if ("getBasicRemote".equals(method.getName())) { + return Proxy.newProxyInstance(RemoteEndpoint.Basic.class.getClassLoader(), + new Class[]{RemoteEndpoint.Basic.class}, + (endpoint, call, callArgs) -> { + if ("sendText".equals(call.getName())) { + outbox.add((String) callArgs[0]); + } + return null; + }); + } + return null; + }); + } }