src/server.bend
emerging-ezhttp@0.19.0.0 · 385 lines · 0 definitions
1# ezhttp/src/server: listen, accept, parse one request, write one response.
2# Uses Base TCP (no TLS in v0). One exchange per connection with
3# Connection: close, one connection at a time. The handler returns IO; a pure
4# one is lifted. Every decision after an effect is step.bend's, so a failed
5# accept, recv or send loses that connection, logs one line and the server
6# takes the next one (HTTP-SRV-4).
7import Base
8import emerging-ezhttp@0.19.0.0/src/http.bend as Http
9import emerging-ezhttp@0.19.0.0/src/cors.bend as Cors
10import emerging-ezhttp@0.19.0.0/src/step.bend as Step
11import bend-kit-wire@0.4.6.1/wire.bend as W
12
13# default recv budget for one request message
14def server.max() -> U32:
15 65536
16
17# listen on a host address and port, e.g. "127.0.0.1" or "0.0.0.0"
18# (RFC 9112 connection establishment stays at the socket)
19def server.listen(host: String, port: U32) -> IO(Listener):
20 IO.try(Listener, TCP.listen(host, port))
21
22# accept one connection; a failed accept is a lost connection
23def server.accept.un(
24 pair: Listener & Result<&1, &1, U32 & String, Socket>
25) -> IO(Listener & Step.Got<Socket>):
26 (lst, res) = pair
27 IO.pure(Listener & Step.Got<Socket>, (lst, Step.step.got(Socket, "accept", res)))
28
29def server.accept(listener: Listener) -> IO(Listener & Step.Got<Socket>):
30 IO.bind(
31 Listener & Result<&1, &1, U32 & String, Socket>,
32 Listener & Step.Got<Socket>,
33 TCP.accept(listener),
34 server.accept.un)
35
36# the short-body refusal, so a partial recv can continue
37def server.read.why() -> String:
38 "the request body stopped before its Content-Length"
39
40# whether this refusal is a body that has not arrived yet
41def server.read.short(why: String) -> Bool:
42 String.eq(why, server.read.why())
43
44# a parsed request still wants more bytes
45def server.read.need(req: Http.Request) -> Bool:
46 match req:
47 case Http.Bad{why}:
48 server.read.short(why)
49 case Http.Request{method, target, headers, body}:
50 False{}
51
52# how many extra recvs to chase a Content-Length
53def server.read.fuel() -> Nat:
54 64n
55
56# parse what has arrived
57def server.read.done(sock: Socket, got: String) -> IO(Socket & Step.Got<Http.Request>):
58 IO.pure(Socket & Step.Got<Http.Request>, (sock, Step.Kept{Http.ask(got)}))
59
60# stop on a finished message, or ask for another recv
61def server.read.next(
62 sock: Socket,
63 got: String,
64 req: Http.Request,
65 rest: Socket -> String -> IO(Socket & Step.Got<Http.Request>),
66 more: Bool
67) -> IO(Socket & Step.Got<Http.Request>):
68 match more:
69 case False{}:
70 IO.pure(Socket & Step.Got<Http.Request>, (sock, Step.Kept{req}))
71 case True{}:
72 rest(sock)(got)
73
74# append a chunk and decide whether it completes the request
75def server.read.grew(
76 sock: Socket,
77 +got: String,
78 rest: Socket -> String -> IO(Socket & Step.Got<Http.Request>)
79) -> IO(Socket & Step.Got<Http.Request>):
80 +req = Http.ask(got)
81 server.read.next(sock, got, req, rest, server.read.need(req))
82
83# peer closed, or the chunk extends the buffer
84def server.read.empty(
85 sock: Socket,
86 got: String,
87 chunk: String,
88 rest: Socket -> String -> IO(Socket & Step.Got<Http.Request>),
89 closed: Bool
90) -> IO(Socket & Step.Got<Http.Request>):
91 match closed:
92 case True{}:
93 server.read.done(sock, got)
94 case False{}:
95 server.read.grew(sock, got ++ chunk, rest)
96
97# one chunk of the request message
98def server.read.use(
99 sock: Socket,
100 got: String,
101 +chunk: String,
102 rest: Socket -> String -> IO(Socket & Step.Got<Http.Request>)
103) -> IO(Socket & Step.Got<Http.Request>):
104 server.read.empty(sock, got, chunk, rest, String.eq(chunk, ""))
105
106# None{} is the peer closing its write side; Some is the bytes that arrived
107def server.read.got(
108 sock: Socket,
109 got: String,
110 rest: Socket -> String -> IO(Socket & Step.Got<Http.Request>),
111 chunk: Maybe<&1, String>
112) -> IO(Socket & Step.Got<Http.Request>):
113 match chunk:
114 case None{}:
115 server.read.done(sock, got)
116 case Some{text}:
117 server.read.use(sock, got, text, rest)
118
119# the bytes from one recv; a failed recv loses the connection
120def server.read.pass(
121 sock: Socket,
122 got: String,
123 rest: Socket -> String -> IO(Socket & Step.Got<Http.Request>),
124 chunk: Step.Got<Maybe<&1, String>>
125) -> IO(Socket & Step.Got<Http.Request>):
126 match chunk:
127 case Step.Lost{line}:
128 IO.pure(Socket & Step.Got<Http.Request>, (sock, Step.Lost{line}))
129 case Step.Kept{text}:
130 server.read.got(sock, got, rest, text)
131
132# unpack recv's socket and result
133def server.read.un(
134 got: String,
135 rest: Socket -> String -> IO(Socket & Step.Got<Http.Request>),
136 pair: Socket & Result<&1, &1, U32 & String, Maybe<&1, String>>
137) -> IO(Socket & Step.Got<Http.Request>):
138 (sock, r) = pair
139 server.read.pass(sock, got, rest,
140 Step.step.got(Maybe<&1, String>, "read", r))
141
142# what a receive brought: the octets, or None{} once the peer has closed
143def server.recv.some(+text: String) -> Maybe<&1, String>:
144 Bool.pick(Maybe<&1, String>, String.is_empty(text), None{}, Some{text})
145
146# a receive's answer in the shape the read loop takes
147def server.recv.un(r: Socket & Result<&1, &1, U32 & String, String>)
148 -> IO(Socket & Result<&1, &1, U32 & String, Maybe<&1, String>>):
149 (sock, res) = r
150 match res:
151 case Fail{e}:
152 IO.pure(Socket & Result<&1, &1, U32 & String, Maybe<&1, String>>,
153 (sock, Fail{e}))
154 case Done{text}:
155 IO.pure(Socket & Result<&1, &1, U32 & String, Maybe<&1, String>>,
156 (sock, Done{server.recv.some(text)}))
157
158# the next octets of a request, one character each (bend-kit-wire; no
159# deadline)
160def server.recv(sock: Socket)
161 -> IO(Socket & Result<&1, &1, U32 & String, Maybe<&1, String>>):
162 IO.bind(Socket & Result<&1, &1, U32 & String, String>,
163 Socket & Result<&1, &1, U32 & String, Maybe<&1, String>>,
164 W.recv(sock, server.max(), 0), server.recv.un)
165
166# recv until Content-Length is met, the peer closes, or fuel runs out.
167# fuel is first so the recursive call is a decreasing self-call.
168def server.read.go(fuel: Nat, sock: Socket, got: String) -> IO(Socket & Step.Got<Http.Request>):
169 match fuel:
170 case 0n:
171 server.read.done(sock, got)
172 case 1n+f:
173 IO.bind(
174 Socket & Result<&1, &1, U32 & String, Maybe<&1, String>>,
175 Socket & Step.Got<Http.Request>,
176 server.recv(sock),
177 pair => server.read.un(got, s => g => server.read.go(f, s, g), pair))
178
179# recv then parse an HTTP request (RFC 9112 §3)
180def server.read(sock: Socket) -> IO(Socket & Step.Got<Http.Request>):
181 server.read.go(server.read.fuel(), sock, "")
182
183# write a structured response; a failed send loses the connection
184def server.write.un(
185 pair: Socket & Result<&1, &1, (U32 & String) & List<&2, U32>, Unit>
186) -> IO(Socket & Step.Got<Unit>):
187 (sock, res) = pair
188 IO.pure(Socket & Step.Got<Unit>, (sock, Step.step.sent(res)))
189
190def server.write(sock: Socket, reply: Http.Reply) -> IO(Socket & Step.Got<Unit>):
191 IO.bind(
192 Socket & Result<&1, &1, (U32 & String) & List<&2, U32>, Unit>,
193 Socket & Step.Got<Unit>,
194 TCP.send(sock, Http.respond.of(reply)),
195 server.write.un)
196
197# turn a bad request into 400; otherwise call the handler
198def server.dispatch(~handle: Http.Request -> Http.Reply, req: Http.Request)
199 -> Http.Reply:
200 match req:
201 case Http.Bad{why}:
202 Http.Reply{400, [], why}
203 case Http.Request{method, target, headers, body}:
204 handle(Http.Request{method, target, headers, body})
205
206# apply HEAD body stripping once the method token is known
207def server.reply.built(~handle: Http.Request -> Http.Reply, method: String,
208 req: Http.Request) -> Http.Reply:
209 Http.reply.for(method, server.dispatch(handle, req))
210
211# HEAD responses are written with an empty body (RFC 9110 §9.3.2)
212def server.reply(~handle: Http.Request -> Http.Reply, req: Http.Request)
213 -> Http.Reply:
214 match req:
215 case Http.Bad{why}:
216 server.dispatch(handle, Http.Bad{why})
217 case Http.Request{+method, target, headers, body}:
218 server.reply.built(handle, method,
219 Http.Request{method, target, headers, body})
220
221# CORS preflight or simple headers, after HEAD bodies are dropped
222def server.cors_reply(
223 ~handle: Http.Request -> Http.Reply,
224 cfg: Cors.Cfg,
225 +req: Http.Request
226) -> Http.Reply:
227 Cors.cors.on(cfg, req, server.reply(handle, req))
228
229# a lost connection logs its line and closes
230def server.lose(sock: Socket, line: String) -> IO(Unit):
231 IO.bind(Unit, Unit, IO.print_err(line), _u => Socket.close(sock))
232
233# after the write: close, logging first when the send failed
234def server.sent.got(sock: Socket, got: Step.Got<Unit>) -> IO(Unit):
235 match got:
236 case Step.Lost{line}:
237 server.lose(sock, line)
238 case Step.Kept{_u}:
239 Socket.close(sock)
240
241def server.sent(pair: Socket & Step.Got<Unit>) -> IO(Unit):
242 (sock, got) = pair
243 server.sent.got(sock, got)
244
245# write a reply shaped for the request's method (HTTP-SRV-3), then close
246def server.answer(sock: Socket, method: String, reply: Http.Reply) -> IO(Unit):
247 IO.bind(Socket & Step.Got<Unit>, Unit,
248 server.write(sock, Http.reply.for(method, reply)), server.sent)
249
250# answer now, or call the handler: step.bend decides which (HTTP-SRV-1, 2)
251def server.step(
252 ~handle: Http.Request -> IO(Http.Reply),
253 sock: Socket,
254 +method: String,
255 step: Step.Step
256) -> IO(Unit):
257 match step:
258 case Step.Answer{reply}:
259 server.answer(sock, method, reply)
260 case Step.Call{req}:
261 IO.bind(Http.Reply, Unit, handle(req),
262 reply => server.answer(sock, method, reply))
263
264# after the read: a lost connection, or a request to step
265def server.exchange.got(
266 ~handle: Http.Request -> IO(Http.Reply),
267 sock: Socket,
268 got: Step.Got<Http.Request>
269) -> IO(Unit):
270 match got:
271 case Step.Lost{line}:
272 server.lose(sock, line)
273 case Step.Kept{got}:
274 +req = Step.step.text(got)
275 server.step(handle, sock, Http.method.of(req), Step.step.of(req))
276
277def server.exchange.un(
278 ~handle: Http.Request -> IO(Http.Reply),
279 pair: Socket & Step.Got<Http.Request>
280) -> IO(Unit):
281 (sock, got) = pair
282 server.exchange.got(handle, sock, got)
283
284# one connection: read, decide, maybe handle, write, close
285def server.exchange(
286 ~handle: Http.Request -> IO(Http.Reply),
287 sock: Socket
288) -> IO(Unit):
289 IO.bind(Socket & Step.Got<Http.Request>, Unit, server.read(sock),
290 pair => server.exchange.un(handle, pair))
291
292# how long a failed accept waits before the next, so EMFILE does not spin
293def server.pause() -> U32:
294 100
295
296# after accept: serve the connection, or log a failed accept and pause, then
297# call `again` with the listener either way. `again` is a runtime closure so
298# it can carry the fuel still left.
299def server.loop.got(
300 ~handle: Http.Request -> IO(Http.Reply),
301 again: Listener -> IO(Unit),
302 lst: Listener,
303 got: Step.Got<Socket>
304) -> IO(Unit):
305 match got:
306 case Step.Lost{line}:
307 IO.bind(Unit, Unit, IO.print_err(line), _u =>
308 IO.bind(Unit, Unit, IO.sleep(server.pause()), _v => again(lst)))
309 case Step.Kept{sock}:
310 IO.bind(Unit, Unit, server.exchange(handle, sock), _u => again(lst))
311
312def server.loop.cont(
313 ~handle: Http.Request -> IO(Http.Reply),
314 again: Listener -> IO(Unit),
315 pair: Listener & Step.Got<Socket>
316) -> IO(Unit):
317 (lst, got) = pair
318 server.loop.got(handle, again, lst, got)
319
320# fuel-bounded accept loop (self-recursive); a failed accept spends fuel too
321def server.loop.go(
322 ~handle: Http.Request -> IO(Http.Reply),
323 fuel: Nat,
324 listener: Listener
325) -> IO(Unit):
326 match fuel:
327 case 0n:
328 Listener.close(listener)
329 case 1n+f:
330 IO.bind(Listener & Step.Got<Socket>, Unit, server.accept(listener),
331 pair => server.loop.cont(handle,
332 lst => server.loop.go(handle, f, lst), pair))
333
334# listen and serve up to `limit` connections with an IO handler
335def server.serve_io(
336 ~handle: Http.Request -> IO(Http.Reply),
337 host: String,
338 port: U32,
339 limit: U32
340) -> IO(Unit):
341 IO.bind(Listener, Unit, server.listen(host, port),
342 lst => server.loop.go(handle, U32.to_nat(limit), lst))
343
344# a pure handler as an IO one
345def server.lift(
346 ~handle: Http.Request -> Http.Reply,
347 req: Http.Request
348) -> IO(Http.Reply):
349 IO.pure(Http.Reply, handle(req))
350
351# listen and serve up to `limit` connections
352def server.serve(
353 ~handle: Http.Request -> Http.Reply,
354 host: String,
355 port: U32,
356 limit: U32
357) -> IO(Unit):
358 server.serve_io(~(req => server.lift(handle, req)), host, port, limit)
359
360# serve exactly one connection then stop
361def server.once(
362 ~handle: Http.Request -> Http.Reply,
363 host: String,
364 port: U32
365) -> IO(Unit):
366 server.serve(handle, host, port, 1)
367
368# the accept loop with no fuel: it runs until the process is stopped
369# (HTTP-TRUST-2)
370@unsafe def server.forever.go(
371 ~handle: Http.Request -> IO(Http.Reply),
372 listener: Listener
373) -> IO(Unit):
374 IO.bind(Listener & Step.Got<Socket>, Unit, server.accept(listener),
375 pair => server.loop.cont(handle,
376 lst => server.forever.go(handle, lst), pair))
377
378# listen and serve with an IO handler until the process is stopped
379@unsafe def server.serve_forever(
380 ~handle: Http.Request -> IO(Http.Reply),
381 host: String,
382 port: U32
383) -> IO(Unit):
384 IO.bind(Listener, Unit, server.listen(host, port),
385 lst => server.forever.go(handle, lst))