src/wire.bend
emerging-ezhttp@0.18.0.0 · 303 lines · 0 definitions
1# ezhttp/src/wire: one request out and one response back, over
2# bend-kit-wire's sockets. Its effects park on the runtime's event loop, so
3# exchanges run side by side under IO.fork. bend-kit-dns resolves the host;
4# TLS is bend-kit-wire's (OpenSSL 3 loaded at run time, BEND_LIBSSL to point
5# at it).
6#
7# An exchange's answer is a status on its first line, then text: "0" and the
8# raw response, or a non-zero status and the reason (wirespec.bend
9# wire.fail). Every decision is wirespec.bend's: the deadline (wire.ms.of),
10# the octets sent (wire.payload), what a failed call is reported as
11# (wire.fail) and when the response is whole (wire.read.next), so a server
12# that keeps the connection open after a complete response is not waited on
13# (SPEC.md HTTP-CLI-1, 6, 10, 13; HTTP-TRUST-3, 5).
14import Base
15import bend-kit-wire@0.4.6.1/wire.bend as W
16import bend-kit-bytes@0.3.2.0/bytes.bend as Bytes
17import bend-kit-dns@0.6.0.2/dns.bend as Dns
18import emerging-ezhttp@0.18.0.0/src/wirespec.bend as Spec
19
20# the most octets one receive asks for
21def wire.max() -> U32:
22 65536
23
24# the most receives one exchange makes
25def wire.reads() -> Nat:
26 65536n
27
28# --- the socket calls, plain or over TLS ---
29
30# close a socket, ending its TLS session when it has one
31def wire.close(secure: Bool, s: Socket) -> IO(Unit):
32 match secure:
33 case True{}:
34 W.tls.close(s)
35 case False{}:
36 Socket.close(s)
37
38# the next octets a socket has received, packed four to a word
39def wire.recv(secure: Bool, s: Socket, ms: U32)
40 -> IO(Socket & Result<&1, &1, U32 & String, U32 & Array<U32>>):
41 match secure:
42 case True{}:
43 W.tls.recv.words(s, wire.max(), ms)
44 case False{}:
45 W.recv.words(s, wire.max(), ms)
46
47# a send.words.deadline answer without its count of octets accepted
48def wire.sent(r: Socket & (U32 & Result<&1, &1, U32 & String, Unit>))
49 -> IO(Socket & Result<&1, &1, U32 & String, Unit>):
50 (s, (n, res)) = r
51 IO.pure(Socket & Result<&1, &1, U32 & String, Unit>, (s, res))
52
53# send octets, the plain send bounded by the deadline
54def wire.send(secure: Bool, s: Socket, +len: U32, buf: Array<U32>, ms: U32)
55 -> IO(Socket & Result<&1, &1, U32 & String, Unit>):
56 match secure:
57 case True{}:
58 W.tls.send.words(s, len, buf)
59 case False{}:
60 IO.bind(Socket & (U32 & Result<&1, &1, U32 & String, Unit>),
61 Socket & Result<&1, &1, U32 & String, Unit>,
62 W.send.words.deadline(s, len, buf, ms), wire.sent)
63
64# the TLS handshake with the host, when the exchange is secure
65def wire.shake(secure: Bool, s: Socket, host: String, ms: U32)
66 -> IO(Socket & Result<&1, &1, U32 & String, Unit>):
67 match secure:
68 case True{}:
69 W.tls.connect(s, host, ms)
70 case False{}:
71 IO.pure(Socket & Result<&1, &1, U32 & String, Unit>, (s, Done{Unit{}}))
72
73# --- reading the answer ---
74
75# a read that answered: its answer
76def wire.end.of(r: Spec.Read) -> String:
77 match r:
78 case Spec.Heard{answer}:
79 answer
80 case Spec.Again{acc}:
81 Spec.wire.spent()
82
83# the held text went in: now the close
84def wire.end.more(r: Spec.Read) -> String:
85 match r:
86 case Spec.Heard{answer}:
87 answer
88 case Spec.Again{acc}:
89 wire.end.of(Spec.wire.read.next(acc, "0\n"))
90
91# what the server's close leaves: the text still held, then the close
92def wire.end(acc: Spec.Got, +text: String, none: Bool) -> String:
93 match none:
94 case True{}:
95 wire.end.of(Spec.wire.read.next(acc, "0\n"))
96 case False{}:
97 wire.end.more(Spec.wire.read.next(acc, "0\n" ++ text))
98
99# a judged read: read again, or close and answer
100def wire.loop.on(+secure: Bool, s: Socket, r: Spec.Read,
101 more: Socket -> Spec.Got -> IO(String)) -> IO(String):
102 match r:
103 case Spec.Again{acc}:
104 more(s)(acc)
105 case Spec.Heard{answer}:
106 IO.bind(Unit, String, wire.close(secure, s),
107 _u => IO.pure(String, answer))
108
109# decoded text once it is known whether it is empty: read again when the
110# octets finished no character, else judge the text
111def wire.loop.judge(+secure: Bool, s: Socket, acc: Spec.Got, +text: String,
112 held: List<&2, U32>, none: Bool,
113 more: Socket -> List<&2, U32> -> Spec.Got -> IO(String)) -> IO(String):
114 match none:
115 case True{}:
116 more(s)(held)(acc)
117 case False{}:
118 wire.loop.on(secure, s, Spec.wire.read.next(acc, "0\n" ++ text),
119 s2 => a2 => more(s2)(held)(a2))
120
121# octets that arrived, decoded with what was held
122def wire.loop.text(+secure: Bool, s: Socket, acc: Spec.Got,
123 t: String & List<&2, U32>,
124 more: Socket -> List<&2, U32> -> Spec.Got -> IO(String)) -> IO(String):
125 (+text, held) = t
126 wire.loop.judge(secure, s, acc, text, held, String.is_empty(text), more)
127
128# the server's close: what was held, decoded as it is, then the close
129def wire.loop.closed(+secure: Bool, s: Socket, held: List<&2, U32>,
130 acc: Spec.Got) -> IO(String):
131 +text = Spec.wire.utf8.flush(held)
132 IO.bind(Unit, String, wire.close(secure, s),
133 _u => IO.pure(String, wire.end(acc, text, String.is_empty(text))))
134
135# a receive's octets, or the server's close
136def wire.loop.got(+secure: Bool, s: Socket, held: List<&2, U32>,
137 acc: Spec.Got, +len: U32, buf: Array<U32>, closed: Bool,
138 more: Socket -> List<&2, U32> -> Spec.Got -> IO(String)) -> IO(String):
139 match closed:
140 case True{}:
141 wire.loop.closed(secure, s, held, acc)
142 case False{}:
143 wire.loop.text(secure, s, acc,
144 Spec.wire.utf8.feed(held, Bytes.to_string(Bytes.Bytes{len, buf})),
145 more)
146
147# a receive's outcome: a failure is the answer
148def wire.loop.un(+secure: Bool, +host: String, +port: U32, +ms: U32,
149 held: List<&2, U32>, acc: Spec.Got,
150 r: Socket & Result<&1, &1, U32 & String, U32 & Array<U32>>,
151 more: Socket -> List<&2, U32> -> Spec.Got -> IO(String)) -> IO(String):
152 (s, res) = r
153 match res:
154 case Fail{(n, why)}:
155 IO.bind(Unit, String, wire.close(secure, s),
156 _u => IO.pure(String,
157 Spec.wire.fail(Spec.Receive{}, host, port, ms, n, why)))
158 case Done{(+len, buf)}:
159 wire.loop.got(secure, s, held, acc, len, buf, U32.is_zero(len), more)
160
161# receive until wirespec.bend says the answer is whole
162def wire.loop(fuel: Nat, +secure: Bool, s: Socket, +host: String, +port: U32,
163 +ms: U32, held: List<&2, U32>, acc: Spec.Got) -> IO(String):
164 match fuel:
165 case 0n:
166 IO.bind(Unit, String, wire.close(secure, s),
167 _u => IO.pure(String, Spec.wire.spent()))
168 case 1n+p:
169 IO.bind(Socket & Result<&1, &1, U32 & String, U32 & Array<U32>>, String,
170 wire.recv(secure, s, ms),
171 r => wire.loop.un(secure, host, port, ms, held, acc, r,
172 s2 => h2 => a2 => wire.loop(p, secure, s2, host, port, ms, h2, a2)))
173
174# --- one exchange ---
175
176# after the send: read the answer, or the send's failure
177def wire.after.send(+secure: Bool, +host: String, +port: U32, +ms: U32,
178 +method: String, r: Socket & Result<&1, &1, U32 & String, Unit>)
179 -> IO(String):
180 (s, res) = r
181 match res:
182 case Fail{(n, why)}:
183 IO.bind(Unit, String, wire.close(secure, s),
184 _u => IO.pure(String,
185 Spec.wire.fail(Spec.Send{}, host, port, ms, n, why)))
186 case Done{_u}:
187 wire.loop(wire.reads(), secure, s, host, port, ms, [],
188 Spec.wire.got(method, []))
189
190# send the octets an exchange carries, then read its answer
191def wire.put(+secure: Bool, s: Socket, +host: String, +port: U32, +ms: U32,
192 +method: String, payload: Bytes.Bytes) -> IO(String):
193 Bytes.Bytes{+len, buf} = payload
194 IO.bind(Socket & Result<&1, &1, U32 & String, Unit>, String,
195 wire.send(secure, s, len, buf, ms),
196 r => wire.after.send(secure, host, port, ms, method, r))
197
198# after the handshake: send, or the handshake's failure (the socket closed
199# plainly, as no TLS session was made)
200def wire.after.shake(+secure: Bool, +host: String, +port: U32, +ms: U32,
201 +method: String, payload: String,
202 r: Socket & Result<&1, &1, U32 & String, Unit>) -> IO(String):
203 (s, res) = r
204 match res:
205 case Fail{(n, why)}:
206 IO.bind(Unit, String, Socket.close(s),
207 _u => IO.pure(String,
208 Spec.wire.fail(Spec.Handshake{}, host, port, ms, n, why)))
209 case Done{_u}:
210 wire.put(secure, s, host, port, ms, method, Bytes.from_string(payload))
211
212# after the connect: the handshake, or the connect's failure
213def wire.after.connect(+secure: Bool, +host: String, +port: U32, +ms: U32,
214 +method: String, payload: String,
215 r: Result<&1, &1, U32 & String, Socket>) -> IO(String):
216 match r:
217 case Fail{(n, why)}:
218 IO.pure(String, Spec.wire.fail(Spec.Connect{}, host, port, ms, n, why))
219 case Done{s}:
220 IO.bind(Socket & Result<&1, &1, U32 & String, Unit>, String,
221 wire.shake(secure, s, host, ms),
222 r2 => wire.after.shake(secure, host, port, ms, method, payload, r2))
223
224# a connect that failed tries the next address; one that worked is kept
225def wire.dial.next(r: Result<&1, &1, U32 & String, Socket>,
226 more: Unit -> IO(Result<&1, &1, U32 & String, Socket>))
227 -> IO(Result<&1, &1, U32 & String, Socket>):
228 match r:
229 case Done{s}:
230 IO.pure(Result<&1, &1, U32 & String, Socket>, Done{s})
231 case Fail{e}:
232 more(Unit{})
233
234# connect to each address in turn until one connects; the last one's
235# failure when none does
236def wire.dial(rest: List<&2, String>, +first: String, +port: U32, +ms: U32)
237 -> IO(Result<&1, &1, U32 & String, Socket>):
238 match rest:
239 case []:
240 W.connect(first, port, ms)
241 case ip <> t:
242 IO.bind(Result<&1, &1, U32 & String, Socket>,
243 Result<&1, &1, U32 & String, Socket>,
244 W.connect(first, port, ms),
245 r => wire.dial.next(r, _u => wire.dial(t, ip, port, ms)))
246
247# after the resolve: connect, or the resolve's failure (no address at all is
248# ENOENT, as bend-kit-dns reports a name with none)
249def wire.after.resolve(+secure: Bool, +host: String, +port: U32, +ms: U32,
250 +method: String, payload: String,
251 r: Result<&1, &1, U32 & String, List<&2, String>>) -> IO(String):
252 match r:
253 case Fail{(n, why)}:
254 IO.pure(String, Spec.wire.fail(Spec.Resolve{}, host, port, ms, n, why))
255 case Done{[]}:
256 IO.pure(String, Spec.wire.fail(Spec.Resolve{}, host, port, ms, 2, ""))
257 case Done{ip <> t}:
258 IO.bind(Result<&1, &1, U32 & String, Socket>, String,
259 wire.dial(t, ip, port, ms),
260 c => wire.after.connect(secure, host, port, ms, method, payload, c))
261
262# the OS resolver's addresses, NUL-separated, as a list
263def wire.looked(r: Result<&1, &1, U32 & String, String>)
264 -> IO(Result<&1, &1, U32 & String, List<&2, String>>):
265 match r:
266 case Fail{e}:
267 IO.pure(Result<&1, &1, U32 & String, List<&2, String>>, Fail{e})
268 case Done{text}:
269 IO.pure(Result<&1, &1, U32 & String, List<&2, String>>,
270 Done{Dns.lookup.split(text)})
271
272# a host's addresses: a numeric address or localhost as it is, else the OS
273# resolver's answer
274def wire.resolve(+host: String, literal: List<&2, String>)
275 -> IO(Result<&1, &1, U32 & String, List<&2, String>>):
276 match literal:
277 case []:
278 IO.bind(Result<&1, &1, U32 & String, String>,
279 Result<&1, &1, U32 & String, List<&2, String>>,
280 Dns.lookup.all(String.to_lower(host)), wire.looked)
281 case ip <> t:
282 IO.pure(Result<&1, &1, U32 & String, List<&2, String>>, Done{ip <> t})
283
284# a request with this method, sent to a host as wirespec.bend says, with
285# its tail of octets after it, every call bounded by the deadline; its whole
286# answer
287def wire.run(+secure: Bool, +ms: U32, +host: String, +port: U32,
288 +method: String, req: String, tail: List<&2, U32>) -> IO(String):
289 IO.bind(Result<&1, &1, U32 & String, List<&2, String>>, String,
290 wire.resolve(host, Dns.resolve.literal.all(host)),
291 r => wire.after.resolve(secure, host, port, ms, method,
292 Spec.wire.payload(req, tail), r))
293
294# a request sent to a host, its tail of octets after it, and its whole answer
295# read back, with the default deadline
296def wire.talk(secure: Bool, host: String, port: U32, method: String,
297 req: String, tail: List<&2, U32>) -> IO(String):
298 wire.run(secure, Spec.wire.ms.of(None{}), host, port, method, req, tail)
299
300# the same exchange, bounded by a timeout in milliseconds
301def wire.talk.timed(secure: Bool, ms: U32, host: String, port: U32,
302 method: String, req: String, tail: List<&2, U32>) -> IO(String):
303 wire.run(secure, Spec.wire.ms.of(Some{ms}), host, port, method, req, tail)