1	import Base
2	import ./json.bend as J
3	
4	type Connection is Type:
5	  Connection{socket: Socket, buffered: String}
6	
7	type Read is Type:
8	  Frame{connection: Connection, value: J.Json}
9	  Broken{message: String}
10	
11	type Cut is Data:
12	  Complete{line: String, rest: String}
13	  Partial{}
14	
15	def cut(s: String, acc: String) -> Cut:
16	  match s:
17	    case SNil{}: Partial{}
18	    case SCon{'\n', rest}: Complete{String.reverse(acc), rest}
19	    case SCon{c, rest}: cut(rest, SCon{c, acc})
20	
21	def broken(socket: Socket, message: String) -> IO(Read):
22	  do IO<Read>:
23	    Socket.close(socket)
24	    return Broken{message}
25	
26	def decoded(result: Result<&1, &1, String, J.Json>, socket: Socket, rest: String) -> IO(Read):
27	  match result:
28	    case Fail{error}: broken(socket, "Invalid bridge JSON: " ++ error)
29	    case Done{value}: IO.pure(Read, Frame{Connection{socket, rest}, value})
30	
31	def chunk_nonempty(empty: Bool, socket: Socket, chunk: String, buffered: String,
32	                   next: Connection -> IO(Read)) -> IO(Read):
33	  match empty:
34	    case True{}: broken(socket, "Bridge closed before a complete frame")
35	    case False{}: next(Connection{socket, buffered ++ chunk})
36	
37	def chunk(result: Socket & Result<&1, &1, U32 & String, String>, buffered: String,
38	          next: Connection -> IO(Read)) -> IO(Read):
39	  (socket, received) = result
40	  match received:
41	    case Fail{(code, message)}: broken(socket, "Bridge receive failed: " ++ message)
42	    case Done{+data}: chunk_nonempty(String.is_empty(data), socket, data, buffered, next)
43	
44	def read_cut(part: Cut, socket: Socket, buffered: String,
45	             next: Connection -> IO(Read)) -> IO(Read):
46	  match part:
47	    case Complete{line, rest}: decoded(J.read(line), socket, rest)
48	    case Partial{}:
49	      do IO<Read>:
50	        received : Socket & Result<&1, &1, U32 & String, String> <- TCP.recv(socket, 4096)
51	        chunk(received, buffered, next)
52	
53	def read_limit(large: Bool, socket: Socket, +buffered: String,
54	               next: Connection -> IO(Read)) -> IO(Read):
55	  match large:
56	    case True{}: broken(socket, "Bridge frame exceeds 4 MiB limit")
57	    case False{}: read_cut(cut(buffered, ""), socket, buffered, next)
58	
59	# Socket progress depends on the external peer, not structural recursion.
60	@unsafe
61	def read(connection: Connection) -> IO(Read):
62	  match connection:
63	    case Connection{socket, +buffered}:
64	      read_limit((U32.from_nat(String.length(buffered)) > 4198400 : U32), socket, buffered, c => read(c))
65	
66	def sent(result: Socket & Result<&1, &1, U32 & String, Unit>) -> IO(Result<&1, &1, String, Connection>):
67	  (socket, result) = result
68	  match result:
69	    case Fail{(code, message)}:
70	      do IO<Result<&1, &1, String, Connection>>:
71	        Socket.close(socket)
72	        return Fail{message}
73	    case Done{unit}: IO.pure(Result<&1, &1, String, Connection>, Done{Connection{socket, ""}})
74	
75	def connected(result: Result<&1, &1, U32 & String, Socket>, body: String) -> IO(Result<&1, &1, String, Connection>):
76	  match result:
77	    case Fail{(code, message)}: IO.pure(Result<&1, &1, String, Connection>, Fail{message})
78	    case Done{socket}:
79	      do IO<Result<&1, &1, String, Connection>>:
80	        result : Socket & Result<&1, &1, U32 & String, Unit> <- TCP.send(socket, body ++ "\n")
81	        sent(result)
82	
83	def connect(port: U32, body: String) -> IO(Result<&1, &1, String, Connection>):
84	  do IO<Result<&1, &1, String, Connection>>:
85	    result : Result<&1, &1, U32 & String, Socket> <- TCP.connect("127.0.0.1", port)
86	    connected(result, body)
87	
88	def close(connection: Connection) -> IO(Unit):
89	  match connection:
90	    case Connection{socket, buffered}: Socket.close(socket)