use crate::errors::*;
use futures_util::{StreamExt, stream::FuturesUnordered, task::SpawnExt};
use russh::{
ChannelOpenFailure, ChannelStream,
server::{ChannelOpenHandle, Msg},
};
use std::fmt;
use tokio::io;
use tokio::net;
use tokio::sync::mpsc;
pub struct Relay {
pub host: String,
pub port: u16,
pub stream: ChannelStream<Msg>,
pub open: ChannelOpenHandle,
}
impl fmt::Display for Relay {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
write!(f, "{:?}:{}", self.host, self.port)
}
}
async fn relay(mut relay: Relay) -> Result<()> {
let mut sock = match net::TcpStream::connect((relay.host.as_str(), relay.port)).await {
Ok(sock) => sock,
Err(err) => {
error!("Failed to connect to {relay}: {err:#}");
relay.open.reject(ChannelOpenFailure::ConnectFailed).await;
return Ok(());
}
};
info!("Connected to {relay}");
relay.open.accept().await;
io::copy_bidirectional(&mut sock, &mut relay.stream).await?;
Ok(())
}
pub async fn run(mut rx: mpsc::Receiver<Relay>) -> Result<()> {
let mut set = FuturesUnordered::new();
loop {
tokio::select! {
req = rx.recv() => if let Some(req) = req {
debug!("Received relay request: {req}");
set.spawn(async {
if let Err(err) = relay(req).await {
error!("Relay error: {err}");
}
})?;
} else {
info!("Relay channel closed");
break;
},
Some(_) = set.next() => {
debug!("Forwarding completed");
}
}
}
Ok(())
}