fix: 🐛 waiting for certificates even if tls is off

This commit is contained in:
Ahmet Kaan GÜMÜŞ 2024-05-18 01:36:56 +03:00
parent e2f3d9eb8f
commit 3432ba68de
2 changed files with 66 additions and 58 deletions

View file

@ -5,11 +5,9 @@ use std::net::SocketAddr;
#[tokio::main] #[tokio::main]
async fn main() { async fn main() {
println!("Hello, world!"); println!("Hello, world!");
let relay_config = get_config().await; let relay_config = get_config().await;
let rustls_config =
RustlsConfig::from_pem_file("certificates/fullchain.pem", "certificates/privkey.pem")
.await
.unwrap();
let state = AppState {}; let state = AppState {};
let app = routing::routing(axum::extract::State(state)).await; let app = routing::routing(axum::extract::State(state)).await;
let addr = SocketAddr::from( let addr = SocketAddr::from(
@ -19,13 +17,21 @@ async fn main() {
.parse::<SocketAddr>() .parse::<SocketAddr>()
.unwrap(), .unwrap(),
); );
println!( tokio::spawn(streaming::start(relay_config.clone()));
"\n\n\tOn Air -> http://{}\n\n", if relay_config.tls {
relay_config.axum_address.clone() let rustls_config =
); RustlsConfig::from_pem_file("certificates/fullchain.pem", "certificates/privkey.pem")
tokio::spawn(streaming::start(relay_config)); .await
axum_server::bind_rustls(addr, rustls_config) .unwrap();
.serve(app.into_make_service())
.await println!("\n\n\tOn Air -> https://{}\n\n", relay_config.axum_address);
.unwrap(); axum_server::bind_rustls(addr, rustls_config)
.serve(app.into_make_service())
.await
.unwrap();
} else {
let listener = tokio::net::TcpListener::bind(addr).await.unwrap();
println!("\n\n\tOn Air -> http://{}\n\n", relay_config.axum_address);
axum::serve(listener, app).await.unwrap();
}
} }

View file

@ -26,7 +26,7 @@ const BUFFER_LENGTH: usize = 1000000;
const MAX_TOLERATED_MESSAGE_COUNT: usize = 10; const MAX_TOLERATED_MESSAGE_COUNT: usize = 10;
pub async fn start(relay_configs: Config) { pub async fn start(relay_configs: Config) {
let timer = Instant::now(); let timer = Instant::now();
let acceptor = tls_configurator().await; let acceptor = None;
loop { loop {
//need to move them for multi streamer //need to move them for multi streamer
let (record_producer, record_consumer) = channel(BUFFER_LENGTH); let (record_producer, record_consumer) = channel(BUFFER_LENGTH);
@ -54,6 +54,7 @@ pub async fn start(relay_configs: Config) {
timer.elapsed() timer.elapsed()
); );
if relay_configs.tls { if relay_configs.tls {
let acceptor = tls_configurator().await;
match acceptor.accept(streamer_tcp).await { match acceptor.accept(streamer_tcp).await {
Ok(streamer_tcp_tls) => { Ok(streamer_tcp_tls) => {
match tokio_tungstenite::accept_async(streamer_tcp_tls).await { match tokio_tungstenite::accept_async(streamer_tcp_tls).await {
@ -118,7 +119,6 @@ pub async fn start(relay_configs: Config) {
let listener_handler_task = tokio::spawn(listener_handler( let listener_handler_task = tokio::spawn(listener_handler(
listener_socket, listener_socket,
acceptor.clone(), acceptor.clone(),
relay_configs.tls,
buffered_producer.clone(), buffered_producer.clone(),
listener_stream_tasks_producer, listener_stream_tasks_producer,
timer, timer,
@ -164,8 +164,7 @@ async fn tls_configurator() -> TlsAcceptor {
} }
async fn listener_handler( async fn listener_handler(
listener_socket: TcpListener, listener_socket: TcpListener,
acceptor: TlsAcceptor, acceptor: Option<TlsAcceptor>,
is_tls: bool,
buffered_producer: Sender<Message>, buffered_producer: Sender<Message>,
listener_stream_tasks_producer: tokio::sync::mpsc::Sender<JoinHandle<()>>, listener_stream_tasks_producer: tokio::sync::mpsc::Sender<JoinHandle<()>>,
timer: Instant, timer: Instant,
@ -179,51 +178,54 @@ async fn listener_handler(
ip: listener_info.ip(), ip: listener_info.ip(),
port: listener_info.port(), port: listener_info.port(),
}; };
if is_tls { match acceptor {
match acceptor.accept(tcp_stream).await { Some(ref acceptor) => {
Ok(listener_tcp_tls) => { match acceptor.accept(tcp_stream).await {
match tokio_tungstenite::accept_async(listener_tcp_tls).await { Ok(listener_tcp_tls) => {
Ok(wss_stream) => { match tokio_tungstenite::accept_async(listener_tcp_tls).await {
let listener_stream_task = tokio::spawn(stream( Ok(wss_stream) => {
new_listener, let listener_stream_task = tokio::spawn(stream(
wss_stream, new_listener,
buffered_producer.subscribe(), wss_stream,
)); buffered_producer.subscribe(),
let _ = listener_stream_tasks_producer ));
.send(listener_stream_task) let _ = listener_stream_tasks_producer
.await; .send(listener_stream_task)
} .await;
Err(err_val) => { }
eprintln!("Error: TCP WSS Listener | {}", err_val); Err(err_val) => {
drop(listener_socket); eprintln!("Error: TCP WSS Listener | {}", err_val);
return; drop(listener_socket);
return;
}
} }
} }
Err(err_val) => {
eprintln!("Error: TCP TLS Listener | {}", err_val);
drop(listener_socket);
return;
}
} }
Err(err_val) => { },
eprintln!("Error: TCP TLS Listener | {}", err_val); None => {
drop(listener_socket); match tokio_tungstenite::accept_async(tcp_stream).await {
return; Ok(ws_stream) => {
let listener_stream_task = tokio::spawn(stream(
new_listener,
ws_stream,
buffered_producer.subscribe(),
));
let _ = listener_stream_tasks_producer
.send(listener_stream_task)
.await;
}
Err(err_val) => {
eprintln!("Error: TCP WS Listener | {}", err_val);
drop(listener_socket);
return;
}
} }
} },
} else {
match tokio_tungstenite::accept_async(tcp_stream).await {
Ok(ws_stream) => {
let listener_stream_task = tokio::spawn(stream(
new_listener,
ws_stream,
buffered_producer.subscribe(),
));
let _ = listener_stream_tasks_producer
.send(listener_stream_task)
.await;
}
Err(err_val) => {
eprintln!("Error: TCP WS Listener | {}", err_val);
drop(listener_socket);
return;
}
}
} }
println!("New Listener: {} | {:#?}", listener_info, timer.elapsed()); println!("New Listener: {} | {:#?}", listener_info, timer.elapsed());
} }