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