server.bend

cswap-bend-cli@0.1.0.0 · 199 lines · 26 definitions

1import Base
2import cswap-bend-cli@0.1.0.0/cmd.bend as Cmd
3import cswap-bend-cli@0.1.0.0/fs.bend as Fs
4import cswap-bend-cli@0.1.0.0/frame.bend as Fr
5import cswap-bend-cli@0.1.0.0/rpc.bend as Rpc
6import cswap-bend-cli@0.1.0.0/fmt/out.bend as O
7
8# `cswap rpc` reads messages from standard input and writes the responses to
9# standard output; `cswap serve --socket PATH` does the same for one client at
10# a time on a unix socket (mode 0600, no TCP). Both end the same way: rpc at
11# end of input, serve on SIGTERM / SIGINT / SIGHUP (the socket file is removed
12# by the handler in effs/usock.c). A client that closes its end is served what
13# it already sent, then the next client is accepted.
14
15# A foreign effect: a non-blocking listening unix socket at `path` (mode 0600);
16# a stale socket file there is replaced, any other file is an error. The
17# Listener takes TCP.accept, and the sockets it accepts take TCP.recv/send.
18def Usock.listen(path: String) -> IO(Result<&1, &1, U32 & String, Listener>):
19  import "./effs/usock.c"
20  import "./effs/usock.js"
21
22def chunk() -> U32:
23  65536
24
25# ---- standard input ----------------------------------------------------------------
26
27type SP is Type:
28  SRead{f: File, buf: String}
29  SFrames{f: File, buf: String}
30  SHandle{f: File, msg: String, rest: String}
31  STail{f: File, buf: String}
32  SDone{f: File}
33
34def s_chunk(has: Bool, f: File, buf: String, c: String) -> SP:
35  match has:
36    case True{}:
37      SFrames{f, buf ++ c}
38    case False{}:
39      STail{f, buf}
40
41def s_read(res: Fs.Res(String), f: File, buf: String) -> SP:
42  match res:
43    case Fail{e}:
44      SDone{f}
45    case Done{+c}:
46      s_chunk(Bool.not(String.is_empty(c)), f, buf, c)
47
48def after_read(r: File & Fs.Res(String), buf: String) -> SP:
49  (f, res) = r
50  s_read(res, f, buf)
51
52def s_cut(m: Maybe<&2, Fr.Cut>, f: File, buf: String) -> SP:
53  match m:
54    case None{}:
55      SRead{f, buf}
56    case Some{Fr.Cut{msg, rest}}:
57      SHandle{f, msg, rest}
58
59def s_tail(blank: Bool, f: File, buf: String) -> SP:
60  match blank:
61    case True{}:
62      SDone{f}
63    case False{}:
64      SHandle{f, buf, ""}
65
66@unsafe def stdin_pump(p: SP, +fmt: O.Fmt) -> IO(Unit):
67  match p:
68    case SRead{f, buf}:
69      IO.bind(File & Fs.Res(String), Unit, File.read(f, chunk()), r => stdin_pump(after_read(r, buf), fmt))
70    case SFrames{f, +buf}:
71      stdin_pump(s_cut(Fr.take(Rpc.is_block(fmt), buf), f, buf), fmt)
72    case SHandle{f, msg, rest}:
73      IO.bind(String, Unit, Rpc.handle(fmt, msg), resp => IO.bind(Unit, Unit, IO.write(resp), u => stdin_pump(SFrames{f, rest}, fmt)))
74    case STail{f, +buf}:
75      stdin_pump(s_tail(Fr.blank(buf), f, buf), fmt)
76    case SDone{f}:
77      File.close(f)
78
79def rpc_start(r: Fs.Res(File), +fmt: O.Fmt) -> IO(Cmd.Out):
80  match r:
81    case Fail{e}:
82      IO.pure(Cmd.Out, Cmd.Out{1, "", "cswap rpc: cannot read standard input\n"})
83    case Done{f}:
84      IO.bind(Unit, Cmd.Out, stdin_pump(SRead{f, ""}, fmt), u => IO.pure(Cmd.Out, Cmd.silent()))
85
86# Serve the messages on standard input; ends at end of input.
87def rpc(+fmt: O.Fmt) -> IO(Cmd.Out):
88  IO.bind(Fs.Res(File), Cmd.Out, File.open("/dev/stdin", "r"), r => rpc_start(r, fmt))
89
90# ---- one connection --------------------------------------------------------------------
91
92type CP is Type:
93  CRecv{s: Socket, buf: String}
94  CFrames{s: Socket, buf: String}
95  CHandle{s: Socket, msg: String, rest: String}
96  CReply{s: Socket, resp: String, rest: String}
97  CSend{s: Socket, resp: String, rest: String}
98  CTail{s: Socket, buf: String}
99  CClose{s: Socket}
100
101def c_recv(res: Result<&1, &1, U32 & String, Maybe<&1, String>>, s: Socket, buf: String) -> CP:
102  match res:
103    case Fail{e}:
104      CClose{s}
105    case Done{None{}}:
106      CTail{s, buf}
107    case Done{Some{c}}:
108      CFrames{s, buf ++ c}
109
110def after_recv(r: Socket & Result<&1, &1, U32 & String, Maybe<&1, String>>, buf: String) -> CP:
111  (s, res) = r
112  c_recv(res, s, buf)
113
114def c_sent(res: Result<&1, &1, (U32 & String) & List<&2, U32>, Unit>, s: Socket, rest: String) -> CP:
115  match res:
116    case Fail{e}:
117      CClose{s}
118    case Done{u}:
119      CFrames{s, rest}
120
121def after_send(r: Socket & Result<&1, &1, (U32 & String) & List<&2, U32>, Unit>, rest: String) -> CP:
122  (s, res) = r
123  c_sent(res, s, rest)
124
125def c_cut(m: Maybe<&2, Fr.Cut>, s: Socket, buf: String) -> CP:
126  match m:
127    case None{}:
128      CRecv{s, buf}
129    case Some{Fr.Cut{msg, rest}}:
130      CHandle{s, msg, rest}
131
132def c_tail(blank: Bool, s: Socket, buf: String) -> CP:
133  match blank:
134    case True{}:
135      CClose{s}
136    case False{}:
137      CHandle{s, buf, ""}
138
139def c_reply(empty: Bool, s: Socket, resp: String, rest: String) -> CP:
140  match empty:
141    case True{}:
142      CFrames{s, rest}
143    case False{}:
144      CSend{s, resp, rest}
145
146@unsafe def conn_pump(p: CP, +fmt: O.Fmt) -> IO(Unit):
147  match p:
148    case CRecv{s, buf}:
149      IO.bind(Socket & Result<&1, &1, U32 & String, Maybe<&1, String>>, Unit, TCP.recv(s, chunk()), r => conn_pump(after_recv(r, buf), fmt))
150    case CFrames{s, +buf}:
151      conn_pump(c_cut(Fr.take(Rpc.is_block(fmt), buf), s, buf), fmt)
152    case CHandle{s, msg, rest}:
153      IO.bind(String, Unit, Rpc.handle(fmt, msg), resp => conn_pump(CReply{s, resp, rest}, fmt))
154    case CReply{s, +resp, rest}:
155      conn_pump(c_reply(String.is_empty(resp), s, resp, rest), fmt)
156    case CSend{s, resp, rest}:
157      IO.bind(Socket & Result<&1, &1, (U32 & String) & List<&2, U32>, Unit>, Unit, TCP.send(s, resp), r => conn_pump(after_send(r, rest), fmt))
158    case CTail{s, +buf}:
159      conn_pump(c_tail(Fr.blank(buf), s, buf), fmt)
160    case CClose{s}:
161      Socket.close(s)
162
163# ---- the listener --------------------------------------------------------------------------
164
165type AP is Type:
166  AAcc{l: Listener}
167  AConn{l: Listener, s: Socket}
168  ARetry{l: Listener}
169
170def a_got(res: Result<&1, &1, U32 & String, Socket>, l: Listener) -> AP:
171  match res:
172    case Fail{e}:
173      ARetry{l}
174    case Done{s}:
175      AConn{l, s}
176
177def after_accept(r: Listener & Result<&1, &1, U32 & String, Socket>) -> AP:
178  (l, res) = r
179  a_got(res, l)
180
181@unsafe def accept_pump(p: AP, +fmt: O.Fmt) -> IO(Unit):
182  match p:
183    case AAcc{l}:
184      IO.bind(Listener & Result<&1, &1, U32 & String, Socket>, Unit, TCP.accept(l), r => accept_pump(after_accept(r), fmt))
185    case AConn{l, s}:
186      IO.bind(Unit, Unit, conn_pump(CRecv{s, ""}, fmt), u => accept_pump(AAcc{l}, fmt))
187    case ARetry{l}:
188      IO.bind(Unit, Unit, IO.sleep(100), u => accept_pump(AAcc{l}, fmt))
189
190def serve_start(r: Result<&1, &1, U32 & String, Listener>, +path: String, +fmt: O.Fmt) -> IO(Cmd.Out):
191  match r:
192    case Fail{(code, msg)}:
193      IO.pure(Cmd.Out, Cmd.Out{1, "", "cswap serve: cannot listen on " ++ path ++ ": " ++ msg ++ "\n"})
194    case Done{l}:
195      IO.bind(Unit, Cmd.Out, accept_pump(AAcc{l}, fmt), u => IO.pure(Cmd.Out, Cmd.silent()))
196
197# Serve clients on the unix socket at `path`, one at a time, until a signal.
198def serve(+path: String, +fmt: O.Fmt) -> IO(Cmd.Out):
199  IO.bind(Result<&1, &1, U32 & String, Listener>, Cmd.Out, Usock.listen(path), r => serve_start(r, path, fmt))