Compare commits
No commits in common. "05ba4822f831cba1e3e15798722da766a8f0f46a" and "f867f32927f3998f282f4b7591215b9516a83ac8" have entirely different histories.
05ba4822f8
...
f867f32927
|
@ -1328,7 +1328,6 @@ dependencies = [
|
||||||
"anyhow",
|
"anyhow",
|
||||||
"base64",
|
"base64",
|
||||||
"futures-util",
|
"futures-util",
|
||||||
"hyper",
|
|
||||||
"quinn",
|
"quinn",
|
||||||
"rcgen",
|
"rcgen",
|
||||||
"rmp-serde",
|
"rmp-serde",
|
||||||
|
|
|
@ -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`
|
|
|
@ -12,7 +12,6 @@ anyhow = "1.0.38"
|
||||||
base64 = "0.13.0"
|
base64 = "0.13.0"
|
||||||
# fltk = "1.1.1"
|
# fltk = "1.1.1"
|
||||||
futures-util = "0.3.9"
|
futures-util = "0.3.9"
|
||||||
hyper = { version = "0.14.4", features = ["http1", "server", "stream", "tcp"] }
|
|
||||||
quinn = "0.7.2"
|
quinn = "0.7.2"
|
||||||
rcgen = "0.8.11"
|
rcgen = "0.8.11"
|
||||||
rmp-serde = "0.15.5"
|
rmp-serde = "0.15.5"
|
||||||
|
|
|
@ -46,10 +46,11 @@ async fn main () -> anyhow::Result <()> {
|
||||||
let server_tcp_port = opt.server_tcp_port.unwrap_or (30382);
|
let server_tcp_port = opt.server_tcp_port.unwrap_or (30382);
|
||||||
let listener = TcpListener::bind (("127.0.0.1", client_tcp_port)).await?;
|
let listener = TcpListener::bind (("127.0.0.1", client_tcp_port)).await?;
|
||||||
|
|
||||||
|
debug! ("Accepting local TCP connections from P1");
|
||||||
|
|
||||||
// End of per-port stuff
|
// End of per-port stuff
|
||||||
// Beginning of per-connection 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 ();
|
||||||
|
@ -78,13 +79,4 @@ async fn main () -> anyhow::Result <()> {
|
||||||
Ok::<_, anyhow::Error> (())
|
Ok::<_, anyhow::Error> (())
|
||||||
});
|
});
|
||||||
}
|
}
|
||||||
|
|
||||||
Ok::<_, anyhow::Error> (())
|
|
||||||
});
|
|
||||||
|
|
||||||
debug! ("Accepting local TCP connections from P1");
|
|
||||||
|
|
||||||
task_tcp_server.await??;
|
|
||||||
|
|
||||||
Ok (())
|
|
||||||
}
|
}
|
||||||
|
|
|
@ -10,41 +10,37 @@ struct Opt {
|
||||||
relay_addr: Option <String>,
|
relay_addr: Option <String>,
|
||||||
#[structopt (long)]
|
#[structopt (long)]
|
||||||
server_id: Option <PeerId>,
|
server_id: Option <PeerId>,
|
||||||
#[structopt (long)]
|
|
||||||
debug_echo: bool,
|
|
||||||
}
|
}
|
||||||
|
|
||||||
#[tokio::main]
|
#[tokio::main]
|
||||||
async fn main () -> anyhow::Result <()> {
|
async fn main () -> anyhow::Result <()> {
|
||||||
tracing_subscriber::fmt::init ();
|
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 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])?;
|
let endpoint = make_client_endpoint ("0.0.0.0:0".parse ()?, &[&server_cert])?;
|
||||||
|
|
||||||
trace! ("Connecting to relay server");
|
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 {
|
let quinn::NewConnection {
|
||||||
mut bi_streams,
|
mut bi_streams,
|
||||||
..
|
..
|
||||||
} = protocol::p4_connect_to_p3 (&endpoint, &relay_addr, &server_id).await?;
|
} = protocol::p4_connect_to_p3 (&endpoint, &relay_addr, &server_id).await?;
|
||||||
|
|
||||||
debug! ("Connected to relay server");
|
|
||||||
trace! ("Accepting bi streams from P3");
|
trace! ("Accepting bi streams from P3");
|
||||||
|
|
||||||
loop {
|
loop {
|
||||||
let (relay_send, relay_recv) = bi_streams.next ().await.ok_or_else (|| anyhow::anyhow! ("Relay server didn't open a bi stream"))??;
|
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 (
|
async fn handle_bi_stream (
|
||||||
opt: Arc <Opt>,
|
|
||||||
relay_send: quinn::SendStream,
|
relay_send: quinn::SendStream,
|
||||||
mut relay_recv: quinn::RecvStream,
|
mut relay_recv: quinn::RecvStream,
|
||||||
) -> anyhow::Result <()>
|
) -> anyhow::Result <()>
|
||||||
|
@ -53,14 +49,13 @@ async fn handle_bi_stream (
|
||||||
protocol::P3ToP4Stream::NewPtthConnection {
|
protocol::P3ToP4Stream::NewPtthConnection {
|
||||||
client_id,
|
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 (())
|
Ok (())
|
||||||
}
|
}
|
||||||
|
|
||||||
async fn handle_new_ptth_connection (
|
async fn handle_new_ptth_connection (
|
||||||
opt: Arc <Opt>,
|
|
||||||
mut relay_send: quinn::SendStream,
|
mut relay_send: quinn::SendStream,
|
||||||
mut relay_recv: quinn::RecvStream,
|
mut relay_recv: quinn::RecvStream,
|
||||||
_client_id: String,
|
_client_id: String,
|
||||||
|
@ -77,12 +72,6 @@ async fn handle_new_ptth_connection (
|
||||||
|
|
||||||
debug! ("Started PTTH connection");
|
debug! ("Started PTTH connection");
|
||||||
|
|
||||||
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?;
|
|
||||||
}
|
|
||||||
else {
|
|
||||||
let stream = TcpStream::connect (("127.0.0.1", p4_to_p5_req.port)).await?;
|
let stream = TcpStream::connect (("127.0.0.1", p4_to_p5_req.port)).await?;
|
||||||
let (local_recv, local_send) = stream.into_split ();
|
let (local_recv, local_send) = stream.into_split ();
|
||||||
|
|
||||||
|
@ -96,7 +85,6 @@ async fn handle_new_ptth_connection (
|
||||||
}.build ();
|
}.build ();
|
||||||
|
|
||||||
ptth_conn.wait_for_close ().await?;
|
ptth_conn.wait_for_close ().await?;
|
||||||
}
|
|
||||||
|
|
||||||
Ok (())
|
Ok (())
|
||||||
}
|
}
|
||||||
|
|
|
@ -1,14 +1,3 @@
|
||||||
use hyper::{
|
|
||||||
Body,
|
|
||||||
Request,
|
|
||||||
Response,
|
|
||||||
Server,
|
|
||||||
service::{
|
|
||||||
make_service_fn,
|
|
||||||
service_fn,
|
|
||||||
},
|
|
||||||
StatusCode,
|
|
||||||
};
|
|
||||||
use structopt::StructOpt;
|
use structopt::StructOpt;
|
||||||
|
|
||||||
use quic_demo::prelude::*;
|
use quic_demo::prelude::*;
|
||||||
|
@ -31,32 +20,9 @@ async fn main () -> anyhow::Result <()> {
|
||||||
println! ("Base64 cert: {}", base64::encode (&server_cert));
|
println! ("Base64 cert: {}", base64::encode (&server_cert));
|
||||||
tokio::fs::write ("quic_server.crt", &server_cert).await?;
|
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 = {
|
|
||||||
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 {
|
while let Some (conn) = incoming.next ().await {
|
||||||
let relay_state = Arc::clone (&relay_state);
|
let relay_state = Arc::clone (&relay_state);
|
||||||
|
|
||||||
|
@ -75,76 +41,7 @@ async fn main () -> anyhow::Result <()> {
|
||||||
});
|
});
|
||||||
}
|
}
|
||||||
|
|
||||||
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 (())
|
||||||
});
|
|
||||||
}
|
|
||||||
|
|
||||||
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<_>> ());
|
|
||||||
}
|
|
||||||
|
|
||||||
let resp = Response::builder ()
|
|
||||||
.status (StatusCode::OK)
|
|
||||||
.header ("content-type", "text/plain")
|
|
||||||
.body (Body::from (debug_string))?;
|
|
||||||
|
|
||||||
Ok (resp)
|
|
||||||
}
|
}
|
||||||
|
|
||||||
#[derive (Default)]
|
#[derive (Default)]
|
||||||
|
@ -199,8 +96,8 @@ struct RequestP2ToP4 {
|
||||||
client_id: String,
|
client_id: String,
|
||||||
}
|
}
|
||||||
|
|
||||||
struct PtthNewConnection <ClientSend: AsyncWrite + Unpin> {
|
struct PtthNewConnection {
|
||||||
client_send: ClientSend,
|
client_send: quinn::SendStream,
|
||||||
client_recv: quinn::RecvStream,
|
client_recv: quinn::RecvStream,
|
||||||
server_send: quinn::SendStream,
|
server_send: quinn::SendStream,
|
||||||
server_recv: quinn::RecvStream,
|
server_recv: quinn::RecvStream,
|
||||||
|
@ -211,7 +108,7 @@ struct PtthConnection {
|
||||||
downlink_task: JoinHandle <anyhow::Result <()>>,
|
downlink_task: JoinHandle <anyhow::Result <()>>,
|
||||||
}
|
}
|
||||||
|
|
||||||
impl <ClientSend: AsyncWrite + Unpin> PtthNewConnection <ClientSend> {
|
impl PtthNewConnection {
|
||||||
fn build (self) -> PtthConnection {
|
fn build (self) -> PtthConnection {
|
||||||
let Self {
|
let Self {
|
||||||
mut client_send,
|
mut client_send,
|
||||||
|
|
|
@ -1,6 +1,5 @@
|
||||||
pub use std::{
|
pub use std::{
|
||||||
collections::*,
|
collections::*,
|
||||||
net::SocketAddr,
|
|
||||||
sync::{
|
sync::{
|
||||||
Arc,
|
Arc,
|
||||||
atomic::{
|
atomic::{
|
||||||
|
@ -18,12 +17,9 @@ pub use anyhow::{
|
||||||
pub use futures_util::StreamExt;
|
pub use futures_util::StreamExt;
|
||||||
pub use tokio::{
|
pub use tokio::{
|
||||||
io::{
|
io::{
|
||||||
AsyncRead,
|
|
||||||
AsyncReadExt,
|
AsyncReadExt,
|
||||||
AsyncWrite,
|
|
||||||
AsyncWriteExt,
|
AsyncWriteExt,
|
||||||
},
|
},
|
||||||
net::TcpListener,
|
|
||||||
sync::{
|
sync::{
|
||||||
Mutex,
|
Mutex,
|
||||||
mpsc,
|
mpsc,
|
||||||
|
|
Loading…
Reference in New Issue