src/wire.bend

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