src/server.bend

emerging-ezhttp@0.21.0.0 · 551 lines · 64 definitions

1# ezhttp/src/server: listen, accept, and serve each connection on its own
2# fork: parse one request, write one response (Connection: close). Sockets
3# are Base's (listen) and bend-kit-wire's (accept with a deadline, recv,
4# send.words), whose effects park on the event loop, so connections overlap.
5# A `_with` serve hands the handler a context value and the request as text
6# with its body's octets, and takes a reply as text, packed octets or a
7# stream of octet chunks; the plain serves are those with a text handler.
8# Every decision after an effect is step.bend's or http.bend's (SPEC.md
9# HTTP-SRV); this module runs the effects in order (HTTP-TRUST-2).
10import Base
11import bend-kit-wire@0.4.6.1/wire.bend as W
12import bend-kit-bytes@0.3.2.0/bytes.bend as Bytes
13import emerging-ezhttp@0.21.0.0/src/http.bend as Http
14import emerging-ezhttp@0.21.0.0/src/cors.bend as Cors
15import emerging-ezhttp@0.21.0.0/src/step.bend as Step
16import emerging-ezhttp@0.21.0.0/src/wirespec.bend as Spec
17
18# what a `_with` handler gets: the request as text (step.text) and its
19# body's octets as they arrived (step.octets)
20type Req is Type:  # noqa: L001 it holds Bytes; step.text and step.octets fill it (HTTP-SRV-9)
21  Req{method: String, target: String, headers: List<&2, Http.Header>,
22    body: String, bytes: Bytes.Bytes}
23
24# what a `_with` handler answers: a text body (written as UTF-8), packed
25# octets written unchanged, or a stream of octet chunks from a channel,
26# written chunked until the channel is closed (SPEC.md HTTP-SRV-6, 7)
27type Answer is Type:  # noqa: L001 it holds Bytes and a channel; HTTP-SRV-6 to 8 cover its writing
28  TextReply{status: U32, headers: List<&2, Http.Header>, body: String}
29  OctetsReply{status: U32, headers: List<&2, Http.Header>, bytes: Bytes.Bytes}
30  StreamReply{status: U32, headers: List<&2, Http.Header>,
31    chunks: Chan(Bytes.Bytes)}
32
33# the most octets one receive asks for
34def server.max() -> U32:
35  65536
36
37# the most receives one request takes
38def server.reads() -> Nat:
39  65536n
40
41# how long one accept waits before it is asked again, in ms
42def server.wait() -> U32:
43  3600000
44
45# how many idle accepts in a row a limited serve allows before it stops
46def server.idles() -> Nat:
47  1000000n
48
49# how long a failed accept waits before the next, so EMFILE does not spin
50def server.pause() -> U32:
51  100
52
53# listen on a host address and port, e.g. "127.0.0.1" or "0.0.0.0"
54def server.listen(host: String, port: U32) -> IO(Listener):
55  IO.try(Listener, TCP.listen(host, port))
56
57# accept one connection, waiting at most server.wait() ms
58def server.accept.un(r: Listener & Result<&1, &1, U32 & String, Socket>)
59  -> IO(Listener & Step.Accept<Socket>):
60  (lst, res) = r
61  IO.pure(Listener & Step.Accept<Socket>, (lst, Step.step.accepted(Socket, res)))
62
63def server.accept(listener: Listener) -> IO(Listener & Step.Accept<Socket>):
64  IO.bind(Listener & Result<&1, &1, U32 & String, Socket>,
65    Listener & Step.Accept<Socket>,
66    W.accept.deadline(listener, server.wait()), server.accept.un)
67
68# a lost connection logs its line and closes
69def server.lose(sock: Socket, line: String) -> IO(Unit):
70  IO.bind(Unit, Unit, IO.print_err(line), _u => Socket.close(sock))
71
72# --- reading a request ---
73
74# the request once it has all arrived (or the peer closed, or the receives
75# ran out): its pieces put together and parsed
76def server.read.done(sock: Socket, ps: List<&2, String>)
77  -> IO(Socket & Step.Got<Http.Request>):
78  IO.pure(Socket & Step.Got<Http.Request>,
79    (sock, Step.Kept{Http.ask(Spec.wire.join(ps))}))
80
81# what the request needs, kept once its head has ended
82def server.read.needs(m: Maybe<&2, Nat>, ps: List<&2, String>)
83  -> Maybe<&2, Nat>:
84  match m:
85    case None{}:
86      Step.step.need(Spec.wire.join(ps))
87    case Some{n}:
88      Some{n}
89
90# whether as much has arrived as the request needs
91def server.read.whole(m: Maybe<&2, Nat>, +have: Nat) -> Bool:
92  match m:
93    case None{}:
94      False{}
95    case Some{n}:
96      Bool.not(Nat.is_lt(have, n))
97
98# octets that arrived: stop when the request is whole, else read again
99def server.read.more(sock: Socket, +ps: List<&2, String>, +have: Nat,
100  +need: Maybe<&2, Nat>, whole: Bool,
101  rest: Socket -> List<&2, String> -> Nat -> Maybe<&2, Nat>
102    -> IO(Socket & Step.Got<Http.Request>))
103  -> IO(Socket & Step.Got<Http.Request>):
104  match whole:
105    case True{}:
106      server.read.done(sock, ps)
107    case False{}:
108      rest(sock)(ps)(have)(need)
109
110# octets added to the pieces: is the request now whole?
111def server.read.add(sock: Socket, +ps: List<&2, String>, +have: Nat,
112  need: Maybe<&2, Nat>,
113  rest: Socket -> List<&2, String> -> Nat -> Maybe<&2, Nat>
114    -> IO(Socket & Step.Got<Http.Request>))
115  -> IO(Socket & Step.Got<Http.Request>):
116  +need2 = server.read.needs(need, ps)
117  server.read.more(sock, ps, have, need2, server.read.whole(need2, have), rest)
118
119# octets that arrived, or none once the peer has closed
120def server.read.text(sock: Socket, +ps: List<&2, String>, +have: Nat,
121  need: Maybe<&2, Nat>, +text: String, closed: Bool,
122  rest: Socket -> List<&2, String> -> Nat -> Maybe<&2, Nat>
123    -> IO(Socket & Step.Got<Http.Request>))
124  -> IO(Socket & Step.Got<Http.Request>):
125  match closed:
126    case True{}:
127      server.read.done(sock, ps)
128    case False{}:
129      server.read.add(sock, text <> ps, Nat.add(have, String.length(text)),
130        need, rest)
131
132# a receive's answer: a failure loses the connection, no octets is the peer
133# closing, octets go onto the pieces
134def server.read.got(+ps: List<&2, String>, +have: Nat, need: Maybe<&2, Nat>,
135  r: Socket & Result<&1, &1, U32 & String, String>,
136  rest: Socket -> List<&2, String> -> Nat -> Maybe<&2, Nat>
137    -> IO(Socket & Step.Got<Http.Request>))
138  -> IO(Socket & Step.Got<Http.Request>):
139  (sock, res) = r
140  match res:
141    case Fail{(code, why)}:
142      IO.pure(Socket & Step.Got<Http.Request>,
143        (sock, Step.Lost{Step.step.line("read", why)}))
144    case Done{+text}:
145      server.read.text(sock, ps, have, need, text, String.is_empty(text), rest)
146
147# receive until the head has ended and its Content-Length has arrived, the
148# peer closes, or the receives run out
149def server.read.go(fuel: Nat, sock: Socket, +ps: List<&2, String>, +have: Nat,
150  +need: Maybe<&2, Nat>) -> IO(Socket & Step.Got<Http.Request>):
151  match fuel:
152    case 0n:
153      server.read.done(sock, ps)
154    case 1n+f:
155      IO.bind(Socket & Result<&1, &1, U32 & String, String>,
156        Socket & Step.Got<Http.Request>,
157        W.recv(sock, server.max(), 0),
158        r => server.read.got(ps, have, need, r,
159          s => p => h => m => server.read.go(f, s, p, h, m)))
160
161# receive then parse an HTTP request (RFC 9112 §3)
162def server.read(sock: Socket) -> IO(Socket & Step.Got<Http.Request>):
163  server.read.go(server.reads(), sock, [], 0n, None{})
164
165# --- writing a reply ---
166
167# after a send: the socket, or the line of a lost connection
168def server.sent.un(r: Socket & Result<&1, &1, U32 & String, Unit>)
169  -> IO(Socket & Step.Got<Unit>):
170  (sock, res) = r
171  IO.pure(Socket & Step.Got<Unit>, (sock, Step.step.got(Unit, "write", res)))
172
173# send octets
174def server.send(sock: Socket, b: Bytes.Bytes) -> IO(Socket & Step.Got<Unit>):
175  Bytes.Bytes{+len, buf} = b
176  IO.bind(Socket & Result<&1, &1, U32 & String, Unit>, Socket & Step.Got<Unit>,
177    W.send.words(sock, len, buf), server.sent.un)
178
179# send text as its UTF-8 octets
180def server.send.text(sock: Socket, text: String) -> IO(Socket & Step.Got<Unit>):
181  server.send(sock, Bytes.from_string(Spec.wire.payload(text)))
182
183# after the last send: close, logging first when it failed
184def server.close(r: Socket & Step.Got<Unit>) -> IO(Unit):
185  (sock, got) = r
186  match got:
187    case Step.Lost{line}:
188      server.lose(sock, line)
189    case Step.Kept{_u}:
190      Socket.close(sock)
191
192# after a send in the middle: go on, or lose the connection
193def server.then(r: Socket & Step.Got<Unit>, more: Socket -> IO(Unit))
194  -> IO(Unit):
195  (sock, got) = r
196  match got:
197    case Step.Lost{line}:
198      server.lose(sock, line)
199    case Step.Kept{_u}:
200      more(sock)
201
202# a text reply shaped for the request's method (HTTP-SRV-3), then close
203def server.answer(sock: Socket, method: String, reply: Http.Reply) -> IO(Unit):
204  IO.bind(Socket & Step.Got<Unit>, Unit,
205    server.send.text(sock, Http.respond.of(Http.reply.for(method, reply))),
206    server.close)
207
208# an octets reply once its length is known: its head, then its octets
209# unless it is bare (HTTP-SRV-6, 8)
210def server.octets.go(sock: Socket, +status: U32, headers: List<&2, Http.Header>,
211  r: Bytes.Bytes & U32, bare: Bool) -> IO(Unit):
212  (b, +n) = r
213  match bare:
214    case True{}:
215      IO.bind(Socket & Step.Got<Unit>, Unit,
216        server.send.text(sock, Http.respond.head(status, headers, 0n)),
217        server.close)
218    case False{}:
219      IO.bind(Socket & Step.Got<Unit>, Unit,
220        server.send(sock, Bytes.append(Bytes.from_string(
221          Http.respond.head(status, headers, U32.to_nat(n))), b)),
222        server.close)
223
224# the most chunks one streamed reply writes
225def server.chunks() -> Nat:
226  U32.to_nat(4294967295)
227
228# one chunk the channel gave: framed and sent, or skipped when empty
229def server.chunk.put(sock: Socket, c: Http.Chunk, b: Bytes.Bytes,
230  more: Socket -> IO(Unit)) -> IO(Unit):
231  match c:
232    case Http.NoChunk{}:
233      more(sock)
234    case Http.Chunk{line}:
235      IO.bind(Socket & Step.Got<Unit>, Unit,
236        server.send(sock, Bytes.append(Bytes.append(Bytes.from_string(line), b),
237          Bytes.from_string("\r\n"))),
238        r => server.then(r, more))
239
240# a chunk with its length known
241def server.chunk.sized(sock: Socket, r: Bytes.Bytes & U32,
242  more: Socket -> IO(Unit)) -> IO(Unit):
243  (b, +n) = r
244  server.chunk.put(sock, Http.chunk.frame(n), b, more)
245
246# what the channel gave: a chunk, or its close, which ends the body
247def server.chunk.got(sock: Socket, m: Maybe<&1, Bytes.Bytes>,
248  more: Socket -> IO(Unit)) -> IO(Unit):
249  match m:
250    case None{}:
251      IO.bind(Socket & Step.Got<Unit>, Unit,
252        server.send.text(sock, Http.chunk.end()), server.close)
253    case Some{b}:
254      server.chunk.sized(sock, Bytes.length(b), more)
255
256# every chunk the channel gives, until it is closed
257def server.stream.go(fuel: Nat, sock: Socket, +ch: Chan(Bytes.Bytes))
258  -> IO(Unit):
259  match fuel:
260    case 0n:
261      IO.bind(Socket & Step.Got<Unit>, Unit,
262        server.send.text(sock, Http.chunk.end()), server.close)
263    case 1n+f:
264      IO.bind(Maybe<&1, Bytes.Bytes>, Unit, Chan.recv(Bytes.Bytes, ch),
265        m => server.chunk.got(sock, m, s => server.stream.go(f, s, ch)))
266
267# a streamed reply after its head: its chunks, or nothing when it is bare
268def server.stream.body(sock: Socket, +ch: Chan(Bytes.Bytes), bare: Bool)
269  -> IO(Unit):
270  match bare:
271    case True{}:
272      IO.bind(Unit, Unit, Chan.close(Bytes.Bytes, ch), _u => Socket.close(sock))
273    case False{}:
274      server.stream.go(server.chunks(), sock, ch)
275
276# a handler's answer, shaped for the request's method (HTTP-SRV-3, 6 to 8)
277def server.reply(sock: Socket, +method: String, a: Answer) -> IO(Unit):
278  match a:
279    case TextReply{status, headers, body}:
280      server.answer(sock, method, Http.Reply{status, headers, body})
281    case OctetsReply{+status, headers, bytes}:
282      server.octets.go(sock, status, headers, Bytes.length(bytes),
283        Step.step.bare(method, status))
284    case StreamReply{+status, +headers, +chunks}:
285      IO.bind(Socket & Step.Got<Unit>, Unit,
286        server.send.text(sock, Http.respond.stream(status, headers)),
287        r => server.then(r,
288          s => server.stream.body(s, chunks, Step.step.bare(method, status))))
289
290# --- one connection ---
291
292# the handler's request: the request as text and its body's octets
293def server.req(req: Http.Request, +octets: String) -> Req:
294  match req:
295    case Http.Bad{why}:
296      Req{"", "", [], why, Bytes.from_string(octets)}
297    case Http.Request{method, target, headers, body}:
298      Req{method, target, headers, body, Bytes.from_string(octets)}
299
300# answer now, or call the handler: step.bend decides which (HTTP-SRV-1, 2, 9)
301def server.step(~C: Data, ~handle: C -> Req -> IO(Answer), +ctx: C,
302  sock: Socket, +method: String, +octets: String, step: Step.Step) -> IO(Unit):
303  match step:
304    case Step.Answer{reply}:
305      server.answer(sock, method, reply)
306    case Step.Call{req}:
307      IO.bind(Answer, Unit, handle(ctx)(server.req(req, octets)),
308        a => server.reply(sock, method, a))
309
310# after the read: a lost connection, or a request to step
311def server.conn.got(~C: Data, ~handle: C -> Req -> IO(Answer), +ctx: C,
312  r: Socket & Step.Got<Http.Request>) -> IO(Unit):
313  (sock, got) = r
314  match got:
315    case Step.Lost{line}:
316      server.lose(sock, line)
317    case Step.Kept{+raw}:
318      +req = Step.step.text(raw)
319      server.step(~C, handle, ctx, sock, Http.method.of(req),
320        Step.step.octets(raw), Step.step.of(req))
321
322# one connection: read, decide, maybe handle, write, close
323def server.conn(~C: Data, ~handle: C -> Req -> IO(Answer), +ctx: C,
324  sock: Socket) -> IO(Unit):
325  IO.bind(Socket & Step.Got<Http.Request>, Unit, server.read(sock),
326    r => server.conn.got(~C, handle, ctx, r))
327
328# --- the accept loop ---
329
330# give a slot back
331def server.slot.give(+slots: Chan(Unit), limited: Bool) -> IO(Unit):
332  match limited:
333    case True{}:
334      IO.bind(Maybe<&1, Unit>, Unit, Chan.recv(Unit, slots),
335        _m => IO.pure(Unit, Unit{}))
336    case False{}:
337      IO.pure(Unit, Unit{})
338
339# take a slot, waiting while every one is taken
340def server.slot.take(+slots: Chan(Unit), limited: Bool) -> IO(Unit):
341  match limited:
342    case True{}:
343      IO.bind(Result<&1, &1, Unit, Unit>, Unit, Chan.send(Unit, slots, Unit{}),
344        _r => IO.pure(Unit, Unit{}))
345    case False{}:
346      IO.pure(Unit, Unit{})
347
348# say a fork is done, when the serve waits for its forks
349def server.done(+done: Chan(Unit), tracked: Bool) -> IO(Unit):
350  match tracked:
351    case True{}:
352      IO.bind(Result<&1, &1, Unit, Unit>, Unit, Chan.send(Unit, done, Unit{}),
353        _r => IO.pure(Unit, Unit{}))
354    case False{}:
355      IO.pure(Unit, Unit{})
356
357# a connection's fork: serve it, give its slot back, and say it is done
358def server.fork.body(~C: Data, ~handle: C -> Req -> IO(Answer), +ctx: C,
359  sock: Socket, +slots: Chan(Unit), +limited: Bool, +done: Chan(Unit),
360  +tracked: Bool) -> IO(Unit):
361  IO.bind(Unit, Unit, server.conn(~C, handle, ctx, sock),
362    _u => IO.bind(Unit, Unit, server.slot.give(slots, limited),
363      _v => server.done(done, tracked)))
364
365# serve an accepted connection on its own fork, a slot taken first
366def server.fork(~C: Data, ~handle: C -> Req -> IO(Answer), +ctx: C,
367  sock: Socket, +slots: Chan(Unit), +limited: Bool, +done: Chan(Unit),
368  +tracked: Bool) -> IO(Unit):
369  IO.bind(Unit, Unit, server.slot.take(slots, limited),
370    _u => IO.spawn(Unit,
371      server.fork.body(~C, handle, ctx, sock, slots, limited, done, tracked)))
372
373# wait for n forks to say they are done
374def server.join(n: Nat, +done: Chan(Unit)) -> IO(Unit):
375  match n:
376    case 0n:
377      IO.pure(Unit, Unit{})
378    case 1n+p:
379      IO.bind(Maybe<&1, Unit>, Unit, Chan.recv(Unit, done),
380        _m => server.join(p, done))
381
382# what an accept brought, for the loop: `took` serves a connection and goes
383# on with one connection fewer, `idle` accepts again, `missed` logs, pauses
384# and goes on with one fewer (a failed accept counts toward the limit)
385def server.loop.got(~C: Data, ~handle: C -> Req -> IO(Answer), +ctx: C,
386  +slots: Chan(Unit), +limited: Bool, +done: Chan(Unit), +tracked: Bool,
387  r: Listener & Step.Accept<Socket>, took: Listener -> IO(Unit),
388  idle: Listener -> IO(Unit), missed: Listener -> IO(Unit)) -> IO(Unit):
389  (lst, got) = r
390  match got:
391    case Step.Took{sock}:
392      IO.bind(Unit, Unit,
393        server.fork(~C, handle, ctx, sock, slots, limited, done, tracked),
394        _u => took(lst))
395    case Step.Idle{}:
396      idle(lst)
397    case Step.Missed{line}:
398      IO.bind(Unit, Unit, IO.print_err(line), _u =>
399        IO.bind(Unit, Unit, IO.sleep(server.pause()), _v => missed(lst)))
400
401# the limited accept loop: `fuel` connections left, `idles` idle accepts in
402# a row left, `forks` forked so far
403def server.loop.go(~C: Data, ~handle: C -> Req -> IO(Answer), +ctx: C,
404  fuel: Nat, idles: Nat, +forks: Nat, +slots: Chan(Unit), +limited: Bool,
405  +done: Chan(Unit), listener: Listener) -> IO(Unit):
406  match fuel idles:
407    case 0n _:
408      IO.bind(Unit, Unit, Listener.close(listener), _u => server.join(forks, done))
409    case 1n+_f 0n:
410      IO.bind(Unit, Unit, Listener.close(listener), _u => server.join(forks, done))
411    case 1n+f0 1n+i:
412      +f = f0
413      IO.bind(Listener & Step.Accept<Socket>, Unit, server.accept(listener),
414        r => server.loop.got(~C, handle, ctx, slots, limited, done, True{}, r,
415          l => server.loop.go(~C, handle, ctx, f, server.idles(), 1n+forks, slots,
416            limited, done, l),
417          l => server.loop.go(~C, handle, ctx, 1n+f, i, forks, slots, limited,
418            done, l),
419          l => server.loop.go(~C, handle, ctx, f, server.idles(), forks, slots,
420            limited, done, l)))
421
422# the slots a concurrency limit allows (max 0: no limit)
423def server.slots(+max: U32) -> IO(Chan(Unit)):
424  Chan.new(Unit, Bool.pick(U32, U32.is_eq(max, 0), 1, max))
425
426# listen and serve up to `limit` connections, each on its own fork, at most
427# `max` at once (0: no limit), with a context value for the handler; returns
428# once every fork is done (SPEC.md HTTP-TRUST-2)
429def server.serve_io_with(~C: Data, ~handle: C -> Req -> IO(Answer), +ctx: C,
430  host: String, port: U32, limit: U32, +max: U32) -> IO(Unit):
431  do IO<Unit>:
432    lst : Listener <- server.listen(host, port)
433    +slots : Chan(Unit) <- server.slots(max)
434    +done : Chan(Unit) <- Chan.new(Unit, 1024)
435    server.loop.go(~C, handle, ctx, U32.to_nat(limit), server.idles(), 0n, slots,
436      Bool.not(U32.is_eq(max, 0)), done, lst)
437
438# a pure handler as an IO one
439def server.lift.with(~C: Data, ~handle: C -> Req -> Answer, +ctx: C, req: Req)
440  -> IO(Answer):
441  IO.pure(Answer, handle(ctx)(req))
442
443# listen and serve with a pure handler and a context value
444def server.serve_with(~C: Data, ~handle: C -> Req -> Answer, +ctx: C,
445  host: String, port: U32, limit: U32, max: U32) -> IO(Unit):
446  server.serve_io_with(~C, ~(c => r => server.lift.with(~C, handle, c, r)), ctx,
447    host, port, limit, max)
448
449# the accept loop with no limit: it runs until the process is stopped
450@unsafe def server.forever.go(~C: Data, ~handle: C -> Req -> IO(Answer),
451  +ctx: C, +slots: Chan(Unit), +limited: Bool, +done: Chan(Unit),
452  listener: Listener) -> IO(Unit):
453  IO.bind(Listener & Step.Accept<Socket>, Unit, server.accept(listener),
454    r => server.loop.got(~C, handle, ctx, slots, limited, done, False{}, r,
455      l => server.forever.go(~C, handle, ctx, slots, limited, done, l),
456      l => server.forever.go(~C, handle, ctx, slots, limited, done, l),
457      l => server.forever.go(~C, handle, ctx, slots, limited, done, l)))
458
459# listen and serve with an IO handler and a context value until the process
460# is stopped, at most `max` connections at once (0: no limit)
461@unsafe def server.serve_forever_with(~C: Data,
462  ~handle: C -> Req -> IO(Answer), +ctx: C, host: String, port: U32, +max: U32)
463  -> IO(Unit):
464  do IO<Unit>:
465    lst : Listener <- server.listen(host, port)
466    +slots : Chan(Unit) <- server.slots(max)
467    +done : Chan(Unit) <- Chan.new(Unit, 1)
468    server.forever.go(~C, handle, ctx, slots, Bool.not(U32.is_eq(max, 0)), done,
469      lst)
470
471# --- the text handlers (no context) ---
472
473# a `_with` request as the text request a plain handler takes
474def server.plain(r: Req) -> Http.Request:
475  Req{method, target, headers, body, bytes} = r
476  Http.Request{method, target, headers, body}
477
478# a text reply as an answer
479def server.text(reply: Http.Reply) -> IO(Answer):
480  match reply:
481    case Http.Torn{why}:
482      IO.pure(Answer, TextReply{500, [], why})
483    case Http.Reply{status, headers, body}:
484      IO.pure(Answer, TextReply{status, headers, body})
485
486# a plain IO handler as a `_with` one
487def server.wrap.io(~handle: Http.Request -> IO(Http.Reply), r: Req)
488  -> IO(Answer):
489  IO.bind(Http.Reply, Answer, handle(server.plain(r)), server.text)
490
491# listen and serve up to `limit` connections with an IO handler
492def server.serve_io(~handle: Http.Request -> IO(Http.Reply), host: String,
493  port: U32, limit: U32) -> IO(Unit):
494  server.serve_io_with(~Unit, ~(_u => r => server.wrap.io(handle, r)), Unit{},
495    host, port, limit, 0)
496
497# a pure handler as an IO one
498def server.lift(~handle: Http.Request -> Http.Reply, req: Http.Request)
499  -> IO(Http.Reply):
500  IO.pure(Http.Reply, handle(req))
501
502# listen and serve up to `limit` connections
503def server.serve(~handle: Http.Request -> Http.Reply, host: String, port: U32,
504  limit: U32) -> IO(Unit):
505  server.serve_io(~(req => server.lift(handle, req)), host, port, limit)
506
507# serve exactly one connection then stop
508def server.once(~handle: Http.Request -> Http.Reply, host: String, port: U32)
509  -> IO(Unit):
510  server.serve(handle, host, port, 1)
511
512# listen and serve with an IO handler until the process is stopped
513@unsafe def server.serve_forever(~handle: Http.Request -> IO(Http.Reply),
514  host: String, port: U32) -> IO(Unit):
515  server.serve_forever_with(~Unit, ~(_u => r => server.wrap.io(handle, r)), Unit{},
516    host, port, 0)
517
518# --- CORS (cors.bend), for pure handlers ---
519
520# turn a bad request into 400; otherwise call the handler
521def server.dispatch(~handle: Http.Request -> Http.Reply, req: Http.Request)
522  -> Http.Reply:
523  match req:
524    case Http.Bad{why}:
525      Http.Reply{400, [], why}
526    case Http.Request{method, target, headers, body}:
527      handle(Http.Request{method, target, headers, body})
528
529# apply HEAD body stripping once the method token is known
530def server.reply.built(~handle: Http.Request -> Http.Reply, method: String,
531  req: Http.Request) -> Http.Reply:
532  Http.reply.for(method, server.dispatch(handle, req))
533
534# HEAD responses are written with an empty body (RFC 9110 §9.3.2)
535def server.reply.text(~handle: Http.Request -> Http.Reply, req: Http.Request)
536  -> Http.Reply:
537  match req:
538    case Http.Bad{why}:
539      server.dispatch(handle, Http.Bad{why})
540    case Http.Request{+method, target, headers, body}:
541      server.reply.built(handle, method,
542        Http.Request{method, target, headers, body})
543
544# CORS preflight or simple headers, after HEAD bodies are dropped
545def server.cors_reply(
546  ~handle: Http.Request -> Http.Reply,
547  cfg: Cors.Cfg,
548  +req: Http.Request
549) -> Http.Reply:
550  Cors.cors.on(cfg, req, server.reply.text(handle, req))
551