From 024743c01447e6fb3f2314e62c9725d87d9a845f Mon Sep 17 00:00:00 2001 From: mateaix <7333791@qq.com> Date: Wed, 16 Sep 2026 21:50:25 +0800 Subject: [PATCH] fix(stream): handle SSE line endings and disable proxy buffering (#636) --- .../vip/mate/channel/web/Utf8SseEmitter.java | 4 ++ .../mate/channel/web/Utf8SseEmitterTest.java | 10 ++++ .../chat/__tests__/useStreamFraming.test.ts | 59 +++++++++++++++++++ mateclaw-ui/src/composables/chat/useStream.ts | 33 +++++++---- 4 files changed, 94 insertions(+), 12 deletions(-) create mode 100644 mateclaw-ui/src/composables/chat/__tests__/useStreamFraming.test.ts diff --git a/mateclaw-server/src/main/java/vip/mate/channel/web/Utf8SseEmitter.java b/mateclaw-server/src/main/java/vip/mate/channel/web/Utf8SseEmitter.java index f3196997..85d0d85d 100644 --- a/mateclaw-server/src/main/java/vip/mate/channel/web/Utf8SseEmitter.java +++ b/mateclaw-server/src/main/java/vip/mate/channel/web/Utf8SseEmitter.java @@ -45,6 +45,10 @@ public class Utf8SseEmitter extends SseEmitter { protected void extendResponse(ServerHttpResponse response) { super.extendResponse(response); HttpHeaders headers = response.getHeaders(); + // Streaming frames must reach the client as they are emitted. Nginx + // honors this response header unless explicitly configured to ignore it. + headers.set("X-Accel-Buffering", "no"); + headers.setCacheControl("no-store, no-transform"); // Spring's default sets Content-Type=text/event-stream without charset. // Only override when no charset is already specified, so callers that // want to roll their own (rare) keep working. diff --git a/mateclaw-server/src/test/java/vip/mate/channel/web/Utf8SseEmitterTest.java b/mateclaw-server/src/test/java/vip/mate/channel/web/Utf8SseEmitterTest.java index cbfac003..54daeab5 100644 --- a/mateclaw-server/src/test/java/vip/mate/channel/web/Utf8SseEmitterTest.java +++ b/mateclaw-server/src/test/java/vip/mate/channel/web/Utf8SseEmitterTest.java @@ -22,6 +22,16 @@ import static org.junit.jupiter.api.Assertions.*; */ class Utf8SseEmitterTest { + @Test + void disablesProxyBufferingAndCachingWhileKeepingUtf8() { + var response = new ServletServerHttpResponse(new MockHttpServletResponse()); + new Utf8SseEmitter().extendResponse(response); + + assertEquals("no", response.getHeaders().getFirst("X-Accel-Buffering")); + assertEquals("no-store, no-transform", response.getHeaders().getCacheControl()); + assertEquals(StandardCharsets.UTF_8, response.getHeaders().getContentType().getCharset()); + } + @Test @DisplayName("extendResponse stamps charset=UTF-8 when Content-Type is unset") void stampsUtf8WhenContentTypeUnset() throws Exception { diff --git a/mateclaw-ui/src/composables/chat/__tests__/useStreamFraming.test.ts b/mateclaw-ui/src/composables/chat/__tests__/useStreamFraming.test.ts new file mode 100644 index 00000000..a5ea6127 --- /dev/null +++ b/mateclaw-ui/src/composables/chat/__tests__/useStreamFraming.test.ts @@ -0,0 +1,59 @@ +import { afterEach, describe, expect, it, vi } from 'vitest' +import { useStream } from '../useStream' + +afterEach(() => vi.unstubAllGlobals()) + +describe('incremental SSE framing', () => { + it.each(['\n', '\r\n', '\r'])('dispatches %j frames before the connection closes', async (newline) => { + let controller!: ReadableStreamDefaultController + const body = new ReadableStream({ start(c) { controller = c } }) + vi.stubGlobal('fetch', vi.fn().mockResolvedValue(new Response(body))) + const stream = useStream({ url: '/test' }) + const received: string[] = [] + stream.on('content_delta', data => received.push(data.delta)) + const connected = stream.connect() + const encoder = new TextEncoder() + try { + for (const delta of ['first', 'second', 'third']) { + const frame = `event: content_delta${newline}data: ${JSON.stringify({ delta })}${newline}${newline}` + // Every possible CRLF pair is split across network reads. + for (const char of frame) controller.enqueue(encoder.encode(char)) + await vi.waitFor(() => expect(received.at(-1)).toBe(delta), { timeout: 300 }) + } + expect(received).toEqual(['first', 'second', 'third']) + } finally { + controller.close() + await connected + } + }) + + it('joins multiple data lines and ignores comment-only frames', async () => { + const payload = ': heartbeat\r\n\r\nevent: content_delta\r\ndata: {"delta":\r\ndata: "你好"}\r\n\r\n' + vi.stubGlobal('fetch', vi.fn().mockResolvedValue(new Response(payload))) + const stream = useStream({ url: '/test' }) + const received: unknown[] = [] + stream.onEvent(event => received.push(event.data)) + await stream.connect() + expect(received).toEqual([{ delta: '你好' }]) + }) + + it('keeps streaming when line endings change after the first event', async () => { + let controller!: ReadableStreamDefaultController + const body = new ReadableStream({ start(c) { controller = c } }) + vi.stubGlobal('fetch', vi.fn().mockResolvedValue(new Response(body))) + const stream = useStream({ url: '/test' }) + const received: string[] = [] + stream.on('content_delta', data => received.push(data.delta)) + const connected = stream.connect() + const encoder = new TextEncoder() + try { + controller.enqueue(encoder.encode('data: {"delta":"first"}\n\n')) + await vi.waitFor(() => expect(received).toEqual(['first'])) + controller.enqueue(encoder.encode('data: {"delta":"second"}\r\n\r\n')) + await vi.waitFor(() => expect(received).toEqual(['first', 'second'])) + } finally { + controller.close() + await connected + } + }) +}) diff --git a/mateclaw-ui/src/composables/chat/useStream.ts b/mateclaw-ui/src/composables/chat/useStream.ts index 66436366..a9b3154b 100644 --- a/mateclaw-ui/src/composables/chat/useStream.ts +++ b/mateclaw-ui/src/composables/chat/useStream.ts @@ -147,9 +147,16 @@ export interface UseStreamReturn { class SSEParser { private buffer = '' private readonly separator = '\n\n' + private trailingCR = false parse(chunk: string): SSEEvent[] { - this.buffer += chunk + // SSE permits LF, CRLF and CR. Normalize before framing, including a + // CRLF pair split across network reads, without manufacturing blank lines. + if (chunk.length > 0) { + const skipLF = this.trailingCR && chunk.startsWith('\n') + this.trailingCR = chunk.endsWith('\r') + this.buffer += (skipLF ? chunk.slice(1) : chunk).replace(/\r\n?/g, '\n') + } const events: SSEEvent[] = [] // 分割事件块 @@ -179,8 +186,7 @@ class SSEParser { private parseEvent(part: string): SSEEvent | null { const lines = part.split('\n') let eventType: SSEEventType = 'content_delta' - let data: any = {} - let hasData = false + const dataLines: string[] = [] let eventId: string | undefined for (const line of lines) { @@ -189,24 +195,27 @@ class SSEParser { const colonIndex = line.indexOf(':') if (colonIndex === -1) continue - const key = line.slice(0, colonIndex).trim() - const value = line.slice(colonIndex + 1).trim() + const key = line.slice(0, colonIndex) + const rawValue = line.slice(colonIndex + 1) + const value = rawValue.startsWith(' ') ? rawValue.slice(1) : rawValue if (key === 'event') { eventType = value as SSEEventType } else if (key === 'id') { eventId = value } else if (key === 'data') { - hasData = true - try { - data = JSON.parse(value) - } catch { - data = value - } + dataLines.push(value) } } - if (!hasData) return null + if (dataLines.length === 0) return null + const rawData = dataLines.join('\n') + let data: any + try { + data = JSON.parse(rawData) + } catch { + data = rawData + } return eventId !== undefined ? { type: eventType, data, id: eventId } : { type: eventType, data }