Cargo.lock

@@ -1328,7 +1328,6 @@ dependencies = [
  "anyhow",
  "base64",
  "futures-util",
- "hyper",
  "quinn",
  "rcgen",
  "rmp-serde",

docs/how-to/test-ptth-quic.md

@@ -1,24 +0,0 @@
-# How-to: Test PTTH_QUIC
-
-## Initial setup
-
-- Open 3 terminals in `prototypes/quic_demo`
-- Use `export RUST_LOG=quic_demo_relay_server=debug` to enable debug logging
-for the terminal that will run the relay server (P3)
-- Use `export RUST_LOG=quic_demo_end_server=debug` for the terminal that
-will run the end server (P4)
-- Use `export RUST_LOG=quic_demo_client=debug` for the terminal that
-will run the client (P2)
-
-When the relay server is running, use curl to get the list of connected
-end servers: `curl 127.0.0.1:4004`
-
-## Test loop - Happy path
-
-- Start a relay `cargo run --bin quic_demo_relay_server`
-- Verify that the relay has no end servers connected 
-- Start an end server `cargo run --bin quic_demo_end_server -- --debug-echo`
-- Verify that the end server connected
-- Start a client `cargo run --bin quic_demo_client`
-- Connect to the client and verify that the debug echo server is running
-`nc 127.0.0.1 30381`

prototypes/quic_demo/Cargo.toml

@@ -12,7 +12,6 @@ anyhow = "1.0.38"
 base64 = "0.13.0"
 # fltk = "1.1.1"
 futures-util = "0.3.9"
-hyper = { version = "0.14.4", features = ["http1", "server", "stream", "tcp"] }
 quinn = "0.7.2"
 rcgen = "0.8.11"
 rmp-serde = "0.15.5"

prototypes/quic_demo/src/bin/quic_demo_client.rs

@@ -46,45 +46,37 @@ async fn main () -> anyhow::Result <()> {
 	let server_tcp_port = opt.server_tcp_port.unwrap_or (30382);
 	let listener = TcpListener::bind (("127.0.0.1", client_tcp_port)).await?;
 	
+	debug! ("Accepting local TCP connections from P1");
+	
 	// End of per-port stuff
 	// Beginning of per-connection stuff
 	
-	let task_tcp_server = tokio::spawn (async move {
+	loop {
-		loop {
+		let (tcp_socket, _) = listener.accept ().await?;
-			let (tcp_socket, _) = listener.accept ().await?;
+		let connection = connection.clone ();
-			let connection = connection.clone ();
+		let server_id = server_id.clone ();
-			let server_id = server_id.clone ();
-			
-			tokio::spawn (async move {
-				let (local_recv, local_send) = tcp_socket.into_split ();
-				
-				debug! ("Starting PTTH connection");
-				
-				let (relay_send, relay_recv) = protocol::p2_connect_to_p5 (&connection, &server_id, server_tcp_port).await?;
-				
-				trace! ("Relaying bytes...");
-				
-				let ptth_conn = quic_demo::connection::NewConnection {
-					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> (())
+		tokio::spawn (async move {
-	});
+			let (local_recv, local_send) = tcp_socket.into_split ();
-	
+			
-	debug! ("Accepting local TCP connections from P1");
+			debug! ("Starting PTTH connection");
-	
+			
-	task_tcp_server.await??;
+			let (relay_send, relay_recv) = protocol::p2_connect_to_p5 (&connection, &server_id, server_tcp_port).await?;
-	
+			
-	Ok (())
+			trace! ("Relaying bytes...");
+			
+			let ptth_conn = quic_demo::connection::NewConnection {
+				local_send,
+				local_recv,
+				relay_send,
+				relay_recv,
+			}.build ();
+			
+			ptth_conn.wait_for_close ().await?;
+			
+			debug! ("Ended PTTH connection");
+			
+			Ok::<_, anyhow::Error> (())
+		});
+	}
 }

prototypes/quic_demo/src/bin/quic_demo_end_server.rs

@@ -10,41 +10,37 @@ struct Opt {
 	relay_addr: Option <String>,
 	#[structopt (long)]
 	server_id: Option <PeerId>,
-	#[structopt (long)]
-	debug_echo: bool,
 }
 
 #[tokio::main]
 async fn main () -> anyhow::Result <()> {
 	tracing_subscriber::fmt::init ();
 	
-	let opt = Arc::new (Opt::from_args ());
+	let opt = Opt::from_args ();
 	
 	let server_cert = tokio::fs::read ("quic_server.crt").await?;
-	let relay_addr = opt.relay_addr.clone ().unwrap_or_else (|| String::from ("127.0.0.1:30380")).parse ()?;
+	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])?;
 	
 	trace! ("Connecting to relay server");
 	
-	let server_id = opt.server_id.clone ().unwrap_or_else (|| "bogus_server".to_string ());
+	let server_id = opt.server_id.unwrap_or_else (|| "bogus_server".to_string ());
 	
 	let quinn::NewConnection {
 		mut bi_streams,
 		..
 	} = protocol::p4_connect_to_p3 (&endpoint, &relay_addr, &server_id).await?;
 	
-	debug! ("Connected to relay server");
 	trace! ("Accepting bi streams from P3");
 	
 	loop {
 		let (relay_send, relay_recv) = 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));
+		tokio::spawn (handle_bi_stream (relay_send, relay_recv));
 	}
 }
 
 async fn handle_bi_stream (
-	opt: Arc <Opt>,
 	relay_send: quinn::SendStream,
 	mut relay_recv: quinn::RecvStream,
 ) -> anyhow::Result <()>
@@ -53,14 +49,13 @@ async fn handle_bi_stream (
 		protocol::P3ToP4Stream::NewPtthConnection {
 			client_id,
 			..
-		} => handle_new_ptth_connection (opt, relay_send, relay_recv, client_id).await?,
+		} => handle_new_ptth_connection (relay_send, relay_recv, client_id).await?,
 	}
 	
 	Ok (())
 }
 
 async fn handle_new_ptth_connection (
-	opt: Arc <Opt>,
 	mut relay_send: quinn::SendStream,
 	mut relay_recv: quinn::RecvStream,
 	_client_id: String,
@@ -77,26 +72,19 @@ async fn handle_new_ptth_connection (
 	
 	debug! ("Started PTTH connection");
 	
-	if opt.debug_echo {
+	let stream = TcpStream::connect (("127.0.0.1", p4_to_p5_req.port)).await?;
-		relay_send.write (b"Connected to P4=P5 debug echo server\n").await?;
+	let (local_recv, local_send) = stream.into_split ();
-		debug! ("Relaying bytes using internal debug echo server (P4=P5)");
+	
-		tokio::io::copy (&mut relay_recv, &mut relay_send).await?;
+	trace! ("Relaying bytes...");
-	}
+	
-	else {
+	let ptth_conn = quic_demo::connection::NewConnection {
-		let stream = TcpStream::connect (("127.0.0.1", p4_to_p5_req.port)).await?;
+		local_send,
-		let (local_recv, local_send) = stream.into_split ();
+		local_recv,
-		
+		relay_send,
-		trace! ("Relaying bytes...");
+		relay_recv,
-		
+	}.build ();
-		let ptth_conn = quic_demo::connection::NewConnection {
+	
-			local_send,
+	ptth_conn.wait_for_close ().await?;
-			local_recv,
-			relay_send,
-			relay_recv,
-		}.build ();
-		
-		ptth_conn.wait_for_close ().await?;
-	}
 	
 	Ok (())
 }

prototypes/quic_demo/src/bin/quic_demo_relay_server.rs

@@ -1,14 +1,3 @@
-use hyper::{
-	Body,
-	Request,
-	Response,
-	Server,
-	service::{
-		make_service_fn,
-		service_fn,
-	},
-	StatusCode,
-};
 use structopt::StructOpt;
 
 use quic_demo::prelude::*;
@@ -31,120 +20,28 @@ async fn main () -> anyhow::Result <()> {
 	println! ("Base64 cert: {}", base64::encode (&server_cert));
 	tokio::fs::write ("quic_server.crt", &server_cert).await?;
 	
-	let relay_state = Arc::new (RelayState::default ());
+	let relay_state = RelayState::default ();
+	let relay_state = Arc::new (relay_state);
 	
-	let make_svc = {
+	while let Some (conn) = incoming.next ().await {
 		let relay_state = Arc::clone (&relay_state);
-		make_service_fn (move |_conn| {
-			let relay_state = Arc::clone (&relay_state);
-			
-			async move {
-				Ok::<_, String> (service_fn (move |req| {
-					let relay_state = Arc::clone (&relay_state);
-					
-					handle_http (req, relay_state)
-				}))
-			}
-		})
-	};
-	
-	let http_addr = SocketAddr::from (([0, 0, 0, 0], 4004));
-	let http_server = Server::bind (&http_addr);
-	
-	let tcp_port = 30382;
-	let tcp_listener = TcpListener::bind (("127.0.0.1", tcp_port)).await?;
-	
-	let task_quic_server = {
-		let relay_state = Arc::clone (&relay_state);
-		tokio::spawn (async move {
-			while let Some (conn) = incoming.next ().await {
-				let relay_state = Arc::clone (&relay_state);
-				
-				// Each new peer QUIC connection gets its own task
-				tokio::spawn (async move {
-					let active = relay_state.stats.quic.connect ();
-					debug! ("QUIC connections: {}", active);
-					
-					match handle_quic_connection (Arc::clone (&relay_state), conn).await {
-						Ok (_) => (),
-						Err (e) => warn! ("handle_quic_connection {:?}", e),
-					}
-					
-					let active = relay_state.stats.quic.disconnect ();
-					debug! ("QUIC connections: {}", active);
-				});
-			}
-			
-			Ok::<_, anyhow::Error> (())
-		})
-	};
-	
-	let task_http_server = tokio::spawn (async move {
-		http_server.serve (make_svc).await?;
-		Ok::<_, anyhow::Error> (())
-	});
-	
-	let task_tcp_server = {
-		let relay_state = Arc::clone (&relay_state);
-		tokio::spawn (async move {
-			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 ();
-					
-					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") {
-						Some (x) => x,
-						None => bail! ("That server isn't connected"),
-					};
-					
-					unimplemented! ();
-					/*
-					p4.req_channel.send (RequestP2ToP4 {
-						client_send,
-						client_recv,
-						client_id: "bogus_client".to_string (),
-					}).await.map_err (|_| anyhow::anyhow! ("Can't send request to P4 server"))?;
-					*/
-					Ok (())
-				});
-			}
-			
-			Ok::<_, anyhow::Error> (())
-		})
-	};
-	
-	debug! ("Serving HTTP on {:?}", http_addr);
-	
-	task_quic_server.await??;
-	task_http_server.await??;
-	task_tcp_server.await??;
-	
-	Ok (())
-}
-
-async fn handle_http (_req: Request <Body>, relay_state: Arc <RelayState>)
--> anyhow::Result <Response <Body>>
-{
-	let debug_string;
-	{
-		let p4_server_proxies = relay_state.p4_server_proxies.lock ().await;
 		
-		debug_string = format! ("{:#?}\n", p4_server_proxies.keys ().collect::<Vec<_>> ());
+		// Each new peer QUIC connection gets its own task
+		tokio::spawn (async move {
+			let active = relay_state.stats.quic.connect ();
+			debug! ("QUIC connections: {}", active);
+			
+			match handle_quic_connection (Arc::clone (&relay_state), conn).await {
+				Ok (_) => (),
+				Err (e) => warn! ("handle_quic_connection {:?}", e),
+			}
+			
+			let active = relay_state.stats.quic.disconnect ();
+			debug! ("QUIC connections: {}", active);
+		});
 	}
 	
-	let resp = Response::builder ()
+	Ok (())
-	.status (StatusCode::OK)
-	.header ("content-type", "text/plain")
-	.body (Body::from (debug_string))?;
-	
-	Ok (resp)
 }
 
 #[derive (Default)]
