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