impl.bend

bend-pipe-session@0.2.0.0 · 241 lines · 18 definitions

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.
3import Base
4import 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.
8type 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.
12type Heard is Type:
13  Line{text: String}
14  Last{text: String}
15  Ended{}
16  Failed{error: U32 & String}
17
18def 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
23def 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).
34def 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.
45def PipeSession.ended() -> U32:
46  32
47
48def 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.
63def 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
98def 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
111def 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.
132def 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.
144def 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.
154def 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.
158def 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
168def 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.
189def 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
210def 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.
235def 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, [])