sunset_async/
async_channel.rs

1//! Presents SSH channels as async
2use core::future::poll_fn;
3
4#[allow(unused_imports)]
5use log::{debug, error, info, log, trace, warn};
6
7use embedded_io_async::{ErrorType, Read, Write};
8
9use crate::*;
10use sunset::{ChanData, ChanNum, Result};
11
12/// Common implementation
13pub(crate) struct ChanIO<'g> {
14    num: ChanNum,
15    dt: ChanData,
16    sunset: &'g dyn async_sunset::ChanCore,
17}
18
19impl<'g> ChanIO<'g> {
20    /// Create a new Normal ChanIO.
21    ///
22    /// Only to be called by add_channel(), which has already set
23    /// the initial refcount = 1.
24    pub(crate) fn new_normal(
25        num: ChanNum,
26        sunset: &'g dyn async_sunset::ChanCore,
27    ) -> Self {
28        Self { num, dt: ChanData::Normal, sunset }
29    }
30
31    pub(crate) fn clone_stderr(&self) -> Self {
32        let mut c = self.clone();
33        c.dt = ChanData::Stderr;
34        c
35    }
36}
37
38impl core::fmt::Debug for ChanIO<'_> {
39    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
40        f.debug_struct("ChanIO")
41            .field("num", &self.num)
42            .field("dt", &self.dt)
43            .finish_non_exhaustive()
44    }
45}
46
47impl ChanIO<'_> {
48    pub async fn until_closed(&self) -> Result<()> {
49        poll_fn(|cx| self.sunset.poll_until_channel_closed(cx, self.num)).await
50    }
51
52    pub async fn term_window_change(
53        &self,
54        winch: sunset::packets::WinChange,
55    ) -> Result<()> {
56        poll_fn(|cx| self.sunset.poll_term_window_change(cx, self.num, &winch)).await
57    }
58}
59
60impl Drop for ChanIO<'_> {
61    fn drop(&mut self) {
62        self.sunset.dec_chan(self.num)
63    }
64}
65
66// ChanIO implements Clone to share between ChanIn/ChanOut/ChanInOut.
67// There's only one waker for each of in/out/ext, so allowing clone
68// on the ChanInOut etc isn't desirable - having two instances polling
69// the same direction/dt will just result in churn between wakers if they're
70// in different tasks.
71impl Clone for ChanIO<'_> {
72    fn clone(&self) -> Self {
73        self.sunset.inc_chan(self.num);
74        Self { num: self.num, dt: self.dt, sunset: self.sunset }
75    }
76}
77
78impl ErrorType for ChanIO<'_> {
79    type Error = sunset::Error;
80}
81
82impl Read for ChanIO<'_> {
83    async fn read(&mut self, buf: &mut [u8]) -> Result<usize, sunset::Error> {
84        poll_fn(|cx| self.sunset.poll_read_channel(cx, self.num, self.dt, buf)).await
85    }
86}
87
88impl Write for ChanIO<'_> {
89    async fn write(&mut self, buf: &[u8]) -> Result<usize, sunset::Error> {
90        poll_fn(|cx| self.sunset.poll_write_channel(cx, self.num, self.dt, buf))
91            .await
92    }
93
94    async fn flush(&mut self) -> Result<()> {
95        // TODO: could this wait for the packet to get sent out?
96        Ok(())
97    }
98}
99
100// Public wrappers for In only
101
102/// An input-only SSH channel.
103///
104/// This is used as stderr for a client.
105///
106/// <div class="warning">
107///
108/// This must be read, otherwise the SSH session will block.
109/// Alternatively drop this `ChanIn` and incoming data will be discarded.
110///
111/// </div>
112///
113/// `Clone` is implemented for convenience, but only one instance each
114/// should be read from.
115/// Otherwise ordering will be arbitrary, and if competing readers or writers
116/// are in different tasks, there will be churn as they continually wake
117/// each other up. Simultaneous single-reader and single-writer is fine.
118#[derive(Debug)]
119pub struct ChanIn<'g>(ChanIO<'g>);
120
121impl<'g> ChanIn<'g> {
122    pub(crate) fn new(io: ChanIO<'g>) -> Self {
123        io.sunset.inc_read_chan(io.num, io.dt);
124        Self(io)
125    }
126
127    /// Return the channel number.
128    pub fn num(&self) -> ChanNum {
129        self.0.num
130    }
131
132    /// Wait until the channel closes.
133    pub async fn until_closed(&self) -> Result<()> {
134        self.0.until_closed().await
135    }
136}
137
138impl Drop for ChanIn<'_> {
139    fn drop(&mut self) {
140        self.0.sunset.dec_read_chan(self.0.num, self.0.dt)
141    }
142}
143
144impl Clone for ChanIn<'_> {
145    fn clone(&self) -> Self {
146        Self::new(self.0.clone())
147    }
148}
149
150impl<'g> From<ChanInOut<'g>> for ChanIn<'g> {
151    fn from(ch: ChanInOut<'g>) -> Self {
152        ChanIn::new(ch.0.clone())
153    }
154}
155
156/// An output-only SSH channel.
157///
158/// This is used as stderr for a server, or can also be obtained using
159/// [`ChanInOut::split()`] for cases where a channel's input should
160/// be discarded.
161///
162/// `Clone` is implemented for convenience, but only one instance each
163/// should be read from or written to (this applies to `split()` instances too).
164/// Otherwise ordering will be arbitrary, and if competing readers or writers
165/// are in different tasks, there will be churn as they continually wake
166/// each other up. Simultaneous single-reader and single-writer is fine.
167#[derive(Debug, Clone)]
168pub struct ChanOut<'g>(ChanIO<'g>);
169
170impl<'g> ChanOut<'g> {
171    pub(crate) fn new(io: ChanIO<'g>) -> Self {
172        Self(io)
173    }
174
175    /// Return the channel number.
176    pub fn num(&self) -> ChanNum {
177        self.0.num
178    }
179
180    /// Wait until the channel closes.
181    pub async fn until_closed(&self) -> Result<()> {
182        self.0.until_closed().await
183    }
184
185    /// Send a terminal size change notification
186    ///
187    /// Only applicable to client shell channels with a PTY
188    pub async fn term_window_change(
189        &self,
190        winch: sunset::packets::WinChange,
191    ) -> Result<()> {
192        self.0.term_window_change(winch).await
193    }
194}
195
196impl<'g> From<ChanInOut<'g>> for ChanOut<'g> {
197    fn from(ch: ChanInOut<'g>) -> Self {
198        ChanOut::new(ch.0.clone())
199    }
200}
201
202/// A bidirectional SSH channel.
203///
204/// Used as stdin/stdout for a shell/exec/subsystem.
205/// Represents other forwarded transports.
206///
207/// <div class="warning">
208///
209/// This must be read, otherwise the SSH session will block.
210/// If input isn't required, use [`split()`](Self::split) and
211/// discard the input half.
212///
213/// </div>
214///
215/// `Clone` is implemented for convenience, but only one instance each
216/// should be read from or written to (this applies to `split()` instances too).
217/// Otherwise ordering will be arbitrary, and if competing readers or writers
218/// are in different tasks, there will be churn as they continually wake
219/// each other up. Simultaneous single-reader and single-writer is fine.
220#[derive(Debug)]
221pub struct ChanInOut<'g>(ChanIO<'g>);
222
223impl<'g> ChanInOut<'g> {
224    pub(crate) fn new(io: ChanIO<'g>) -> Self {
225        io.sunset.inc_read_chan(io.num, io.dt);
226        Self(io)
227    }
228
229    /// Return the channel number.
230    pub fn num(&self) -> ChanNum {
231        self.0.num
232    }
233
234    /// Convert this into separate input and output.
235    ///
236    /// Note the warning above against simultaneous use and `Clone`.
237    pub fn split(&self) -> (ChanIn<'g>, ChanOut<'g>) {
238        (ChanIn::new(self.0.clone()), ChanOut::new(self.0.clone()))
239    }
240
241    /// Wait until the channel closes.
242    pub async fn until_closed(&self) -> Result<()> {
243        self.0.until_closed().await
244    }
245
246    /// Send a terminal size change notification
247    ///
248    /// Only applicable to client shell channels with a PTY
249    pub async fn term_window_change(
250        &self,
251        winch: sunset::packets::WinChange,
252    ) -> Result<()> {
253        self.0.term_window_change(winch).await
254    }
255}
256
257impl Drop for ChanInOut<'_> {
258    fn drop(&mut self) {
259        self.0.sunset.dec_read_chan(self.0.num, self.0.dt)
260    }
261}
262
263impl Clone for ChanInOut<'_> {
264    fn clone(&self) -> Self {
265        Self::new(self.0.clone())
266    }
267}
268
269impl ErrorType for ChanInOut<'_> {
270    type Error = sunset::Error;
271}
272
273impl ErrorType for ChanIn<'_> {
274    type Error = sunset::Error;
275}
276
277impl ErrorType for ChanOut<'_> {
278    type Error = sunset::Error;
279}
280
281impl Read for ChanInOut<'_> {
282    async fn read(&mut self, buf: &mut [u8]) -> Result<usize, sunset::Error> {
283        self.0.read(buf).await
284    }
285}
286
287impl Write for ChanInOut<'_> {
288    async fn write(&mut self, buf: &[u8]) -> Result<usize, sunset::Error> {
289        self.0.write(buf).await
290    }
291
292    async fn flush(&mut self) -> Result<()> {
293        // TODO: could this wait for the packet to get sent out?
294        Ok(())
295    }
296}
297
298impl Read for ChanIn<'_> {
299    async fn read(&mut self, buf: &mut [u8]) -> Result<usize, sunset::Error> {
300        self.0.read(buf).await
301    }
302}
303
304impl Write for ChanOut<'_> {
305    async fn write(&mut self, buf: &[u8]) -> Result<usize, sunset::Error> {
306        self.0.write(buf).await
307    }
308
309    async fn flush(&mut self) -> Result<()> {
310        // TODO: could this wait for the packet to get sent out?
311        Ok(())
312    }
313}
sunset_async/async_channel.rs

sunset_async/
async_channel.rs
