From 19fdd1324f67b13f9a3a9edf748983809661bdc3 Mon Sep 17 00:00:00 2001 From: Marve10s Date: Fri, 16 Oct 2026 04:22:27 +0300 Subject: [PATCH] fix: preserve multiline SSE events --- src/create-proxy.ts | 21 +++++-- test/create-proxy.test.ts | 113 ++++++++++++++++++++++++++++++++++++++ 2 files changed, 130 insertions(+), 4 deletions(-) create mode 100644 test/create-proxy.test.ts diff --git a/src/create-proxy.ts b/src/create-proxy.ts index e9b1790..e4f12c3 100644 --- a/src/create-proxy.ts +++ b/src/create-proxy.ts @@ -6,6 +6,7 @@ import type { HttpClientResponse } from 'effect/http/HttpClientResponse' import type { ManagedRuntime } from 'effect/ManagedRuntime' import * as Effect from 'effect/Effect' +import * as Sse from 'effect/encoding/Sse' import * as Fiber from 'effect/Fiber' import * as Schedule from 'effect/Schedule' import * as Stream from 'effect/Stream' @@ -171,21 +172,33 @@ export function createTanstackQueryOptionsProxy( const source = yield* response.stream.pipe( Stream.decodeText, - Stream.splitLines, - Stream.filter((str) => str.length > 0), Stream.share({ capacity: 'unbounded' }) ) const keepAliveStream = source.pipe( + Stream.splitLines, Stream.filter((data) => data.startsWith(keepAlive?.message ?? ':keep-alive') ), Stream.timeout(keepAlive?.timeout ?? '10 seconds') ) + let buffer: string[] = [] + const parser = Sse.makeParser((event) => { + if (event._tag === 'Event') buffer.push(event.data) + }) + const dataStream = yield* source.pipe( - Stream.filter((line) => line.startsWith('data:')), - Stream.map((line) => line.slice(5).trim()), + Stream.mapEffect((chunk) => + Effect.suspend(() => { + const error = parser.feed(chunk) + if (error !== undefined) return Effect.fail(error) + const parsed = buffer + buffer = [] + return Effect.succeed(parsed) + }) + ), + Stream.flattenIterable, Stream.filter((data) => data.length > 0), Stream.mapEffect((data) => Effect.orElseSucceed( diff --git a/test/create-proxy.test.ts b/test/create-proxy.test.ts new file mode 100644 index 0000000..9c6772d --- /dev/null +++ b/test/create-proxy.test.ts @@ -0,0 +1,113 @@ +import { expect, test } from 'bun:test' +import * as Context from 'effect/Context' +import * as Effect from 'effect/Effect' +import * as HttpApi from 'effect/http-api/HttpApi' +import * as HttpApiClient from 'effect/http-api/HttpApiClient' +import * as HttpApiEndpoint from 'effect/http-api/HttpApiEndpoint' +import * as HttpApiGroup from 'effect/http-api/HttpApiGroup' +import * as HttpApiSchema from 'effect/http-api/HttpApiSchema' +import * as HttpClient from 'effect/http/HttpClient' +import * as HttpClientResponse from 'effect/http/HttpClientResponse' +import * as Layer from 'effect/Layer' +import * as ManagedRuntime from 'effect/ManagedRuntime' +import * as Schema from 'effect/Schema' + +import { createTanstackQueryOptionsProxy } from '../src/index' + +const Api = HttpApi.make('SseTest').add( + HttpApiGroup.make('events').add( + HttpApiEndpoint.get('values', '/values', { + success: HttpApiSchema.StreamSse({ data: Schema.Unknown }), + }) + ) +) + +class ApiClient extends Context.Service< + ApiClient, + HttpApiClient.ForApi +>()('SseTestClient') {} + +test.each([ + { + name: 'single-line JSON events', + text: 'data: {"value":42}\n\ndata: {"value":43}\n\n', + fragmented: false, + expected: [{ value: 42 }, { value: 43 }], + }, + { + name: 'multiline JSON event', + text: 'data: {\ndata: "value":42\ndata: }\n\n', + fragmented: false, + expected: [{ value: 42 }], + }, + { + name: 'fragmented UTF-8, fields, CRLF and event boundaries', + text: 'data: {\r\ndata: "value":"café"\r\ndata: }\r\n\r\ndata: 43\r\n\r\n', + fragmented: true, + expected: [{ value: 'café' }, 43], + }, + { + name: 'raw strings, retry directives and comments', + text: ':keep-alive\nretry: 1000\n\ndata: hello\ndata: world\n\n:comment\nretry: 2000\ndata: {"value":42}\n\n', + fragmented: false, + expected: ['hello\nworld', { value: 42 }], + }, +])('$name', async ({ text, fragmented, expected }) => { + let cancelled = false + const bytes = new TextEncoder().encode(`${text}data: "complete"\n\n`) + const body = new ReadableStream({ + start(controller) { + if (fragmented) + for (const byte of bytes) controller.enqueue(Uint8Array.of(byte)) + else controller.enqueue(bytes) + }, + cancel() { + cancelled = true + }, + }) + const transport = HttpClient.make((request) => + Effect.succeed( + HttpClientResponse.fromWeb( + request, + new Response(body, { headers: { 'content-type': 'text/event-stream' } }) + ) + ) + ) + const client = HttpApiClient.makeWith(Api, { + httpClient: transport, + baseUrl: 'https://in-memory.invalid', + }) + const runtime = ManagedRuntime.make(Layer.effect(ApiClient, client)) + try { + const proxy = createTanstackQueryOptionsProxy(ApiClient, runtime) + const options = proxy.events.values.subscriptionOptions() + const data: unknown[] = [] + const errors: unknown[] = [] + const done = Promise.withResolvers() + const stop = options.subscriptionFn({ + onData: (value) => { + if (value === 'complete') done.resolve(value) + else data.push(value) + }, + onError: (error) => { + errors.push(error) + done.reject(error) + }, + }) + const timer = setTimeout( + () => done.reject(new Error('Subscription did not deliver all events')), + 2000 + ) + try { + await done.promise + } finally { + clearTimeout(timer) + await stop() + } + expect(errors).toEqual([]) + expect(cancelled).toBe(true) + expect(data).toEqual([...expected]) + } finally { + await runtime.dispose() + } +})