YGGverse/aquatic
1
0
Fork 0
mirror of https://github.com/YGGverse/aquatic.git synced 2026-03-31 09:45:31 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

http: start awaiting conn close message before tls setup

Browse source
This commit is contained in:
Joakim Frostegård 2024-01-20 09:43:55 +01:00
parent 75c1402394
commit 22e151d0f0
2 changed files with 21 additions and 29 deletions
Download patch file Download diff file Expand all files Collapse all files

22
crates/http/src/workers/socket/connection.rs
View file

@ -16,11 +16,9 @@ use aquatic_http_protocol::response::{
use arc_swap::ArcSwap; use arc_swap::ArcSwap;
use either::Either; use either::Either;
use futures::stream::FuturesUnordered; use futures::stream::FuturesUnordered;
use futures_lite::future::race;
use futures_lite::{AsyncReadExt, AsyncWriteExt, StreamExt}; use futures_lite::{AsyncReadExt, AsyncWriteExt, StreamExt};
use futures_rustls::TlsAcceptor; use futures_rustls::TlsAcceptor;
use glommio::channels::channel_mesh::Senders; use glommio::channels::channel_mesh::Senders;
use glommio::channels::local_channel::LocalReceiver;
use glommio::channels::shared_channel::{self, SharedReceiver}; use glommio::channels::shared_channel::{self, SharedReceiver};
use glommio::net::TcpStream; use glommio::net::TcpStream;
use once_cell::sync::Lazy; use once_cell::sync::Lazy;
@ -76,7 +74,6 @@ pub(super) async fn run_connection(
server_start_instant: ServerStartInstant, server_start_instant: ServerStartInstant,
opt_tls_config: Option<Arc<ArcSwap<RustlsConfig>>>, opt_tls_config: Option<Arc<ArcSwap<RustlsConfig>>>,
valid_until: Rc<RefCell<ValidUntil>>, valid_until: Rc<RefCell<ValidUntil>>,
close_conn_receiver: LocalReceiver<()>,
stream: TcpStream, stream: TcpStream,
) -> Result<(), ConnectionError> { ) -> Result<(), ConnectionError> {
let access_list_cache = create_access_list_cache(&access_list); let access_list_cache = create_access_list_cache(&access_list);
@ -119,7 +116,7 @@ pub(super) async fn run_connection(
stream, stream,
}; };
conn.run(close_conn_receiver).await?; conn.run().await
} else { } else {
let mut conn = Connection { let mut conn = Connection {
config, config,
@ -135,10 +132,8 @@ pub(super) async fn run_connection(
stream, stream,
}; };
conn.run(close_conn_receiver).await?; conn.run().await
} }
Ok(())
} }
struct Connection<S> { struct Connection<S> {
@ -159,18 +154,7 @@ impl<S> Connection<S>
where where
S: futures::AsyncRead + futures::AsyncWrite + Unpin + 'static, S: futures::AsyncRead + futures::AsyncWrite + Unpin + 'static,
{ {
async fn run(&mut self, close_conn_receiver: LocalReceiver<()>) -> Result<(), ConnectionError> { async fn run(&mut self) -> Result<(), ConnectionError> {
let f1 = async { self.run_request_response_loop().await };
let f2 = async {
close_conn_receiver.recv().await;
Err(ConnectionError::Inactive)
};
race(f1, f2).await
}
async fn run_request_response_loop(&mut self) -> Result<(), ConnectionError> {
loop { loop {
let response = match self.read_request().await? { let response = match self.read_request().await? {
Either::Left(response) => Response::Failure(response), Either::Left(response) => Response::Failure(response),

28
crates/http/src/workers/socket/mod.rs
View file

@ -12,6 +12,7 @@ use aquatic_common::privileges::PrivilegeDropper;
use aquatic_common::rustls_config::RustlsConfig; use aquatic_common::rustls_config::RustlsConfig;
use aquatic_common::{CanonicalSocketAddr, PanicSentinel, ServerStartInstant}; use aquatic_common::{CanonicalSocketAddr, PanicSentinel, ServerStartInstant};
use arc_swap::ArcSwap; use arc_swap::ArcSwap;
use futures_lite::future::race;
use futures_lite::StreamExt; use futures_lite::StreamExt;
use glommio::channels::channel_mesh::{MeshBuilder, Partial, Role}; use glommio::channels::channel_mesh::{MeshBuilder, Partial, Role};
use glommio::channels::local_channel::{new_bounded, LocalSender}; use glommio::channels::local_channel::{new_bounded, LocalSender};
@ -97,16 +98,23 @@ pub async fn run_socket_worker(
"worker_index" => worker_index.to_string(), "worker_index" => worker_index.to_string(),
); );
let result = run_connection( let f1 = async { run_connection(
config, config,
access_list, access_list,
request_senders, request_senders,
server_start_instant, server_start_instant,
opt_tls_config, opt_tls_config,
valid_until.clone(), valid_until.clone(),
close_conn_receiver, stream,
stream, ).await
).await; };
let f2 = async {
close_conn_receiver.recv().await;
Err(ConnectionError::Inactive)
};
let result = race(f1, f2).await;
#[cfg(feature = "metrics")] #[cfg(feature = "metrics")]
::metrics::decrement_gauge!( ::metrics::decrement_gauge!(