@@ -199,8 +96,8 @@ struct RequestP2ToP4 {
 	client_id: String,
 }
 
-struct PtthNewConnection <ClientSend: AsyncWrite + Unpin> {
+struct PtthNewConnection {
-	client_send: ClientSend,
+	client_send: quinn::SendStream,
 	client_recv: quinn::RecvStream,
 	server_send: quinn::SendStream,
 	server_recv: quinn::RecvStream,
@@ -211,7 +108,7 @@ struct PtthConnection {
 	downlink_task: JoinHandle <anyhow::Result <()>>,
 }
 
-impl <ClientSend: AsyncWrite + Unpin> PtthNewConnection <ClientSend> {
+impl PtthNewConnection {
 	fn build (self) -> PtthConnection {
 		let Self {
 			mut client_send,

prototypes/quic_demo/src/prelude.rs

@@ -1,6 +1,5 @@
 pub use std::{
 	collections::*,
-	net::SocketAddr,
 	sync::{
 		Arc,
 		atomic::{
@@ -18,12 +17,9 @@ pub use anyhow::{
 pub use futures_util::StreamExt;
 pub use tokio::{
 	io::{
-		AsyncRead,
 		AsyncReadExt,
-		AsyncWrite,
 		AsyncWriteExt,
 	},
-	net::TcpListener,
 	sync::{
 		Mutex,
 		mpsc,