use std::{net::SocketAddr, sync::Arc, time::Duration};
use async_trait::async_trait;
use futures::{stream::FuturesUnordered, FutureExt, StreamExt, TryFutureExt};
use safelog::sensitive as sv;
use tor_error::bad_api_usage;
use tor_linkspec::{ChannelMethod, HasChanMethod, OwnedChanTarget};
use tor_rtcompat::{Runtime, TcpProvider};
use crate::Error;
#[derive(Clone, Debug)]
pub(crate) struct DefaultTransport<R: Runtime> {
runtime: R,
}
impl<R: Runtime> DefaultTransport<R> {
pub(crate) fn new(runtime: R) -> Self {
Self { runtime }
}
}
#[async_trait]
impl<R: Runtime> crate::transport::TransportImplHelper for DefaultTransport<R> {
type Stream = <R as TcpProvider>::TcpStream;
async fn connect(
&self,
target: &OwnedChanTarget,
) -> crate::Result<(OwnedChanTarget, Self::Stream)> {
let direct_addrs: Vec<_> = match target.chan_method() {
ChannelMethod::Direct(addrs) => addrs,
#[allow(unreachable_patterns)]
_ => {
return Err(Error::UnusableTarget(bad_api_usage!(
"Used default transport implementation for an unsupported transport."
)))
}
};
let (stream, addr) = connect_to_one(&self.runtime, &direct_addrs).await?;
let mut using_target = target.clone();
let _ignore = using_target.chan_method_mut().retain_addrs(|a| a == &addr);
Ok((using_target, stream))
}
}
static CONNECTION_DELAY: Duration = Duration::from_millis(150);
async fn connect_to_one<R: Runtime>(
rt: &R,
addrs: &[SocketAddr],
) -> crate::Result<(<R as TcpProvider>::TcpStream, SocketAddr)> {
if addrs.is_empty() {
return Err(Error::UnusableTarget(bad_api_usage!(
"No addresses for chosen relay"
)));
}
let mut connections = addrs
.iter()
.enumerate()
.map(|(i, a)| {
let delay = rt.sleep(CONNECTION_DELAY * i as u32);
delay.then(move |_| {
tracing::debug!("Connecting to {}", a);
rt.connect(a)
.map_ok(move |stream| (stream, *a))
.map_err(move |e| (e, *a))
})
})
.collect::<FuturesUnordered<_>>();
let mut ret = None;
let mut errors = vec![];
while let Some(result) = connections.next().await {
match result {
Ok(s) => {
ret = Some(s);
break;
}
Err((e, a)) => {
tracing::warn!("Connection to {} failed: {}", sv(a), e);
errors.push((e, a));
}
}
}
drop(connections);
ret.ok_or_else(|| Error::ChannelBuild {
addresses: errors
.into_iter()
.map(|(e, a)| (sv(a), Arc::new(e)))
.collect(),
})
}
#[cfg(test)]
mod test {
#![allow(clippy::bool_assert_comparison)]
#![allow(clippy::clone_on_copy)]
#![allow(clippy::dbg_macro)]
#![allow(clippy::print_stderr)]
#![allow(clippy::print_stdout)]
#![allow(clippy::single_char_pattern)]
#![allow(clippy::unwrap_used)]
use std::str::FromStr;
use tor_rtcompat::{test_with_one_runtime, SleepProviderExt};
use tor_rtmock::net::MockNetwork;
use super::*;
#[test]
fn test_connect_one() {
let client_addr = "192.0.1.16".parse().unwrap();
let addr1 = SocketAddr::from_str("192.0.2.17:443").unwrap();
let addr2 = SocketAddr::from_str("192.0.3.18:443").unwrap();
let addr3 = SocketAddr::from_str("192.0.4.19:443").unwrap();
let addr4 = SocketAddr::from_str("192.0.9.9:443").unwrap();
test_with_one_runtime!(|rt| async move {
let network = MockNetwork::new();
let client_rt = network
.builder()
.add_address(client_addr)
.runtime(rt.clone());
let server_rt = network
.builder()
.add_address(addr1.ip())
.add_address(addr4.ip())
.runtime(rt.clone());
let _listener = server_rt.mock_net().listen(&addr1).await.unwrap();
let _listener2 = server_rt.mock_net().listen(&addr4).await.unwrap();
network.add_blackhole(addr3).unwrap();
let failure = connect_to_one(&client_rt, &[]).await;
assert!(failure.is_err());
for addresses in [
&[addr1][..],
&[addr1, addr2][..],
&[addr2, addr1][..],
&[addr1, addr3][..],
&[addr3, addr1][..],
&[addr1, addr2, addr3][..],
&[addr3, addr2, addr1][..],
] {
let (_conn, addr) = connect_to_one(&client_rt, addresses).await.unwrap();
assert_eq!(addr, addr1);
}
for addresses in [
&[addr2][..],
&[addr2, addr3][..],
&[addr3, addr2][..],
&[addr3][..],
] {
let expect_timeout = addresses.contains(&addr3);
let failure = rt
.timeout(
Duration::from_millis(300),
connect_to_one(&client_rt, addresses),
)
.await;
if expect_timeout {
assert!(failure.is_err());
} else {
assert!(failure.unwrap().is_err());
}
}
let (_conn, addr) = connect_to_one(&client_rt, &[addr1, addr4]).await.unwrap();
assert_eq!(addr, addr1);
let (_conn, addr) = connect_to_one(&client_rt, &[addr4, addr1]).await.unwrap();
assert_eq!(addr, addr4);
});
}
}