reactor
/
ptth
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

♻️ refactor: extract TCP-QUIC relay connection

main
_ 2021-07-18 18:58:59 +00:00
parent 4ba35ee7d1
commit 69c4afe456
4 changed files with 80 additions and 131 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

67
prototypes/quic_demo/src/bin/quic_demo_client.rs
View File

@ -54,15 +54,14 @@ async fn main () -> anyhow::Result <()> {
trace! ("Relaying bytes...");
let ptth_conn = PtthNewConnection {
let ptth_conn = quic_demo::connection::NewConnection {
local_send,
local_recv,
relay_send,
relay_recv,
}.build ();
ptth_conn.uplink_task.await??;
ptth_conn.downlink_task.await??;
ptth_conn.wait_for_close ().await?;
debug! ("Ended PTTH connection");
@ -70,65 +69,3 @@ async fn main () -> anyhow::Result <()> {
});
}
}
struct PtthNewConnection {
local_send: tokio::net::tcp::OwnedWriteHalf,
local_recv: tokio::net::tcp::OwnedReadHalf,
relay_send: quinn::SendStream,
relay_recv: quinn::RecvStream,
}
struct PtthConnection {
uplink_task: JoinHandle <anyhow::Result <()>>,
downlink_task: JoinHandle <anyhow::Result <()>>,
}
impl PtthNewConnection {
fn build (self) -> PtthConnection {
let Self {
mut local_send,
mut local_recv,
mut relay_send,
mut relay_recv,
} = self;
let uplink_task = tokio::spawn (async move {
// Uplink - local client to relay server
let mut buf = vec! [0u8; 65_536];
loop {
let bytes_read = local_recv.read (&mut buf).await?;
if bytes_read == 0 {
break;
}
let buf_slice = &buf [0..bytes_read];
trace! ("Uplink relaying {} bytes", bytes_read);
relay_send.write_all (buf_slice).await?;
}
trace! ("Uplink closed");
Ok::<_, anyhow::Error> (())
});
let downlink_task = tokio::spawn (async move {
// Downlink - Relay server to local client
let mut buf = vec! [0u8; 65_536];
while let Some (bytes_read) = relay_recv.read (&mut buf).await? {
let buf_slice = &buf [0..bytes_read];
trace! ("Downlink relaying {} bytes", bytes_read);
local_send.write_all (buf_slice).await?;
}
trace! ("Downlink closed");
Ok::<_, anyhow::Error> (())
});
PtthConnection {
uplink_task,
downlink_task,
}
}
}

69
prototypes/quic_demo/src/bin/quic_demo_end_server.rs
View File

@ -84,77 +84,14 @@ async fn handle_p2_connection (
trace! ("Relaying bytes...");
let ptth_conn = PtthNewConnection {
let ptth_conn = quic_demo::connection::NewConnection {
local_send,
local_recv,
relay_send,
relay_recv,
}.build ();
ptth_conn.uplink_task.await??;
ptth_conn.downlink_task.await??;
ptth_conn.wait_for_close ().await?;
Ok (())
}
struct PtthNewConnection {
local_send: tokio::net::tcp::OwnedWriteHalf,
local_recv: tokio::net::tcp::OwnedReadHalf,
relay_send: quinn::SendStream,
relay_recv: quinn::RecvStream,
}
struct PtthConnection {
uplink_task: JoinHandle <anyhow::Result <()>>,
downlink_task: JoinHandle <anyhow::Result <()>>,
}
impl PtthNewConnection {
fn build (self) -> PtthConnection {
let Self {
mut local_send,
mut local_recv,
mut relay_send,
mut relay_recv,
} = self;
let uplink_task = tokio::spawn (async move {
// Downlink - Relay server to local client
let mut buf = vec! [0u8; 65_536];
while let Some (bytes_read) = relay_recv.read (&mut buf).await? {
let buf_slice = &buf [0..bytes_read];
trace! ("Uplink relaying {} bytes", bytes_read);
local_send.write_all (buf_slice).await?;
}
trace! ("Downlink closed");
Ok::<_, anyhow::Error> (())
});
let downlink_task = tokio::spawn (async move {
// Uplink - local client to relay server
let mut buf = vec! [0u8; 65_536];
loop {
let bytes_read = local_recv.read (&mut buf).await?;
if bytes_read == 0 {
break;
}
let buf_slice = &buf [0..bytes_read];
trace! ("Downlink relaying {} bytes", bytes_read);
relay_send.write_all (buf_slice).await?;
}
trace! ("Uplink closed");
Ok::<_, anyhow::Error> (())
});
PtthConnection {
uplink_task,
downlink_task,
}
}
}

74
prototypes/quic_demo/src/connection.rs Normal file
View File

@ -0,0 +1,74 @@
use crate::prelude::*;
pub struct NewConnection {
pub local_send: tokio::net::tcp::OwnedWriteHalf,
pub local_recv: tokio::net::tcp::OwnedReadHalf,
pub relay_send: quinn::SendStream,
pub relay_recv: quinn::RecvStream,
}
pub struct Connection {
// Blue and green because they're not necessarily uplink nor downlink.
// It depends on whether the client or server is using us.
task_blue: JoinHandle <anyhow::Result <()>>,
task_green: JoinHandle <anyhow::Result <()>>,
}
impl NewConnection {
pub fn build (self) -> Connection {
let Self {
mut local_send,
mut local_recv,
mut relay_send,
mut relay_recv,
} = self;
let task_blue = tokio::spawn (async move {
// Downlink - Relay server to local client
let mut buf = vec! [0u8; 65_536];
while let Some (bytes_read) = relay_recv.read (&mut buf).await? {
let buf_slice = &buf [0..bytes_read];
trace! ("Uplink relaying {} bytes", bytes_read);
local_send.write_all (buf_slice).await?;
}
trace! ("Downlink closed");
Ok::<_, anyhow::Error> (())
});
let task_green = tokio::spawn (async move {
// Uplink - local client to relay server
let mut buf = vec! [0u8; 65_536];
loop {
let bytes_read = local_recv.read (&mut buf).await?;
if bytes_read == 0 {
break;
}
let buf_slice = &buf [0..bytes_read];
trace! ("Downlink relaying {} bytes", bytes_read);
relay_send.write_all (buf_slice).await?;
}
trace! ("Uplink closed");
Ok::<_, anyhow::Error> (())
});
Connection {
task_blue,
task_green,
}
}
}
impl Connection {
pub async fn wait_for_close (self) -> anyhow::Result <()> {
self.task_blue.await??;
self.task_green.await??;
Ok (())
}
}

1
prototypes/quic_demo/src/lib.rs
View File

@ -1,3 +1,4 @@
pub mod connection;
pub mod prelude;
pub mod protocol;
pub mod quinn_utils;