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    // TODO: not sure how easy end-to-end flush is
95    // async fn flush(&mut self) -> Result<(), Self::Error> {
96    // }
97}
98
99// Public wrappers for In only
100
101/// An input-only SSH channel.
102///
103/// This is used as stderr for a client.
104///
105/// <div class="warning">
106///
107/// This must be read, otherwise the SSH session will block.
108///
109/// </div>
110///
111/// `Clone` is implemented for convenience, but only one instance each
112/// should be read from.
113/// Otherwise ordering will be arbitrary, and if competing readers or writers
114/// are in different tasks, there will be churn as they continually wake
115/// each other up. Simultaneous single-reader and single-writer is fine.
116#[derive(Debug, Clone)]
117pub struct ChanIn<'g>(ChanIO<'g>);
118
119impl<'g> ChanIn<'g> {
120    pub(crate) fn new(io: ChanIO<'g>) -> Self {
121        io.sunset.inc_read_chan(io.num, io.dt);
122        Self(io)
123    }
124
125    /// Wait until the channel closes.
126    pub async fn until_closed(&self) -> Result<()> {
127        self.0.until_closed().await
128    }
129}
130
131impl Drop for ChanIn<'_> {
132    fn drop(&mut self) {
133        self.0.sunset.dec_read_chan(self.0.num, self.0.dt)
134    }
135}
136
137/// An output-only SSH channel.
138///
139/// This is used as stderr for a server, or can also be obtained using
140/// [`ChanInOut::split()`] for cases where a channel's input should
141/// be discarded.
142///
143/// `Clone` is implemented for convenience, but only one instance each
144/// should be read from or written to (this applies to `split()` instances too).
145/// Otherwise ordering will be arbitrary, and if competing readers or writers
146/// are in different tasks, there will be churn as they continually wake
147/// each other up. Simultaneous single-reader and single-writer is fine.
148#[derive(Debug, Clone)]
149pub struct ChanOut<'g>(ChanIO<'g>);
150
151impl<'g> ChanOut<'g> {
152    pub(crate) fn new(io: ChanIO<'g>) -> Self {
153        Self(io)
154    }
155
156    /// Wait until the channel closes.
157    pub async fn until_closed(&self) -> Result<()> {
158        self.0.until_closed().await
159    }
160
161    /// Send a terminal size change notification
162    ///
163    /// Only applicable to client shell channels with a PTY
164    pub async fn term_window_change(
165        &self,
166        winch: sunset::packets::WinChange,
167    ) -> Result<()> {
168        self.0.term_window_change(winch).await
169    }
170}
171
172/// A bidirectional SSH channel.
173///
174/// Used as stdin/stdout for a shell/exec/subsystem.
175/// Represents other forwarded transports.
176///
177/// <div class="warning">
178///
179/// This must be read, otherwise the SSH session will block.
180/// If input isn't required, use [`split()`](Self::split) and
181/// discard the input half.
182///
183/// </div>
184///
185/// `Clone` is implemented for convenience, but only one instance each
186/// should be read from or written to (this applies to `split()` instances too).
187/// Otherwise ordering will be arbitrary, and if competing readers or writers
188/// are in different tasks, there will be churn as they continually wake
189/// each other up. Simultaneous single-reader and single-writer is fine.
190#[derive(Debug, Clone)]
191pub struct ChanInOut<'g>(ChanIO<'g>);
192
193impl<'g> ChanInOut<'g> {
194    pub(crate) fn new(io: ChanIO<'g>) -> Self {
195        io.sunset.inc_read_chan(io.num, io.dt);
196        Self(io)
197    }
198
199    /// Convert this into separate input and output.
200    ///
201    /// Note the warning above against simultaneous use and `Clone`.
202    pub fn split(&self) -> (ChanIn<'g>, ChanOut<'g>) {
203        (ChanIn::new(self.0.clone()), ChanOut::new(self.0.clone()))
204    }
205
206    /// Wait until the channel closes.
207    pub async fn until_closed(&self) -> Result<()> {
208        self.0.until_closed().await
209    }
210
211    /// Send a terminal size change notification
212    ///
213    /// Only applicable to client shell channels with a PTY
214    pub async fn term_window_change(
215        &self,
216        winch: sunset::packets::WinChange,
217    ) -> Result<()> {
218        self.0.term_window_change(winch).await
219    }
220}
221
222impl Drop for ChanInOut<'_> {
223    fn drop(&mut self) {
224        self.0.sunset.dec_read_chan(self.0.num, self.0.dt)
225    }
226}
227
228impl ErrorType for ChanInOut<'_> {
229    type Error = sunset::Error;
230}
231
232impl ErrorType for ChanIn<'_> {
233    type Error = sunset::Error;
234}
235
236impl ErrorType for ChanOut<'_> {
237    type Error = sunset::Error;
238}
239
240impl Read for ChanInOut<'_> {
241    async fn read(&mut self, buf: &mut [u8]) -> Result<usize, sunset::Error> {
242        self.0.read(buf).await
243    }
244}
245
246impl Write for ChanInOut<'_> {
247    async fn write(&mut self, buf: &[u8]) -> Result<usize, sunset::Error> {
248        self.0.write(buf).await
249    }
250}
251
252impl Read for ChanIn<'_> {
253    async fn read(&mut self, buf: &mut [u8]) -> Result<usize, sunset::Error> {
254        self.0.read(buf).await
255    }
256}
257
258impl Write for ChanOut<'_> {
259    async fn write(&mut self, buf: &[u8]) -> Result<usize, sunset::Error> {
260        self.0.write(buf).await
261    }
262}
sunset_async/async_channel.rs

sunset_async/
async_channel.rs
