lukeye/doc-parser-server
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix

Browse Source
This commit is contained in:
luke 2025-06-23 19:29:40 +08:00
parent 58a78d33c1
commit 56e988435f
1 changed files with 8 additions and 12 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

20
src/main/java/com/knowledge/base/infrastructure/util/http/FilteredSseOutputAdapter.java
View File

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