diff --git a/src/main/java/com/knowledge/base/infrastructure/util/http/FilteredSseOutputAdapter.java b/src/main/java/com/knowledge/base/infrastructure/util/http/FilteredSseOutputAdapter.java index f3e84fb..59726d9 100644 --- a/src/main/java/com/knowledge/base/infrastructure/util/http/FilteredSseOutputAdapter.java +++ b/src/main/java/com/knowledge/base/infrastructure/util/http/FilteredSseOutputAdapter.java @@ -24,8 +24,8 @@ public class FilteredSseOutputAdapter implements WriterAdapter { @Override public void writeLine(String line) throws IOException { - if (StrUtil.isBlank(line)) { - return; + if (StrUtil.isBlank(line) || closed) { + return; // 如果已关闭,不再处理 } if (line.startsWith("data:")) { @@ -36,42 +36,38 @@ public class FilteredSseOutputAdapter implements WriterAdapter { Map original = mapper.readValue(line, Map.class); Map filtered = new HashMap<>(); - // textResponse 累积 if (original.containsKey("textResponse")) { filtered.put("textResponse", original.get("textResponse")); lastChunk.put("textResponse", original.get("textResponse")); } - // sources 只保存不立刻发 if (original.containsKey("sources")) { lastChunk.put("sources", original.get("sources")); } - // close 也缓存并判断是否最后一条 if (original.containsKey("close")) { lastChunk.put("close", original.get("close")); } if (Boolean.TRUE.equals(original.get("close"))) { - // 合并缓存数据并一次性输出 filtered.putAll(lastChunk); lastChunk.clear(); String payload = mapper.writeValueAsString(filtered); emitter.send(SseEmitter.event().data(payload)); - - if (!closed) { - closed = true; - emitter.complete(); - } + closed = true; + emitter.complete(); } else if (!filtered.isEmpty()) { - // 中间段 String payload = mapper.writeValueAsString(filtered); emitter.send(SseEmitter.event().data(payload)); } + } catch (IllegalStateException e) { + log.warn("SSE连接已关闭,忽略发送: {}", e.getMessage()); + closed = true; } catch (Exception e) { log.warn("SSE数据解析失败:{}", line, e); } } + }