1	# Talks in lines to a started program, over pipes, with or without a deadline.
2	# Send lines, then read lines until one starts with a given text. It knows no protocol.
3	import Base
4	import bend-stdio-pipes@0.2.0.0/impl.bend as Stdio
5	
6	# A running program: its process id, the File that writes its stdin, the File that reads its stdout,
7	# and the most bytes in one of its lines, with its line end.
8	type PipeSession is Type:
9	  PipeSession{pid: U32, input: File, out: File, line_cap: U32}
10	
11	# A line of the program, as PipeSession.ask sorts it.
12	type Heard is Type:
13	  Line{text: String}
14	  Last{text: String}
15	  Ended{}
16	  Failed{error: U32 & String}
17	
18	def PipeSession.pid(session: PipeSession) -> PipeSession & U32:
19	  match session:
20	    case PipeSession{+pid, input, out, line_cap}:
21	      (PipeSession{pid, input, out, line_cap}, pid)
22	
23	def PipeSession.started(
24	  opened: Result<&1, &1, U32 & String, U32 & (File & File)>,
25	  line_cap: U32
26	) -> Result<&1, &1, U32 & String, PipeSession>:
27	  match opened:
28	    case Fail{error}:
29	      Fail{error}
30	    case Done{(pid, (input, out))}:
31	      Done{PipeSession{pid, input, out, line_cap}}
32	
33	# Starts program with args. A longer line of it fails with E2BIG (7).
34	def PipeSession.start(
35	  program: String,
36	  args: List<String>,
37	  line_cap: U32
38	) -> IO(Result<&1, &1, U32 & String, PipeSession>):
39	  do IO<Result<&1, &1, U32 & String, PipeSession>>:
40	    opened: Result<&1, &1, U32 & String, U32 & (File & File)> <- Stdio.open(program, args)
41	    return PipeSession.started(opened, line_cap)
42	
43	# The code of the error that PipeSession.ask gives when the program ended before its answer:
44	# EPIPE.
45	def PipeSession.ended() -> U32:
46	  32
47	
48	def Heard.read(
49	  read: File & Result<&1, &1, U32 & String, Maybe<String>>,
50	  +until: String
51	) -> File
52	  & Heard:
53	  match read:
54	    case (out, Fail{error}):
55	      (out, Failed{error})
56	    case (out, Done{None{}}):
57	      (out, Ended{})
58	    case (out, Done{Some{line}}):
59	      +text = String.trim_end(line)
60	      (out, Bool.pick(Heard, String.starts_with(text, until), Last{text}, Line{text}))
61	
62	# said holds the lines so far, the last first; fuel bounds the lines, far beyond any answer.
63	def PipeSession.ask.go(
64	  fuel: Nat,
65	  at: File & Heard,
66	  +until: String,
67	  +line_cap: U32,
68	  said: List<&2, String>
69	) -> IO(File & Result<&1, &1, U32 & String, List<&2, String>>):
70	  match fuel:
71	    case 0n:
72	      match at:
73	        case (out, _):
74	          IO.pure(
75	            File & Result<&1, &1, U32 & String, List<&2, String>>,
76	            (out, Fail{(5, "the program said too many lines before " ++ until)})
77	          )
78	    case 1n+rest:
79	      match at:
80	        case (out, Failed{error}):
81	          IO.pure(File & Result<&1, &1, U32 & String, List<&2, String>>, (out, Fail{error}))
82	        case (out, Ended{}):
83	          IO.pure(
84	            File & Result<&1, &1, U32 & String, List<&2, String>>,
85	            (out, Fail{(PipeSession.ended(), "the program ended before it said " ++ until)})
86	          )
87	        case (out, Last{text}):
88	          IO.pure(
89	            File & Result<&1, &1, U32 & String, List<&2, String>>,
90	            (out, Done{List.reverse(&2, String, text <> said)})
91	          )
92	        case (out, Line{text}):
93	          do IO<File & Result<&1, &1, U32 & String, List<&2, String>>>:
94	            read: File
95	              & Result<&1, &1, U32 & String, Maybe<String>> <- Stdio.read_line(out, line_cap)
96	            PipeSession.ask.go(rest, Heard.read(read, until), until, line_cap, text <> said)
97	
98	def PipeSession.asked(
99	  pid: U32,
100	  input: File,
101	  line_cap: U32,
102	  answer: File & Result<&1, &1, U32 & String, List<&2, String>>
103	) -> IO(PipeSession & Result<&1, &1, U32 & String, List<&2, String>>):
104	  match answer:
105	    case (out, result):
106	      IO.pure(
107	        PipeSession & Result<&1, &1, U32 & String, List<&2, String>>,
108	        (PipeSession{pid, input, out, line_cap}, result)
109	      )
110	
111	def PipeSession.wrote(
112	  pid: U32,
113	  wrote: File & Result<&1, &1, U32 & String, Unit>,
114	  out: File,
115	  +line_cap: U32,
116	  +until: String
117	) -> IO(PipeSession & Result<&1, &1, U32 & String, List<&2, String>>):
118	  match wrote:
119	    case (input, Fail{error}):
120	      IO.pure(
121	        PipeSession & Result<&1, &1, U32 & String, List<&2, String>>,
122	        (PipeSession{pid, input, out, line_cap}, Fail{error})
123	      )
124	    case (input, Done{_}):
125	      do IO<PipeSession & Result<&1, &1, U32 & String, List<&2, String>>>:
126	        read: File & Result<&1, &1, U32 & String, Maybe<String>> <- Stdio.read_line(out, line_cap)
127	        answer: File & Result<&1, &1, U32 & String, List<&2, String>> <-
128	          PipeSession.ask.go(4294967295n, Heard.read(read, until), until, line_cap, [])
129	        PipeSession.asked(pid, input, line_cap, answer)
130	
131	# Sends lines, then answers the program's lines up to the first that starts with until.
132	def PipeSession.ask(
133	  session: PipeSession,
134	  lines: List<String>,
135	  +until: String
136	) -> IO(PipeSession & Result<&1, &1, U32 & String, List<&2, String>>):
137	  match session:
138	    case PipeSession{pid, input, out, line_cap}:
139	      do IO<PipeSession & Result<&1, &1, U32 & String, List<&2, String>>>:
140	        wrote: File & Result<&1, &1, U32 & String, Unit> <- Stdio.write_lines(input, lines)
141	        PipeSession.wrote(pid, wrote, out, line_cap, until)
142	
143	# Kills the program, whether or not it has ended, and closes its Files.
144	def PipeSession.kill(session: PipeSession) -> IO(Unit):
145	  match session:
146	    case PipeSession{pid, input, out, _}:
147	      do IO<Unit>:
148	        killed: Result<&1, &1, U32 & String, Unit> <- Stdio.kill(pid)
149	        File.close(input)
150	        File.close(out)
151	        return Unit{}
152	
153	# What PipeSession.ask gives.
154	def PipeSession.Answer() -> Type:
155	  PipeSession & Result<&1, &1, U32 & String, List<&2, String>>
156	
157	# The answer that came once the program was killed: its Files close.
158	def PipeSession.ask.late(late: Maybe<&1, PipeSession.Answer()>) -> IO(Maybe<PipeSession.Answer()>):
159	  match late:
160	    case Some{(PipeSession{_, input, out, _}, _)}:
161	      do IO<Maybe<PipeSession.Answer()>>:
162	        File.close(input)
163	        File.close(out)
164	        return None{}
165	    case None{}:
166	      IO.pure(Maybe<PipeSession.Answer()>, None{})
167	
168	def PipeSession.ask.polled(
169	  +pid: U32,
170	  +chan: Chan(PipeSession.Answer()),
171	  polled: Poll<&1, &1, Maybe<&1, PipeSession.Answer()>, Unit>
172	) -> IO(Maybe<PipeSession.Answer()>):
173	  match polled:
174	    case Ready{Some{answer}}:
175	      do IO<Maybe<PipeSession.Answer()>>:
176	        Chan.close(PipeSession.Answer(), chan)
177	        return Some{answer}
178	    case Ready{None{}}:
179	      IO.pure(Maybe<PipeSession.Answer()>, None{})
180	    case Wait{_}:
181	      do IO<Maybe<PipeSession.Answer()>>:
182	        killed: Result<&1, &1, U32 & String, Unit> <- Stdio.kill(pid)
183	        late: Maybe<&1, PipeSession.Answer()> <- Chan.recv(PipeSession.Answer(), chan)
184	        Chan.close(PipeSession.Answer(), chan)
185	        PipeSession.ask.late(late)
186	
187	# As PipeSession.ask, within ms; None when the program has not answered by then: it is killed, and
188	# the session is gone. Nothing of it runs on after.
189	def PipeSession.ask.within(
190	  session: PipeSession,
191	  lines: List<String>,
192	  +until: String,
193	  ms: U32
194	) -> IO(Maybe<PipeSession.Answer()>):
195	  match session:
196	    case PipeSession{+pid, input, out, line_cap}:
197	      do IO<Maybe<PipeSession.Answer()>>:
198	        +chan: Chan(PipeSession.Answer()) <- Chan.new(PipeSession.Answer(), 1)
199	        forked: Chan(
200	          PipeSession.Answer()
201	        ) <- IO.fork.go(
202	          PipeSession.Answer(),
203	          PipeSession.ask(PipeSession{pid, input, out, line_cap}, lines, until),
204	          chan
205	        )
206	        polled: Poll<&1, &1, Maybe<&1, PipeSession.Answer()>, Unit> <-
207	          Chan.try_recv(PipeSession.Answer(), chan, ms)
208	        PipeSession.ask.polled(pid, chan, polled)
209	
210	def PipeSession.end.go(
211	  fuel: Nat,
212	  read: File & Result<&1, &1, U32 & String, Maybe<String>>,
213	  +line_cap: U32,
214	  said: List<&2, String>
215	) -> IO(Result<&1, &1, U32 & String, List<&2, String>>):
216	  match fuel read:
217	    case 0n Tuple{out, _}:
218	      do IO<Result<&1, &1, U32 & String, List<&2, String>>>:
219	        File.close(out)
220	        return Fail{(5, "the program said too many lines after its input ended")}
221	    case 1n+rest Tuple{out, Fail{error}}:
222	      do IO<Result<&1, &1, U32 & String, List<&2, String>>>:
223	        File.close(out)
224	        return Fail{error}
225	    case 1n+rest Tuple{out, Done{None{}}}:
226	      do IO<Result<&1, &1, U32 & String, List<&2, String>>>:
227	        File.close(out)
228	        return Done{List.reverse(&2, String, said)}
229	    case 1n+rest Tuple{out, Done{Some{line}}}:
230	      do IO<Result<&1, &1, U32 & String, List<&2, String>>>:
231	        next: File & Result<&1, &1, U32 & String, Maybe<String>> <- Stdio.read_line(out, line_cap)
232	        PipeSession.end.go(rest, next, line_cap, String.trim_end(line) <> said)
233	
234	# Closes the program's stdin, then answers what it said until it ended.
235	def PipeSession.end(session: PipeSession) -> IO(Result<&1, &1, U32 & String, List<&2, String>>):
236	  match session:
237	    case PipeSession{_, input, out, +line_cap}:
238	      do IO<Result<&1, &1, U32 & String, List<&2, String>>>:
239	        File.close(input)
240	        read: File & Result<&1, &1, U32 & String, Maybe<String>> <- Stdio.read_line(out, line_cap)
241	        PipeSession.end.go(4294967295n, read, line_cap, [])