diff --git a/libs/sdk-js/src/client.ts b/libs/sdk-js/src/client.ts index b2f1efc1d..ae68bc180 100644 --- a/libs/sdk-js/src/client.ts +++ b/libs/sdk-js/src/client.ts @@ -1014,8 +1014,8 @@ export class RunsClient< const stream: ReadableStream<{ event: any; data: any }> = ( response.body || new ReadableStream({ start: (ctrl) => ctrl.close() }) ) - .pipeThrough(new BytesLineDecoder()) - .pipeThrough(new SSEDecoder()); + .pipeThrough(BytesLineDecoder()) + .pipeThrough(SSEDecoder()); yield* IterableReadableStream.fromReadableStream(stream); } @@ -1318,8 +1318,8 @@ export class RunsClient< const stream: ReadableStream<{ event: string; data: any }> = ( response.body || new ReadableStream({ start: (ctrl) => ctrl.close() }) ) - .pipeThrough(new BytesLineDecoder()) - .pipeThrough(new SSEDecoder()); + .pipeThrough(BytesLineDecoder()) + .pipeThrough(SSEDecoder()); yield* IterableReadableStream.fromReadableStream(stream); } diff --git a/libs/sdk-js/src/tests/sse.test.ts b/libs/sdk-js/src/tests/sse.test.ts index e22f1e0d7..1dd6a6d94 100644 --- a/libs/sdk-js/src/tests/sse.test.ts +++ b/libs/sdk-js/src/tests/sse.test.ts @@ -20,7 +20,7 @@ describe("BytesLineDecoder", () => { test("handles single line with newline", async () => { const input = createStream([textEncoder.encode("hello\n")]); - const decoded = input.pipeThrough(new BytesLineDecoder()); + const decoded = input.pipeThrough(BytesLineDecoder()); const results = await gather(decoded); expect(results.length).toBe(1); @@ -29,7 +29,7 @@ describe("BytesLineDecoder", () => { test("handles multiple lines", async () => { const input = createStream([textEncoder.encode("line1\nline2\nline3\n")]); - const decoded = input.pipeThrough(new BytesLineDecoder()); + const decoded = input.pipeThrough(BytesLineDecoder()); const results = await gather(decoded); expect(results.length).toBe(3); @@ -44,7 +44,7 @@ describe("BytesLineDecoder", () => { textEncoder.encode("ne1\nli"), textEncoder.encode("ne2\n"), ]); - const decoded = input.pipeThrough(new BytesLineDecoder()); + const decoded = input.pipeThrough(BytesLineDecoder()); const results = await gather(decoded); expect(results.length).toBe(2); @@ -54,7 +54,7 @@ describe("BytesLineDecoder", () => { test("handles CR LF line endings", async () => { const input = createStream([textEncoder.encode("line1\r\nline2\r\n")]); - const decoded = input.pipeThrough(new BytesLineDecoder()); + const decoded = input.pipeThrough(BytesLineDecoder()); const results = await gather(decoded); expect(results.length).toBe(2); @@ -67,7 +67,7 @@ describe("BytesLineDecoder", () => { textEncoder.encode("line1\r"), textEncoder.encode("\nline2\r\n"), ]); - const decoded = input.pipeThrough(new BytesLineDecoder()); + const decoded = input.pipeThrough(BytesLineDecoder()); const results = await gather(decoded); expect(results.length).toBe(2); @@ -77,7 +77,7 @@ describe("BytesLineDecoder", () => { test("handles stale line", async () => { const input = createStream([textEncoder.encode("hello")]); - const decoded = input.pipeThrough(new BytesLineDecoder()); + const decoded = input.pipeThrough(BytesLineDecoder()); const results = await gather(decoded); expect(results.length).toBe(1); @@ -99,8 +99,8 @@ describe("SSEDecoder", () => { "\n", ]); const decoded = input - .pipeThrough(new BytesLineDecoder()) - .pipeThrough(new SSEDecoder()); + .pipeThrough(BytesLineDecoder()) + .pipeThrough(SSEDecoder()); const results = await gather(decoded); expect(results.length).toBe(1); @@ -117,8 +117,8 @@ describe("SSEDecoder", () => { 'data: {"message": "hello"}\n', ]); const decoded = input - .pipeThrough(new BytesLineDecoder()) - .pipeThrough(new SSEDecoder()); + .pipeThrough(BytesLineDecoder()) + .pipeThrough(SSEDecoder()); const results = await gather(decoded); expect(results.length).toBe(1); @@ -138,8 +138,8 @@ describe("SSEDecoder", () => { "\n", ]); const decoded = input - .pipeThrough(new BytesLineDecoder()) - .pipeThrough(new SSEDecoder()); + .pipeThrough(BytesLineDecoder()) + .pipeThrough(SSEDecoder()); const results = await gather(decoded); expect(results.length).toBe(2); @@ -156,8 +156,8 @@ describe("SSEDecoder", () => { test("end event without data", async () => { const input = createStream(["event: test\n"]); const decoded = input - .pipeThrough(new BytesLineDecoder()) - .pipeThrough(new SSEDecoder()); + .pipeThrough(BytesLineDecoder()) + .pipeThrough(SSEDecoder()); const results = await gather(decoded); expect(results.length).toBe(1); @@ -170,8 +170,8 @@ describe("SSEDecoder", () => { test("end event without newline", async () => { const input = createStream(["event: end"]); const decoded = input - .pipeThrough(new BytesLineDecoder()) - .pipeThrough(new SSEDecoder()); + .pipeThrough(BytesLineDecoder()) + .pipeThrough(SSEDecoder()); const results = await gather(decoded); expect(results.length).toBe(1); diff --git a/libs/sdk-js/src/utils/sse.ts b/libs/sdk-js/src/utils/sse.ts index d0482f7e5..a080922f5 100644 --- a/libs/sdk-js/src/utils/sse.ts +++ b/libs/sdk-js/src/utils/sse.ts @@ -6,90 +6,88 @@ const SPACE = " ".charCodeAt(0); const TRAILING_NEWLINE = [CR, LF]; -export class BytesLineDecoder extends TransformStream { - constructor() { - let buffer: Uint8Array[] = []; - let trailingCr = false; +export function BytesLineDecoder() { + let buffer: Uint8Array[] = []; + let trailingCr = false; - super({ - start() { - buffer = []; + return new TransformStream({ + start() { + buffer = []; + trailingCr = false; + }, + + transform(chunk, controller) { + // See https://docs.python.org/3/glossary.html#term-universal-newlines + let text = chunk; + + // Handle trailing CR from previous chunk + if (trailingCr) { + text = joinArrays([[CR], text]); trailingCr = false; - }, + } - transform(chunk, controller) { - // See https://docs.python.org/3/glossary.html#term-universal-newlines - let text = chunk; + // Check for trailing CR in current chunk + if (text.length > 0 && text.at(-1) === CR) { + trailingCr = true; + text = text.subarray(0, -1); + } - // Handle trailing CR from previous chunk - if (trailingCr) { - text = joinArrays([[CR], text]); - trailingCr = false; - } + if (!text.length) return; + const trailingNewline = TRAILING_NEWLINE.includes(text.at(-1)!); - // Check for trailing CR in current chunk - if (text.length > 0 && text.at(-1) === CR) { - trailingCr = true; - text = text.subarray(0, -1); - } + const lastIdx = text.length - 1; + const { lines } = text.reduce<{ lines: Uint8Array[]; from: number }>( + (acc, cur, idx) => { + if (acc.from > idx) return acc; - if (!text.length) return; - const trailingNewline = TRAILING_NEWLINE.includes(text.at(-1)!); - - const lastIdx = text.length - 1; - const { lines } = text.reduce<{ lines: Uint8Array[]; from: number }>( - (acc, cur, idx) => { - if (acc.from > idx) return acc; - - if (cur === CR || cur === LF) { - acc.lines.push(text.subarray(acc.from, idx)); - if (cur === CR && text[idx + 1] === LF) { - acc.from = idx + 2; - } else { - acc.from = idx + 1; - } + if (cur === CR || cur === LF) { + acc.lines.push(text.subarray(acc.from, idx)); + if (cur === CR && text[idx + 1] === LF) { + acc.from = idx + 2; + } else { + acc.from = idx + 1; } + } - if (idx === lastIdx && acc.from <= lastIdx) { - acc.lines.push(text.subarray(acc.from)); - } + if (idx === lastIdx && acc.from <= lastIdx) { + acc.lines.push(text.subarray(acc.from)); + } - return acc; - }, - { lines: [], from: 0 }, - ); + return acc; + }, + { lines: [], from: 0 }, + ); - if (lines.length === 1 && !trailingNewline) { - buffer.push(lines[0]); - return; - } + if (lines.length === 1 && !trailingNewline) { + buffer.push(lines[0]); + return; + } - if (buffer.length) { - // Include existing buffer in first line - buffer.push(lines[0]); - lines[0] = joinArrays(buffer); - buffer = []; - } + if (buffer.length) { + // Include existing buffer in first line + buffer.push(lines[0]); + lines[0] = joinArrays(buffer); + buffer = []; + } - if (!trailingNewline) { - // If the last segment is not newline terminated, - // buffer it for the next chunk - if (lines.length) buffer = [lines.pop()!]; - } + if (!trailingNewline) { + // If the last segment is not newline terminated, + // buffer it for the next chunk + if (lines.length) buffer = [lines.pop()!]; + } - // Enqueue complete lines - for (const line of lines) { - controller.enqueue(line); - } - }, + // Enqueue complete lines + for (const line of lines) { + controller.enqueue(line); + } + }, - flush(controller) { - if (buffer.length) { - controller.enqueue(joinArrays(buffer)); - } - }, - }); - } + flush(controller) { + if (buffer.length) { + controller.enqueue(joinArrays(buffer)); + } + }, + }); } interface StreamPart { @@ -98,69 +96,67 @@ interface StreamPart { data: unknown; } -export class SSEDecoder extends TransformStream { - constructor() { - let event = ""; - let data: Uint8Array[] = []; - let lastEventId = ""; - let retry: number | null = null; +export function SSEDecoder() { + let event = ""; + let data: Uint8Array[] = []; + let lastEventId = ""; + let retry: number | null = null; - const decoder = new TextDecoder(); + const decoder = new TextDecoder(); - super({ - transform(chunk, controller) { - // Handle empty line case - if (!chunk.length) { - if (!event && !data.length && !lastEventId && retry == null) return; + return new TransformStream({ + transform(chunk, controller) { + // Handle empty line case + if (!chunk.length) { + if (!event && !data.length && !lastEventId && retry == null) return; - const sse = { - id: lastEventId || undefined, - event, - data: data.length ? decodeArraysToJson(decoder, data) : null, - }; + const sse = { + id: lastEventId || undefined, + event, + data: data.length ? decodeArraysToJson(decoder, data) : null, + }; - // NOTE: as per the SSE spec, do not reset lastEventId - event = ""; - data = []; - retry = null; + // NOTE: as per the SSE spec, do not reset lastEventId + event = ""; + data = []; + retry = null; - controller.enqueue(sse); - return; - } + controller.enqueue(sse); + return; + } - // Ignore comments - if (chunk[0] === COLON) return; + // Ignore comments + if (chunk[0] === COLON) return; - const sepIdx = chunk.indexOf(COLON); - if (sepIdx === -1) return; + const sepIdx = chunk.indexOf(COLON); + if (sepIdx === -1) return; - const fieldName = decoder.decode(chunk.subarray(0, sepIdx)); - let value = chunk.subarray(sepIdx + 1); - if (value[0] === SPACE) value = value.subarray(1); + const fieldName = decoder.decode(chunk.subarray(0, sepIdx)); + let value = chunk.subarray(sepIdx + 1); + if (value[0] === SPACE) value = value.subarray(1); - if (fieldName === "event") { - event = decoder.decode(value); - } else if (fieldName === "data") { - data.push(value); - } else if (fieldName === "id") { - if (value.indexOf(NULL) === -1) lastEventId = decoder.decode(value); - } else if (fieldName === "retry") { - const retryNum = Number.parseInt(decoder.decode(value)); - if (!Number.isNaN(retryNum)) retry = retryNum; - } - }, + if (fieldName === "event") { + event = decoder.decode(value); + } else if (fieldName === "data") { + data.push(value); + } else if (fieldName === "id") { + if (value.indexOf(NULL) === -1) lastEventId = decoder.decode(value); + } else if (fieldName === "retry") { + const retryNum = Number.parseInt(decoder.decode(value)); + if (!Number.isNaN(retryNum)) retry = retryNum; + } + }, - flush(controller) { - if (event) { - controller.enqueue({ - id: lastEventId || undefined, - event, - data: data.length ? decodeArraysToJson(decoder, data) : null, - }); - } - }, - }); - } + flush(controller) { + if (event) { + controller.enqueue({ + id: lastEventId || undefined, + event, + data: data.length ? decodeArraysToJson(decoder, data) : null, + }); + } + }, + }); } function joinArrays(data: ArrayLike[]) {