Cargo.lock

@@ -93,9 +93,9 @@ checksum = "904dfeac50f3cdaba28fc6f57fdcddb75f49ed61346676a78c4ffe55877802fd"
 
 [[package]]
 name = "bitflags"
-version = "1.3.2"
+version = "1.2.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "bef38d45163c2f1dde094a7dfd33ccf595c92905c8f8f4fdc18d06fb1037718a"
+checksum = "cf1de2fe8c75bc145a2f577add951f8134889b4795d47466a54a5c846d691693"
 
 [[package]]
 name = "blake3"
@@ -257,9 +257,9 @@ dependencies = [
 
 [[package]]
 name = "ctrlc"
-version = "3.2.1"
+version = "3.2.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "a19c6cedffdc8c03a3346d723eb20bd85a13362bb96dc2ac000842c6381ec7bf"
+checksum = "377c9b002a72a0b2c1a18c62e2f3864bdfea4a015e3683a96e24aa45dd6c02d1"
 dependencies = [
  "nix",
  "winapi",
@@ -875,9 +875,9 @@ dependencies = [
 
 [[package]]
 name = "nix"
-version = "0.23.0"
+version = "0.22.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f305c2c2e4c39a82f7bf0bf65fb557f9070ce06781d4f2454295cc34b1c43188"
+checksum = "d3bb9a13fa32bc5aeb64150cd3f32d6cf4c748f8f8a417cce5d2eb976a8370ba"
 dependencies = [
  "bitflags",
  "cc",
@@ -1228,7 +1228,6 @@ name = "ptth_multi_call_server"
 version = "0.1.0"
 dependencies = [
  "anyhow",
- "ctrlc",
  "ptth_file_server",
  "ptth_server",
  "quic_demo",
@@ -1365,7 +1364,6 @@ version = "0.1.0"
 dependencies = [
  "anyhow",
  "base64",
- "ctrlc",
  "futures-util",
  "hyper",
  "quinn",

crates/ptth_multi_call_server/Cargo.toml

@@ -7,7 +7,6 @@ license = "AGPL-3.0"
 
 [dependencies]
 anyhow = "1.0.38"
-ctrlc = "3.2.1"
 ptth_file_server = { path = "../ptth_file_server_bin" }
 ptth_server = { path = "../ptth_server" }
 quic_demo = { path = "../../prototypes/quic_demo" }

crates/ptth_multi_call_server/src/main.rs

@@ -3,8 +3,6 @@ use std::{
 	iter::FromIterator,
 };
 
-use tokio::sync::watch;
-
 #[derive (Clone, Copy, Debug, PartialEq)]
 enum Subcommand {
 	PtthServer,
@@ -12,30 +10,6 @@ enum Subcommand {
 	PtthQuicEndServer,
 }
 
-#[tokio::main]
-async fn main () -> anyhow::Result <()> {
-	tracing_subscriber::fmt::init ();
-	
-	let args = Vec::from_iter (std::env::args_os ());
-	
-	let (subcommand, args) = parse_args (&args)?;
-	match subcommand {
-		Subcommand::PtthServer => ptth_server::executable::main (&args).await,
-		Subcommand::PtthFileServer => ptth_file_server::main (&args).await,
-		Subcommand::PtthQuicEndServer => {
-			let (shutdown_tx, shutdown_rx) = watch::channel (false);
-			
-			ctrlc::set_handler (move || {
-				shutdown_tx.send (true).expect ("Couldn't forward Ctrl+C signal");
-			})?;
-			tracing::trace! ("Set Ctrl+C handler");
-			quic_demo::executable_end_server::main (&args, Some (shutdown_rx)).await?;
-			
-			Ok (())
-		}
-	}
-}
-
 fn parse_subcommand (arg: &str) -> Option <Subcommand>
 {
 	use Subcommand::*;
@@ -84,6 +58,20 @@ fn parse_args (args: &[OsString]) -> anyhow::Result <(Subcommand, &[OsString])>
 	anyhow::bail! ("Subcommand must be either arg 0 (exe name) or arg 1")
 }
 
+#[tokio::main]
+async fn main () -> anyhow::Result <()> {
+	tracing_subscriber::fmt::init ();
+	
+	let args = Vec::from_iter (std::env::args_os ());
+	
+	let (subcommand, args) = parse_args (&args)?;
+	match subcommand {
+		Subcommand::PtthServer => ptth_server::executable::main (&args).await,
+		Subcommand::PtthFileServer => ptth_file_server::main (&args).await,
+		Subcommand::PtthQuicEndServer => quic_demo::executable_end_server::main (&args).await,
+	}
+}
+
 #[cfg (test)]
 mod tests {
 	use super::*;

prototypes/ptth_quic_client_gui/src/main.rs

@@ -10,7 +10,6 @@ use fltk::{
 	window::Window
 };
 use structopt::StructOpt;
-use tokio::runtime::Runtime;
 
 use quic_demo::{
 	client_proxy::*,
@@ -36,69 +35,9 @@ enum Message {
 	ClosePort (usize),
 }
 
-struct GuiClient <'a> {
-	rt: &'a Runtime,
-	frame_status: Frame,
-	forwarding_instances: Vec <Option <ForwardingInstance>>,
-	gui_ports: Vec <GuiPort>,
-}
-
-impl GuiClient <'_> {
-	pub fn open_port (
-		&mut self, 
-		connection_p2_p3: quinn::Connection, 
-		port_idx: usize,
-	) -> anyhow::Result <()>
-	{
-		let params = self.gui_ports [port_idx].get_params ()?;
-		
-		let _guard = self.rt.enter ();
-		let forwarding_instance = self.rt.block_on (ForwardingInstance::new (
-			connection_p2_p3,
-			params,
-		))?;
-		
-		self.forwarding_instances [port_idx].replace (forwarding_instance);
-		
-		self.gui_ports [port_idx].set_forwarding (true);
-		self.sync_status ();
-		
-		Ok (())
-	}
-	
-	pub fn close_port (&mut self, port_idx: usize) -> anyhow::Result <()> {
-		if let Some (old_instance) = self.forwarding_instances [port_idx].take () {
-			self.rt.block_on (async {
-				old_instance.close ()
-				.await
-				.context ("closing ForwardingInstance")?;
-				
-				Ok::<_, anyhow::Error> (())
-			})?;
-		}
-		
-		self.gui_ports [port_idx].set_forwarding (false);
-		self.sync_status ();
-		
-		Ok (())
-	}
-	
-	fn open_ports (&self) -> usize {
-		self.forwarding_instances.iter ()
-		.map (|x| if x.is_some () { 1 } else { 0 })
-		.sum ()
-	}
-	
-	fn sync_status (&mut self) {
-		let open_ports = self.open_ports ();
-		
-		self.frame_status.set_label (&format! ("Forwarding {} ports", open_ports));
-	}
-}
-
 fn main () -> anyhow::Result <()> {
 	tracing_subscriber::fmt::init ();
-	let rt = Runtime::new ()?;
+	let rt = tokio::runtime::Runtime::new ()?;
 	
 	let opt = Opt::from_args ();
 	
@@ -114,7 +53,7 @@ fn main () -> anyhow::Result <()> {
 	let mut x = margin;
 	let mut y = margin;
 	
-	let frame_status = Frame::new (x, y, 800 - 20, h, "Forwarding 0 ports");
+	let mut frame_status = Frame::new (x, y, 800 - 20, h, "Forwarding 0 ports");
 	
 	y += h + margin;
 	x = margin;
@@ -148,25 +87,18 @@ fn main () -> anyhow::Result <()> {
 	// y += h + margin;
 	// x = margin;
 	
-	let gui_ports = vec! [
+	let mut gui_ports = vec! [
 		gui_port_0,
 		gui_port_1,
 		gui_port_2,
 	];
 	
-	let forwarding_instances = vec! [
+	let mut forwarding_instances = vec! [
 		None,
 		None,
 		None,
 	];
 	
-	let mut gui_client = GuiClient {
-		rt: &rt,
-		frame_status,
-		forwarding_instances,
-		gui_ports,
-	};
-	
 	// y += h + margin;
 	
 	wind.end ();
@@ -175,7 +107,7 @@ fn main () -> anyhow::Result <()> {
 	let connection_p2_p3 = rt.block_on (async move {
 		let server_cert = match opt.cert_url.as_ref () {
 			Some (url) => reqwest::get (url).await?.bytes ().await?,
-			None => tokio::fs::read ("ptth_quic_output/quic_server.crt").await?.into (),
+			None => tokio::fs::read ("quic_server.crt").await?.into (),
 		};
 		
 		let relay_addr = opt.relay_addr
@@ -200,13 +132,27 @@ fn main () -> anyhow::Result <()> {
 	while app.wait () {
 		match fltk_rx.recv () {
 			Some (Message::OpenPort (port_idx)) => {
-				match gui_client.open_port (connection_p2_p3.clone (), port_idx) {
+				if let Ok (params) = gui_ports [port_idx].get_params () {
-					Err (e) => error! ("{:?}", e),
+					let connection_p2_p3 = connection_p2_p3.clone ();
-					_ => (),
+					
-				};
+					let _guard = rt.enter ();
+					forwarding_instances [port_idx].replace (ForwardingInstance::new (
+						connection_p2_p3,
+						params,
+					));
+					
+					gui_ports [port_idx].set_forwarding (true);
+					
+					frame_status.set_label ("Forwarding 1 port");
+				}
 			},
 			Some (Message::ClosePort (port_idx)) => {
-				gui_client.close_port (port_idx)?;
+				if let Some (old_instance) = forwarding_instances [port_idx].take () {
+					rt.block_on (old_instance.close ())?;
+				}
+				
+				gui_ports [port_idx].set_forwarding (false);
+				frame_status.set_label ("Forwarding 0 ports");
 			},
 			None => (),
 		}
@@ -261,9 +207,9 @@ impl GuiPort {
 		input_server_id.set_value ("bogus_server");
 		input_server_port.set_value ("5900");
 		
-		but_open.set_trigger (CallbackTrigger::Release);
+		but_open.set_trigger (CallbackTrigger::Changed);
 		but_open.emit (fltk_tx, Message::OpenPort (port_idx));
-		but_close.set_trigger (CallbackTrigger::Release);
+		but_close.set_trigger (CallbackTrigger::Changed);
 		but_close.emit (fltk_tx, Message::ClosePort (port_idx));
 		
 		set_active (&mut but_open, true);

prototypes/quic_demo/Cargo.toml

@@ -10,7 +10,6 @@ license = "AGPL-3.0"
 [dependencies]
 anyhow = "1.0.38"
 base64 = "0.13.0"
-ctrlc = "3.2.1"
 # fltk = "1.1.1"
 futures-util = "0.3.9"
 hyper = { version = "0.14.4", features = ["http1", "server", "stream", "tcp"] }

prototypes/quic_demo/src/bin/quic_demo_client.rs

@@ -1,15 +1,7 @@
 use structopt::StructOpt;
-use tokio::{
+use tokio::net::TcpListener;
-	sync::watch,
-};
 
-use quic_demo::{
+use quic_demo::prelude::*;
-	client_proxy::{
-		ForwardingParams,
-		forward_port,
-	},
-	prelude::*,
-};
 use protocol::PeerId;
 
 #[derive (Debug, StructOpt)]
@@ -31,109 +23,68 @@ async fn main () -> anyhow::Result <()> {
 	tracing_subscriber::fmt::init ();
 	
 	let opt = Opt::from_args ();
-	let conf = opt.into_config ().await?;
 	
-	let client = P2Client::connect (conf)?;
+	let server_cert = tokio::fs::read ("quic_server.crt").await?;
-	client.run ().await?;
+	let relay_addr = opt.relay_addr.unwrap_or_else (|| String::from ("127.0.0.1:30380")).parse ()?;
+	let endpoint = make_client_endpoint ("0.0.0.0:0".parse ()?, &[&server_cert])?;
 	
-	Ok (())
+	debug! ("Connecting to relay server");
-}
 	
-pub struct P2Client {
+	let client_id = opt.client_id.unwrap_or_else (|| "bogus_client".to_string ());
-	endpoint: quinn::Endpoint,
-	conf: Arc <Config>,
-}
-
-impl P2Client {
-	pub fn connect (conf: Config) -> anyhow::Result <Self> {
-		let endpoint = make_client_endpoint ("0.0.0.0:0".parse ()?, &[&conf.relay_cert])?;
-		let conf = Arc::new (conf);
-		
-		Ok (Self {
-			endpoint,
-			conf,
-		})
-	}
-	
-	pub async fn run (&self) -> anyhow::Result <()> {
-		debug! ("P2 client connecting to P3 relay server");
-		
-		let conf = Arc::clone (&self.conf);
 	
 	let quinn::NewConnection {
 		connection,
 		..
-		} = protocol::p2_connect_to_p3 (&self.endpoint, &conf.relay_addr, &conf.client_id).await?;
+	} = protocol::p2_connect_to_p3 (&endpoint, &relay_addr, &client_id).await?;
 	
-		let client_tcp_port = conf.client_tcp_port;
+	// End of per-client stuff
+	// Beginning of per-port stuff
 	
-		debug! ("Accepting local TCP connections from P1 at {}", client_tcp_port);
+	let server_id = opt.server_id.unwrap_or_else (|| "bogus_server".to_string ());
+	
+	let client_tcp_port = opt.client_tcp_port.unwrap_or (30381);
+	let server_tcp_port = opt.server_tcp_port.unwrap_or (30382);
+	let listener = TcpListener::bind (("127.0.0.1", client_tcp_port)).await?;
 	
 	// End of per-port stuff
 	// Beginning of per-connection stuff
 	
-		let (_shutdown_flag_tx, shutdown_flag_rx) = watch::channel (true);
+	let task_tcp_server = tokio::spawn (async move {
-		
+		loop {
-		let task_tcp_server = {
+			let (tcp_socket, _) = listener.accept ().await?;
 			let connection = connection.clone ();
-			let server_id = conf.server_id.clone ();
+			let server_id = server_id.clone ();
-			let server_tcp_port = conf.server_tcp_port;
-			
-			let listener = TcpListener::bind (("127.0.0.1", client_tcp_port)).await?;
-			trace! ("Accepting local TCP connections from P1 on {}", client_tcp_port);
 			
 			tokio::spawn (async move {
-				forward_port (
+				let (local_recv, local_send) = tcp_socket.into_split ();
-					listener,
+				
-					connection,
+				debug! ("Starting PTTH connection");
-					ForwardingParams {
+				
-						client_tcp_port,
+				let (relay_send, relay_recv) = protocol::p2_connect_to_p5 (&connection, &server_id, server_tcp_port).await?;
-						server_id,
+				
-						server_tcp_port,
+				trace! ("Relaying bytes...");
-					},
+				
-					shutdown_flag_rx,
+				let ptth_conn = quic_demo::connection::NewConnection {
-				).await?;
+					local_send,
+					local_recv,
+					relay_send,
+					relay_recv,
+				}.build ();
+				
+				ptth_conn.wait_for_close ().await?;
+				
+				debug! ("Ended PTTH connection");
 				
 				Ok::<_, anyhow::Error> (())
-			})
+			});
-		};
+		}
+		
+		Ok::<_, anyhow::Error> (())
+	});
+	
+	debug! ("Accepting local TCP connections from P1");
 	
 	task_tcp_server.await??;
 	
 	Ok (())
-	}
-}
-
-/// A filled-out config for constructing a P2 client
-#[derive (Clone)]
-pub struct Config {
-	client_tcp_port: u16,
-	server_tcp_port: u16,
-	client_id: String,
-	server_id: String,
-	relay_addr: SocketAddr,
-	relay_cert: Vec <u8>,
-}
-
-impl Opt {
-	pub async fn into_config (self) -> anyhow::Result <Config> {
-		let client_tcp_port = self.client_tcp_port.unwrap_or (30381);
-		let server_tcp_port = self.server_tcp_port.unwrap_or (30382);
-		let client_id = self.client_id.unwrap_or_else (|| "bogus_client".to_string ());
-		let server_id = self.server_id.unwrap_or_else (|| "bogus_server".to_string ());
-		let relay_addr = self.relay_addr.unwrap_or_else (|| String::from ("127.0.0.1:30380")).parse ()?;
-		
-		// Begin I/O
-		
-		let relay_cert = tokio::fs::read ("ptth_quic_output/quic_server.crt").await?;
-		
-		Ok (Config {
-			client_tcp_port,
-			server_tcp_port,
-			client_id, 
-			server_id,
-			relay_addr,
-			relay_cert,
-		})
-	}
 }

prototypes/quic_demo/src/bin/quic_demo_end_server.rs

@@ -2,22 +2,11 @@ use std::{
 	iter::FromIterator,
 };
 
-use tokio::sync::watch;
-
-use quic_demo::prelude::*;
-
 #[tokio::main]
 async fn main () -> anyhow::Result <()> {
 	tracing_subscriber::fmt::init ();
 	
 	let args = Vec::from_iter (std::env::args_os ());
 	
-	let (shutdown_tx, shutdown_rx) = watch::channel (false);
+	quic_demo::executable_end_server::main (&args).await
-	
-	ctrlc::set_handler (move || {
-		shutdown_tx.send (true).expect ("Couldn't forward Ctrl+C signal");
-	})?;
-	trace! ("Set Ctrl+C handler");
-	
-	quic_demo::executable_end_server::main (&args, Some (shutdown_rx)).await
 }

prototypes/quic_demo/src/bin/quic_demo_relay_server.rs

@@ -10,7 +10,6 @@ use hyper::{
 	StatusCode,
 };
 use structopt::StructOpt;
-use tokio::sync::watch;
 
 use quic_demo::prelude::*;
 use protocol::PeerId;
@@ -57,8 +56,6 @@ async fn main () -> anyhow::Result <()> {
 	let tcp_port = 30382;
 	let tcp_listener = TcpListener::bind (("127.0.0.1", tcp_port)).await?;
 	
-	let (_running_tx, running_rx) = watch::channel (true);
-	
 	let task_quic_server = {
 		let relay_state = Arc::clone (&relay_state);
 		tokio::spawn (async move {
@@ -92,22 +89,24 @@ async fn main () -> anyhow::Result <()> {
 	let task_tcp_server = {
 		let relay_state = Arc::clone (&relay_state);
 		tokio::spawn (async move {
-			while *running_rx.borrow () {
+			loop {
 				let (tcp_socket, _) = tcp_listener.accept ().await?;
 				
+				let server_id = "bogus_server".to_string ();
+				
 				let relay_state = Arc::clone (&relay_state);
 				tokio::spawn (async move {
-					let (_client_recv, _client_send) = tcp_socket.into_split ();
+					let (client_recv, client_send) = tcp_socket.into_split ();
 					
 					debug! ("Accepted direct TCP connection P1 --> P3");
 					
 					let p4_server_proxies = relay_state.p4_server_proxies.lock ().await;
-					let _p4 = match p4_server_proxies.get ("bogus_server") {
+					let p4 = match p4_server_proxies.get ("bogus_server") {
 						Some (x) => x,
 						None => bail! ("That server isn't connected"),
 					};
 					
-					// unimplemented! ();
+					unimplemented! ();
 					/*
 					p4.req_channel.send (RequestP2ToP4 {
 						client_send,

prototypes/quic_demo/src/client_proxy.rs

@@ -12,37 +12,28 @@ pub struct ForwardingInstance {
 }
 
 impl ForwardingInstance {
-	pub async fn new (
+	pub fn new (
 		connection_p2_p3: quinn::Connection,
 		params: ForwardingParams,
-	) -> anyhow::Result <Self>
+	) -> Self 
 	{
 		let (shutdown_flag, shutdown_flag_rx) = tokio::sync::watch::channel (true);
 		
-		let listener = TcpListener::bind (("127.0.0.1", params.client_tcp_port)).await?;
-		trace! ("Accepting local TCP connections from P1 on {}", params.client_tcp_port);
-		
 		let task = tokio::spawn (forward_port (
-			listener,
 			connection_p2_p3,
 			params,
 			shutdown_flag_rx
 		));
 		
-		Ok (Self {
+		Self {
 			task,
 			shutdown_flag,
-		})
+		}
 	}
 	
 	pub async fn close (self) -> anyhow::Result <()> {
-		match self.shutdown_flag.send (false) {
+		self.shutdown_flag.send (false)?;
-			Err (_) => warn! ("Trying to gracefully shutdown forwarding task but it appears to already be shut down"),
+		self.task.await??;
-			_ => (),
-		}
-		self.task.await
-		.context ("awaiting ForwardingInstance task")?
-		.context ("inside ForwardingInstance task")?;
 		Ok (())
 	}
 }
@@ -53,14 +44,10 @@ pub struct ForwardingParams {
 	pub server_tcp_port: u16,
 }
 
-/// Starts a TCP listener that can forward any number of TCP streams to
+async fn forward_port (
-/// the same client:server port combination
-
-pub async fn forward_port (
-	listener: TcpListener,
 	connection_p2_p3: quinn::Connection,
 	params: ForwardingParams,
-	mut shutdown_flag_rx: tokio::sync::watch::Receiver <bool>,
+	shutdown_flag_rx: tokio::sync::watch::Receiver <bool>,
 ) -> anyhow::Result <()>
 {
 	let ForwardingParams {
@@ -69,7 +56,13 @@ pub async fn forward_port (
 		server_tcp_port,
 	} = params;
 	
+	let listener = TcpListener::bind (("127.0.0.1", client_tcp_port)).await?;
+	
+	trace! ("Accepting local TCP connections from P1 on {}", client_tcp_port);
+	
 	while *shutdown_flag_rx.borrow () {
+		let mut shutdown_flag_rx_2 = shutdown_flag_rx.clone ();
+		
 		tokio::select! {
 			x = listener.accept () => {
 				let (tcp_socket, _) = x?;
@@ -79,7 +72,7 @@ pub async fn forward_port (
 				
 				tokio::spawn (handle_p1 (connection, server_id, server_tcp_port, tcp_socket, shutdown_flag_rx));
 			},
-			_ = shutdown_flag_rx.changed () => (),
+			_ = shutdown_flag_rx_2.changed () => (),
 		};
 	}
 

prototypes/quic_demo/src/executable_end_server.rs

@@ -1,14 +1,9 @@
 use structopt::StructOpt;
-use tokio::{
+use tokio::net::TcpStream;
-	net::TcpStream,
-	sync::watch,
-};
 
 use crate::prelude::*;
 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)]
 struct Opt {
 	#[structopt (long)]
@@ -21,145 +16,37 @@ struct Opt {
 	cert_url: Option <String>,
 }
 
-pub async fn main (args: &[OsString], shutdown_rx: Option <watch::Receiver <bool>>) -> anyhow::Result <()> {
+pub async fn main (args: &[OsString]) -> anyhow::Result <()> {
-	let opt = Opt::from_iter (args);
+	let opt = Arc::new (Opt::from_iter (args));
-	let conf = opt.into_config ().await?;
 	
-	let end_server = Arc::new (P4EndServer::connect (conf)?);
+	let server_cert = match opt.cert_url.as_ref () {
-	
+		Some (url) => reqwest::get (url).await?.bytes ().await?,
-	let run_task = {
+		None => tokio::fs::read ("quic_server.crt").await?.into (),
-		let end_server = Arc::clone (&end_server);
-		tokio::spawn (async move {
-			end_server.run ().await?;
-			Ok::<_, anyhow::Error> (())
-		})
 	};
+	let relay_addr = opt.relay_addr.clone ().unwrap_or_else (|| String::from ("127.0.0.1:30380")).parse ()?;
+	let endpoint = make_client_endpoint ("0.0.0.0:0".parse ()?, &[&server_cert])?;
 	
-	if let Some (mut shutdown_rx) = shutdown_rx {
+	trace! ("Connecting to relay server");
-		while ! *shutdown_rx.borrow () {
-			shutdown_rx.changed ().await?;
-		}
-		end_server.shut_down ()?;
-	}
 	
-	run_task.await??;
+	let server_id = opt.server_id.clone ().unwrap_or_else (|| "bogus_server".to_string ());
 	
-	trace! ("P4 end server shut down gracefully.");
-	
-	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 (),
-			None => tokio::fs::read ("ptth_quic_output/quic_server.crt").await?,
-		};
-		
-		Ok (Config {
-			debug_echo: self.debug_echo,
-			id,
-			relay_addr,
-			relay_cert,
-		})
-	}
-}
-
-pub struct P4EndServer {
-	endpoint: quinn::Endpoint,
-	conf: Arc <Config>,
-	shutdown_tx: watch::Sender <bool>,
-	shutdown_rx: watch::Receiver <bool>,
-}
-
-impl P4EndServer {
-	pub 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])?;
-		
-		let (shutdown_tx, shutdown_rx) = watch::channel (false);
-		
-		Ok (P4EndServer {
-			conf: Arc::new (conf),
-			endpoint,
-			shutdown_tx,
-			shutdown_rx,
-		})
-	}
-	
-	pub fn config (&self) -> &Config {
-		&*self.conf
-	}
-	
-	pub async fn run (&self) -> anyhow::Result <()> {
-		trace! ("P4 end server connecting to P3 relay server");
 	let quinn::NewConnection {
 		mut bi_streams,
 		..
-		} = protocol::p4_connect_to_p3 (
+	} = protocol::p4_connect_to_p3 (&endpoint, &relay_addr, &server_id).await?;
-			&self.endpoint, 
-			&self.conf.relay_addr, 
-			&self.conf.id
-		).await?;
 	
 	debug! ("Connected to relay server");
-		
 	trace! ("Accepting bi streams from P3");
 	
-		let mut shutdown_rx = self.shutdown_rx.clone ();
-		
 	loop {
-			tokio::select! {
+		let (relay_send, relay_recv) = bi_streams.next ().await.ok_or_else (|| anyhow::anyhow! ("Relay server didn't open a bi stream"))??;
-				_ = shutdown_rx.changed () => {
-					if *shutdown_rx.borrow () {
-						trace! ("P4 incoming bi streams task caught graceful shutdown");
-						break;
-					}
-				}
-				stream_opt = bi_streams.next () => {
-					let (relay_send, relay_recv) = stream_opt.ok_or_else (|| anyhow::anyhow! ("P4 ran out of incoming streams. Maybe P3 shut down or disconnected?"))??;
 		
-					tokio::spawn (handle_bi_stream (Arc::clone (&self.conf), relay_send, relay_recv));
+		tokio::spawn (handle_bi_stream (Arc::clone (&opt), relay_send, relay_recv));
-				}
-			};
-		}
-		
-		Ok (())
-	}
-	
-	pub fn shut_down (&self) -> anyhow::Result <()> {
-		trace! ("P4 end server shutting down...");
-		Ok (self.shutdown_tx.send (true)?)
-	}
-	
-	pub fn shutting_down (&self) -> bool {
-		*self.shutdown_rx.borrow ()
 	}
 }
 
 async fn handle_bi_stream (
-	conf: Arc <Config>,
+	opt: Arc <Opt>,
 	relay_send: quinn::SendStream,
 	mut relay_recv: quinn::RecvStream,
 ) -> anyhow::Result <()>
@@ -168,14 +55,14 @@ async fn handle_bi_stream (
 		protocol::P3ToP4Stream::NewPtthConnection {
 			client_id,
 			..
-		} => handle_new_ptth_connection (conf, relay_send, relay_recv, client_id).await?,
+		} => handle_new_ptth_connection (opt, relay_send, relay_recv, client_id).await?,
 	}
 	
 	Ok (())
 }
 
 async fn handle_new_ptth_connection (
-	conf: Arc <Config>,
+	opt: Arc <Opt>,
 	mut relay_send: quinn::SendStream,
 	mut relay_recv: quinn::RecvStream,
 	_client_id: String,
@@ -192,7 +79,7 @@ async fn handle_new_ptth_connection (
 	
 	debug! ("Started PTTH connection");
 	
-	if conf.debug_echo {
+	if opt.debug_echo {
 		relay_send.write (b"Connected to P4=P5 debug echo server\n").await?;
 		debug! ("Relaying bytes using internal debug echo server (P4=P5)");
 		tokio::io::copy (&mut relay_recv, &mut relay_send).await?;

prototypes/quic_demo/src/protocol.rs

@@ -61,7 +61,7 @@ pub async fn p2_connect_to_p5 (
 	let cmd_type = Command::CONNECT_P2_TO_P4.0;
 	
 	send.write_all (&[cmd_type, 0, 0, 0]).await?;
-	send_lv_string (&mut send, server_id).await?;
+	send_lv_string (&mut send, &server_id).await?;
 	
 	expect_exact_response (&mut recv, [Command::OKAY.0, cmd_type, 0, 0]).await
 	.context ("P2 didn't get OK response when asking P3 to connect P2 to P4")?;

prototypes/quic_demo/src/quinn_utils.rs

@@ -58,7 +58,7 @@ pub fn make_server_endpoint(bind_addr: SocketAddr) -> anyhow::Result<(Incoming,
 fn configure_client(server_certs: &[&[u8]]) -> anyhow::Result<ClientConfig> {
 	let mut cfg_builder = ClientConfigBuilder::default();
 	for cert in server_certs {
-		cfg_builder.add_certificate_authority(Certificate::from_der(cert)?)?;
+		cfg_builder.add_certificate_authority(Certificate::from_der(&cert)?)?;
 	}
 	Ok(cfg_builder.build())
 }