impl.bend
bend-pipe-session@0.1.0.0 · 175 lines · 0 definitions
1# THIS MODULE IS PROJECT-AGNOSTIC. KEEP IT THAT WAY.
2# A started program that we talk to in lines, over pipes: send lines, then read lines until one
3# starts with a given text. It knows no protocol.
4import Base
5import bend-stdio-pipes@0.1.0.0/impl.bend as Stdio
6
7# A running program: its process id, the File that writes its stdin, the File that reads its stdout,
8# and the most bytes in one of its lines, with its line end.
9type PipeSession is Type:
10 PipeSession{pid: U32, input: File, out: File, line_cap: U32}
11
12# A line of the program, as PipeSession.ask sorts it.
13type Heard is Type:
14 Line{text: String}
15 Last{text: String}
16 Ended{}
17 Failed{error: U32 & String}
18
19def PipeSession.pid(session: PipeSession) -> PipeSession & U32:
20 match session:
21 case PipeSession{+pid, input, out, line_cap}:
22 (PipeSession{pid, input, out, line_cap}, pid)
23
24def PipeSession.started(
25 opened: Result<&1, &1, U32 & String, U32 & (File & File)>,
26 line_cap: U32
27) -> Result<&1, &1, U32 & String, PipeSession>:
28 match opened:
29 case Fail{error}:
30 Fail{error}
31 case Done{(pid, (input, out))}:
32 Done{PipeSession{pid, input, out, line_cap}}
33
34# Starts program with args. A longer line of it fails with E2BIG (7).
35def PipeSession.start(
36 program: String,
37 args: List<String>,
38 line_cap: U32
39) -> IO(Result<&1, &1, U32 & String, PipeSession>):
40 do IO<Result<&1, &1, U32 & String, PipeSession>>:
41 opened: Result<&1, &1, U32 & String, U32 & (File & File)> <- Stdio.open(program, args)
42 return PipeSession.started(opened, line_cap)
43
44# The code of the error that PipeSession.ask gives when the program ended before its answer:
45# EPIPE.
46def PipeSession.ended() -> U32:
47 32
48
49def Heard.read(
50 read: File & Result<&1, &1, U32 & String, Maybe<String>>,
51 +until: String
52) -> File
53 & Heard:
54 match read:
55 case (out, Fail{error}):
56 (out, Failed{error})
57 case (out, Done{None{}}):
58 (out, Ended{})
59 case (out, Done{Some{line}}):
60 +text = String.trim_end(line)
61 (out, Bool.pick(Heard, String.starts_with(text, until), Last{text}, Line{text}))
62
63# said holds the lines so far, the last first; fuel bounds the lines, far beyond any answer.
64def PipeSession.ask.go(
65 fuel: Nat,
66 at: File & Heard,
67 +until: String,
68 +line_cap: U32,
69 said: List<&2, String>
70) -> IO(File & Result<&1, &1, U32 & String, List<&2, String>>):
71 match fuel:
72 case 0n:
73 match at:
74 case (out, _):
75 IO.pure(
76 File & Result<&1, &1, U32 & String, List<&2, String>>,
77 (out, Fail{(5, "the program said too many lines before " ++ until)})
78 )
79 case 1n+rest:
80 match at:
81 case (out, Failed{error}):
82 IO.pure(File & Result<&1, &1, U32 & String, List<&2, String>>, (out, Fail{error}))
83 case (out, Ended{}):
84 IO.pure(
85 File & Result<&1, &1, U32 & String, List<&2, String>>,
86 (out, Fail{(PipeSession.ended(), "the program ended before it said " ++ until)})
87 )
88 case (out, Last{text}):
89 IO.pure(
90 File & Result<&1, &1, U32 & String, List<&2, String>>,
91 (out, Done{List.reverse(&2, String, text <> said)})
92 )
93 case (out, Line{text}):
94 do IO<File & Result<&1, &1, U32 & String, List<&2, String>>>:
95 read: File
96 & Result<&1, &1, U32 & String, Maybe<String>> <- Stdio.read_line(out, line_cap)
97 PipeSession.ask.go(rest, Heard.read(read, until), until, line_cap, text <> said)
98
99def PipeSession.asked(
100 pid: U32,
101 input: File,
102 line_cap: U32,
103 answer: File & Result<&1, &1, U32 & String, List<&2, String>>
104) -> IO(PipeSession & Result<&1, &1, U32 & String, List<&2, String>>):
105 match answer:
106 case (out, result):
107 IO.pure(
108 PipeSession & Result<&1, &1, U32 & String, List<&2, String>>,
109 (PipeSession{pid, input, out, line_cap}, result)
110 )
111
112def PipeSession.wrote(
113 pid: U32,
114 wrote: File & Result<&1, &1, U32 & String, Unit>,
115 out: File,
116 +line_cap: U32,
117 +until: String
118) -> IO(PipeSession & Result<&1, &1, U32 & String, List<&2, String>>):
119 match wrote:
120 case (input, Fail{error}):
121 IO.pure(
122 PipeSession & Result<&1, &1, U32 & String, List<&2, String>>,
123 (PipeSession{pid, input, out, line_cap}, Fail{error})
124 )
125 case (input, Done{_}):
126 do IO<PipeSession & Result<&1, &1, U32 & String, List<&2, String>>>:
127 read: File & Result<&1, &1, U32 & String, Maybe<String>> <- Stdio.read_line(out, line_cap)
128 answer: File & Result<&1, &1, U32 & String, List<&2, String>> <-
129 PipeSession.ask.go(4294967295n, Heard.read(read, until), until, line_cap, [])
130 PipeSession.asked(pid, input, line_cap, answer)
131
132# Sends lines, then answers the program's lines up to the first that starts with until.
133def PipeSession.ask(
134 session: PipeSession,
135 lines: List<String>,
136 +until: String
137) -> IO(PipeSession & Result<&1, &1, U32 & String, List<&2, String>>):
138 match session:
139 case PipeSession{pid, input, out, line_cap}:
140 do IO<PipeSession & Result<&1, &1, U32 & String, List<&2, String>>>:
141 wrote: File & Result<&1, &1, U32 & String, Unit> <- Stdio.write_lines(input, lines)
142 PipeSession.wrote(pid, wrote, out, line_cap, until)
143
144def PipeSession.end.go(
145 fuel: Nat,
146 read: File & Result<&1, &1, U32 & String, Maybe<String>>,
147 +line_cap: U32,
148 said: List<&2, String>
149) -> IO(Result<&1, &1, U32 & String, List<&2, String>>):
150 match fuel read:
151 case 0n Tuple{out, _}:
152 do IO<Result<&1, &1, U32 & String, List<&2, String>>>:
153 File.close(out)
154 return Fail{(5, "the program said too many lines after its input ended")}
155 case 1n+rest Tuple{out, Fail{error}}:
156 do IO<Result<&1, &1, U32 & String, List<&2, String>>>:
157 File.close(out)
158 return Fail{error}
159 case 1n+rest Tuple{out, Done{None{}}}:
160 do IO<Result<&1, &1, U32 & String, List<&2, String>>>:
161 File.close(out)
162 return Done{List.reverse(&2, String, said)}
163 case 1n+rest Tuple{out, Done{Some{line}}}:
164 do IO<Result<&1, &1, U32 & String, List<&2, String>>>:
165 next: File & Result<&1, &1, U32 & String, Maybe<String>> <- Stdio.read_line(out, line_cap)
166 PipeSession.end.go(rest, next, line_cap, String.trim_end(line) <> said)
167
168# Closes the program's stdin, then answers what it said until it ended.
169def PipeSession.end(session: PipeSession) -> IO(Result<&1, &1, U32 & String, List<&2, String>>):
170 match session:
171 case PipeSession{_, input, out, +line_cap}:
172 do IO<Result<&1, &1, U32 & String, List<&2, String>>>:
173 File.close(input)
174 read: File & Result<&1, &1, U32 & String, Maybe<String>> <- Stdio.read_line(out, line_cap)
175 PipeSession.end.go(4294967295n, read, line_cap, [])