1use super::*;
2use buntrail_protocol::terminal as wire;
3impl RemoteServices {
4 fn terminal_owner(
5 &self,
6 owner: &WorkspaceLease,
7 runtime: &RuntimeGeneration,
8 ) -> Result<(), CoreError> {
9 if runtime != &self.transport.runtime || owner != &self.lease {
10 return Err(CoreError::new(
11 ErrorCode::TargetChanged,
12 "terminal owner or runtime changed",
13 ));
14 }
15 Ok(())
16 }
17 fn ordered_controls(&self) -> Result<(), CoreError> {
18 if !self
19 .negotiated_capabilities()
20 .iter()
21 .any(|capability| capability == "terminal.ordered_controls")
22 {
23 return Err(CoreError::new(
24 ErrorCode::Unavailable,
25 "helper cannot guarantee ordered terminal renderer and resize admission",
26 ));
27 }
28 Ok(())
29 }
30 pub(crate) fn admit_terminal_input(
32 &self,
33 session: TerminalSessionID,
34 bytes: Vec<u8>,
35 ) -> Result<ServiceFuture<'static, Result<(), CoreError>>, CoreError> {
36 if bytes.len() > 65536 {
37 return Err(CoreError::new(
38 ErrorCode::Busy,
39 "terminal input chunk exceeds 64 KiB; input not sent",
40 ));
41 }
42 self.ordered_controls()?;
43 self.admit_terminal_control(
44 w::Call::TerminalInput {
45 lease_id: w::uuid(self.lease.lease.0),
46 session_id: w::uuid(session.0),
47 },
48 bytes,
49 )
50 }
51 pub(crate) fn admit_terminal_resize(
52 &self,
53 session: TerminalSessionID,
54 size: TerminalSize,
55 ) -> Result<ServiceFuture<'static, Result<(), CoreError>>, CoreError> {
56 self.ordered_controls()?;
57 self.admit_terminal_control(
58 w::Call::TerminalResize {
59 lease_id: w::uuid(self.lease.lease.0),
60 session_id: w::uuid(session.0),
61 columns: size.columns.get(),
62 rows: size.rows.get(),
63 },
64 Vec::new(),
65 )
66 }
67 fn admit_terminal_control(
68 &self,
69 call: w::Call,
70 bytes: Vec<u8>,
71 ) -> Result<ServiceFuture<'static, Result<(), CoreError>>, CoreError> {
72 let mut stream = self.stream_payload(call, None, bytes)?;
73 let service = self.clone();
74 Ok(Box::pin(async move {
75 loop {
76 let frame = service.next_reply(&mut stream.request)?;
77 if let Some(reply) = stream.accept(frame)? {
78 return match reply? {
79 w::Reply::TerminalAck => Ok(()),
80 _ => Err(w::invalid()),
81 };
82 }
83 }
84 }))
85 }
86 fn terminal_ack(&self, call: w::Call) -> Result<(), CoreError> {
87 match self.call(call, None)? {
88 w::Reply::TerminalAck => Ok(()),
89 _ => Err(w::invalid()),
90 }
91 }
92}
93impl TerminalService for RemoteServices {
94 fn open(&self, r: TerminalOpenRequest) -> ServiceFuture<'_, OperationReceipt> {
95 Box::pin(async move {
96 let call = (|| {
97 self.check(&r.target)?;
98 self.terminal_owner(&r.owner, &r.runtime)?;
99 Ok(w::Call::TerminalOpen {
100 lease_id: w::uuid(r.owner.lease.0),
101 session_id: w::uuid(r.session.0),
102 directory_base64: w::path(&r.directory)?,
103 shell_base64: w::path(&r.shell)?,
104 columns: r.size.columns.get(),
105 rows: r.size.rows.get(),
106 })
107 })();
108 match call {
109 Ok(call) => self.mutation_call(call, r.operation, &r.target),
110 Err(e) => buntrail_services::files::failure(r.operation, r.target, e),
111 }
112 })
113 }
114 fn attach(
115 &self,
116 r: TerminalAttachRequest,
117 ) -> ServiceFuture<'_, Result<EventStream<TerminalEvent>, CoreError>> {
118 Box::pin(async move {
119 self.ordered_controls()?;
120 self.terminal_owner(&r.owner, &r.runtime)?;
121 Ok(Box::pin(Terminal {
122 source: self.stream(
123 w::Call::TerminalAttach {
124 lease_id: w::uuid(r.owner.lease.0),
125 session_id: w::uuid(r.session.0),
126 last_output_sequence: r.last_output_sequence,
127 },
128 None,
129 )?,
130 disconnected: false,
131 }) as EventStream<TerminalEvent>)
132 })
133 }
134 fn input(
135 &self,
136 session: TerminalSessionID,
137 bytes: Vec<u8>,
138 ) -> ServiceFuture<'_, Result<(), CoreError>> {
139 match self.admit_terminal_input(session, bytes) {
140 Ok(completion) => completion,
141 Err(error) => Box::pin(async move { Err(error) }),
142 }
143 }
144 fn resize(
145 &self,
146 session: TerminalSessionID,
147 size: TerminalSize,
148 ) -> ServiceFuture<'_, Result<(), CoreError>> {
149 match self.admit_terminal_resize(session, size) {
150 Ok(completion) => completion,
151 Err(error) => Box::pin(async move { Err(error) }),
152 }
153 }
154 fn end(
155 &self,
156 session: TerminalSessionID,
157 intent: TerminalEndIntent,
158 ) -> ServiceFuture<'_, Result<(), CoreError>> {
159 Box::pin(async move {
160 self.terminal_ack(w::Call::TerminalEnd {
161 lease_id: w::uuid(self.lease.lease.0),
162 session_id: w::uuid(session.0),
163 intent: wire::EndIntent::from_core(intent),
164 })
165 })
166 }
167}
168struct Terminal {
169 source: Replies,
170 disconnected: bool,
171}
172impl Stream for Terminal {
173 type Item = Result<TerminalEvent, CoreError>;
174 fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
175 if self.disconnected {
176 return Poll::Ready(None);
177 }
178 match Pin::new(&mut self.source).poll_next(cx) {
179 Poll::Ready(Some(Ok(w::Reply::Terminal { event }))) => Poll::Ready(Some(event.core())),
180 Poll::Ready(Some(Ok(_))) => Poll::Ready(Some(Err(w::invalid()))),
181 Poll::Ready(Some(Err(_))) if self.source.request.connection_lost() => {
182 self.disconnected = true;
183 Poll::Ready(Some(Ok(TerminalEvent::Disconnected {
184 expires_at_unix_seconds: std::time::SystemTime::now()
185 .duration_since(std::time::UNIX_EPOCH)
186 .unwrap_or_default()
187 .as_secs()
188 + RETENTION_SECONDS,
189 })))
190 }
191 other => other.map(|v| v.map(|r| r.and_then(|_| Err(w::invalid())))),
192 }
193 }
194}