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.
10	import Base
11	import bend-kit-bytes@0.3.2.0/bytes.bend as Bytes
12	
13	type Envelope is Type:
14	  Envelope{flags: U32, data: Bytes.Bytes}
15	
16	def flag_compressed() -> U32:
17	  1
18	
19	def flag_end() -> U32:
20	  2
21	
22	# 5-byte header followed by the message.
23	def 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	
29	def 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.
34	def 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	
42	def encode_all(es: List<&1, Envelope>) -> Bytes.Bytes:
43	  Bytes.concat(encode_list(es))
44	
45	type Step is Type:
46	  Need{buf: Bytes.Bytes}
47	  Got{env: Envelope, rest: Bytes.Bytes}
48	  Bad{reason: String}
49	
50	def decode.too_big(len: U32, max: U32) -> String:
51	  "message size " ++ U32.show(len) ++ " is larger than configured max " ++ U32.show(max)
52	
53	def decode.rest(r: Bytes.Bytes & Bytes.Bytes, +flags: U32, data: Bytes.Bytes) -> Step:
54	  (b, rest) = r
55	  Got{Envelope{flags, data}, rest}
56	
57	def 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	
61	def decode.cut(buf: Bytes.Bytes, +flags: U32, +len: U32, +n: U32) -> Step:
62	  decode.body(Bytes.slice(buf, 5, len), flags, len, n)
63	
64	def 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	
71	def 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	
78	def 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	
86	def 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	
94	def 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.
103	def 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	
107	def decode(buf: Bytes.Bytes, +max: U32) -> Step:
108	  decode.start(Bytes.length(buf), max)
109	
110	def 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	
117	def 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	
124	def 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.
129	def 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	
142	def 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.
147	def decode_all(buf: Bytes.Bytes, +max: U32) -> Result<&2, &1, String, List<&1, Envelope>>:
148	  decode_all.start(Bytes.length(buf), max)