Open History to see the last commit
1import { expect, it } from "@effect/vitest";
2import * as Effect from "effect/Effect";
3import * as Stream from "effect/Stream";
4
5import { responsePayloads } from "./mcpResponsePayloads.ts";
6
7it.effect("reads SSE line endings across byte-chunk boundaries without splitting CRLF", () =>
8 Effect.gen(function* () {
9 for (const separator of [
10 "\n\n",
11 "\r\r",
12 "\r\n\r\n",
13 "\r\n\n",
14 "\n\r",
15 "\n\r\n",
16 "\r\r\n",
17 "\r\n\r",
18 ]) {
19 for (const dataLineSeparator of ["\n", "\r", "\r\n"]) {
20 const bytes = new TextEncoder().encode(
21 `: keep-alive${separator}data: {"id":1,${dataLineSeparator}data: "result":{"text":"café😀"}}${separator}` +
22 'data: {"id":2,"result":{}}\n\ndata: {"id":3}',
23 );
24 for (let cut = 0; cut <= bytes.length; cut += 1) {
25 const response = new Response(
26 new ReadableStream<Uint8Array>({
27 start(controller) {
28 controller.enqueue(bytes.subarray(0, cut));
29 controller.enqueue(bytes.subarray(cut));
30 controller.close();
31 },
32 }),
33 { headers: { "content-type": "text/event-stream" } },
34 );
35 const payloads = yield* Stream.runCollect(
36 responsePayloads(response, (cause) => ({
37 _tag: "ResponsePayloadError" as const,
38 cause,
39 separator,
40 dataLineSeparator,
41 cut,
42 })),
43 );
44 expect(
45 Array.from(payloads),
46 JSON.stringify({ separator, dataLineSeparator, cut }),
47 ).toEqual([
48 { id: 1, result: { text: "café😀" } },
49 { id: 2, result: {} },
50 ]);
51 }
52 }
53 const response = new Response(
54 new ReadableStream<Uint8Array>({
55 start(controller) {
56 controller.enqueue(new TextEncoder().encode(`data: {"id":4}${separator}`));
57 },
58 }),
59 { headers: { "content-type": "text/event-stream" } },
60 );
61 const payloads = yield* Stream.runCollect(
62 responsePayloads(response, (cause) => ({
63 _tag: "ResponsePayloadError" as const,
64 cause,
65 separator,
66 })).pipe(Stream.take(1)),
67 );
68 expect(Array.from(payloads)).toEqual([{ id: 4 }]);
69 }
70 }),
71);
72
73it.effect("picks the SSE reader from the media type alone, whatever its case", () =>
74 Effect.gen(function* () {
75 const response = new Response('data: {"id":1}\n\n', {
76 headers: { "content-type": "Text/Event-Stream; charset=utf-8" },
77 });
78 const payloads = yield* Stream.runCollect(
79 responsePayloads(response, (cause) => ({ _tag: "ResponsePayloadError" as const, cause })),
80 );
81 expect(Array.from(payloads)).toEqual([{ id: 1 }]);
82
83 const json = new Response('{"jsonrpc":"2.0","id":1,"result":{}}', {
84 headers: {
85 "content-type": 'application/json; profile="https://example.com/Text/Event-Stream"',
86 },
87 });
88 const jsonPayloads = yield* Stream.runCollect(
89 responsePayloads(json, (cause) => ({ _tag: "ResponsePayloadError" as const, cause })),
90 );
91 expect(Array.from(jsonPayloads)).toEqual([{ jsonrpc: "2.0", id: 1, result: {} }]);
92 }),
93);