fix(ai): SSE 帧优先投给在途流式请求,修「生成后按钮永久不可点」
WebSocketServer.send 原本 WS 优先:只要同一账号还有 WS 会话(问答助手页会建 连接),文稿工作台那条 /stream 的帧就全被投到 WS 上,SSE 一帧收不到——正文只能 等末帧,收尾的 __END__ 永远不来,前端 loading 不复位,生成按钮卡死。 改为 SSE 优先:ChatStreamRegistry 是按请求登记的,登记期间发出的帧都属于那条 请求;没有在途请求时 emit 返回 false,照旧走 WS。握手回执「连接成功」改由 register() 直接写给该会话,避免被派发到别的请求。 测试:补两条回归(有在途 SSE 时即使 WS 在线也投 SSE;没有在途 SSE 时仍走 WS), ai 包 158 条全绿。
This commit is contained in:
@@ -58,7 +58,9 @@ public class WebSocketServer {
|
|||||||
logger.info("websocket connected, userId={}", userId);
|
logger.info("websocket connected, userId={}", userId);
|
||||||
|
|
||||||
try {
|
try {
|
||||||
send(userId, "连接成功");
|
// 握手回应属于这条连接本身,不能走 {@link #send}:那条路会优先投给
|
||||||
|
// 在途的 HTTP 流,可能把「连接成功」送到别的请求上,端上就等不到握手回执。
|
||||||
|
session.getBasicRemote().sendText("连接成功");
|
||||||
} catch (IOException e) {
|
} catch (IOException e) {
|
||||||
logger.warn("websocket 握手回应发送失败, userId={}", userId, e);
|
logger.warn("websocket 握手回应发送失败, userId={}", userId, e);
|
||||||
}
|
}
|
||||||
@@ -124,17 +126,25 @@ public class WebSocketServer {
|
|||||||
/**
|
/**
|
||||||
* 向指定用户推送一条文本;用户不在线时静默丢弃。
|
* 向指定用户推送一条文本;用户不在线时静默丢弃。
|
||||||
*
|
*
|
||||||
* <p>优先走 WebSocket;没有截活会话时退到 HTTP 流式通道
|
* <p><b>先投在途的 HTTP 流通道(SSE),没有才退到 WebSocket。</b>
|
||||||
* ({@link ChatStreamRegistry})。生产网关在 {@code /api} 前丢 Upgrade 头,
|
*
|
||||||
* WS 握手根本进不来,端上因此一直在用 HTTP 流那条路。
|
* <p>HTTP 流通道是按请求登记的:登记期间发出的每一帧,都属于那条正在跑的
|
||||||
|
* {@code /stream} 请求。反过来按「这个用户有没有 WS 会话」优先就会出事——同一个账号
|
||||||
|
* 在问答助手开着(WS 会话在 {@code SESSIONS} 里)时,再到文稿工作台生成,文稿页
|
||||||
|
* 那条 SSE 一帧都收不到:正文只能等末帧,收尾的 {@code __END__} 永远不来,
|
||||||
|
* 生成按钮就卡在加载态、点不动(见 issue 20)。
|
||||||
|
*
|
||||||
|
* <p>没有在途请求时 {@link ChatStreamRegistry#emit} 返回 {@code false},照旧走 WS。
|
||||||
|
* 生产网关在 {@code /api} 前丢 Upgrade 头、WS 握手进不来,端上用的正是 SSE 那条路。
|
||||||
*/
|
*/
|
||||||
static void send(String userId, String message) throws IOException {
|
static void send(String userId, String message) throws IOException {
|
||||||
|
if (ChatStreamRegistry.emit(userId, message)) {
|
||||||
|
return;
|
||||||
|
}
|
||||||
Session target = SESSIONS.get(userId);
|
Session target = SESSIONS.get(userId);
|
||||||
if (target != null) {
|
if (target != null) {
|
||||||
target.getBasicRemote().sendText(message);
|
target.getBasicRemote().sendText(message);
|
||||||
return;
|
|
||||||
}
|
}
|
||||||
ChatStreamRegistry.emit(userId, message);
|
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
@@ -2,7 +2,10 @@ package com.gxwebsoft.ai.config;
|
|||||||
|
|
||||||
import org.junit.jupiter.api.Test;
|
import org.junit.jupiter.api.Test;
|
||||||
|
|
||||||
|
import javax.websocket.RemoteEndpoint;
|
||||||
|
import javax.websocket.Session;
|
||||||
import java.io.IOException;
|
import java.io.IOException;
|
||||||
|
import java.lang.reflect.Proxy;
|
||||||
import java.util.ArrayList;
|
import java.util.ArrayList;
|
||||||
import java.util.Collections;
|
import java.util.Collections;
|
||||||
import java.util.List;
|
import java.util.List;
|
||||||
@@ -97,4 +100,67 @@ class ChatStreamRegistryTest {
|
|||||||
|
|
||||||
assertEquals(Collections.singletonList("{\"answer\":\"fallback\"}"), received);
|
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<String> received = new ArrayList<>();
|
||||||
|
List<String> 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<String> 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<String> 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;
|
||||||
|
});
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user