starla_controller/
channel_stream.rs

1//! Async stream adapter for SSH channels
2//!
3//! Bridges a `russh::Channel<Msg>` into a standard `DuplexStream` that
4//! implements `AsyncRead + AsyncWrite`, usable anywhere that expects a
5//! tokio async stream.
6
7use russh::client::Msg;
8use russh::{Channel, ChannelMsg};
9use tokio::io::{AsyncReadExt, AsyncWriteExt, DuplexStream};
10use tracing::{debug, trace};
11
12/// Create a bidirectional async stream from an SSH channel.
13///
14/// Returns a `DuplexStream` that can be used as a normal async stream.
15/// A background task bridges data between the SSH channel and the stream.
16/// The task exits when either side closes.
17pub fn channel_to_stream(mut channel: Channel<Msg>) -> DuplexStream {
18    let (app_side, bridge_side) = tokio::io::duplex(64 * 1024);
19    let (mut bridge_read, mut bridge_write) = tokio::io::split(bridge_side);
20
21    tokio::spawn(async move {
22        let mut local_buf = [0u8; 8192];
23
24        loop {
25            tokio::select! {
26                biased;
27
28                // SSH channel -> app
29                msg = channel.wait() => {
30                    match msg {
31                        Some(ChannelMsg::Data { data }) => {
32                            trace!("SSH channel -> stream: {} bytes", data.len());
33                            if bridge_write.write_all(&data).await.is_err() {
34                                break;
35                            }
36                            let _ = bridge_write.flush().await;
37                        }
38                        Some(ChannelMsg::Eof) | None => {
39                            debug!("SSH channel closed");
40                            let _ = bridge_write.shutdown().await;
41                            break;
42                        }
43                        _ => {} // ignore other messages
44                    }
45                }
46
47                // App -> SSH channel
48                result = bridge_read.read(&mut local_buf) => {
49                    match result {
50                        Ok(0) => {
51                            debug!("Stream closed by app");
52                            let _ = channel.eof().await;
53                            break;
54                        }
55                        Ok(n) => {
56                            trace!("Stream -> SSH channel: {} bytes", n);
57                            if channel.data(&local_buf[..n]).await.is_err() {
58                                break;
59                            }
60                        }
61                        Err(_) => break,
62                    }
63                }
64            }
65        }
66
67        // Ensure SSH channel is closed when bridge exits
68        let _ = channel.eof().await;
69        debug!("Channel stream bridge ended");
70    });
71
72    app_side
73}
starla_controller/channel_stream.rs

starla_controller/
channel_stream.rs
