🚧 the GUI can forward a port, but it can't seem to stop forwarding it

prototypes/quic_demo/src/bin/client_gui.rs

@@ -7,6 +7,19 @@ use fltk::{
 	prelude::*, 
 	window::Window
 };
+use structopt::StructOpt;
+use tokio::net::TcpListener;
+
+use quic_demo::prelude::*;
+use protocol::PeerId;
+
+#[derive (Debug, StructOpt)]
+struct Opt {
+	#[structopt (long)]
+	relay_addr: Option <String>,
+	#[structopt (long)]
+	client_id: Option <PeerId>,
+}
 
 #[derive (Clone, Copy)]
 enum Message {
@@ -18,6 +31,8 @@ fn main () -> anyhow::Result <()> {
 	tracing_subscriber::fmt::init ();
 	let rt = tokio::runtime::Runtime::new ()?;
 	
+	let opt = Opt::from_args ();
+	
 	let (fltk_tx, fltk_rx) = app::channel::<Message> ();
 	
 	let app = app::App::default();
@@ -29,6 +44,9 @@ fn main () -> anyhow::Result <()> {
 	let mut y = margin;
 	
 	let frame_status;
+	let mut input_server_id;
+	let mut but_open;
+	let mut but_close;
 	
 	{
 		frame_status = Frame::new (10, 10, 800 - 20, 30, "0 ports forwarded");
@@ -47,7 +65,7 @@ fn main () -> anyhow::Result <()> {
 		
 		let w = 80;
 		let frame_server_port = Frame::new (x, y, 80, h, "Server port");
-		x += w + margin;
+		// x += w + margin;
 	}
 	
 	y += h + margin;
@@ -58,7 +76,7 @@ fn main () -> anyhow::Result <()> {
 		x += w + margin;
 		
 		let w = 120;
-		let mut input_server_id = Input::new (100, y, w, h, "");
+		input_server_id = Input::new (100, y, w, h, "");
 		x += w + margin;
 		
 		let w = 80;
@@ -66,12 +84,12 @@ fn main () -> anyhow::Result <()> {
 		x += w + margin;
 		
 		let w = 80;
-		let mut but_open = Button::new (320, y, w, h, "Open");
+		but_open = Button::new (320, y, w, h, "Open");
 		x += w + margin;
 		
 		let w = 80;
-		let mut but_close = Button::new (410, y, w, h, "Close");
-		x += w + margin;
+		but_close = Button::new (410, y, w, h, "Close");
+		// x += w + margin;
 		
 		input_client_port.set_value ("5901");
 		input_server_id.set_value ("bogus_server");
@@ -86,20 +104,91 @@ fn main () -> anyhow::Result <()> {
 		set_active (&mut but_close, false);
 	}
 	
-	y += h + margin;
+	// y += h + margin;
 	
 	wind.end ();
 	wind.show ();
 	
+	let connection_p2_p3 = rt.block_on (async move {
+		let server_cert = tokio::fs::read ("quic_server.crt").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])?;
+		
+		trace! ("Connecting to relay server");
+		
+		let client_id = opt.client_id.unwrap_or_else (|| "bogus_client".to_string ());
+		
+		let quinn::NewConnection {
+			connection,
+			..
+		} = protocol::p2_connect_to_p3 (&endpoint, &relay_addr, &client_id).await?;
+		
+		Ok::<_, anyhow::Error> (connection)
+	})?;
 	
+	let mut forwarding_instance = None;
 	
 	while app.wait () {
 		match fltk_rx.recv () {
 			Some (Message::OpenPort) => {
+				let connection_p2_p3 = connection_p2_p3.clone ();
+				let server_id = input_server_id.value ().to_string ();
 				
+				let task = rt.spawn (async move {
+					let client_tcp_port = 30381;
+					let server_tcp_port = 30382;
+					let listener = TcpListener::bind (("127.0.0.1", client_tcp_port)).await?;
+					
+					trace! ("Accepting local TCP connections from P1 on {}", client_tcp_port);
+					
+					loop {
+						let (tcp_socket, _) = listener.accept ().await?;
+						let connection = connection_p2_p3.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> (())
+				});
+				
+				forwarding_instance.replace (Some (ForwardingInstance {
+					task,
+				}));
+				
+				set_active (&mut but_open, false);
+				set_active (&mut but_close, true);
+				but_open.set (true);
+				but_close.set (false);
 			},
 			Some (Message::ClosePort) => {
+				forwarding_instance.replace (None);
 				
+				set_active (&mut but_open, true);
+				set_active (&mut but_close, false);
+				but_open.set (false);
+				but_close.set (true);
 			},
 			None => (),
 		}
@@ -116,3 +205,7 @@ fn set_active <W: WidgetExt> (w: &mut W, b: bool) {
 		w.deactivate ();
 	}
 }
+
+struct ForwardingInstance {
+	task: tokio::task::JoinHandle <anyhow::Result <()>>,
+}