src/wire.bend
emerging-ezhttp@0.19.0.0 · 253 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.19.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 judged read: read again, or close and answer
76def wire.loop.on(+secure: Bool, s: Socket, r: Spec.Read,
77 more: Socket -> Spec.Got -> IO(String)) -> IO(String):
78 match r:
79 case Spec.Again{acc}:
80 more(s)(acc)
81 case Spec.Heard{answer}:
82 IO.bind(Unit, String, wire.close(secure, s),
83 _u => IO.pure(String, answer))
84
85# a receive's octets, or the server's close (no octets)
86def wire.loop.got(+secure: Bool, s: Socket, acc: Spec.Got, len: U32,
87 buf: Array<U32>, more: Socket -> Spec.Got -> IO(String)) -> IO(String):
88 wire.loop.on(secure, s,
89 Spec.wire.read.next(acc, "0\n" ++ Bytes.to_string(Bytes.Bytes{len, buf})),
90 more)
91
92# a receive's outcome: a failure is the answer
93def wire.loop.un(+secure: Bool, +host: String, +port: U32, +ms: U32,
94 acc: Spec.Got,
95 r: Socket & Result<&1, &1, U32 & String, U32 & Array<U32>>,
96 more: Socket -> Spec.Got -> IO(String)) -> IO(String):
97 (s, res) = r
98 match res:
99 case Fail{(n, why)}:
100 IO.bind(Unit, String, wire.close(secure, s),
101 _u => IO.pure(String,
102 Spec.wire.fail(Spec.Receive{}, host, port, ms, n, why)))
103 case Done{(len, buf)}:
104 wire.loop.got(secure, s, acc, len, buf, more)
105
106# receive until wirespec.bend says the answer is whole
107def wire.loop(fuel: Nat, +secure: Bool, s: Socket, +host: String, +port: U32,
108 +ms: U32, acc: Spec.Got) -> IO(String):
109 match fuel:
110 case 0n:
111 IO.bind(Unit, String, wire.close(secure, s),
112 _u => IO.pure(String, Spec.wire.spent()))
113 case 1n+p:
114 IO.bind(Socket & Result<&1, &1, U32 & String, U32 & Array<U32>>, String,
115 wire.recv(secure, s, ms),
116 r => wire.loop.un(secure, host, port, ms, acc, r,
117 s2 => a2 => wire.loop(p, secure, s2, host, port, ms, a2)))
118
119# --- one exchange ---
120
121# after the send: read the answer, or the send's failure
122def wire.after.send(+secure: Bool, +host: String, +port: U32, +ms: U32,
123 +method: String, r: Socket & Result<&1, &1, U32 & String, Unit>)
124 -> IO(String):
125 (s, res) = r
126 match res:
127 case Fail{(n, why)}:
128 IO.bind(Unit, String, wire.close(secure, s),
129 _u => IO.pure(String,
130 Spec.wire.fail(Spec.Send{}, host, port, ms, n, why)))
131 case Done{_u}:
132 wire.loop(wire.reads(), secure, s, host, port, ms,
133 Spec.wire.got(method, []))
134
135# send the octets an exchange carries, then read its answer
136def wire.put(+secure: Bool, s: Socket, +host: String, +port: U32, +ms: U32,
137 +method: String, payload: Bytes.Bytes) -> IO(String):
138 Bytes.Bytes{+len, buf} = payload
139 IO.bind(Socket & Result<&1, &1, U32 & String, Unit>, String,
140 wire.send(secure, s, len, buf, ms),
141 r => wire.after.send(secure, host, port, ms, method, r))
142
143# after the handshake: send, or the handshake's failure (the socket closed
144# plainly, as no TLS session was made)
145def wire.after.shake(+secure: Bool, +host: String, +port: U32, +ms: U32,
146 +method: String, payload: Bytes.Bytes,
147 r: Socket & Result<&1, &1, U32 & String, Unit>) -> IO(String):
148 (s, res) = r
149 match res:
150 case Fail{(n, why)}:
151 IO.bind(Unit, String, Socket.close(s),
152 _u => IO.pure(String,
153 Spec.wire.fail(Spec.Handshake{}, host, port, ms, n, why)))
154 case Done{_u}:
155 wire.put(secure, s, host, port, ms, method, payload)
156
157# after the connect: the handshake, or the connect's failure
158def wire.after.connect(+secure: Bool, +host: String, +port: U32, +ms: U32,
159 +method: String, payload: Bytes.Bytes,
160 r: Result<&1, &1, U32 & String, Socket>) -> IO(String):
161 match r:
162 case Fail{(n, why)}:
163 IO.pure(String, Spec.wire.fail(Spec.Connect{}, host, port, ms, n, why))
164 case Done{s}:
165 IO.bind(Socket & Result<&1, &1, U32 & String, Unit>, String,
166 wire.shake(secure, s, host, ms),
167 r2 => wire.after.shake(secure, host, port, ms, method, payload, r2))
168
169# a connect that failed tries the next address; one that worked is kept
170def wire.dial.next(r: Result<&1, &1, U32 & String, Socket>,
171 more: Unit -> IO(Result<&1, &1, U32 & String, Socket>))
172 -> IO(Result<&1, &1, U32 & String, Socket>):
173 match r:
174 case Done{s}:
175 IO.pure(Result<&1, &1, U32 & String, Socket>, Done{s})
176 case Fail{e}:
177 more(Unit{})
178
179# connect to each address in turn until one connects; the last one's
180# failure when none does
181def wire.dial(rest: List<&2, String>, +first: String, +port: U32, +ms: U32)
182 -> IO(Result<&1, &1, U32 & String, Socket>):
183 match rest:
184 case []:
185 W.connect(first, port, ms)
186 case ip <> t:
187 IO.bind(Result<&1, &1, U32 & String, Socket>,
188 Result<&1, &1, U32 & String, Socket>,
189 W.connect(first, port, ms),
190 r => wire.dial.next(r, _u => wire.dial(t, ip, port, ms)))
191
192# after the resolve: connect, or the resolve's failure (no address at all is
193# ENOENT, as bend-kit-dns reports a name with none)
194def wire.after.resolve(+secure: Bool, +host: String, +port: U32, +ms: U32,
195 +method: String, payload: Bytes.Bytes,
196 r: Result<&1, &1, U32 & String, List<&2, String>>) -> IO(String):
197 match r:
198 case Fail{(n, why)}:
199 IO.pure(String, Spec.wire.fail(Spec.Resolve{}, host, port, ms, n, why))
200 case Done{[]}:
201 IO.pure(String, Spec.wire.fail(Spec.Resolve{}, host, port, ms, 2, ""))
202 case Done{ip <> t}:
203 IO.bind(Result<&1, &1, U32 & String, Socket>, String,
204 wire.dial(t, ip, port, ms),
205 c => wire.after.connect(secure, host, port, ms, method, payload, c))
206
207# the OS resolver's addresses, NUL-separated, as a list
208def wire.looked(r: Result<&1, &1, U32 & String, String>)
209 -> IO(Result<&1, &1, U32 & String, List<&2, String>>):
210 match r:
211 case Fail{e}:
212 IO.pure(Result<&1, &1, U32 & String, List<&2, String>>, Fail{e})
213 case Done{text}:
214 IO.pure(Result<&1, &1, U32 & String, List<&2, String>>,
215 Done{Dns.lookup.split(text)})
216
217# a host's addresses: a numeric address or localhost as it is, else the OS
218# resolver's answer
219def wire.resolve(+host: String, literal: List<&2, String>)
220 -> IO(Result<&1, &1, U32 & String, List<&2, String>>):
221 match literal:
222 case []:
223 IO.bind(Result<&1, &1, U32 & String, String>,
224 Result<&1, &1, U32 & String, List<&2, String>>,
225 Dns.lookup.all(String.to_lower(host)), wire.looked)
226 case ip <> t:
227 IO.pure(Result<&1, &1, U32 & String, List<&2, String>>, Done{ip <> t})
228
229# the exchange once its octets are packed
230def wire.run.with(+secure: Bool, +ms: U32, +host: String, +port: U32,
231 +method: String, payload: Bytes.Bytes) -> IO(String):
232 IO.bind(Result<&1, &1, U32 & String, List<&2, String>>, String,
233 wire.resolve(host, Dns.resolve.literal.all(host)),
234 r => wire.after.resolve(secure, host, port, ms, method, payload, r))
235
236# a request with this method, sent to a host as wirespec.bend says, with
237# its tail of octets after it, every call bounded by the deadline; its whole
238# answer
239def wire.run(+secure: Bool, +ms: U32, +host: String, +port: U32,
240 +method: String, req: String, tail: Bytes.Bytes) -> IO(String):
241 wire.run.with(secure, ms, host, port, method,
242 Bytes.append(Bytes.from_string(Spec.wire.payload(req)), tail))
243
244# a request sent to a host, its tail of octets after it, and its whole answer
245# read back, with the default deadline
246def wire.talk(secure: Bool, host: String, port: U32, method: String,
247 req: String, tail: Bytes.Bytes) -> IO(String):
248 wire.run(secure, Spec.wire.ms.of(None{}), host, port, method, req, tail)
249
250# the same exchange, bounded by a timeout in milliseconds
251def wire.talk.timed(secure: Bool, ms: U32, host: String, port: U32,
252 method: String, req: String, tail: Bytes.Bytes) -> IO(String):
253 wire.run(secure, Spec.wire.ms.of(Some{ms}), host, port, method, req, tail)