http2.bend
bend-kit-http2@0.3.0.0 · 1,244 lines · 0 definitions
1# RFC 9113 HTTP/2 client and frames, with RFC 7541 HPACK over packed Bytes.
2import Base
3import bend-kit-bytes@0.3.2.0/bytes.bend as Bytes
4import bend-kit-http2@0.3.0.0/hpack.bend as Hpack
5
6type Frame is Type:
7 Frame{kind: U32, flags: U32, stream: U32, payload: Bytes.Bytes}
8
9type Error is Data:
10 Error{code: U32, connection: Bool, stream: U32}
11
12type Decode is Type:
13 Need{input: Bytes.Bytes}
14 Got{frame: Frame, rest: Bytes.Bytes}
15 Bad{error: Error}
16
17type Checked is Type:
18 Valid{frame: Frame}
19 Invalid{error: Error}
20
21# Error codes from RFC 9113 §7: 1=PROTOCOL_ERROR, 3=FLOW_CONTROL_ERROR,
22# 6=FRAME_SIZE_ERROR. A connection error uses stream 0.
23def fail(+code: U32, +connection: Bool, +stream: U32) -> Maybe<&2, Error>:
24 Some{Error{code, connection, Bool.pick(U32, connection, 0, stream)}}
25
26def reject(bad: Bool, +code: U32, +connection: Bool, +stream: U32) -> Maybe<&2, Error>:
27 match bad:
28 case True{}:
29 fail(code, connection, stream)
30 case False{}:
31 None{}
32
33def first(m: Maybe<&2, Error>, n: Maybe<&2, Error>) -> Maybe<&2, Error>:
34 match m:
35 case Some{e}:
36 Some{e}
37 case None{}:
38 n
39
40def number(m: Maybe<&2, U32>) -> U32:
41 match m:
42 case Some{x}:
43 x
44 case None{}:
45 0
46
47def with_number(-R: Type, r: Bytes.Bytes & Maybe<&2, U32>, k: Bytes.Bytes -> U32 -> R) -> R:
48 (b, m) = r
49 k(b, number(m))
50
51def with_slice(-R: Type, r: Bytes.Bytes & Bytes.Bytes, k: Bytes.Bytes -> Bytes.Bytes -> R) -> R:
52 (b, part) = r
53 k(b, part)
54
55def has(+flags: U32, +bit: U32) -> Bool:
56 U32.is_ne((flags .&. bit : U32), 0)
57
58# Field blocks, SETTINGS, and stream 0 make frame-size faults connection-wide.
59def size.scope(+kind: U32, +stream: U32) -> Bool:
60 match kind:
61 case 1:
62 True{}
63 case 4:
64 True{}
65 case 5:
66 True{}
67 case 9:
68 True{}
69 case _:
70 U32.is_eq(stream, 0)
71
72def size.error(+kind: U32, +stream: U32) -> Error:
73 +connection = size.scope(kind, stream)
74 Error{6, connection, Bool.pick(U32, connection, 0, stream)}
75
76def size.reject(oversized: Bool, +kind: U32, +stream: U32) -> Maybe<&2, Error>:
77 match oversized:
78 case True{}:
79 Some{size.error(kind, stream)}
80 case False{}:
81 None{}
82
83# A PADDED payload must contain the pad-length octet and all the padding.
84def padding(+flags: U32, +len: U32, +fixed: U32, +pad: U32) -> Bool:
85 Bool.and(has(flags, 8), Bool.or(U32.is_le(len, fixed), U32.is_gt(pad, (len - fixed - 1 : U32))))
86
87def setting(+id: U32, +value: U32) -> Maybe<&2, Error>:
88 match id:
89 case 2:
90 reject(U32.is_gt(value, 1), 1, True{}, 0)
91 case 4:
92 reject(U32.is_gt(value, 2147483647), 3, True{}, 0)
93 case 5:
94 reject(Bool.or(U32.is_lt(value, 16384), U32.is_gt(value, 16777215)), 1, True{}, 0)
95 case _:
96 None{}
97
98def settings.go(n: Nat, b: Bytes.Bytes, +at: U32, m: Maybe<&2, Error>) -> Bytes.Bytes & Maybe<&2, Error>:
99 match n:
100 case 0n:
101 (b, m)
102 case 1n+p:
103 with_number(Bytes.Bytes & Maybe<&2, Error>, Bytes.get.u16be(b, at), b => id =>
104 with_number(Bytes.Bytes & Maybe<&2, Error>, Bytes.get.u32be(b, (at + 2 : U32)), b => value =>
105 settings.go(p, b, (at + 6 : U32), first(m, setting(id, value)))))
106
107def rules(+kind: U32, +flags: U32, +stream: U32, +len: U32, +pad: U32, +promised: U32, +value: U32, m: Maybe<&2, Error>) -> Maybe<&2, Error>:
108 match kind:
109 case 0:
110 first(reject(U32.is_eq(stream, 0), 1, True{}, stream),
111 first(reject(Bool.and(has(flags, 8), U32.is_eq(len, 0)), 6, False{}, stream),
112 reject(padding(flags, len, 0, pad), 1, True{}, stream)))
113 case 1:
114 +fixed = Bool.pick(U32, has(flags, 32), 5, 0)
115 +prefix = Bool.pick(U32, has(flags, 8), 1, 0)
116 first(reject(U32.is_eq(stream, 0), 1, True{}, stream),
117 first(reject(U32.is_lt(len, (fixed + prefix : U32)), 6, True{}, stream),
118 reject(padding(flags, len, fixed, pad), 1, True{}, stream)))
119 case 2:
120 first(reject(U32.is_ne(len, 5), 6, U32.is_eq(stream, 0), stream),
121 reject(U32.is_eq(stream, 0), 1, True{}, stream))
122 case 3:
123 first(reject(U32.is_ne(len, 4), 6, True{}, stream),
124 reject(U32.is_eq(stream, 0), 1, True{}, stream))
125 case 4:
126 first(reject(U32.is_ne(stream, 0), 1, True{}, stream),
127 first(reject(Bool.and(has(flags, 1), U32.is_ne(len, 0)), 6, True{}, stream),
128 first(reject(U32.is_ne((len % 6 : U32), 0), 6, True{}, stream), m)))
129 case 5:
130 +prefix = Bool.pick(U32, has(flags, 8), 1, 0)
131 first(reject(U32.is_eq(stream, 0), 1, True{}, stream),
132 first(reject(U32.is_lt(len, (prefix + 4 : U32)), 6, True{}, stream),
133 first(reject(padding(flags, len, 4, pad), 1, True{}, stream),
134 reject(U32.is_eq(promised, 0), 1, True{}, stream))))
135 case 6:
136 first(reject(U32.is_ne(len, 8), 6, True{}, stream),
137 reject(U32.is_ne(stream, 0), 1, True{}, stream))
138 case 7:
139 first(reject(U32.is_lt(len, 8), 6, True{}, stream),
140 reject(U32.is_ne(stream, 0), 1, True{}, stream))
141 case 8:
142 first(reject(U32.is_ne(len, 4), 6, U32.is_eq(stream, 0), stream),
143 reject(U32.is_eq((value .&. 2147483647 : U32), 0), 1, U32.is_eq(stream, 0), stream))
144 case 9:
145 reject(U32.is_eq(stream, 0), 1, True{}, stream)
146 case _:
147 None{}
148
149def checked(m: Maybe<&2, Error>, +kind: U32, +flags: U32, +stream: U32, b: Bytes.Bytes) -> Checked:
150 match m:
151 case Some{e}:
152 Invalid{e}
153 case None{}:
154 Valid{Frame{kind, flags, stream, b}}
155
156def with_settings(r: Bytes.Bytes & Maybe<&2, Error>, +flags: U32, +stream: U32) -> Checked:
157 (b, m) = r
158 checked(m, 4, flags, stream, b)
159
160def inspect.settings(m: Maybe<&2, Error>, +flags: U32, +stream: U32, +len: U32, b: Bytes.Bytes) -> Checked:
161 match m:
162 case Some{e}:
163 Invalid{e}
164 case None{}:
165 with_settings(settings.go(U32.to_nat((len / 6 : U32)), b, 0, None{}), flags, stream)
166
167def inspect.promised(+kind: U32, +flags: U32, +stream: U32, +len: U32, +pad: U32, b: Bytes.Bytes) -> Checked:
168 match kind:
169 case 5:
170 +prefix = Bool.pick(U32, has(flags, 8), 1, 0)
171 with_number(Checked, Bytes.get.u32be(b, prefix), b => promised =>
172 checked(rules(kind, flags, stream, len, pad, (promised .&. 2147483647 : U32), 0, None{}), kind, flags, stream, b))
173 case _:
174 checked(rules(kind, flags, stream, len, pad, 0, 0, None{}), kind, flags, stream, b)
175
176def inspect.padded.if(padded: Bool, +kind: U32, +flags: U32, +stream: U32, +len: U32, b: Bytes.Bytes) -> Checked:
177 match padded:
178 case True{}:
179 with_number(Checked, Bytes.get(b, 0), b => pad =>
180 inspect.promised(kind, flags, stream, len, pad, b))
181 case False{}:
182 inspect.promised(kind, flags, stream, len, 0, b)
183
184def inspect.padded(+kind: U32, +flags: U32, +stream: U32, +len: U32, b: Bytes.Bytes) -> Checked:
185 inspect.padded.if(has(flags, 8), kind, flags, stream, len, b)
186
187def inspect.detail(+kind: U32, +flags: U32, +stream: U32, +len: U32, b: Bytes.Bytes) -> Checked:
188 match kind:
189 case 0:
190 inspect.padded(kind, flags, stream, len, b)
191 case 1:
192 inspect.padded(kind, flags, stream, len, b)
193 case 4:
194 inspect.settings(rules(kind, flags, stream, len, 0, 0, 0, None{}), flags, stream, len, b)
195 case 5:
196 inspect.padded(kind, flags, stream, len, b)
197 case 8:
198 with_number(Checked, Bytes.get.u32be(b, 0), b => value =>
199 checked(rules(kind, flags, stream, len, 0, 0, value, None{}), kind, flags, stream, b))
200 case _:
201 checked(rules(kind, flags, stream, len, 0, 0, 0, None{}), kind, flags, stream, b)
202
203
204def inspect.basic(m: Maybe<&2, Error>, +kind: U32, +flags: U32, +stream: U32, +len: U32, b: Bytes.Bytes) -> Checked:
205 match m:
206 case Some{e}:
207 Invalid{e}
208 case None{}:
209 inspect.detail(kind, flags, stream, len, b)
210
211def inspect(max: U32, +kind: U32, +flags: U32, +stream: U32, b: Bytes.Bytes) -> Checked:
212 Bytes.Bytes{+len, buf} = b
213 +basic = first(size.reject(Bool.or(U32.is_gt(len, max), U32.is_gt(len, 16777215)), kind, stream),
214 first(reject(U32.is_gt(kind, 255), 1, True{}, stream),
215 first(reject(U32.is_gt(flags, 255), 1, True{}, stream),
216 reject(U32.is_gt(stream, 2147483647), 1, True{}, stream))))
217 inspect.basic(basic, kind, flags, stream, len, Bytes.Bytes{len, buf})
218
219# max is the peer's SETTINGS_MAX_FRAME_SIZE, initially 16384.
220def allowed(+kind: U32) -> U32:
221 match kind:
222 case 0:
223 9
224 case 1:
225 45
226 case 2:
227 0
228 case 3:
229 0
230 case 4:
231 1
232 case 5:
233 12
234 case 6:
235 1
236 case 7:
237 0
238 case 8:
239 0
240 case 9:
241 4
242 case _:
243 255
244
245def encode.buffer(b: Bytes.Bytes, src: Array<U32>, +len: U32) -> Result<&1, &1, Error, Bytes.Bytes>:
246 Bytes.Bytes{_, dst} = b
247 Done{Bytes.Bytes{(len + 9 : U32), Bytes.dst(Bytes.copy(len, src, dst, 0, 9))}}
248
249def encode.flags(ok: Bool, +kind: U32, +flags: U32, +stream: U32, payload: Bytes.Bytes) -> Result<&1, &1, Error, Bytes.Bytes>:
250 match ok:
251 case False{}:
252 Fail{Error{1, True{}, 0}}
253 case True{}:
254 Bytes.Bytes{+len, src} = payload
255 encode.buffer(Bytes.set.u32be(Bytes.set(Bytes.set(Bytes.put(Bytes.new((len + 9 : U32)), 0, 3, True{}, len), 3, kind), 4, flags), 5, stream), src, len)
256
257def encode.checked(c: Checked) -> Result<&1, &1, Error, Bytes.Bytes>:
258 match c:
259 case Invalid{e}:
260 Fail{e}
261 case Valid{Frame{+kind, +flags, +stream, payload}}:
262 encode.flags(U32.is_eq((flags .&. U32.not(allowed(kind)) : U32), 0), kind, flags, stream, payload)
263def encode(max: U32, f: Frame) -> Result<&1, &1, Error, Bytes.Bytes>:
264 Frame{+kind, +flags, +stream, payload} = f
265 encode.checked(inspect(max, kind, flags, stream, payload))
266
267def parse.result(c: Checked, b: Bytes.Bytes, +len: U32) -> Decode:
268 match c:
269 case Invalid{e}:
270 Bad{e}
271 case Valid{f}:
272 Bytes.Bytes{+n, buf} = b
273 with_slice(Decode, Bytes.slice(Bytes.Bytes{n, buf}, (len + 9 : U32), (n - len - 9 : U32)),
274 unused => rest => Got{f, rest})
275
276def parse.payload(max: U32, +len: U32, +kind: U32, +flags: U32, +stream: U32, b: Bytes.Bytes) -> Decode:
277 with_slice(Decode, Bytes.slice(b, 9, len), b => payload =>
278 parse.result(inspect(max, kind, flags, stream, payload), b, len))
279
280def parse.full(enough: Bool, max: U32, +len: U32, +kind: U32, +flags: U32, +stream: U32, b: Bytes.Bytes) -> Decode:
281 match enough:
282 case True{}:
283 parse.payload(max, len, kind, flags, stream, b)
284 case False{}:
285 Need{b}
286
287def parse.size(oversized: Bool, max: U32, +len: U32, +kind: U32, +flags: U32, +stream: U32, b: Bytes.Bytes) -> Decode:
288 match oversized:
289 case True{}:
290 Bad{size.error(kind, stream)}
291 case False{}:
292 Bytes.Bytes{+n, buf} = b
293 parse.full(U32.is_le(len, (n - 9 : U32)), max, len, kind, flags, stream, Bytes.Bytes{n, buf})
294
295def parse.header(+max: U32, b: Bytes.Bytes) -> Decode:
296 with_number(Decode, Bytes.uint(b, 0, 3, True{}), b => +len =>
297 with_number(Decode, Bytes.get(b, 3), b => kind =>
298 with_number(Decode, Bytes.get(b, 4), b => flags =>
299 with_number(Decode, Bytes.get.u32be(b, 5), b => stream =>
300 parse.size(U32.is_gt(len, max), max, len, kind, flags, (stream .&. 2147483647 : U32), b)))))
301
302def parse.start(short: Bool, max: U32, b: Bytes.Bytes) -> Decode:
303 match short:
304 case True{}:
305 Need{b}
306 case False{}:
307 parse.header(max, b)
308
309# A partial header or payload returns Need with the original input intact.
310def parse(max: U32, b: Bytes.Bytes) -> Decode:
311 Bytes.Bytes{+n, buf} = b
312 parse.start(U32.is_lt(n, 9), max, Bytes.Bytes{n, buf})
313
314# Send windows use U32 two's-complement; a SETTINGS change can make a stream window negative.
315type Stream is Type:
316 Stream{id: U32, send_window: U32, recv_window: U32, local_end: Bool, remote_end: Bool,
317 head: Bool, headers: List<&2, Hpack.Field>, partial: Bytes.Bytes, body: Bytes.Bytes, outgoing: Bytes.Bytes}
318
319type Control is Data:
320 Control{next: U32, active: U32, frame_max: U32, initial: U32, max_streams: Maybe<&2, U32>,
321 send_window: U32, recv_window: U32, first: Bool, ack: Bool,
322 continuation: Maybe<&2, U32>, goaway: Maybe<&2, U32>}
323
324type Client is Type:
325 Client{input: Bytes.Bytes, streams: List<&1, Stream>,
326 encoder: Hpack.State, decoder: Hpack.State, control: Control}
327
328type ClientEvent is Type:
329 Response{id: U32, headers: List<&2, Hpack.Field>, body: Bytes.Bytes}
330 Reset{id: U32, code: U32}
331 Shutdown{last: U32, code: U32}
332 Pong{data: Bytes.Bytes}
333
334type Reply is Type:
335 Advanced{client: Client, writes: List<&1, Bytes.Bytes>, events: List<&1, ClientEvent>}
336 Failed{error: Error}
337
338def client.start() -> Client & Bytes.Bytes:
339 (Client{Bytes.new(0), Nil{}, Hpack.new(4096), Hpack.new(4096),
340 Control{1, 0, 16384, 65535, None{}, 65535, 65535, True{}, True{}, None{}, None{}}},
341 Bytes.append(Bytes.from_string("PRI * HTTP/2.0\u{d}\u{a}\u{d}\u{a}SM\u{d}\u{a}\u{d}\u{a}"),
342 Bytes.set.u16be(Bytes.set(Bytes.set(Bytes.new(15), 2, 6), 3, 4), 9, 2)))
343
344def client.frame(r: Result<&1, &1, Error, Bytes.Bytes>) -> Maybe<&1, Bytes.Bytes>:
345 match r:
346 case Fail{error}:
347 None{}
348 case Done{bytes}:
349 Some{bytes}
350
351def client.headers.empty(first: Bool, end_stream: Bool, id: U32, max: U32, rev: List<&1, Bytes.Bytes>) -> Maybe<&1, List<&1, Bytes.Bytes>>:
352 match first:
353 case False{}:
354 Some{List.reverse(&1, Bytes.Bytes, rev)}
355 case True{}:
356 Maybe.map(&1, Bytes.Bytes, List<&1, Bytes.Bytes>,
357 b => List.reverse(&1, Bytes.Bytes, Con{b, rev}),
358 client.frame(encode(max, Frame{1, (4 + Bool.pick(U32, end_stream, 1, 0) : U32), id, Bytes.new(0)})))
359
360def client.headers.go(n: Nat, +first: Bool, +end_stream: Bool, +id: U32, +max: U32, block: Bytes.Bytes, rev: List<&1, Bytes.Bytes>) -> Maybe<&1, List<&1, Bytes.Bytes>>:
361 match n:
362 case 0n:
363 None{}
364 case 1n+p:
365 match block:
366 case Bytes.Bytes{0, buf}:
367 client.headers.empty(first, end_stream, id, max, rev)
368 case Bytes.Bytes{+len, buf}:
369 +take = Bool.pick(U32, U32.is_le(len, max), len, max)
370 +last = U32.is_le(len, max)
371 with_slice(Maybe<&1, List<&1, Bytes.Bytes>>, Bytes.slice(Bytes.Bytes{len, buf}, 0, take), whole => part =>
372 with_slice(Maybe<&1, List<&1, Bytes.Bytes>>, Bytes.slice(whole, take, (len - take : U32)), unused => rest =>
373 Maybe.bind(&1, Bytes.Bytes, List<&1, Bytes.Bytes>,
374 client.frame(encode(max, Frame{Bool.pick(U32, first, 1, 9),
375 (Bool.pick(U32, last, 4, 0) + Bool.pick(U32, Bool.and(first, end_stream), 1, 0) : U32), id, part})),
376 packet => client.headers.go(p, False{}, end_stream, id, max, rest, Con{packet, rev}))))
377
378def client.headers(end_stream: Bool, +id: U32, +max: U32, block: Bytes.Bytes) -> Maybe<&1, List<&1, Bytes.Bytes>>:
379 Bytes.Bytes{+len, buf} = block
380 client.headers.go(1n+U32.to_nat(len), True{}, end_stream, id, max, Bytes.Bytes{len, buf}, Nil{})
381
382type BodySent is Type:
383 BodySent{remaining: Bytes.Bytes, ended: Bool, stream_window: U32,
384 connection_window: U32, writes: List<&1, Bytes.Bytes>}
385
386def client.min(+a: U32, +b: U32) -> U32:
387 Bool.pick(U32, U32.is_le(a, b), a, b)
388
389def client.positive(+w: U32) -> U32:
390 Bool.pick(U32, U32.is_lt(w, 2147483648), w, 0)
391
392def client.available(+len: U32, +max: U32, +connection: U32, +stream: U32) -> U32:
393 client.min(client.min(len, max), client.min(client.positive(connection), client.positive(stream)))
394
395def client.data.go(n: Nat, body: Bytes.Bytes, take: U32, +id: U32, +max: U32,
396 +connection: U32, +stream: U32, rev: List<&1, Bytes.Bytes>) -> Maybe<&1, BodySent>:
397 match n:
398 case 0n:
399 None{}
400 case 1n+p:
401 match body:
402 case Bytes.Bytes{0, buf}:
403 Some{BodySent{Bytes.Bytes{0, buf}, True{}, stream, connection, List.reverse(&1, Bytes.Bytes, rev)}}
404 case Bytes.Bytes{+len, buf}:
405 match take:
406 case 0:
407 Some{BodySent{Bytes.Bytes{len, buf}, False{}, stream, connection, List.reverse(&1, Bytes.Bytes, rev)}}
408 case +count:
409 with_slice(Maybe<&1, BodySent>, Bytes.slice(Bytes.Bytes{len, buf}, 0, count), whole => part =>
410 with_slice(Maybe<&1, BodySent>, Bytes.slice(whole, count, (len - count : U32)), unused => rest =>
411 Maybe.bind(&1, Bytes.Bytes, BodySent,
412 client.frame(encode(max, Frame{0, Bool.pick(U32, U32.is_eq(count, len), 1, 0), id, part})), packet =>
413 +new_connection = (connection - count : U32)
414 +new_stream = (stream - count : U32)
415 +remaining = (len - count : U32)
416 client.data.go(p, rest, client.available(remaining, max, new_connection, new_stream),
417 id, max, new_connection, new_stream, Con{packet, rev}))))
418
419def client.data(+id: U32, +max: U32, +connection: U32, +stream: U32, body: Bytes.Bytes) -> Maybe<&1, BodySent>:
420 Bytes.Bytes{+len, buf} = body
421 client.data.go(1n+U32.to_nat(len), Bytes.Bytes{len, buf},
422 client.available(len, max, connection, stream), id, max, connection, stream, Nil{})
423
424def client.limit(m: Maybe<&2, U32>, active: U32) -> Bool:
425 match m:
426 case None{}:
427 True{}
428 case Some{max}:
429 U32.is_lt(active, max)
430
431def client.can(c: Control) -> Bool:
432 Control{+next, active, frame_max, initial, max_streams, send_window, recv_window, first, ack, continuation, goaway} = c
433 Bool.and(U32.is_le(next, 2147483647),
434 Bool.and(Maybe.is_none(&2, U32, goaway), client.limit(max_streams, active)))
435
436def client.id(c: Control) -> U32:
437 Control{next, active, frame_max, initial, max_streams, send_window, recv_window, first, ack, continuation, goaway} = c
438 next
439
440def client.frame_max(c: Control) -> U32:
441 Control{next, active, frame_max, initial, max_streams, send_window, recv_window, first, ack, continuation, goaway} = c
442 frame_max
443
444def client.request.done(sent: BodySent, headers: List<&1, Bytes.Bytes>, encoder: Hpack.State, c: Client) -> Reply:
445 BodySent{remaining, ended, stream_window, connection_window, data} = sent
446 Client{input, streams, old, decoder,
447 Control{+next, +active, frame_max, initial, max_streams, previous, recv_window, first, ack, continuation, goaway}} = c
448 Advanced{Client{input,
449 Con{Stream{next, stream_window, 65535, ended, False{}, False{}, Nil{}, Bytes.new(0), Bytes.new(0), remaining}, streams},
450 encoder, decoder,
451 Control{(next + 2 : U32), (active + 1 : U32), frame_max, initial, max_streams,
452 connection_window, recv_window, first, ack, continuation, goaway}},
453 List.append(&1, Bytes.Bytes, headers, data), Nil{}}
454
455def client.request.data(m: Maybe<&1, BodySent>, headers: List<&1, Bytes.Bytes>, encoder: Hpack.State, c: Client) -> Reply:
456 match m:
457 case None{}:
458 Failed{Error{1, True{}, 0}}
459 case Some{sent}:
460 client.request.done(sent, headers, encoder, c)
461
462def client.request.frames(m: Maybe<&1, List<&1, Bytes.Bytes>>, body: Bytes.Bytes, encoder: Hpack.State, c: Client) -> Reply:
463 match m:
464 case None{}:
465 Failed{Error{6, True{}, 0}}
466 case Some{headers}:
467 Client{input, streams, old, decoder,
468 +control} = c
469 Control{+next, active, +frame_max, initial, max_streams, +connection, recv_window, first, ack, continuation, goaway} = control
470 client.request.data(client.data(next, frame_max, connection, initial, body),
471 headers, encoder, Client{input, streams, old, decoder, control})
472
473def client.request.body(r: Bytes.Bytes & U32, encoder: Hpack.State, block: Bytes.Bytes, c: Client) -> Reply:
474 (body, +len) = r
475 Client{input, streams, old, decoder, +control} = c
476 client.request.frames(client.headers(U32.is_eq(len, 0), client.id(control), client.frame_max(control), block),
477 body, encoder, Client{input, streams, old, decoder, control})
478
479def client.request.encoded(m: Maybe<&1, Hpack.State & Bytes.Bytes>, c: Client, body: Bytes.Bytes) -> Reply:
480 match m:
481 case None{}:
482 Failed{Error{1, True{}, 0}}
483 case Some{(encoder, block)}:
484 client.request.body(Bytes.length(body), encoder, block, c)
485
486def client.request.ready(valid: Bool, c: Client, fields: List<&2, Hpack.Field>, body: Bytes.Bytes) -> Reply:
487 match valid:
488 case False{}:
489 Failed{Error{7, True{}, 0}}
490 case True{}:
491 Client{input, streams, +encoder, decoder, control} = c
492 client.request.encoded(Hpack.encode(fields, False{}, encoder),
493 Client{input, streams, encoder, decoder, control}, body)
494
495def client.request(c: Client, fields: List<&2, Hpack.Field>, body: Bytes.Bytes) -> Reply:
496 Client{input, streams, encoder, decoder, +control} = c
497 client.request.ready(client.can(control), Client{input, streams, encoder, decoder, control}, fields, body)
498
499def client.fail(+code: U32, +connection: Bool, +stream: U32) -> Reply:
500 Failed{Error{code, connection, Bool.pick(U32, connection, 0, stream)}}
501
502def client.ok(c: Client) -> Reply:
503 Advanced{c, Nil{}, Nil{}}
504
505def client.after(r: Reply, k: Client -> List<&1, Bytes.Bytes> -> List<&1, ClientEvent> -> Reply) -> Reply:
506 match r:
507 case Failed{error}:
508 Failed{error}
509 case Advanced{client, writes, events}:
510 k(client, writes, events)
511
512def client.write(m: Maybe<&1, Bytes.Bytes>, c: Client) -> Reply:
513 match m:
514 case None{}:
515 client.fail(1, True{}, 0)
516 case Some{packet}:
517 Advanced{c, [packet], Nil{}}
518
519def client.ping(+flags: U32, payload: Bytes.Bytes, c: Client) -> Reply:
520 match flags:
521 case 0:
522 client.write(client.frame(encode(16384, Frame{6, 1, 0, payload})), c)
523 case 1:
524 Advanced{c, Nil{}, [Pong{payload}]}
525 case _:
526 client.fail(1, True{}, 0)
527
528def client.ping.request(c: Client, payload: Bytes.Bytes) -> Reply:
529 client.write(client.frame(encode(16384, Frame{6, 0, 0, payload})), c)
530
531def client.window.add(valid: Bool, +w: U32, +delta: U32) -> Maybe<&1, U32>:
532 match valid:
533 case False{}:
534 None{}
535 case True{}:
536 Some{(w + delta : U32)}
537
538def client.window.adjust.sign(positive: Bool, +w: U32, +delta: U32) -> Maybe<&1, U32>:
539 match positive:
540 case True{}:
541 client.window.add(Bool.or(U32.is_ge(w, 2147483648),
542 U32.is_le(w, (2147483647 - delta : U32))), w, delta)
543 case False{}:
544 client.window.add(Bool.or(U32.is_lt(w, 2147483648),
545 U32.is_ge(w, (2147483648 - delta : U32))), w, delta)
546
547def client.window.adjust(+w: U32, +delta: U32) -> Maybe<&1, U32>:
548 client.window.adjust.sign(U32.is_lt(delta, 2147483648), w, delta)
549
550def client.window.plus.sign(positive: Bool, +w: U32, +value: U32) -> Maybe<&1, U32>:
551 match positive:
552 case True{}:
553 client.window.add(U32.is_le(value, (2147483647 - w : U32)), w, value)
554 case False{}:
555 Some{(w + value : U32)}
556
557def client.window.plus(+w: U32, value: U32) -> Maybe<&1, U32>:
558 client.window.plus.sign(U32.is_lt(w, 2147483648), w, value)
559
560def client.stream.adjust(s: Stream, delta: U32) -> Maybe<&1, Stream>:
561 Stream{id, send_window, recv_window, local_end, remote_end, head, headers, partial, body, outgoing} = s
562 Maybe.map(&1, U32, Stream,
563 next => Stream{id, next, recv_window, local_end, remote_end, head, headers, partial, body, outgoing},
564 client.window.adjust(send_window, delta))
565
566def client.streams.adjust(xs: List<&1, Stream>, +delta: U32) -> Maybe<&1, List<&1, Stream>>:
567 match xs:
568 case Nil{}:
569 Some{Nil{}}
570 case Con{first, rest}:
571 Maybe.bind(&1, Stream, List<&1, Stream>, client.stream.adjust(first, delta), updated =>
572 Maybe.map(&1, List<&1, Stream>, List<&1, Stream>, tail => Con{updated, tail},
573 client.streams.adjust(rest, delta)))
574
575def client.settings.initial.done(m: Maybe<&1, List<&1, Stream>>, input: Bytes.Bytes,
576 encoder: Hpack.State, decoder: Hpack.State, control: Control) -> Reply:
577 match m:
578 case None{}:
579 client.fail(3, True{}, 0)
580 case Some{streams}:
581 client.ok(Client{input, streams, encoder, decoder, control})
582
583def client.settings.initial(+value: U32, c: Client) -> Reply:
584 Client{input, streams, encoder, decoder,
585 Control{next, active, frame_max, +initial, max_streams, send_window, recv_window,
586 first, ack, continuation, goaway}} = c
587 +delta = (value - initial : U32)
588 client.settings.initial.done(client.streams.adjust(streams, delta),
589 input, encoder, decoder,
590 Control{next, active, frame_max, value, max_streams, send_window, recv_window,
591 first, ack, continuation, goaway})
592
593def client.settings.item(id: U32, +value: U32, c: Client) -> Reply:
594 match id:
595 case 1:
596 Client{input, streams, encoder, decoder, control} = c
597 client.ok(Client{input, streams, Hpack.set_limit(value, encoder), decoder, control})
598 case 2:
599 client.fail(1, True{}, 0)
600 case 3:
601 Client{input, streams, encoder, decoder,
602 Control{next, active, frame_max, initial, max_streams, send_window, recv_window,
603 first, ack, continuation, goaway}} = c
604 client.ok(Client{input, streams, encoder, decoder,
605 Control{next, active, frame_max, initial, Some{value}, send_window, recv_window,
606 first, ack, continuation, goaway}})
607 case 4:
608 client.settings.initial(value, c)
609 case 5:
610 Client{input, streams, encoder, decoder,
611 Control{next, active, frame_max, initial, max_streams, send_window, recv_window,
612 first, ack, continuation, goaway}} = c
613 client.ok(Client{input, streams, encoder, decoder,
614 Control{next, active, value, initial, max_streams, send_window, recv_window,
615 first, ack, continuation, goaway}})
616 case _:
617 client.ok(c)
618
619def client.settings.go(n: Nat, b: Bytes.Bytes, +at: U32, c: Client) -> Reply:
620 match n:
621 case 0n:
622 client.write(client.frame(encode(16384, Frame{4, 1, 0, Bytes.new(0)})), c)
623 case 1n+p:
624 with_number(Reply, Bytes.get.u16be(b, at), read_id => id =>
625 with_number(Reply, Bytes.get.u32be(read_id, (at + 2 : U32)), rest => value =>
626 client.after(client.settings.item(id, value, c), next => writes => events =>
627 client.settings.go(p, rest, (at + 6 : U32), next))))
628
629def client.settings.mark(c: Client) -> Client:
630 Client{input, streams, encoder, decoder,
631 Control{next, active, frame_max, initial, max_streams, send_window, recv_window,
632 first, ack, continuation, goaway}} = c
633 Client{input, streams, encoder, decoder,
634 Control{next, active, frame_max, initial, max_streams, send_window, recv_window,
635 False{}, ack, continuation, goaway}}
636
637def client.settings.ack(c: Client) -> Reply:
638 Client{input, streams, encoder, decoder,
639 Control{next, active, frame_max, initial, max_streams, send_window, recv_window,
640 first, ack, continuation, goaway}} = c
641 match ack:
642 case False{}:
643 client.fail(1, True{}, 0)
644 case True{}:
645 client.ok(Client{input, streams, encoder, decoder,
646 Control{next, active, frame_max, initial, max_streams, send_window, recv_window,
647 first, False{}, continuation, goaway}})
648
649def client.settings(flags: U32, payload: Bytes.Bytes, c: Client) -> Reply:
650 match flags:
651 case 1:
652 client.settings.ack(c)
653 case 0:
654 Bytes.Bytes{+len, buf} = payload
655 client.settings.go(U32.to_nat(U32.div(len, 6)), Bytes.Bytes{len, buf}, 0, client.settings.mark(c))
656 case _:
657 client.fail(1, True{}, 0)
658
659def client.streams.prefix(previous: Maybe<&1, Stream>, r: Reply) -> Reply:
660 match previous:
661 case None{}:
662 r
663 case Some{s}:
664 match r:
665 case Failed{error}:
666 Failed{error}
667 case Advanced{Client{input, streams, encoder, decoder, control}, writes, events}:
668 Advanced{Client{input, Con{s, streams}, encoder, decoder, control}, writes, events}
669
670def client.streams.go(xs: List<&1, Stream>, found: Bool, previous: Maybe<&1, Stream>,
671 +target: U32, k: Stream -> List<&1, Stream> -> Reply) -> Reply:
672 match xs:
673 case Nil{}:
674 match found:
675 case False{}:
676 client.fail(5, False{}, target)
677 case True{}:
678 match previous:
679 case None{}:
680 client.fail(1, True{}, 0)
681 case Some{s}:
682 k(s, Nil{})
683 case Con{s, rest}:
684 Stream{+id, send_window, recv_window, local_end, remote_end, head, headers, partial, body, outgoing} = s
685 match found:
686 case True{}:
687 match previous:
688 case None{}:
689 client.fail(1, True{}, 0)
690 case Some{matched}:
691 k(matched, Con{Stream{id, send_window, recv_window, local_end, remote_end,
692 head, headers, partial, body, outgoing}, rest})
693 case False{}:
694 client.streams.prefix(previous,
695 client.streams.go(rest, U32.is_eq(id, target),
696 Some{Stream{id, send_window, recv_window, local_end, remote_end, head, headers, partial, body, outgoing}},
697 target, k))
698
699def client.streams.use(xs: List<&1, Stream>, target: U32, k: Stream -> List<&1, Stream> -> Reply) -> Reply:
700 client.streams.go(xs, False{}, None{}, target, k)
701
702type Flushed is Type:
703 Flushed{streams: List<&1, Stream>, connection: U32, writes: List<&1, Bytes.Bytes>}
704
705def client.flush.prepend(s: Stream, packets: List<&1, Bytes.Bytes>, tail: Flushed) -> Flushed:
706 Flushed{more, window, writes} = tail
707 Flushed{Con{s, more}, window, List.append(&1, Bytes.Bytes, packets, writes)}
708
709def client.with_sent(-R: Type, sent: BodySent,
710 k: Bytes.Bytes -> Bool -> U32 -> U32 -> List<&1, Bytes.Bytes> -> R) -> R:
711 BodySent{remaining, ended, stream, connection, packets} = sent
712 k(remaining, ended, stream, connection, packets)
713
714def client.flush.go(xs: List<&1, Stream>, +connection: U32, +max: U32) -> Maybe<&1, Flushed>:
715 match xs:
716 case Nil{}:
717 Some{Flushed{Nil{}, connection, Nil{}}}
718 case Con{Stream{+id, +stream_window, recv_window, local_end, remote_end,
719 head, headers, partial, body, outgoing}, rest}:
720 match outgoing:
721 case Bytes.Bytes{0, buf}:
722 Maybe.map(&1, Flushed, Flushed, tail =>
723 client.flush.prepend(Stream{id, stream_window, recv_window, local_end, remote_end,
724 head, headers, partial, body, Bytes.Bytes{0, buf}}, Nil{}, tail),
725 client.flush.go(rest, connection, max))
726 case Bytes.Bytes{+len, buf}:
727 Maybe.bind(&1, BodySent, Flushed,
728 client.data(id, max, connection, stream_window, Bytes.Bytes{len, buf}), sent =>
729 client.with_sent(Maybe<&1, Flushed>, sent, remaining => ended => next_stream => next_connection => packets =>
730 Maybe.map(&1, Flushed, Flushed, tail =>
731 client.flush.prepend(Stream{id, next_stream, recv_window, ended, remote_end,
732 head, headers, partial, body, remaining}, packets, tail),
733 client.flush.go(rest, next_connection, max))))
734
735def client.flush.done(m: Maybe<&1, Flushed>, input: Bytes.Bytes, encoder: Hpack.State,
736 decoder: Hpack.State, control: Control) -> Reply:
737 match m:
738 case None{}:
739 client.fail(1, True{}, 0)
740 case Some{Flushed{streams, +connection, writes}}:
741 Control{next, active, frame_max, initial, max_streams, previous, recv_window,
742 first, ack, continuation, goaway} = control
743 Advanced{Client{input, streams, encoder, decoder,
744 Control{next, active, frame_max, initial, max_streams, connection, recv_window,
745 first, ack, continuation, goaway}}, writes, Nil{}}
746
747def client.window.connection.valid(m: Maybe<&1, U32>, c: Client) -> Reply:
748 match m:
749 case None{}:
750 client.fail(3, True{}, 0)
751 case Some{w}:
752 Client{input, streams, encoder, decoder, +control} = c
753 Control{next, active, +frame_max, initial, max_streams, send_window, recv_window,
754 first, ack, continuation, goaway} = control
755 client.flush.done(client.flush.go(streams, w, frame_max),
756 input, encoder, decoder, control)
757
758def client.window.connection(+value: U32, c: Client) -> Reply:
759 Client{input, streams, encoder, decoder, +control} = c
760 Control{next, active, frame_max, initial, max_streams, +send_window, recv_window,
761 first, ack, continuation, goaway} = control
762 client.window.connection.valid(client.window.plus(send_window, value),
763 Client{input, streams, encoder, decoder, control})
764
765def client.window.stream.valid(m: Maybe<&1, U32>, +id: U32, s: Stream, tail: List<&1, Stream>,
766 input: Bytes.Bytes, encoder: Hpack.State, decoder: Hpack.State, +control: Control) -> Reply:
767 match m:
768 case None{}:
769 client.fail(3, False{}, id)
770 case Some{w}:
771 Stream{id, previous, recv_window, local_end, remote_end, head, headers, partial, body, outgoing} = s
772 Control{next, active, +frame_max, initial, max_streams, +connection, recv, first, ack, continuation, goaway} = control
773 client.flush.done(client.flush.go(
774 Con{Stream{id, w, recv_window, local_end, remote_end,
775 head, headers, partial, body, outgoing}, tail}, connection, frame_max),
776 input, encoder, decoder, control)
777
778def client.window.stream.handle(+value: U32, s: Stream, tail: List<&1, Stream>,
779 input: Bytes.Bytes, encoder: Hpack.State, decoder: Hpack.State, control: Control) -> Reply:
780 Stream{+id, +send_window, recv_window, local_end, remote_end, head, headers, partial, body, outgoing} = s
781 client.window.stream.valid(client.window.plus(send_window, value), id,
782 Stream{id, send_window, recv_window, local_end, remote_end, head, headers, partial, body, outgoing},
783 tail, input, encoder, decoder, control)
784
785def client.window.stream(id: U32, +value: U32, c: Client) -> Reply:
786 Client{input, streams, encoder, decoder, control} = c
787 client.streams.use(streams, id, s => tail =>
788 client.window.stream.handle(value, s, tail, input, encoder, decoder, control))
789
790def client.window.route(stream: U32, value: U32, c: Client) -> Reply:
791 match stream:
792 case 0:
793 client.window.connection(value, c)
794 case +id:
795 client.window.stream(id, value, c)
796
797def client.window.update(stream: U32, payload: Bytes.Bytes, c: Client) -> Reply:
798 with_number(Reply, Bytes.get.u32be(payload, 0), rest => raw =>
799 client.window.route(stream, (raw .&. 2147483647 : U32), c))
800
801def client.rst.stream(s: Stream, tail: List<&1, Stream>, +code: U32, input: Bytes.Bytes,
802 encoder: Hpack.State, decoder: Hpack.State, control: Control) -> Reply:
803 Stream{+id, send_window, recv_window, local_end, remote_end, head, headers, partial, body, outgoing} = s
804 Control{next, +active, frame_max, initial, max_streams, connection, recv,
805 first, ack, continuation, goaway} = control
806 Advanced{Client{input, tail, encoder, decoder,
807 Control{next, (active - 1 : U32), frame_max, initial, max_streams,
808 connection, recv, first, ack, continuation, goaway}}, Nil{}, [Reset{id, code}]}
809
810def client.rst.code(id: U32, +code: U32, c: Client) -> Reply:
811 Client{input, streams, encoder, decoder, control} = c
812 client.streams.use(streams, id, s => tail =>
813 client.rst.stream(s, tail, code, input, encoder, decoder, control))
814
815def client.rst(id: U32, payload: Bytes.Bytes, c: Client) -> Reply:
816 with_number(Reply, Bytes.get.u32be(payload, 0), rest => code =>
817 client.rst.code(id, code, c))
818
819type Gone is Type:
820 Gone{streams: List<&1, Stream>, active: U32, events: List<&1, ClientEvent>}
821
822def client.goaway.cons(keep: Bool, s: Stream, id: U32, tail: Gone) -> Gone:
823 match keep:
824 case True{}:
825 Gone{streams, +active, events} = tail
826 Gone{Con{s, streams}, (active + 1 : U32), events}
827 case False{}:
828 Gone{streams, active, events} = tail
829 Gone{streams, active, Con{Reset{id, 7}, events}}
830
831def client.goaway.filter(xs: List<&1, Stream>, +last: U32) -> Gone:
832 match xs:
833 case Nil{}:
834 Gone{Nil{}, 0, Nil{}}
835 case Con{Stream{+id, send_window, recv_window, local_end, remote_end,
836 head, headers, partial, body, outgoing}, rest}:
837 client.goaway.cons(U32.is_le(id, last),
838 Stream{id, send_window, recv_window, local_end, remote_end, head, headers, partial, body, outgoing},
839 id, client.goaway.filter(rest, last))
840
841def client.goaway.allowed(previous: Maybe<&2, U32>, last: U32) -> Bool:
842 match previous:
843 case None{}:
844 True{}
845 case Some{older}:
846 U32.is_le(last, older)
847
848def client.goaway.finish(g: Gone, +last: U32, code: U32, input: Bytes.Bytes,
849 encoder: Hpack.State, decoder: Hpack.State, control: Control) -> Reply:
850 Gone{remaining, count, events} = g
851 Control{next, active, frame_max, initial, max_streams, connection, recv,
852 first, ack, continuation, previous} = control
853 Advanced{Client{input, remaining, encoder, decoder,
854 Control{next, count, frame_max, initial, max_streams, connection, recv,
855 first, ack, continuation, Some{last}}}, Nil{},
856 Con{Shutdown{last, code}, events}}
857
858def client.goaway.valid(valid: Bool, +last: U32, code: U32, c: Client) -> Reply:
859 match valid:
860 case False{}:
861 client.fail(1, True{}, 0)
862 case True{}:
863 Client{input, streams, encoder, decoder, control} = c
864 client.goaway.finish(client.goaway.filter(streams, last), last, code,
865 input, encoder, decoder, control)
866
867def client.goaway.body(+last: U32, +code: U32, c: Client) -> Reply:
868 Client{input, streams, encoder, decoder, +control} = c
869 Control{next, active, frame_max, initial, max_streams, connection, recv,
870 first, ack, continuation, +previous} = control
871 client.goaway.valid(client.goaway.allowed(previous, last), last, code,
872 Client{input, streams, encoder, decoder, control})
873
874def client.goaway(payload: Bytes.Bytes, c: Client) -> Reply:
875 with_number(Reply, Bytes.get.u32be(payload, 0), b => raw =>
876 with_number(Reply, Bytes.get.u32be(b, 4), rest => code =>
877 client.goaway.body((raw .&. 2147483647 : U32), code, c)))
878
879def client.complete.ready(+id: U32, headers: List<&2, Hpack.Field>, body: Bytes.Bytes,
880 tail: List<&1, Stream>, input: Bytes.Bytes, encoder: Hpack.State,
881 decoder: Hpack.State, control: Control, writes: List<&1, Bytes.Bytes>) -> Reply:
882 Control{next, +active, frame_max, initial, max_streams, connection, recv,
883 first, ack, continuation, goaway} = control
884 Advanced{Client{input, tail, encoder, decoder,
885 Control{next, (active - 1 : U32), frame_max, initial, max_streams,
886 connection, recv, first, ack, continuation, goaway}},
887 writes, [Response{id, headers, body}]}
888
889def client.complete.packet(m: Maybe<&1, Bytes.Bytes>, id: U32, headers: List<&2, Hpack.Field>,
890 body: Bytes.Bytes, tail: List<&1, Stream>, input: Bytes.Bytes,
891 encoder: Hpack.State, decoder: Hpack.State, control: Control) -> Reply:
892 match m:
893 case None{}:
894 client.fail(1, True{}, 0)
895 case Some{packet}:
896 client.complete.ready(id, headers, body, tail, input, encoder, decoder, control, [packet])
897
898def client.complete(s: Stream, tail: List<&1, Stream>, input: Bytes.Bytes,
899 encoder: Hpack.State, decoder: Hpack.State, control: Control) -> Reply:
900 Stream{+id, send_window, recv_window, local_end, remote_end, head,
901 headers, partial, body, outgoing} = s
902 match local_end:
903 case True{}:
904 client.complete.ready(id, headers, body, tail, input, encoder, decoder, control, Nil{})
905 case False{}:
906 client.complete.packet(client.frame(encode(16384,
907 Frame{3, 0, id, Bytes.set.u32be(Bytes.new(4), 0, 8)})),
908 id, headers, body, tail, input, encoder, decoder, control)
909
910def client.fragment.slice(b: Bytes.Bytes, +offset: U32, +pad: U32) -> Bytes.Bytes:
911 Bytes.Bytes{+len, buf} = b
912 with_slice(Bytes.Bytes, Bytes.slice(Bytes.Bytes{len, buf}, offset, (len - offset - pad : U32)),
913 whole => part => part)
914
915def client.fragment.padded(padded: Bool, +fixed: U32, b: Bytes.Bytes) -> Bytes.Bytes:
916 match padded:
917 case False{}:
918 client.fragment.slice(b, fixed, 0)
919 case True{}:
920 with_number(Bytes.Bytes, Bytes.get(b, 0), rest => pad =>
921 client.fragment.slice(rest, (fixed + 1 : U32), pad))
922
923def client.fragment(+kind: U32, +flags: U32, b: Bytes.Bytes) -> Bytes.Bytes:
924 client.fragment.padded(has(flags, 8),
925 Bool.pick(U32, Bool.and(U32.is_eq(kind, 1), has(flags, 32)), 5, 0), b)
926
927def client.fragments.join(first: Bytes.Bytes, second: Bytes.Bytes) -> Bytes.Bytes:
928 Bytes.Bytes{len, buf} = first
929 match len:
930 case 0:
931 second
932 case +count:
933 Bytes.append(Bytes.Bytes{count, buf}, second)
934
935def client.headers.interim.status(s: String) -> Bool:
936 match s:
937 case SCon{Chr{49}, rest}:
938 True{}
939 case _:
940 False{}
941
942def client.headers.interim(xs: List<&2, Hpack.Field>) -> Bool:
943 match xs:
944 case Nil{}:
945 False{}
946 case Con{Hpack.Field{name, value, mode}, rest}:
947 Bool.and(String.eq(name, ":status"), client.headers.interim.status(value))
948
949def client.headers.interim.valid(valid: Bool, id: U32, fields: List<&2, Hpack.Field>,
950 s: Stream, tail: List<&1, Stream>, input: Bytes.Bytes,
951 encoder: Hpack.State, decoder: Hpack.State, control: Control) -> Reply:
952 match valid:
953 case False{}:
954 client.fail(1, False{}, id)
955 case True{}:
956 Advanced{Client{input, Con{s, tail}, encoder, decoder, control}, Nil{},
957 [Response{id, fields, Bytes.new(0)}]}
958
959def client.headers.fields.select(interim: Bool, fields: List<&2, Hpack.Field>, s: Stream,
960 tail: List<&1, Stream>, input: Bytes.Bytes, encoder: Hpack.State,
961 decoder: Hpack.State, control: Control) -> Reply:
962 match interim:
963 case True{}:
964 Stream{+id, send_window, recv_window, local_end, +remote_end, head,
965 headers, partial, body, outgoing} = s
966 client.headers.interim.valid(Bool.not(remote_end), id, fields,
967 Stream{id, send_window, recv_window, local_end, remote_end, head,
968 headers, partial, body, outgoing}, tail, input, encoder, decoder, control)
969 case False{}:
970 Stream{id, send_window, recv_window, local_end, remote_end, head,
971 headers, partial, body, outgoing} = s
972 match remote_end:
973 case True{}:
974 client.complete(Stream{id, send_window, recv_window, local_end, True{}, True{},
975 List.append(&2, Hpack.Field, headers, fields), partial, body, outgoing},
976 tail, input, encoder, decoder, control)
977 case False{}:
978 client.ok(Client{input,
979 Con{Stream{id, send_window, recv_window, local_end, False{}, True{},
980 List.append(&2, Hpack.Field, headers, fields), partial, body, outgoing}, tail},
981 encoder, decoder, control})
982
983def client.headers.fields(+fields: List<&2, Hpack.Field>, s: Stream, tail: List<&1, Stream>,
984 input: Bytes.Bytes, encoder: Hpack.State, decoder: Hpack.State, control: Control) -> Reply:
985 client.headers.fields.select(client.headers.interim(fields),
986 fields, s, tail, input, encoder, decoder, control)
987
988def client.headers.decoded(m: Maybe<&1, Hpack.State & List<&2, Hpack.Field>>,
989 s: Stream, tail: List<&1, Stream>, input: Bytes.Bytes,
990 encoder: Hpack.State, control: Control) -> Reply:
991 match m:
992 case None{}:
993 client.fail(9, True{}, 0)
994 case Some{(decoder, fields)}:
995 client.headers.fields(fields, s, tail, input, encoder, decoder, control)
996
997def client.headers.finish(end_headers: Bool, s: Stream, tail: List<&1, Stream>,
998 input: Bytes.Bytes, encoder: Hpack.State, decoder: Hpack.State, control: Control) -> Reply:
999 match end_headers:
1000 case False{}:
1001 Stream{+id, send_window, recv_window, local_end, remote_end, head,
1002 headers, partial, body, outgoing} = s
1003 Control{next, active, frame_max, initial, max_streams, connection, recv,
1004 first, ack, continuation, goaway} = control
1005 client.ok(Client{input,
1006 Con{Stream{id, send_window, recv_window, local_end, remote_end,
1007 head, headers, partial, body, outgoing}, tail},
1008 encoder, decoder,
1009 Control{next, active, frame_max, initial, max_streams, connection, recv,
1010 first, ack, Some{id}, goaway}})
1011 case True{}:
1012 Stream{id, send_window, recv_window, local_end, remote_end, head,
1013 headers, partial, body, outgoing} = s
1014 Control{next, active, frame_max, initial, max_streams, connection, recv,
1015 first, ack, continuation, goaway} = control
1016 client.headers.decoded(Hpack.decode(partial, decoder),
1017 Stream{id, send_window, recv_window, local_end, remote_end, head,
1018 headers, Bytes.new(0), body, outgoing}, tail, input, encoder,
1019 Control{next, active, frame_max, initial, max_streams, connection, recv,
1020 first, ack, None{}, goaway})
1021
1022def client.headers.accumulate(end_headers: Bool, fragment: Bytes.Bytes,
1023 s: Stream, tail: List<&1, Stream>, input: Bytes.Bytes,
1024 encoder: Hpack.State, decoder: Hpack.State, control: Control) -> Reply:
1025 Stream{id, send_window, recv_window, local_end, remote_end, head,
1026 headers, partial, body, outgoing} = s
1027 client.headers.finish(end_headers,
1028 Stream{id, send_window, recv_window, local_end, remote_end, head,
1029 headers, client.fragments.join(partial, fragment), body, outgoing},
1030 tail, input, encoder, decoder, control)
1031
1032def client.headers.open.valid(valid: Bool, +id: U32, end_headers: Bool, end_stream: Bool,
1033 fragment: Bytes.Bytes, s: Stream, tail: List<&1, Stream>, input: Bytes.Bytes,
1034 encoder: Hpack.State, decoder: Hpack.State, control: Control) -> Reply:
1035 match valid:
1036 case False{}:
1037 client.fail(5, False{}, id)
1038 case True{}:
1039 Stream{stream, send_window, recv_window, local_end, remote_end, head,
1040 headers, partial, body, outgoing} = s
1041 client.headers.accumulate(end_headers, fragment,
1042 Stream{stream, send_window, recv_window, local_end, end_stream, head,
1043 headers, partial, body, outgoing}, tail, input, encoder, decoder, control)
1044
1045def client.headers.open(end_headers: Bool, +end_stream: Bool, fragment: Bytes.Bytes,
1046 s: Stream, tail: List<&1, Stream>, input: Bytes.Bytes,
1047 encoder: Hpack.State, decoder: Hpack.State, control: Control) -> Reply:
1048 Stream{+id, send_window, recv_window, local_end, +remote_end, +head,
1049 headers, partial, body, outgoing} = s
1050 client.headers.open.valid(Bool.and(Bool.not(remote_end), Bool.or(Bool.not(head), end_stream)),
1051 id, end_headers, end_stream, fragment,
1052 Stream{id, send_window, recv_window, local_end, remote_end, head,
1053 headers, partial, body, outgoing}, tail, input, encoder, decoder, control)
1054
1055def client.headers.receive(+id: U32, +flags: U32, payload: Bytes.Bytes, c: Client) -> Reply:
1056 Client{input, streams, encoder, decoder, control} = c
1057 +end_headers = has(flags, 4)
1058 +end_stream = has(flags, 1)
1059 fragment = client.fragment(1, flags, payload)
1060 client.streams.use(streams, id, s => tail =>
1061 client.headers.open(end_headers, end_stream, fragment, s, tail, input, encoder, decoder, control))
1062
1063def client.continuation(+id: U32, +flags: U32, payload: Bytes.Bytes, c: Client) -> Reply:
1064 Client{input, streams, encoder, decoder, control} = c
1065 client.streams.use(streams, id, s => tail =>
1066 client.headers.accumulate(has(flags, 4), payload, s, tail, input, encoder, decoder, control))
1067
1068def client.data.updates.end(end_stream: Bool, +count: U32, id: U32,
1069 connection: Bytes.Bytes) -> Maybe<&1, List<&1, Bytes.Bytes>>:
1070 match end_stream:
1071 case True{}:
1072 Some{[connection]}
1073 case False{}:
1074 Maybe.map(&1, Bytes.Bytes, List<&1, Bytes.Bytes>,
1075 stream => [connection, stream],
1076 client.frame(encode(16384, Frame{8, 0, id, Bytes.set.u32be(Bytes.new(4), 0, count)})))
1077
1078def client.data.updates(+len: U32, end_stream: Bool, +id: U32) -> Maybe<&1, List<&1, Bytes.Bytes>>:
1079 match len:
1080 case 0:
1081 Some{Nil{}}
1082 case +count:
1083 Maybe.bind(&1, Bytes.Bytes, List<&1, Bytes.Bytes>,
1084 client.frame(encode(16384, Frame{8, 0, 0, Bytes.set.u32be(Bytes.new(4), 0, count)})), connection =>
1085 client.data.updates.end(end_stream, count, id, connection))
1086
1087def client.data.content(end_stream: Bool, content: Bytes.Bytes, writes: List<&1, Bytes.Bytes>,
1088 s: Stream, tail: List<&1, Stream>, input: Bytes.Bytes,
1089 encoder: Hpack.State, decoder: Hpack.State, control: Control) -> Reply:
1090 match end_stream:
1091 case False{}:
1092 Stream{id, send_window, recv_window, local_end, remote_end, head,
1093 headers, partial, body, outgoing} = s
1094 Advanced{Client{input,
1095 Con{Stream{id, send_window, recv_window, local_end, False{}, head,
1096 headers, partial, Bytes.append(body, content), outgoing}, tail},
1097 encoder, decoder, control}, writes, Nil{}}
1098 case True{}:
1099 Stream{id, send_window, recv_window, local_end, remote_end, head,
1100 headers, partial, body, outgoing} = s
1101 client.after(client.complete(Stream{id, send_window, recv_window, local_end, True{}, head,
1102 headers, partial, Bytes.append(body, content), outgoing}, tail, input, encoder, decoder, control),
1103 next => extra => events =>
1104 Advanced{next, List.append(&1, Bytes.Bytes, writes, extra), events})
1105
1106def client.data.content.updated(m: Maybe<&1, List<&1, Bytes.Bytes>>, end_stream: Bool,
1107 content: Bytes.Bytes, s: Stream, tail: List<&1, Stream>, input: Bytes.Bytes,
1108 encoder: Hpack.State, decoder: Hpack.State, control: Control) -> Reply:
1109 match m:
1110 case None{}:
1111 client.fail(1, True{}, 0)
1112 case Some{writes}:
1113 client.data.content(end_stream, content, writes, s, tail, input, encoder, decoder, control)
1114
1115def client.data.stream.state(open: Bool, available: Bool, +id: U32, +len: U32,
1116 +end_stream: Bool, content: Bytes.Bytes, s: Stream, tail: List<&1, Stream>,
1117 input: Bytes.Bytes, encoder: Hpack.State, decoder: Hpack.State, control: Control) -> Reply:
1118 match open:
1119 case False{}:
1120 client.fail(1, False{}, id)
1121 case True{}:
1122 match available:
1123 case False{}:
1124 client.fail(3, False{}, id)
1125 case True{}:
1126 client.data.content.updated(client.data.updates(len, end_stream, id),
1127 end_stream, content, s, tail, input, encoder, decoder, control)
1128
1129def client.data.stream(+len: U32, end_stream: Bool, content: Bytes.Bytes,
1130 s: Stream, tail: List<&1, Stream>, input: Bytes.Bytes,
1131 encoder: Hpack.State, decoder: Hpack.State, control: Control) -> Reply:
1132 Stream{+id, send_window, +recv_window, local_end, +remote_end, +head,
1133 headers, partial, body, outgoing} = s
1134 client.data.stream.state(Bool.and(head, Bool.not(remote_end)), U32.is_le(len, recv_window),
1135 id, len, end_stream, content,
1136 Stream{id, send_window, recv_window, local_end, remote_end, head,
1137 headers, partial, body, outgoing}, tail, input, encoder, decoder, control)
1138
1139def client.data.connection(valid: Bool, +id: U32, +flags: U32, +len: U32,
1140 payload: Bytes.Bytes, c: Client) -> Reply:
1141 match valid:
1142 case False{}:
1143 client.fail(3, True{}, 0)
1144 case True{}:
1145 Client{input, streams, encoder, decoder, control} = c
1146 content = client.fragment(0, flags, payload)
1147 client.streams.use(streams, id, s => tail =>
1148 client.data.stream(len, has(flags, 1), content, s, tail, input, encoder, decoder, control))
1149
1150def client.data.receive(+id: U32, +flags: U32, payload: Bytes.Bytes, c: Client) -> Reply:
1151 Bytes.Bytes{+len, buf} = payload
1152 Client{input, streams, encoder, decoder, +control} = c
1153 Control{next, active, frame_max, initial, max_streams, connection, +recv_window,
1154 first, ack, continuation, goaway} = control
1155 client.data.connection(U32.is_le(len, recv_window), id, flags, len, Bytes.Bytes{len, buf},
1156 Client{input, streams, encoder, decoder, control})
1157
1158def client.continuation.expected(m: Maybe<&2, U32>, +kind: U32, id: U32) -> Bool:
1159 match m:
1160 case None{}:
1161 U32.is_ne(kind, 9)
1162 case Some{expected}:
1163 Bool.and(U32.is_eq(kind, 9), U32.is_eq(expected, id))
1164
1165def client.apply.allowed(c: Control, +kind: U32, flags: U32, id: U32) -> Bool:
1166 Control{next, active, frame_max, initial, max_streams, connection, recv,
1167 first, ack, continuation, goaway} = c
1168 Bool.and(Bool.or(Bool.not(first), Bool.and(U32.is_eq(kind, 4), U32.is_eq(flags, 0))),
1169 client.continuation.expected(continuation, kind, id))
1170
1171def client.apply.frame(kind: U32, flags: U32, id: U32, payload: Bytes.Bytes, c: Client) -> Reply:
1172 match kind:
1173 case 0:
1174 client.data.receive(id, flags, payload, c)
1175 case 1:
1176 client.headers.receive(id, flags, payload, c)
1177 case 2:
1178 client.ok(c)
1179 case 3:
1180 client.rst(id, payload, c)
1181 case 4:
1182 client.settings(flags, payload, c)
1183 case 5:
1184 client.fail(1, True{}, 0)
1185 case 6:
1186 client.ping(flags, payload, c)
1187 case 7:
1188 client.goaway(payload, c)
1189 case 8:
1190 client.window.update(id, payload, c)
1191 case 9:
1192 client.continuation(id, flags, payload, c)
1193 case _:
1194 client.ok(c)
1195
1196def client.apply.checked(valid: Bool, kind: U32, flags: U32, id: U32,
1197 payload: Bytes.Bytes, c: Client) -> Reply:
1198 match valid:
1199 case False{}:
1200 client.fail(1, True{}, 0)
1201 case True{}:
1202 client.apply.frame(kind, flags, id, payload, c)
1203
1204def client.apply(f: Frame, c: Client) -> Reply:
1205 Frame{+kind, +flags, +id, payload} = f
1206 Client{input, streams, encoder, decoder, +control} = c
1207 client.apply.checked(client.apply.allowed(control, kind, flags, id), kind, flags, id,
1208 payload, Client{input, streams, encoder, decoder, control})
1209
1210def client.receive.advance(c: Client,
1211 k: Decode -> List<&1, Stream> -> Hpack.State -> Hpack.State -> Control -> Reply) -> Reply:
1212 Client{input, streams, encoder, decoder, control} = c
1213 k(parse(16384, input), streams, encoder, decoder, control)
1214
1215def client.receive.go(n: Nat, d: Decode, streams: List<&1, Stream>,
1216 encoder: Hpack.State, decoder: Hpack.State, control: Control,
1217 rev_writes: List<&1, Bytes.Bytes>, rev_events: List<&1, ClientEvent>) -> Reply:
1218 match n:
1219 case 0n:
1220 client.fail(1, True{}, 0)
1221 case 1n+p:
1222 match d:
1223 case Need{input}:
1224 Advanced{Client{input, streams, encoder, decoder, control},
1225 List.reverse(&1, Bytes.Bytes, rev_writes), List.reverse(&1, ClientEvent, rev_events)}
1226 case Bad{error}:
1227 Failed{error}
1228 case Got{frame, rest}:
1229 client.after(client.apply(frame, Client{rest, streams, encoder, decoder, control}),
1230 next => writes => events =>
1231 client.receive.advance(next, more => streams2 => enc => dec => ctl =>
1232 client.receive.go(p, more, streams2, enc, dec, ctl,
1233 List.append(&1, Bytes.Bytes, List.reverse(&1, Bytes.Bytes, writes), rev_writes),
1234 List.append(&1, ClientEvent, List.reverse(&1, ClientEvent, events), rev_events))))
1235
1236def client.receive.buffer(b: Bytes.Bytes, streams: List<&1, Stream>,
1237 encoder: Hpack.State, decoder: Hpack.State, control: Control) -> Reply:
1238 Bytes.Bytes{+len, buf} = b
1239 client.receive.go(1n+U32.to_nat(len), parse(16384, Bytes.Bytes{len, buf}),
1240 streams, encoder, decoder, control, Nil{}, Nil{})
1241
1242def client.receive(c: Client, chunk: Bytes.Bytes) -> Reply:
1243 Client{input, streams, encoder, decoder, control} = c
1244 client.receive.buffer(Bytes.append(input, chunk), streams, encoder, decoder, control)