src/envelope.bend

bendonnect-rpc@0.1.0.0 · 148 lines · 25 definitions

1# Connect streaming envelope: 1 flags byte + 4-byte big-endian length + message.
2# https://connectrpc.com/docs/protocol/#streaming-request
3#
4# Design notes:
5# - Reserved flag bits (anything besides 1 = compressed, 2 = end-stream) are KEPT in
6#   `flags` by decode (never rejected); the caller decides what to do with them.
7#   encode writes the low 8 bits of `flags`.
8# - decode reports Bad as soon as a header announces a message larger than `max`,
9#   before the body has arrived.
10import Base
11import bend-kit-bytes@0.3.2.0/bytes.bend as Bytes
12
13type Envelope is Type:
14  Envelope{flags: U32, data: Bytes.Bytes}
15
16def flag_compressed() -> U32:
17  1
18
19def flag_end() -> U32:
20  2
21
22# 5-byte header followed by the message.
23def encode.of(+flags: U32, r: Bytes.Bytes & U32) -> Bytes.Bytes:
24  (data, len) = r
25  h1 = Bytes.set(Bytes.new(5), 0, (flags .&. 255 : U32))
26  h2 = Bytes.set.u32be(h1, 1, len)
27  Bytes.append(h2, data)
28
29def encode(e: Envelope) -> Bytes.Bytes:
30  Envelope{+flags, data} = e
31  encode.of(flags, Bytes.length(data))
32
33# Encodes every envelope in parallel, then concatenates.
34def encode_list(es: List<&1, Envelope>) -> List<&1, Bytes.Bytes>:
35  match es:
36    case Nil{}:
37      Nil{}
38    case Con{h, t}:
39      a b = encode(h) encode_list(t)
40      a <> b
41
42def encode_all(es: List<&1, Envelope>) -> Bytes.Bytes:
43  Bytes.concat(encode_list(es))
44
45type Step is Type:
46  Need{buf: Bytes.Bytes}
47  Got{env: Envelope, rest: Bytes.Bytes}
48  Bad{reason: String}
49
50def decode.too_big(len: U32, max: U32) -> String:
51  "message size " ++ U32.show(len) ++ " is larger than configured max " ++ U32.show(max)
52
53def decode.rest(r: Bytes.Bytes & Bytes.Bytes, +flags: U32, data: Bytes.Bytes) -> Step:
54  (b, rest) = r
55  Got{Envelope{flags, data}, rest}
56
57def decode.body(r: Bytes.Bytes & Bytes.Bytes, +flags: U32, +len: U32, +n: U32) -> Step:
58  (b, data) = r
59  decode.rest(Bytes.slice(b, (5 + len : U32), (n - 5 - len : U32)), flags, data)
60
61def decode.cut(buf: Bytes.Bytes, +flags: U32, +len: U32, +n: U32) -> Step:
62  decode.body(Bytes.slice(buf, 5, len), flags, len, n)
63
64def decode.fit(short: Bool, buf: Bytes.Bytes, +flags: U32, +len: U32, +n: U32) -> Step:
65  match short:
66    case True{}:
67      Need{buf}
68    case False{}:
69      decode.cut(buf, flags, len, n)
70
71def decode.size(big: Bool, buf: Bytes.Bytes, +flags: U32, +len: U32, +n: U32, +max: U32) -> Step:
72  match big:
73    case True{}:
74      Bad{decode.too_big(len, max)}
75    case False{}:
76      decode.fit(U32.is_lt((n - 5 : U32), len), buf, flags, len, n)
77
78def decode.len(r: Bytes.Bytes & Maybe<&2, U32>, +flags: U32, +n: U32, +max: U32) -> Step:
79  (buf, m) = r
80  match m:
81    case None{}:
82      Bad{"truncated envelope header"}
83    case Some{+len}:
84      decode.size(U32.is_gt(len, max), buf, flags, len, n, max)
85
86def decode.flags(r: Bytes.Bytes & Maybe<&2, U32>, +n: U32, +max: U32) -> Step:
87  (buf, m) = r
88  match m:
89    case None{}:
90      Bad{"truncated envelope header"}
91    case Some{+flags}:
92      decode.len(Bytes.get.u32be(buf, 1), flags, n, max)
93
94def decode.hdr(short: Bool, buf: Bytes.Bytes, +n: U32, +max: U32) -> Step:
95  match short:
96    case True{}:
97      Need{buf}
98    case False{}:
99      decode.flags(Bytes.get(buf, 0), n, max)
100
101# Reads one envelope off the front of buf: Need when more bytes are required, Got with
102# the remaining bytes, Bad when the announced size exceeds max.
103def decode.start(r: Bytes.Bytes & U32, +max: U32) -> Step:
104  (b, +n) = r
105  decode.hdr(U32.is_lt(n, 5), b, n, max)
106
107def decode(buf: Bytes.Bytes, +max: U32) -> Step:
108  decode.start(Bytes.length(buf), max)
109
110def decode_all.cons(env: Envelope, r: Result<&2, &1, String, List<&1, Envelope>>) -> Result<&2, &1, String, List<&1, Envelope>>:
111  match r:
112    case Fail{e}:
113      Fail{e}
114    case Done{l}:
115      Done{env <> l}
116
117def decode_all.end(empty: Bool) -> Result<&2, &1, String, List<&1, Envelope>>:
118  match empty:
119    case True{}:
120      Done{Nil{}}
121    case False{}:
122      Fail{"incomplete envelope at end of body"}
123
124def decode_all.fin(r: Bytes.Bytes & U32) -> Result<&2, &1, String, List<&1, Envelope>>:
125  (b, +n) = r
126  decode_all.end(U32.is_eq(n, 0))
127
128# fuel: every envelope takes at least 5 bytes, so (bytes + 1) steps always suffice.
129def decode_all.go(fuel: Nat, s: Step, +max: U32) -> Result<&2, &1, String, List<&1, Envelope>>:
130  match fuel:
131    case 0n:
132      Fail{"envelope decoder out of fuel"}
133    case 1n+p:
134      match s:
135        case Need{b}:
136          decode_all.fin(Bytes.length(b))
137        case Bad{r}:
138          Fail{r}
139        case Got{env, rest}:
140          decode_all.cons(env, decode_all.go(p, decode(rest, max), max))
141
142def decode_all.start(r: Bytes.Bytes & U32, +max: U32) -> Result<&2, &1, String, List<&1, Envelope>>:
143  (b, +n) = r
144  decode_all.go(1n+U32.to_nat(n), decode(b, max), max)
145
146# Every envelope of a complete body; a trailing partial envelope is a Fail.
147def decode_all(buf: Bytes.Bytes, +max: U32) -> Result<&2, &1, String, List<&1, Envelope>>:
148  decode_all.start(Bytes.length(buf), max)