1import { assert, describe, it } from "@effect/vitest"
2import { Effect, Stream } from "effect"
3import * as Sse from "effect/encoding/Sse"
4import * as Schema from "effect/Schema"
5
6describe("Sse", () => {
7 it("parses mixed CRLF and LF line endings", () => {
8 const events: Array<Sse.AnyEvent> = []
9 const parser = Sse.makeParser((event) => events.push(event))
10
11 parser.feed("data: first\r\ndata: second\n\n")
12
13 assert.deepStrictEqual(events, [{
14 _tag: "Event",
15 event: "message",
16 id: undefined,
17 data: "first\nsecond"
18 }])
19 })
20
21 it("treats CRLF split across chunks as one line ending", () => {
22 const events: Array<Sse.AnyEvent> = []
23 const parser = Sse.makeParser((event) => events.push(event))
24
25 parser.feed("data: first\r")
26 parser.feed("\ndata: second\n\n")
27
28 assert.deepStrictEqual(events, [{
29 _tag: "Event",
30 event: "message",
31 id: undefined,
32 data: "first\nsecond"
33 }])
34 })
35
36 it("retains the last event ID for later events", () => {
37 const events: Array<Sse.AnyEvent> = []
38 const parser = Sse.makeParser((event) => events.push(event))
39
40 parser.feed("id: 1\ndata: first\n\ndata: second\n\n")
41
42 assert.deepStrictEqual(events, [
43 {
44 _tag: "Event",
45 id: "1",
46 event: "message",
47 data: "first"
48 },
49 {
50 _tag: "Event",
51 id: "1",
52 event: "message",
53 data: "second"
54 }
55 ])
56 })
57
58 it("roundtrips an SSE event with empty data", () => {
59 const input: Sse.Event = { _tag: "Event", event: "message", id: undefined, data: "" }
60 const events: Array<Sse.AnyEvent> = []
61 const parser = Sse.makeParser((event) => events.push(event))
62
63 parser.feed(Sse.encoder.write(input))
64
65 assert.deepStrictEqual(events, [input])
66 })
67
68 it("strips a UTF-8 BOM from the start of an SSE stream", () => {
69 const events: Array<Sse.AnyEvent> = []
70 const parser = Sse.makeParser((event) => events.push(event))
71
72 parser.feed("\uFEFFdata: ok\n\n")
73
74 assert.deepStrictEqual(events, [{
75 _tag: "Event",
76 event: "message",
77 id: undefined,
78 data: "ok"
79 }])
80 })
81
82 it("uses message for an empty SSE event type", () => {
83 const events: Array<Sse.AnyEvent> = []
84 const parser = Sse.makeParser((event) => events.push(event))
85
86 parser.feed("event:\ndata: ok\n\n")
87
88 assert.strictEqual(events.length, 1)
89 assert.strictEqual((events[0] as Sse.Event).event, "message")
90 })
91
92 it("ignores retry values that are not ASCII digits", () => {
93 const events: Array<Sse.AnyEvent> = []
94 for (const value of ["123x", "+123", "-123", "1.5"]) {
95 Sse.makeParser((event) => events.push(event)).feed(`retry: ${value}\n`)
96 }
97
98 assert.deepStrictEqual(events, [])
99 })
100
101 it("Event preserves string payloads", () => {
102 const decode = Schema.decodeUnknownSync(Sse.Event)
103 const encode = Schema.encodeSync(Sse.Event)
104
105 const event = decode({
106 _tag: "Event",
107 event: "message",
108 id: undefined,
109 data: "{\"type\":\"ok\"}"
110 })
111
112 assert.strictEqual(event.data, "{\"type\":\"ok\"}")
113 assert.deepStrictEqual(encode(event), {
114 _tag: "Event",
115 event: "message",
116 id: undefined,
117 data: "{\"type\":\"ok\"}"
118 })
119 })
120
121 it("EventEncoded consumers can explicitly decode json payloads", () => {
122 const EventWithJsonPayload = Schema.Struct({
123 ...Sse.EventEncoded.fields,
124 data: Schema.fromJsonString(Schema.Struct({
125 type: Schema.String
126 }))
127 })
128 const decode = Schema.decodeUnknownSync(EventWithJsonPayload)
129
130 const event = decode({
131 event: "message",
132 id: undefined,
133 data: "{\"type\":\"ok\"}"
134 })
135
136 assert.deepStrictEqual(event.data, { type: "ok" })
137 })
138
139 it.effect("decodeDataSchema decodes json payload from SSE stream", () =>
140 Effect.gen(function*() {
141 const events = yield* Stream.make(
142 "event: message\ndata: {\"type\":\"ok\"}\n\n"
143 ).pipe(
144 Stream.pipeThroughChannel(Sse.decodeDataSchema(Schema.Struct({ type: Schema.String }))),
145 Stream.runCollect
146 )
147
148 assert.deepStrictEqual([...events], [{
149 event: "message",
150 id: undefined,
151 data: {
152 type: "ok"
153 }
154 }])
155 }))
156
157 it.effect("fails when an unterminated line exceeds maxEventSize", () =>
158 Effect.gen(function*() {
159 const error = yield* Stream.make("12345").pipe(
160 Stream.pipeThroughChannel(Sse.decode({ maxEventSize: 4 })),
161 Stream.runCollect,
162 Effect.flip
163 )
164
165 assert.instanceOf(error, Sse.SseError)
166 assert.instanceOf(error.reason, Sse.EventTooLarge)
167 assert.strictEqual(error.reason.maxEventSize, 4)
168 }))
169
170 it.effect("fails when pending data exceeds maxEventSize", () =>
171 Effect.gen(function*() {
172 const error = yield* Stream.make("data: a\n", "data: b\n").pipe(
173 Stream.pipeThroughChannel(Sse.decode({ maxEventSize: 3 })),
174 Stream.runCollect,
175 Effect.flip
176 )
177
178 assert.instanceOf(error, Sse.SseError)
179 assert.instanceOf(error.reason, Sse.EventTooLarge)
180 assert.strictEqual(error.reason.maxEventSize, 3)
181 }))
182
183 it.effect("parses pending state just under maxEventSize", () =>
184 Effect.gen(function*() {
185 const events = yield* Stream.make("data: a\ndata: b", "\n\n").pipe(
186 Stream.pipeThroughChannel(Sse.decode({ maxEventSize: 10 })),
187 Stream.runCollect
188 )
189
190 assert.deepStrictEqual([...events], [{
191 _tag: "Event",
192 event: "message",
193 id: undefined,
194 data: "a\nb"
195 }])
196 }))
197
198 it.effect("parses well-formed events split across chunks", () =>
199 Effect.gen(function*() {
200 const events = yield* Stream.make(
201 "id: 1\nevent: up",
202 "date\ndata: hel",
203 "lo\n",
204 "\n"
205 ).pipe(
206 Stream.pipeThroughChannel(Sse.decode()),
207 Stream.runCollect
208 )
209
210 assert.deepStrictEqual([...events], [{
211 _tag: "Event",
212 event: "update",
213 id: "1",
214 data: "hello"
215 }])
216 }))
217})