1	import Base
2	import ./cmd.bend as Cmd
3	import ./fs.bend as Fs
4	import ./frame.bend as Fr
5	import ./rpc.bend as Rpc
6	import ./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.
18	def Usock.listen(path: String) -> IO(Result<&1, &1, U32 & String, Listener>):
19	  import "./effs/usock.c"
20	  import "./effs/usock.js"
21	
22	def chunk() -> U32:
23	  65536
24	
25	# ---- standard input ----------------------------------------------------------------
26	
27	type 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	
34	def 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	
41	def 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	
48	def after_read(r: File & Fs.Res(String), buf: String) -> SP:
49	  (f, res) = r
50	  s_read(res, f, buf)
51	
52	def 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	
59	def 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	
79	def 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.
87	def 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	
92	type 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	
101	def 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	
110	def 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	
114	def 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	
121	def 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	
125	def 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	
132	def 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	
139	def 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	
165	type AP is Type:
166	  AAcc{l: Listener}
167	  AConn{l: Listener, s: Socket}
168	  ARetry{l: Listener}
169	
170	def 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	
177	def 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	
190	def 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.
198	def 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))