1 # Server-Sent Events parser (WHATWG HTML ยง9.2.6), fed in pieces. Source: https://github.com/paymog/bend-kit/tree/main/llm
2 import Base
3 # By hash, as http imports it: bend-kit-bytes@0.3.0.0.
4 import 0x49814d83de8f70993a43e1002be29ecd/bytes.bend as Bytes
5
6 # name: the event type, "message" when the stream sets none. data: the data lines joined by LF.
7 # id: the last event ID the stream set. name and id hold one Char per octet.
8 type Message is Type:
9 Message{name: String, data: Bytes.Bytes, id: String}
10
11 # The event so far: its type, its data lines each ended by LF, the last event ID, and the last retry in ms.
12 type St is Type:
13 St{name: String, data: Bytes.Bytes, id: String, retry: Maybe<&2, U32>}
14
15 # buf from pos: bytes not yet read as lines. No CR or LF is between pos and seen.
16 # cr: the last line ended with a CR at the end of buf, so an LF first belongs to it.
17 type Reader is Type:
18 Reader{buf: Bytes.Bytes, pos: U32, seen: U32, cr: Bool, st: St}
19
20 type Next is Type:
21 Got{e: Message}
22 Want{}
23
24 # ponytail: a leading U+FEFF is not dropped; the LLM APIs never send one.
25 def reader() -> Reader:
26 Reader{Bytes.new(0), 0, 0, False{}, St{"", Bytes.new(0), "", None{}}}
27
28 def feed.at(zero: Bool, buf: Bytes.Bytes, +pos: U32, +seen: U32, cr: Bool, st: St, chunk: Bytes.Bytes) -> Reader:
29 match zero:
30 case True{}:
31 Reader{Bytes.append(buf, chunk), 0, seen, cr, st}
32 case False{}:
33 Reader{Bytes.append(Bytes.snd(Bytes.slice(buf, pos, 4294967295)), chunk), 0, (U32.max(seen, pos) - pos : U32), cr, st}
34
35 # Drops the lines already read, then adds chunk.
36 def feed(r: Reader, chunk: Bytes.Bytes) -> Reader:
37 Reader{buf, +pos, +seen, cr, st} = r
38 feed.at(U32.is_zero(pos), buf, pos, seen, cr, st, chunk)
39
40 # The last retry field, in ms.
41 def retry(r: Reader) -> Reader & Maybe<&2, U32>:
42 Reader{buf, pos, seen, cr, St{name, data, id, +m}} = r
43 (Reader{buf, pos, seen, cr, St{name, data, id, m}}, m)
44
45 # The first CR or LF at or after i, or the end. f counts the bytes from i to the end; r holds byte i.
46 def eol(f: Nat, r: Array & U32, +i: U32) -> Array & U32:
47 match f:
48 case 0n:
49 (a, c) = r
50 (a, i)
51 case 1n+p:
52 (a, +c) = r
53 match c:
54 case 10:
55 (a, i)
56 case 13:
57 (a, i)
58 case _:
59 eol(p, Bytes.peek(a, (i + 1 : U32)), (i + 1 : U32))
60
61 # The first colon at or after i, or the line's end. f counts the bytes from i to the line's end; r holds byte i.
62 def colon(f: Nat, r: Array & U32, +i: U32) -> Array & U32:
63 match f:
64 case 0n:
65 (a, c) = r
66 (a, i)
67 case 1n+p:
68 (a, +c) = r
69 match c:
70 case 58:
71 (a, i)
72 case _:
73 colon(p, Bytes.peek(a, (i + 1 : U32)), (i + 1 : U32))
74
75 # Base-ten digits, at most 9 of them; None for any other text.
76 def digits(s: String, +acc: U32, +n: U32, bad: Bool) -> Maybe<&2, U32>:
77 match s:
78 case SNil{}:
79 Bool.pick(Maybe<&2, U32>, Bool.or(bad, Bool.or(U32.is_zero(n), U32.is_lt(9, n))), None{}, Some{acc})
80 case SCon{Chr{+c}, t}:
81 digits(t, (acc * 10 + (c - 48 : U32) : U32), (n + 1 : U32), Bool.or(bad, Bool.or(U32.is_lt(c, 48), U32.is_lt(57, c))))
82
83 def has_nul(s: String) -> Bool:
84 match s:
85 case SNil{}:
86 False{}
87 case SCon{Chr{+c}, t}:
88 Bool.or(U32.is_zero(c), has_nul(t))
89
90 def lf() -> Bytes.Bytes:
91 Bytes.from_string("\n")
92
93 def kind.is(+k: U32, r: Array & Bool) -> Array & U32:
94 (a, +hit) = r
95 (a, Bool.pick(U32, hit, k, 0))
96
97 def kind.five(+s: U32, r: Array & Bool) -> Array & U32:
98 (a, hit) = r
99 match hit:
100 case True{}:
101 (a, 2)
102 case False{}:
103 kind.is(4, Bytes.at(a, "retry", s))
104
105 # The kind of the field whose n-byte name starts at s: 1 data, 2 event, 3 id, 4 retry, 0 any other.
106 def kind(+n: U32, a: Array, +s: U32) -> Array & U32:
107 match n:
108 case 2:
109 kind.is(3, Bytes.at(a, "id", s))
110 case 4:
111 kind.is(1, Bytes.at(a, "data", s))
112 case 5:
113 kind.five(s, Bytes.at(a, "event", s))
114 case _:
115 (a, 0)
116
117 def set.retry(m: Maybe<&2, U32>, old: Maybe<&2, U32>) -> Maybe<&2, U32>:
118 match m:
119 case None{}:
120 old
121 case Some{x}:
122 Some{x}
123
124 # The event after a field of kind k with value v. An id with NUL and a retry that is not digits change nothing.
125 def set(+k: U32, v: Bytes.Bytes, st: St) -> St:
126 match k:
127 case 1:
128 St{name, data, id, retry} = st
129 St{name, Bytes.append(Bytes.append(data, v), lf()), id, retry}
130 case 2:
131 St{old, data, id, retry} = st
132 St{Bytes.to_string(v), data, id, retry}
133 case 3:
134 St{name, data, old, retry} = st
135 +s = Bytes.to_string(v)
136 St{name, data, Bool.pick(String, has_nul(s), old, s), retry}
137 case 4:
138 St{name, data, id, old} = st
139 St{name, data, id, set.retry(digits(Bytes.to_string(v), 0, 0, False{}), old)}
140 case _:
141 st
142
143 def field.put(+k: U32, st: St, r: Bytes.Bytes & Bytes.Bytes) -> Array & St:
144 (Bytes.Bytes{l, a}, v) = r
145 (a, set(k, v, st))
146
147 def field.set(+len: U32, +v: U32, +e: U32, st: St, r: Array & U32) -> Array & St:
148 (a, +k) = r
149 field.put(k, st, Bytes.slice(Bytes.Bytes{len, a}, v, (e - v : U32)))
150
151 # r holds the byte after the colon at c. The value starts after the colon and one space, if there is one.
152 def field.at(+s: U32, +c: U32, +e: U32, +len: U32, st: St, r: Array & U32) -> Array & St:
153 (a, +b) = r
154 +v = Bool.pick(U32, U32.is_eq(c, e), e, Bool.pick(U32, Bool.and(U32.is_lt((c + 1 : U32), e), U32.is_eq(b, 32)), (c + 2 : U32), (c + 1 : U32)))
155 field.set(len, v, e, st, kind((c - s : U32), a, s))
156
157 def field.colon(+s: U32, +e: U32, +len: U32, st: St, r: Array & U32) -> Array & St:
158 (a, +c) = r
159 field.at(s, c, e, len, st, Bytes.peek(a, (c + 1 : U32)))
160
161 # The line from s to e as a field: the name up to the first colon, the value after it. A comment's name is empty.
162 def field(a: Array, +len: U32, +s: U32, +e: U32, st: St) -> Array & St:
163 field.colon(s, e, len, st, colon(U32.to_nat((e - s : U32)), Bytes.peek(a, s), s))
164
165 # Drops the last LF. Bytes past len stay 0.
166 def trim(b: Bytes.Bytes) -> Bytes.Bytes:
167 Bytes.Bytes{+n, buf} = b
168 Bytes.Bytes{(n - 1 : U32), Bytes.poke(buf, (n - 1 : U32), 0)}
169
170 def dispatch.of(empty: Bool, +name: String, data: Bytes.Bytes, +id: String, retry: Maybe<&2, U32>) -> St & Maybe<&1, Message>:
171 match empty:
172 case True{}:
173 (St{"", data, id, retry}, None{})
174 case False{}:
175 (St{"", Bytes.new(0), id, retry}, Some{Message{Bool.pick(String, String.is_empty(name), "message", name), trim(data), id}})
176
177 # A blank line: the event, when it has data. Either way the event type starts over.
178 def dispatch(st: St) -> St & Maybe<&1, Message>:
179 St{name, Bytes.Bytes{+n, buf}, id, retry} = st
180 dispatch.of(U32.is_zero(n), name, Bytes.Bytes{n, buf}, id, retry)
181
182 # A line was read and more may follow; the bytes ran out first; or a line ended an event.
183 type Line is Type:
184 LMore{a: Array, pos: U32, cr: Bool, st: St}
185 LWant{a: Array, pos: U32, cr: Bool, st: St}
186 LGot{a: Array, pos: U32, cr: Bool, st: St, e: Message}
187
188 def line.got(a: Array, +next: U32, cr: Bool, r: St & Maybe<&1, Message>) -> Line:
189 (st, m) = r
190 match m:
191 case None{}:
192 LMore{a, next, cr, st}
193 case Some{e}:
194 LGot{a, next, cr, st, e}
195
196 def line.set(+next: U32, cr: Bool, r: Array & St) -> Line:
197 (a, st) = r
198 LMore{a, next, cr, st}
199
200 # The line from s to e; the next line starts at next.
201 def line.body(blank: Bool, a: Array, +len: U32, +s: U32, +e: U32, +next: U32, cr: Bool, st: St) -> Line:
202 match blank:
203 case True{}:
204 line.got(a, next, cr, dispatch(st))
205 case False{}:
206 line.set(next, cr, field(a, len, s, e, st))
207
208 def line.crlf(+len: U32, +s: U32, +e: U32, st: St, r: Array & U32) -> Line:
209 (a, +c) = r
210 line.body(U32.is_eq(s, e), a, len, s, e, Bool.pick(U32, U32.is_eq(c, 10), (e + 2 : U32), (e + 1 : U32)), False{}, st)
211
212 # A CR at e. With nothing after it yet, an LF may still come.
213 def line.cr(more: Bool, a: Array, +len: U32, +s: U32, +e: U32, st: St) -> Line:
214 match more:
215 case True{}:
216 line.crlf(len, s, e, st, Bytes.peek(a, (e + 1 : U32)))
217 case False{}:
218 line.body(U32.is_eq(s, e), a, len, s, e, (e + 1 : U32), True{}, st)
219
220 def line.term(+len: U32, +s: U32, +e: U32, st: St, r: Array & U32) -> Line:
221 (a, +t) = r
222 match t:
223 case 13:
224 line.cr(U32.is_lt((e + 1 : U32), len), a, len, s, e, st)
225 case _:
226 line.body(U32.is_eq(s, e), a, len, s, e, (e + 1 : U32), False{}, st)
227
228 def line.end(want: Bool, a: Array, +len: U32, +s: U32, +e: U32, cr: Bool, st: St) -> Line:
229 match want:
230 case True{}:
231 LWant{a, s, cr, st}
232 case False{}:
233 line.term(len, s, e, st, Bytes.peek(a, e))
234
235 def line.eol(+len: U32, +s: U32, cr: Bool, st: St, r: Array & U32) -> Line:
236 (a, +e) = r
237 line.end(U32.is_eq(e, len), a, len, s, e, cr, st)
238
239 # The line that starts at s. No CR or LF lies between s and seen.
240 def line(a: Array, +len: U32, +s: U32, +seen: U32, cr: Bool, st: St) -> Line:
241 +i = U32.max(s, seen)
242 line.eol(len, s, cr, st, eol(U32.to_nat((len - i : U32)), Bytes.peek(a, i), i))
243
244 def stop(l: Line, +len: U32) -> Reader & Next:
245 match l:
246 case LMore{a, +pos, cr, st}:
247 (Reader{Bytes.Bytes{len, a}, pos, pos, cr, st}, Want{})
248 case LWant{a, +pos, cr, st}:
249 (Reader{Bytes.Bytes{len, a}, pos, len, cr, st}, Want{})
250 case LGot{a, +pos, cr, st, e}:
251 (Reader{Bytes.Bytes{len, a}, pos, pos, cr, st}, Got{e})
252
253 # f: at least the lines left, plus one.
254 def loop(f: Nat, l: Line, +len: U32) -> Reader & Next:
255 match f:
256 case 0n:
257 stop(l, len)
258 case 1n+p:
259 match l:
260 case LMore{a, +pos, cr, st}:
261 loop(p, line(a, len, pos, pos, cr, st), len)
262 case LWant{a, +pos, cr, st}:
263 stop(LWant{a, pos, cr, st}, len)
264 case LGot{a, +pos, cr, st, e}:
265 stop(LGot{a, pos, cr, st, e}, len)
266
267 def skip.byte(+pos: U32, r: Array & U32) -> Array & U32 & Bool:
268 (a, +c) = r
269 (a, Bool.pick(U32, U32.is_eq(c, 10), (pos + 1 : U32), pos), False{})
270
271 def skip.at(some: Bool, a: Array, +pos: U32) -> Array & U32 & Bool:
272 match some:
273 case True{}:
274 skip.byte(pos, Bytes.peek(a, pos))
275 case False{}:
276 (a, pos, True{})
277
278 # After a CR that ended the last piece, an LF first belongs to that CR.
279 def skip(cr: Bool, a: Array, +len: U32, +pos: U32) -> Array & U32 & Bool:
280 match cr:
281 case True{}:
282 skip.at(U32.is_lt(pos, len), a, pos)
283 case False{}:
284 (a, pos, False{})
285
286 def next.at(+len: U32, +seen: U32, st: St, r: Array & U32 & Bool) -> Reader & Next:
287 (a, +pos, cr) = r
288 loop(Nat.add(U32.to_nat((len - pos : U32)), 2n), line(a, len, pos, seen, cr, st), len)
289
290 # The next whole event, or Want when the bytes run out first. A blank line with no data before it gives no event.
291 def next(r: Reader) -> Reader & Next:
292 Reader{Bytes.Bytes{+len, a}, +pos, +seen, cr, st} = r
293 next.at(len, seen, st, skip(cr, a, len, pos))