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

🚧

main
_ 2021-07-17 06:50:54 +00:00
parent ce817eab1a
commit 1024c15fbb
3 changed files with 62 additions and 39 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -19,49 +19,59 @@ async fn main () -> anyhow::Result <()> {
.. ..
} = endpoint.connect (&server_addr, "localhost")?.await?; } = endpoint.connect (&server_addr, "localhost")?.await?;
debug! ("Waiting for local TCP client to connect to us"); let (mut send, mut recv) = connection.open_bi ().await?;
let listener = TcpListener::bind ("127.0.0.1:30381").await?; let req_buf = [2u8, 42, 0, 0];
let (tcp_socket, _) = listener.accept ().await?; send.write_all (&req_buf).await?;
let (mut local_recv, mut local_send) = tcp_socket.into_split ();
debug! ("Connecting to end server"); let mut resp_buf = [0u8, 0, 0, 0];
recv.read_exact (&mut resp_buf).await?;
let (mut relay_send, mut relay_recv) = connection.open_bi ().await?; if false {
debug! ("Waiting for local TCP client to connect to us");
debug! ("Relaying bytes..."); let listener = TcpListener::bind ("127.0.0.1:30381").await?;
let (tcp_socket, _) = listener.accept ().await?;
let (mut local_recv, mut local_send) = tcp_socket.into_split ();
let uplink_task = tokio::spawn (async move { debug! ("Connecting to end server");
// Uplink - local client to relay server
let mut buf = vec! [0u8; 65_536]; let (mut relay_send, mut relay_recv) = connection.open_bi ().await?;
loop {
let bytes_read = local_recv.read (&mut buf).await?;
let buf_slice = &buf [0..bytes_read];
relay_send.write_all (buf_slice).await?;
}
debug! ("Uplink closed"); debug! ("Relaying bytes...");
Ok::<_, anyhow::Error> (()) let uplink_task = tokio::spawn (async move {
}); // Uplink - local client to relay server
let downlink_task = tokio::spawn (async move { let mut buf = vec! [0u8; 65_536];
// Downlink - Relay server to local client loop {
let bytes_read = local_recv.read (&mut buf).await?;
let buf_slice = &buf [0..bytes_read];
relay_send.write_all (buf_slice).await?;
}
let mut buf = vec! [0u8; 65_536]; debug! ("Uplink closed");
while let Some (bytes_read) = relay_recv.read (&mut buf).await? {
let buf_slice = &buf [0..bytes_read];
local_send.write_all (buf_slice).await?;
}
debug! ("Downlink closed"); Ok::<_, anyhow::Error> (())
});
Ok::<_, anyhow::Error> (()) let downlink_task = tokio::spawn (async move {
}); // Downlink - Relay server to local client
uplink_task.await??; let mut buf = vec! [0u8; 65_536];
downlink_task.await??; while let Some (bytes_read) = relay_recv.read (&mut buf).await? {
let buf_slice = &buf [0..bytes_read];
local_send.write_all (buf_slice).await?;
}
debug! ("Downlink closed");
Ok::<_, anyhow::Error> (())
});
uplink_task.await??;
downlink_task.await??;
}
Ok (()) Ok (())
} }

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

@ -20,7 +20,7 @@ async fn main () -> anyhow::Result <()> {
let (mut send, mut recv) = connection.open_bi ().await?; let (mut send, mut recv) = connection.open_bi ().await?;
let req_buf = [4u8, 0, 0, 0]; let req_buf = [4u8, 43, 0, 0];
send.write_all (&req_buf).await?; send.write_all (&req_buf).await?;
let mut resp_buf = [0u8, 0, 0, 0]; let mut resp_buf = [0u8, 0, 0, 0];

21
prototypes/quic_demo/src/bin/quic_demo_relay_server.rs
View File

@ -14,7 +14,7 @@ async fn main () -> anyhow::Result <()> {
while let Some (conn) = incoming.next ().await { while let Some (conn) = incoming.next ().await {
let relay_state = Arc::clone (&relay_state); let relay_state = Arc::clone (&relay_state);
// Each new connection gets its own task // Each new peer QUIC connection gets its own task
tokio::spawn (async move { tokio::spawn (async move {
let quinn::NewConnection { let quinn::NewConnection {
connection, connection,
@ -31,15 +31,28 @@ async fn main () -> anyhow::Result <()> {
let mut req_buf = [0u8; 4]; let mut req_buf = [0u8; 4];
recv.read_exact (&mut req_buf).await?; recv.read_exact (&mut req_buf).await?;
match req_buf [0] { let peer_type = req_buf [0];
4 => debug! ("Server-side proxy (P4) connected"), let peer_id = req_buf [1];
2 => debug! ("Client-side proxy (P2) connected"),
match peer_type {
4 => debug! ("Server-side proxy (P4) connected, ID {}", peer_id),
2 => debug! ("Client-side proxy (P2) connected, ID {}", peer_id),
_ => bail! ("Unknown QUIC client type"), _ => bail! ("Unknown QUIC client type"),
} }
let resp_buf = [20u8, 0, 0, 0]; let resp_buf = [20u8, 0, 0, 0];
send.write_all (&resp_buf).await?; send.write_all (&resp_buf).await?;
match peer_type {
2 => while let Some (bi_stream) = bi_streams.next ().await {
let (mut send, mut recv) = bi_stream?;
// Each new
},
4 => (),
_ => bail! ("Unknown QUIC client type"),
}
Ok::<_, anyhow::Error> (()) Ok::<_, anyhow::Error> (())
}); });
} }