♻️ refactor

main
_ 2021-10-10 15:19:48 +00:00
parent 953254e550
commit a2d4ae81e0
1 changed files with 77 additions and 42 deletions

View File

@ -4,6 +4,8 @@ use tokio::net::TcpStream;
use crate::prelude::*; use crate::prelude::*;
use protocol::PeerId; use protocol::PeerId;
/// A partially-filled-out config that structopt can deal with
/// Try to turn this into a Config as soon as possible.
#[derive (Debug, StructOpt)] #[derive (Debug, StructOpt)]
struct Opt { struct Opt {
#[structopt (long)] #[structopt (long)]
@ -17,65 +19,98 @@ struct Opt {
} }
pub async fn main (args: &[OsString]) -> anyhow::Result <()> { pub async fn main (args: &[OsString]) -> anyhow::Result <()> {
let opt = Arc::new (Opt::from_iter (args)); let opt = Opt::from_iter (args);
let conf = opt.into_config ().await?;
let relay_cert: Vec <u8> = match opt.cert_url.as_ref () { let mut end_server = P4EndServer::connect (conf).await?;
end_server.run ().await?;
Ok (())
}
/// A filled-out config for constructing an end server
#[derive (Clone)]
pub struct Config {
pub debug_echo: bool,
pub id: String,
pub relay_addr: SocketAddr,
pub relay_cert: Vec <u8>,
}
impl Opt {
/// Converts self into a Config that the server can use.
/// Performs I/O to load the relay cert from disk or from HTTP.
/// Fails if arguments can't be parsed or if I/O fails.
pub async fn into_config (self) -> anyhow::Result <Config> {
let id = self.server_id.clone ().unwrap_or_else (|| "bogus_server".to_string ());
let relay_addr: SocketAddr = self.relay_addr.clone ().unwrap_or_else (|| String::from ("127.0.0.1:30380")).parse ()?;
// Do I/O after all parsing is done.
// We don't want to waste a network request only to come back and error
// out on like "127.oooo.1" not parsing into a relay address.
let relay_cert: Vec <u8> = match self.cert_url.as_ref () {
Some (url) => reqwest::get (url).await?.bytes ().await?.into_iter ().collect (), Some (url) => reqwest::get (url).await?.bytes ().await?.into_iter ().collect (),
None => tokio::fs::read ("quic_server.crt").await?, None => tokio::fs::read ("quic_server.crt").await?,
}; };
let relay_addr: SocketAddr = opt.relay_addr.clone ().unwrap_or_else (|| String::from ("127.0.0.1:30380")).parse ()?;
let id = opt.server_id.clone ().unwrap_or_else (|| "bogus_server".to_string ()); Ok (Config {
debug_echo: self.debug_echo,
let endpoint = make_client_endpoint ("0.0.0.0:0".parse ()?, &[&relay_cert])?;
trace! ("Connecting to relay server");
let quinn::NewConnection {
bi_streams,
..
} = protocol::p4_connect_to_p3 (&endpoint, &relay_addr, &id).await?;
let mut end_server = P4EndServer {
bi_streams,
id, id,
relay_addr, relay_addr,
relay_cert, relay_cert,
}; })
debug! ("Connected to relay server");
trace! ("Accepting bi streams from P3");
loop {
let (relay_send, relay_recv) = end_server.bi_streams.next ().await.ok_or_else (|| anyhow::anyhow! ("Relay server didn't open a bi stream"))??;
tokio::spawn (handle_bi_stream (Arc::clone (&opt), relay_send, relay_recv));
} }
} }
/// An end server that is connected to its relay server
pub struct P4EndServer { pub struct P4EndServer {
bi_streams: quinn::IncomingBiStreams, bi_streams: quinn::IncomingBiStreams,
id: String, conf: Arc <Config>,
relay_addr: SocketAddr,
relay_cert: Vec <u8>,
} }
impl P4EndServer { impl P4EndServer {
pub fn id (&self) -> &str { /// Tries to connect to a relay server and return an end server.
&self.id /// Fails if there's a network issue while making the QUIC connection
/// to the relay server.
pub async fn connect (conf: Config) -> anyhow::Result <Self> {
trace! ("P4 end server making its QUIC endpoint");
let endpoint = make_client_endpoint ("0.0.0.0:0".parse ()?, &[&conf.relay_cert])?;
trace! ("P4 end server connecting to P3 relay server");
let quinn::NewConnection {
bi_streams,
..
} = protocol::p4_connect_to_p3 (&endpoint, &conf.relay_addr, &conf.id).await?;
debug! ("Connected to relay server");
Ok (P4EndServer {
bi_streams,
conf: Arc::new (conf),
})
} }
pub fn relay_addr (&self) -> SocketAddr { pub fn config (&self) -> &Config {
self.relay_addr &*self.conf
} }
pub fn relay_cert (&self) -> &[u8] { pub async fn run (&mut self) -> anyhow::Result <()> {
&self.relay_cert trace! ("Accepting bi streams from P3");
loop {
let (relay_send, relay_recv) = self.bi_streams.next ().await.ok_or_else (|| anyhow::anyhow! ("Relay server didn't open a bi stream"))??;
tokio::spawn (handle_bi_stream (Arc::clone (&self.conf), relay_send, relay_recv));
}
} }
} }
async fn handle_bi_stream ( async fn handle_bi_stream (
opt: Arc <Opt>, conf: Arc <Config>,
relay_send: quinn::SendStream, relay_send: quinn::SendStream,
mut relay_recv: quinn::RecvStream, mut relay_recv: quinn::RecvStream,
) -> anyhow::Result <()> ) -> anyhow::Result <()>
@ -84,14 +119,14 @@ async fn handle_bi_stream (
protocol::P3ToP4Stream::NewPtthConnection { protocol::P3ToP4Stream::NewPtthConnection {
client_id, client_id,
.. ..
} => handle_new_ptth_connection (opt, relay_send, relay_recv, client_id).await?, } => handle_new_ptth_connection (conf, relay_send, relay_recv, client_id).await?,
} }
Ok (()) Ok (())
} }
async fn handle_new_ptth_connection ( async fn handle_new_ptth_connection (
opt: Arc <Opt>, conf: Arc <Config>,
mut relay_send: quinn::SendStream, mut relay_send: quinn::SendStream,
mut relay_recv: quinn::RecvStream, mut relay_recv: quinn::RecvStream,
_client_id: String, _client_id: String,
@ -108,7 +143,7 @@ async fn handle_new_ptth_connection (
debug! ("Started PTTH connection"); debug! ("Started PTTH connection");
if opt.debug_echo { if conf.debug_echo {
relay_send.write (b"Connected to P4=P5 debug echo server\n").await?; relay_send.write (b"Connected to P4=P5 debug echo server\n").await?;
debug! ("Relaying bytes using internal debug echo server (P4=P5)"); debug! ("Relaying bytes using internal debug echo server (P4=P5)");
tokio::io::copy (&mut relay_recv, &mut relay_send).await?; tokio::io::copy (&mut relay_recv, &mut relay_send).await?;