starred/moli
1
0
Fork 0
mirror of https://github.com/lexmount/moli.git synced 2026-10-30 08:07:38 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor(curl): separate request handles from runtime ownership

Browse Source
This commit is contained in:
ldm0
2026-09-10 18:49:19 +08:00
parent ec8095653c
commit 5a8970dce8
21 changed files with 306 additions and 157 deletions
Download Patch File Download Diff File Expand all files Collapse all files
moli-curl/src/lib.rs
+2 -2
View File
@@ -7,7 +7,7 @@ pub mod websocket;
pub use dns_adapter::CurlDnsResolution;
pub use runtime::{
CurlMultiCompletion, CurlMultiJob, CurlMultiRuntime, CurlMultiRuntimeConfig, CurlOriginKey,
CurlSubmitError, CurlTransferId,
CurlHttpSender, CurlMultiCompletion, CurlMultiJob, CurlMultiRuntime, CurlMultiRuntimeConfig,
CurlOriginKey, CurlSubmitError, CurlTransferId,
};
pub use tls::CurlTlsConfig;
moli-curl/src/runtime.rs
+97 -65
View File
@@ -106,31 +106,34 @@ impl<H: Handler, C> fmt::Debug for CurlSubmitError<H, C> {
}
}
/// Cloneable handle for a single libcurl multi owner thread.
/// Owns one native thread. Request handles cannot extend its lifetime or join it.
#[derive(Debug)]
pub struct CurlMultiRuntime<H: Handler + Send + 'static, C: Send + 'static> {
inner: Arc<CurlMultiRuntimeInner<H, C>>,
}
impl<H: Handler + Send + 'static, C: Send + 'static> Clone for CurlMultiRuntime<H, C> {
fn clone(&self) -> Self {
Self {
inner: Arc::clone(&self.inner),
}
}
}
#[derive(Debug)]
struct CurlMultiRuntimeInner<H: Handler + Send + 'static, C: Send + 'static> {
command_tx: Sender<CurlRuntimeCommand<H, C>>,
owner_waker: MultiWaker,
shutdown_requested: Arc<AtomicBool>,
http: CurlHttpSender<H, C>,
websocket_connector: CurlWebSocketConnector,
#[cfg(test)]
owner_started: Arc<AtomicBool>,
owner_handle: Mutex<Option<thread::JoinHandle<()>>>,
}
/// Submits HTTP work to one owner. Clones carry no shutdown or join authority.
#[derive(Debug)]
pub struct CurlHttpSender<H: Handler + Send + 'static, C: Send + 'static> {
command_tx: Sender<CurlRuntimeCommand<H, C>>,
owner_waker: MultiWaker,
shutdown_requested: Arc<AtomicBool>,
}
impl<H: Handler + Send + 'static, C: Send + 'static> Clone for CurlHttpSender<H, C> {
fn clone(&self) -> Self {
Self {
command_tx: self.command_tx.clone(),
owner_waker: self.owner_waker.clone(),
shutdown_requested: self.shutdown_requested.clone(),
}
}
}
impl<H: Handler + Send + 'static, C: Send + 'static> CurlMultiRuntime<H, C> {
pub fn new(
config: CurlMultiRuntimeConfig,
@@ -162,28 +165,61 @@ impl<H: Handler + Send + 'static, C: Send + 'static> CurlMultiRuntime<H, C> {
.recv()
.context("curl multi runtime owner did not publish a waker")?;
let runtime = Self {
inner: Arc::new(CurlMultiRuntimeInner {
websocket_connector: CurlWebSocketConnector::new(
websocket_tx,
owner_waker.clone(),
shutdown_requested.clone(),
),
http: CurlHttpSender {
command_tx,
websocket_connector: CurlWebSocketConnector::new(
websocket_tx,
owner_waker.clone(),
shutdown_requested.clone(),
),
owner_waker,
shutdown_requested,
#[cfg(test)]
owner_started,
owner_handle: Mutex::new(Some(owner_handle)),
}),
},
#[cfg(test)]
owner_started,
owner_handle: Mutex::new(Some(owner_handle)),
};
Ok((runtime, completion_rx))
}
pub fn http_sender(&self) -> CurlHttpSender<H, C> {
self.http.clone()
}
pub fn shutdown(&self) {
if !self.http.shutdown_requested.swap(true, Ordering::SeqCst) {
let _ = self.http.command_tx.send(CurlRuntimeCommand::Shutdown);
let _ = self.http.owner_waker.wakeup();
}
if let Some(owner_handle) = self.owner_handle.lock().take() {
let _ = owner_handle.join();
}
}
/// Creates WebSockets on this runtime's owner and Multi. The capability
/// cannot keep the runtime alive or shut down unrelated HTTP transfers.
pub fn websocket_connector(&self) -> CurlWebSocketConnector {
self.websocket_connector.clone()
}
#[cfg(test)]
pub fn owner_count_for_testing(&self) -> usize {
usize::from(self.owner_started.load(Ordering::SeqCst))
}
}
impl<H: Handler + Send + 'static, C: Send + 'static> Drop for CurlMultiRuntime<H, C> {
fn drop(&mut self) {
self.shutdown();
}
}
impl<H: Handler + Send + 'static, C: Send + 'static> CurlHttpSender<H, C> {
pub fn submit(
&self,
job: CurlMultiJob<H, C>,
) -> std::result::Result<CurlTransferId, CurlSubmitError<H, C>> {
if self.inner.shutdown_requested.load(Ordering::SeqCst) {
if self.shutdown_requested.load(Ordering::SeqCst) {
return Err(CurlSubmitError {
job,
error: anyhow!("curl multi runtime is shutting down"),
@@ -194,12 +230,11 @@ impl<H: Handler + Send + 'static, C: Send + 'static> CurlMultiRuntime<H, C> {
Err(error) => return Err(CurlSubmitError { job, error }),
};
match self
.inner
.command_tx
.send(CurlRuntimeCommand::Request { transfer_id, job })
{
Ok(()) => {
let _ = self.inner.owner_waker.wakeup();
let _ = self.owner_waker.wakeup();
Ok(transfer_id)
}
Err(error) => {
@@ -213,41 +248,6 @@ impl<H: Handler + Send + 'static, C: Send + 'static> CurlMultiRuntime<H, C> {
}
}
}
pub fn shutdown(&self) {
self.inner.shutdown();
}
/// Creates WebSockets on this runtime's owner and Multi. The capability
/// cannot keep the runtime alive or shut down unrelated HTTP transfers.
pub fn websocket_connector(&self) -> CurlWebSocketConnector {
self.inner.websocket_connector.clone()
}
#[cfg(test)]
pub fn owner_count_for_testing(&self) -> usize {
usize::from(self.inner.owner_started.load(Ordering::SeqCst))
}
}
impl<H: Handler + Send + 'static, C: Send + 'static> Drop for CurlMultiRuntimeInner<H, C> {
fn drop(&mut self) {
self.shutdown();
}
}
impl<H: Handler + Send + 'static, C: Send + 'static> CurlMultiRuntimeInner<H, C> {
fn shutdown(&self) {
if self.shutdown_requested.swap(true, Ordering::SeqCst) {
return;
}
let _ = self.command_tx.send(CurlRuntimeCommand::Shutdown);
let _ = self.owner_waker.wakeup();
let Some(owner_handle) = self.owner_handle.lock().take() else {
return;
};
let _ = owner_handle.join();
}
}
#[cfg(test)]
@@ -297,6 +297,7 @@ mod tests {
easy.url(&format!("http://{address}/identity"))
.expect("test curl URL should be valid");
let transfer_id = runtime
.http_sender()
.submit(CurlMultiJob {
easy,
context: "matching-context".to_owned(),
@@ -321,4 +322,35 @@ mod tests {
runtime.shutdown();
server.join().expect("test HTTP server should finish");
}
#[test]
fn http_sender_does_not_keep_owner_alive_and_returns_rejected_job() {
let (runtime, completed) =
CurlMultiRuntime::<TestHandler, Vec<u8>>::new(Default::default()).unwrap();
let sender = runtime.http_sender();
let retained = sender.clone();
drop(runtime);
for sender in [sender, retained] {
let mut easy = Easy2::new(TestHandler);
easy.url("http://127.0.0.1:1/must-not-connect").unwrap();
let error = sender
.submit(CurlMultiJob {
easy,
context: vec![7; 1024],
origin: None,
deadline: None,
dns_resolution: CurlDnsResolution::curl_managed(),
priority: 1,
label: "closed".into(),
})
.unwrap_err();
assert!(error.error.to_string().contains("shutting down"));
assert_eq!(error.job.context, vec![7; 1024]);
assert_eq!(error.job.label, "closed");
}
assert!(matches!(
completed.try_recv(),
Err(crossbeam_channel::TryRecvError::Disconnected)
));
}
}
moli-curl/src/websocket.rs
+1 -1
View File
@@ -377,7 +377,7 @@ impl CurlWebSocketConnector {
/// Standalone owner for callers without an existing HTTP runtime. Uses the
/// same Multi driver as HTTP, with no HTTP submissions.
#[derive(Clone, Debug)]
#[derive(Debug)]
pub struct CurlWebSocketRuntime {
runtime: CurlMultiRuntime<StandaloneHandler, ()>,
}
moli-curl/src/websocket/tests/shared.rs
+4
View File
@@ -136,6 +136,7 @@ async fn shared_owner_keeps_http_and_websocket_pools_independent() {
connections.push(connection);
}
runtime
.http_sender()
.submit(request(
&(base.clone() + "/ok"),
HttpCapture::default(),
@@ -167,6 +168,7 @@ async fn shared_owner_keeps_http_and_websocket_pools_independent() {
}
// Retiring WS does not retire the shared owner or its HTTP pool.
runtime
.http_sender()
.submit(request(
&(base.clone() + "/after"),
HttpCapture::default(),
@@ -200,6 +202,7 @@ async fn shared_owner_preserves_http_connection_caps_with_live_websocket() {
opened(&mut connection).await;
let (headers, ready) = tokio::sync::oneshot::channel();
let first = runtime
.http_sender()
.submit(request(
&(base.clone() + "/held"),
HttpCapture {
@@ -213,6 +216,7 @@ async fn shared_owner_preserves_http_connection_caps_with_live_websocket() {
// Both HTTP jobs fit max_active=2. Only the connection cap keeps the
// second waiting until its own deadline, while the first is held open.
let second = runtime
.http_sender()
.submit(request(
&(base.clone() + "/queued"),
HttpCapture::default(),
moli-fetch/src/runtime.rs
+8 -4
View File
@@ -25,7 +25,8 @@ use moli_cookie_jar::{
advance_cookie_request_context,
};
use moli_curl::{
CurlMultiCompletion, CurlMultiJob, CurlMultiRuntime, CurlMultiRuntimeConfig, CurlOriginKey,
CurlHttpSender, CurlMultiCompletion, CurlMultiJob, CurlMultiRuntime, CurlMultiRuntimeConfig,
CurlOriginKey,
};
use moli_url_policy::ensure_http_network_transport_url;
use parking_lot::Mutex;
@@ -390,12 +391,14 @@ impl FetchRuntimeOwner {
let (curl_runtime, curl_completion_rx) = CurlMultiRuntime::new(curl_runtime_config(config))
.expect("failed to start fetch curl multi runtime");
let websocket_connector = curl_runtime.websocket_connector();
let curl_http = curl_runtime.http_sender();
let owner = RuntimeOwner {
config: config.clone(),
cookie_store,
client_hint_preferences,
request_rx,
curl_runtime,
curl_http,
curl_completion_rx,
shutdown_requested: Arc::clone(&shutdown_requested),
#[cfg(test)]
@@ -704,6 +707,7 @@ struct RuntimeOwner {
client_hint_preferences: SharedClientHintPreferences,
request_rx: Receiver<RuntimeCommand>,
curl_runtime: CurlMultiRuntime<FetchTransferHandler, ActiveTransferContext>,
curl_http: CurlHttpSender<FetchTransferHandler, ActiveTransferContext>,
curl_completion_rx: Receiver<RuntimeCurlCompletion>,
shutdown_requested: Arc<AtomicBool>,
#[cfg(test)]
@@ -942,7 +946,7 @@ impl RuntimeOwner {
label,
context: ActiveTransferContext::Buffered(Box::new(context)),
};
match self.curl_runtime.submit(curl_job) {
match self.curl_http.submit(curl_job) {
Ok(_) => Ok(JobOutcome::Submitted),
Err(error) => Err((
error
@@ -1128,7 +1132,7 @@ impl RuntimeOwner {
label,
context: ActiveTransferContext::Streaming(Box::new(context)),
};
match self.curl_runtime.submit(curl_job) {
match self.curl_http.submit(curl_job) {
Ok(_) => Ok(StreamingJobOutcome::Submitted),
Err(error) => {
let context = error
@@ -1327,7 +1331,7 @@ impl RuntimeOwner {
label,
context: ActiveTransferContext::StreamingRaw(Box::new(context)),
};
match self.curl_runtime.submit(curl_job) {
match self.curl_http.submit(curl_job) {
Ok(_) => Ok(StreamingJobOutcome::Submitted),
Err(error) => {
let context = error
moli-renderer-v8/src/native_bridge/context_host/websockets/registration.rs
+52 -22
View File
@@ -9,7 +9,8 @@ use crate::types::{
};
use moli_websocket::{
ConnectOptions as WebSocketConnectOptions, spawn_connection,
spawn_connection_with_handshake_pause, spawn_failed_connection, spawn_synthetic_connection,
spawn_connection_with_handshake_pause, spawn_failed_connection, spawn_standalone_connection,
spawn_standalone_connection_with_handshake_pause, spawn_synthetic_connection,
websocket_cookie_url,
};
use url::Url;
@@ -49,8 +50,8 @@ impl JsContextHost {
.as_ref()
.ok()
.and_then(|header| header.clone());
let connector = loader.map(|loader| loader.websocket_connector());
let context = WebSocketConnectOptions {
connector: loader.map(|loader| loader.websocket_connector()),
origin: moli_url::origin_ascii_serialization(&document_url),
user_agent: loader
.map(|loader| loader.user_agent().to_owned())
@@ -91,6 +92,7 @@ impl JsContextHost {
v8::Global::new(scope, scope.get_current_context()),
dispatch_scope,
PendingWebSocketConnection {
connector,
socket_id,
protocols,
connect_options: context,
@@ -113,13 +115,23 @@ impl JsContextHost {
(None, Some(internal_id))
} else {
let connection = match cookie_header {
Ok(_) => spawn_connection(
socket_id,
url.to_string(),
protocols,
context,
self.page_websocket_sender().event_sender(),
),
Ok(_) => match connector {
Some(connector) => spawn_connection(
connector,
socket_id,
url.to_string(),
protocols,
context,
self.page_websocket_sender().event_sender(),
),
None => spawn_standalone_connection(
socket_id,
url.to_string(),
protocols,
context,
self.page_websocket_sender().event_sender(),
),
},
Err(error) => spawn_failed_connection(
socket_id,
format!("failed to build WebSocket cookie header: {error}"),
@@ -166,25 +178,43 @@ impl JsContextHost {
context.cookie_header = None;
}
let (connection, handshake_controller) = if intercept_response {
let (connection, controller) = spawn_connection_with_handshake_pause(
pending.socket_id,
url.to_string(),
pending.protocols,
context,
event_sender,
);
(connection, Some(controller))
} else {
(
spawn_connection(
let (connection, controller) = match pending.connector {
Some(connector) => spawn_connection_with_handshake_pause(
connector,
pending.socket_id,
url.to_string(),
pending.protocols,
context,
event_sender,
),
None,
)
None => spawn_standalone_connection_with_handshake_pause(
pending.socket_id,
url.to_string(),
pending.protocols,
context,
event_sender,
),
};
(connection, Some(controller))
} else {
let connection = match pending.connector {
Some(connector) => spawn_connection(
connector,
pending.socket_id,
url.to_string(),
pending.protocols,
context,
event_sender,
),
None => spawn_standalone_connection(
pending.socket_id,
url.to_string(),
pending.protocols,
context,
event_sender,
),
};
(connection, None)
};
state.handshake_controller = handshake_controller;
state.url = url;
moli-renderer-v8/src/network/request_client/tests/websocket.rs
+2 -2
View File
@@ -11,10 +11,10 @@ async fn page_and_worker_websocket_connectors_follow_resource_runtime_owner() ->
let mut connections = Vec::new();
let mut peers = Vec::new();
for (index, client) in [&page, &worker].into_iter().enumerate() {
let mut context = test_websocket_context();
context.connector = Some(client.websocket_connector());
let context = test_websocket_context();
let (events, incoming) = tokio::sync::mpsc::channel(8);
let connection = spawn_connection(
client.websocket_connector(),
3000 + index as u64,
url.clone(),
Vec::new(),
moli-renderer-v8/src/types.rs
+1
View File
@@ -307,6 +307,7 @@ impl ImageRequestKey {
}
}
pub(super) struct PendingWebSocketConnection {
pub(super) connector: Option<moli_fetch::CurlWebSocketConnector>,
pub(super) socket_id: u64,
pub(super) protocols: Vec<String>,
pub(super) connect_options: moli_websocket::ConnectOptions,
moli-renderer-v8/src/worker/global_scope/fetch.rs
+1 -1
View File
@@ -1276,7 +1276,6 @@ pub(crate) fn register_worker_websocket<'s>(
.ok()
.and_then(|header| header.clone());
let context = WebSocketConnectOptions {
connector: Some(loader.request_client().websocket_connector()),
origin: moli_url::origin_ascii_serialization(&document_url),
user_agent: loader.request_client().user_agent().to_owned(),
extra_headers: extra_http_headers,
@@ -1327,6 +1326,7 @@ pub(crate) fn register_worker_websocket<'s>(
None,
)?;
let connection = spawn_connection(
loader.request_client().websocket_connector(),
socket_id,
url.to_string(),
protocols,
moli-websocket/src/connection.rs
+2 -1
View File
@@ -10,6 +10,7 @@ use crate::{
};
pub(crate) async fn run_websocket_connection(
connector: moli_curl::websocket::CurlWebSocketConnector,
socket_id: u64,
url: String,
protocols: Vec<String>,
@@ -51,7 +52,7 @@ pub(crate) async fn run_websocket_connection(
// The handshake future owns the pending transport. End its scope before
// waiting for terminal delivery, including Close received while opening.
let opened = {
let handshake = open_websocket_connection(request, &context);
let handshake = open_websocket_connection(&connector, request, &context);
tokio::pin!(handshake);
loop {
tokio::select! {
moli-websocket/src/lib.rs
+1
View File
@@ -28,6 +28,7 @@ pub use protocol::{
};
pub use runtime::{
spawn_connection, spawn_connection_with_handshake_pause, spawn_failed_connection,
spawn_standalone_connection, spawn_standalone_connection_with_handshake_pause,
spawn_synthetic_connection,
};
pub(crate) use types::Command;
moli-websocket/src/native_tests.rs
+9 -8
View File
@@ -1,4 +1,4 @@
use crate::{Event, FrameOpcode, spawn_connection, test_support::*};
use crate::{Event, FrameOpcode, spawn_standalone_connection, test_support::*};
use futures_util::{SinkExt, StreamExt};
use tokio::{
io::AsyncWriteExt,
@@ -25,7 +25,7 @@ async fn native_send_accounting_does_not_require_a_server_echo() {
let _ = socket.flush().await;
});
let (tx, mut rx) = mpsc::channel(8);
let handle = spawn_connection(70, url, Vec::new(), test_websocket_context(), tx);
let handle = spawn_standalone_connection(70, url, Vec::new(), test_websocket_context(), tx);
recv_open_event(&mut rx).await;
handle.send_text("without echo".to_owned()).unwrap();
assert_send_completed(&mut rx, 70, FrameOpcode::Text, 12).await;
@@ -51,7 +51,7 @@ async fn native_local_close_without_peer_close_is_abnormal_once() {
// Intentionally drop without flushing tungstenite's queued Close reply.
});
let (tx, mut rx) = mpsc::channel(8);
let handle = spawn_connection(71, url, Vec::new(), test_websocket_context(), tx);
let handle = spawn_standalone_connection(71, url, Vec::new(), test_websocket_context(), tx);
recv_open_event(&mut rx).await;
handle
.close(Some(3001), "not acknowledged".to_owned())
@@ -105,7 +105,7 @@ async fn native_fragmented_utf8_and_ping_are_assembled_as_one_message() {
})
.await;
let (tx, mut rx) = mpsc::channel(8);
let handle = spawn_connection(72, url, Vec::new(), test_websocket_context(), tx);
let handle = spawn_standalone_connection(72, url, Vec::new(), test_websocket_context(), tx);
recv_open_event(&mut rx).await;
assert_text_message(&mut rx, 72, "🙂").await;
assert_text_message(&mut rx, 72, "").await;
@@ -154,7 +154,8 @@ async fn native_invalid_frames_and_messages_fail_before_delivery() {
})
.await;
let (tx, mut rx) = mpsc::channel(8);
let _handle = spawn_connection(73, url, Vec::new(), test_websocket_context(), tx);
let _handle =
spawn_standalone_connection(73, url, Vec::new(), test_websocket_context(), tx);
recv_open_event(&mut rx).await;
match timeout(Duration::from_secs(3), rx.recv()).await.unwrap() {
Some(Event::Error { message, .. }) => {
@@ -209,7 +210,7 @@ async fn native_message_size_limit_counts_fragments_and_resets_after_delivery()
})
.await;
let (tx, mut rx) = mpsc::channel(8);
let handle = spawn_connection(76, url, Vec::new(), test_websocket_context(), tx);
let handle = spawn_standalone_connection(76, url, Vec::new(), test_websocket_context(), tx);
recv_open_event(&mut rx).await;
let event = timeout(Duration::from_secs(10), rx.recv()).await.unwrap();
if overflow {
@@ -260,7 +261,7 @@ async fn native_large_incoming_frame_keeps_chunk_offsets_and_utf8() {
})
.await;
let (tx, mut rx) = mpsc::channel(8);
let handle = spawn_connection(74, url, Vec::new(), test_websocket_context(), tx);
let handle = spawn_standalone_connection(74, url, Vec::new(), test_websocket_context(), tx);
recv_open_event(&mut rx).await;
assert_text_message(&mut rx, 74, &expected).await;
handle.close(Some(1000), String::new()).unwrap();
@@ -307,7 +308,7 @@ async fn native_close_handshake_finishes_while_message_sink_is_blocked() {
}
}
});
let handle = spawn_connection(75, url, Vec::new(), test_websocket_context(), sink);
let handle = spawn_standalone_connection(75, url, Vec::new(), test_websocket_context(), sink);
recv_open_event(&mut rx).await;
timeout(Duration::from_secs(3), blocked.notified())
.await
moli-websocket/src/runtime.rs
+71 -2
View File
@@ -1,3 +1,4 @@
use moli_curl::websocket::{CurlWebSocketConnector, CurlWebSocketRuntime};
use std::sync::OnceLock;
use crate::commands::command_channel;
@@ -9,17 +10,27 @@ use crate::{
};
pub fn spawn_connection(
connector: CurlWebSocketConnector,
socket_id: u64,
url: String,
protocols: Vec<String>,
context: ConnectOptions,
event_tx: impl Into<EventSender>,
) -> ConnectionHandle {
spawn_native_connection(socket_id, url, protocols, context, event_tx.into(), None)
spawn_native_connection(
Ok(connector),
socket_id,
url,
protocols,
context,
event_tx.into(),
None,
)
}
/// Starts a native connection whose browser Open awaits one explicit decision.
pub fn spawn_connection_with_handshake_pause(
connector: CurlWebSocketConnector,
socket_id: u64,
url: String,
protocols: Vec<String>,
@@ -28,6 +39,7 @@ pub fn spawn_connection_with_handshake_pause(
) -> (ConnectionHandle, HandshakeController) {
let (decision, receiver) = tokio::sync::oneshot::channel();
let connection = spawn_native_connection(
Ok(connector),
socket_id,
url,
protocols,
@@ -38,7 +50,56 @@ pub fn spawn_connection_with_handshake_pause(
(connection, HandshakeController::new(decision))
}
/// Uses the process runtime for callers without a network owner.
pub fn spawn_standalone_connection(
socket_id: u64,
url: String,
protocols: Vec<String>,
context: ConnectOptions,
event_tx: impl Into<EventSender>,
) -> ConnectionHandle {
spawn_native_connection(
standalone_connector(),
socket_id,
url,
protocols,
context,
event_tx.into(),
None,
)
}
pub fn spawn_standalone_connection_with_handshake_pause(
socket_id: u64,
url: String,
protocols: Vec<String>,
context: ConnectOptions,
event_tx: impl Into<EventSender>,
) -> (ConnectionHandle, HandshakeController) {
let (decision, receiver) = tokio::sync::oneshot::channel();
let connection = spawn_native_connection(
standalone_connector(),
socket_id,
url,
protocols,
context,
event_tx.into(),
Some(receiver),
);
(connection, HandshakeController::new(decision))
}
pub(super) fn standalone_connector() -> Result<CurlWebSocketConnector, String> {
static RUNTIME: OnceLock<Result<CurlWebSocketRuntime, String>> = OnceLock::new();
RUNTIME
.get_or_init(|| CurlWebSocketRuntime::new().map_err(|error| error.to_string()))
.as_ref()
.map(CurlWebSocketRuntime::connector)
.map_err(Clone::clone)
}
fn spawn_native_connection(
connector: Result<CurlWebSocketConnector, String>,
socket_id: u64,
url: String,
protocols: Vec<String>,
@@ -48,8 +109,16 @@ fn spawn_native_connection(
) -> ConnectionHandle {
let (command_tx, command_rx) = command_channel();
let task = websocket_runtime().spawn(async move {
let connector = match connector {
Ok(connector) => connector,
Err(error) => {
drop(command_rx);
let _ = send_error_and_close(&event_tx, socket_id, error).await;
return;
}
};
let _ = run_websocket_connection(
socket_id, url, protocols, context, command_rx, event_tx, decision,
connector, socket_id, url, protocols, context, command_rx, event_tx, decision,
)
.await;
});
moli-websocket/src/runtime_tests.rs
+6 -6
View File
@@ -19,10 +19,10 @@ async fn next(events: &mut mpsc::Receiver<Event>) -> Event {
async fn provided_native_connector_shutdown_ends_browser_session() {
let (url, server) = spawn_text_echo_websocket_server().await;
let runtime = CurlWebSocketRuntime::new().unwrap();
let mut context = test_websocket_context();
context.connector = Some(runtime.connector());
let context = test_websocket_context();
let connector = runtime.connector();
let (events, mut incoming) = mpsc::channel(8);
let handle = spawn_connection(1000, url, Vec::new(), context, events);
let handle = spawn_connection(connector, 1000, url, Vec::new(), context, events);
assert!(matches!(next(&mut incoming).await, Event::Open { .. }));
handle.send_text("shared owner".into()).unwrap();
assert!(matches!(
@@ -52,11 +52,11 @@ async fn closed_provided_connector_never_falls_back_to_global_owner() {
listener.set_nonblocking(true).unwrap();
let url = format!("ws://{}/must-not-connect", listener.local_addr().unwrap());
let runtime = CurlWebSocketRuntime::new().unwrap();
let mut context = test_websocket_context();
context.connector = Some(runtime.connector());
let context = test_websocket_context();
let connector = runtime.connector();
drop(runtime);
let (events, mut incoming) = mpsc::channel(8);
let handle = spawn_connection(1001, url, Vec::new(), context, events);
let handle = spawn_connection(connector, 1001, url, Vec::new(), context, events);
assert!(
matches!(next(&mut incoming).await, Event::Error { message, .. } if message.contains("curl WebSocket runtime is closed"))
);
moli-websocket/src/session/tests.rs
+4 -4
View File
@@ -1,5 +1,5 @@
use super::CLOSE_TIMEOUT;
use crate::{Command, Event, FrameOpcode, spawn_connection, test_support::*};
use crate::{Command, Event, FrameOpcode, spawn_standalone_connection, test_support::*};
use futures_util::{SinkExt, StreamExt};
use tokio::{
net::{TcpListener, TcpSocket},
@@ -39,7 +39,7 @@ async fn native_close_waits_for_queued_data_before_starting_timeout() {
socket.flush().await.unwrap();
});
let (tx, mut rx) = mpsc::channel(8);
let handle = spawn_connection(80, url, Vec::new(), test_websocket_context(), tx);
let handle = spawn_standalone_connection(80, url, Vec::new(), test_websocket_context(), tx);
recv_open_event(&mut rx).await;
handle
.send_binary((0..MESSAGE_BYTES).map(|i| (i % 251) as u8).collect())
@@ -78,7 +78,7 @@ async fn native_close_still_times_out_when_peer_never_answers_close() {
let _ = finish_rx.await;
});
let (tx, mut rx) = mpsc::channel(8);
let handle = spawn_connection(81, url, Vec::new(), test_websocket_context(), tx);
let handle = spawn_standalone_connection(81, url, Vec::new(), test_websocket_context(), tx);
recv_open_event(&mut rx).await;
handle.close(Some(1000), String::new()).unwrap();
assert_closing(&mut rx, 81).await;
@@ -151,7 +151,7 @@ async fn native_fragmented_sends_complete_while_incoming_sink_is_blocked() {
}
}
});
let handle = spawn_connection(82, url, Vec::new(), test_websocket_context(), sink);
let handle = spawn_standalone_connection(82, url, Vec::new(), test_websocket_context(), sink);
recv_open_event(&mut rx).await;
timeout(Duration::from_secs(3), blocked.notified())
.await
moli-websocket/src/terminal_tests.rs
+3
View File
@@ -106,6 +106,7 @@ async fn failed_handshake_closes_commands_before_blocked_error() {
let (commands, receiver) = command_channel();
let (sink, mut events, release) = blocked_terminal_sink();
let task = tokio::spawn(run_websocket_connection(
crate::runtime::standalone_connector().unwrap(),
99,
url,
Vec::new(),
@@ -138,6 +139,7 @@ async fn close_during_handshake_releases_tcp_before_blocked_error() {
let (commands, receiver) = command_channel();
let (sink, mut events, release) = blocked_terminal_sink();
let task = tokio::spawn(run_websocket_connection(
crate::runtime::standalone_connector().unwrap(),
100,
url,
Vec::new(),
@@ -197,6 +199,7 @@ async fn rejected_handshake_decision_closes_commands_before_blocked_error() {
let (decision, decide) = oneshot::channel();
let (sink, mut events, release) = blocked_terminal_sink();
let task = tokio::spawn(run_websocket_connection(
crate::runtime::standalone_connector().unwrap(),
101,
url,
Vec::new(),
moli-websocket/src/test_support.rs
+4 -3
View File
@@ -16,7 +16,8 @@ use tokio_tungstenite::tungstenite::{
};
use crate::{
ConnectOptions, Event, FrameOpcode, proxy::websocket_proxy_url_with_env, spawn_connection,
ConnectOptions, Event, FrameOpcode, proxy::websocket_proxy_url_with_env,
spawn_standalone_connection,
};
struct InfallibleHandshakeCallback<F>(F);
@@ -32,7 +33,6 @@ where
pub fn test_websocket_context() -> ConnectOptions {
ConnectOptions {
connector: None,
origin: "https://example.com".to_owned(),
user_agent: "Moli-WebSocket-Test/1.0".to_owned(),
extra_headers: Vec::new(),
@@ -84,7 +84,8 @@ pub async fn websocket_computed_accept_handshake_failure_message(
async fn websocket_handshake_failure_message(url: String, protocols: Vec<String>) -> String {
let (event_tx, mut event_rx) = mpsc::channel(32);
let _command_tx = spawn_connection(90, url, protocols, test_websocket_context(), event_tx);
let _command_tx =
spawn_standalone_connection(90, url, protocols, test_websocket_context(), event_tx);
recv_handshake_failure_events(&mut event_rx).await
}
moli-websocket/src/tests.rs
+32 -18
View File
@@ -451,7 +451,8 @@ async fn websocket_last_handle_drop_releases_pending_handshake() {
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let url = format!("ws://{}/pending", listener.local_addr().unwrap());
let (event_tx, _event_rx) = mpsc::channel(1);
let handle = spawn_connection(61, url, Vec::new(), test_websocket_context(), event_tx);
let handle =
spawn_standalone_connection(61, url, Vec::new(), test_websocket_context(), event_tx);
let retained = handle.clone();
let (mut server, _) = timeout(Duration::from_secs(3), listener.accept())
.await
@@ -495,6 +496,7 @@ async fn websocket_rejected_handshake_or_open_event_releases_transport() {
assert!(terminal.is_none() || terminal.is_some_and(|result| result.is_err()));
};
let client = crate::connection::run_websocket_connection(
crate::runtime::standalone_connector().unwrap(),
64,
url,
Vec::new(),
@@ -535,6 +537,7 @@ async fn websocket_rejected_message_event_releases_transport_and_writer() {
assert!(terminal.is_none() || terminal.is_some_and(|result| result.is_err()));
};
let client = crate::connection::run_websocket_connection(
crate::runtime::standalone_connector().unwrap(),
65,
url,
Vec::new(),
@@ -614,7 +617,7 @@ async fn websocket_cancel_releases_blocked_sink_and_open_writer() {
true
}
});
let handle = spawn_connection(62, url, Vec::new(), test_websocket_context(), sink);
let handle = spawn_standalone_connection(62, url, Vec::new(), test_websocket_context(), sink);
let retained = handle.clone();
let (stream, _) = timeout(Duration::from_secs(3), listener.accept())
.await
@@ -820,7 +823,8 @@ async fn websocket_synthetic_connection_can_receive_frames_and_server_close() {
async fn websocket_transport_close_while_connecting_fails_before_open() {
let (url, server) = spawn_sleeping_handshake_websocket_server().await;
let (event_tx, mut event_rx) = mpsc::channel(32);
let command_tx = spawn_connection(91, url, Vec::new(), test_websocket_context(), event_tx);
let command_tx =
spawn_standalone_connection(91, url, Vec::new(), test_websocket_context(), event_tx);
command_tx
.close(None, String::new())
@@ -836,7 +840,7 @@ async fn websocket_transport_close_while_connecting_fails_before_open() {
async fn websocket_transport_can_pause_after_handshake_before_open() {
let (url, server) = spawn_text_echo_websocket_server().await;
let (event_tx, mut event_rx) = mpsc::channel(32);
let (command_tx, controller) = spawn_connection_with_handshake_pause(
let (command_tx, controller) = spawn_standalone_connection_with_handshake_pause(
92,
url,
Vec::new(),
@@ -890,7 +894,8 @@ async fn websocket_transport_handshake_applies_context_headers_and_preserves_con
];
context.cookie_header = Some("sid=server".to_owned());
let command_tx = spawn_connection(1, url, vec!["chat".to_owned()], context, event_tx);
let command_tx =
spawn_standalone_connection(1, url, vec!["chat".to_owned()], context, event_tx);
let headers = timeout(Duration::from_secs(3), headers_rx)
.await
.expect("websocket headers should arrive")
@@ -939,7 +944,7 @@ async fn websocket_transport_uses_explicit_http_proxy_connect_without_forwarding
context.proxy_bearer_token = Some("proxy-token".to_owned());
let user_agent = context.user_agent.clone();
let command_tx = spawn_connection(2, url.clone(), Vec::new(), context, event_tx);
let command_tx = spawn_standalone_connection(2, url.clone(), Vec::new(), context, event_tx);
let proxy_request = timeout(Duration::from_secs(3), proxy_request_rx)
.await
.expect("proxy CONNECT should arrive")
@@ -985,7 +990,7 @@ async fn websocket_transport_rejects_non_200_http_proxy_connect() {
context.http_proxy = Some(proxy_url);
context.http_no_proxy = Some(String::new());
let _command_tx = spawn_connection(
let _command_tx = spawn_standalone_connection(
3,
"ws://example.test/socket".to_owned(),
Vec::new(),
@@ -1016,7 +1021,7 @@ async fn websocket_transport_respects_disabled_tls_verify_for_self_signed_wss()
let mut context = test_websocket_context();
context.tls.verify = false;
let command_tx = spawn_connection(3, url, Vec::new(), context, event_tx);
let command_tx = spawn_standalone_connection(3, url, Vec::new(), context, event_tx);
let headers = timeout(Duration::from_secs(3), headers_rx)
.await
.expect("websocket TLS headers should arrive")
@@ -1041,7 +1046,7 @@ async fn websocket_transport_wss_allows_server_to_omit_response_subprotocol() {
let mut context = test_websocket_context();
context.tls.verify = false;
let _command_tx = spawn_connection(
let _command_tx = spawn_standalone_connection(
4,
url,
vec!["chat".to_owned(), "superchat".to_owned()],
@@ -1175,7 +1180,7 @@ async fn websocket_transport_allows_server_to_omit_response_subprotocol() {
let (url, server) = spawn_text_binary_echo_websocket_server().await;
let (event_tx, mut event_rx) = mpsc::channel(32);
let command_tx = spawn_connection(
let command_tx = spawn_standalone_connection(
4,
url,
vec!["chat".to_owned(), "superchat".to_owned()],
@@ -1200,7 +1205,8 @@ async fn websocket_transport_allows_server_to_omit_response_subprotocol() {
async fn websocket_transport_sends_text_binary_and_reports_buffered_amount_consumption() {
let (url, server) = spawn_text_binary_echo_websocket_server().await;
let (event_tx, mut event_rx) = mpsc::channel(32);
let command_tx = spawn_connection(4, url, Vec::new(), test_websocket_context(), event_tx);
let command_tx =
spawn_standalone_connection(4, url, Vec::new(), test_websocket_context(), event_tx);
let open = recv_open_event(&mut event_rx).await;
assert_eq!(open.socket_id, 4);
@@ -1231,7 +1237,8 @@ async fn websocket_transport_sends_text_binary_and_reports_buffered_amount_consu
async fn websocket_transport_reads_while_sending_many_large_messages() {
let (url, server) = spawn_backpressure_websocket_server().await;
let (event_tx, mut event_rx) = mpsc::channel(32);
let command_tx = spawn_connection(40, url, Vec::new(), test_websocket_context(), event_tx);
let command_tx =
spawn_standalone_connection(40, url, Vec::new(), test_websocket_context(), event_tx);
let open = recv_open_event(&mut event_rx).await;
assert_eq!(open.socket_id, 40);
@@ -1280,7 +1287,8 @@ async fn websocket_transport_reads_while_sending_many_large_messages() {
async fn websocket_transport_reports_server_initiated_close_frame() {
let (url, server) = spawn_server_close_websocket_server().await;
let (event_tx, mut event_rx) = mpsc::channel(32);
let _command_tx = spawn_connection(5, url, Vec::new(), test_websocket_context(), event_tx);
let _command_tx =
spawn_standalone_connection(5, url, Vec::new(), test_websocket_context(), event_tx);
let open = recv_open_event(&mut event_rx).await;
assert_eq!(open.socket_id, 5);
@@ -1299,7 +1307,8 @@ async fn websocket_transport_handles_close_frame_in_handshake_packet() {
)
.await;
let (event_tx, mut event_rx) = mpsc::channel(32);
let _command_tx = spawn_connection(6, url, Vec::new(), test_websocket_context(), event_tx);
let _command_tx =
spawn_standalone_connection(6, url, Vec::new(), test_websocket_context(), event_tx);
let open = recv_open_event(&mut event_rx).await;
assert_eq!(open.socket_id, 6);
@@ -1320,7 +1329,7 @@ async fn websocket_dropped_handshake_controller_fails_open_and_releases_transpor
assert!(socket.next().await.unwrap().is_err());
});
let (events, mut receiver) = mpsc::channel(4);
let (_connection, controller) = spawn_connection_with_handshake_pause(
let (_connection, controller) = spawn_standalone_connection_with_handshake_pause(
95,
url,
Vec::new(),
@@ -1387,8 +1396,13 @@ async fn websocket_queued_close_precedes_handshake_continuation() {
accepted
}
});
let (connection, controller) =
spawn_connection_with_handshake_pause(97, url, Vec::new(), test_websocket_context(), sink);
let (connection, controller) = spawn_standalone_connection_with_handshake_pause(
97,
url,
Vec::new(),
test_websocket_context(),
sink,
);
assert!(matches!(
timeout(Duration::from_secs(3), receiver.recv())
.await
@@ -1432,7 +1446,7 @@ async fn websocket_native_handshake_generates_keys_and_reports_actual_headers()
"{header} belongs to native handshake generation"
);
}
let connection = spawn_connection(socket_id, url, Vec::new(), context, events);
let connection = spawn_standalone_connection(socket_id, url, Vec::new(), context, events);
let actual = timeout(Duration::from_secs(3), captured)
.await
.unwrap()
moli-websocket/src/tests/tls.rs
+1 -1
View File
@@ -12,7 +12,7 @@ fn tls_context(tls: CurlTlsConfig) -> ConnectOptions {
async fn exercise_tls_connection(url: String, context: ConnectOptions, opens: bool) {
let (events, mut receiver) = mpsc::channel(32);
let connection = spawn_connection(501, url, Vec::new(), context, events);
let connection = spawn_standalone_connection(501, url, Vec::new(), context, events);
if !opens {
assert!(
!recv_handshake_failure_events(&mut receiver)
moli-websocket/src/transport.rs
+5 -14
View File
@@ -8,11 +8,10 @@ use crate::{
use moli_curl::{
CurlDnsResolution,
websocket::{
CurlWebSocketConnection, CurlWebSocketEvent, CurlWebSocketRequest, CurlWebSocketRuntime,
CurlWebSocketConnection, CurlWebSocketConnector, CurlWebSocketEvent, CurlWebSocketRequest,
},
};
use moli_dns_resolver::DnsTarget;
use std::sync::OnceLock;
pub(crate) struct HandshakeInfo {
pub request_headers: http::HeaderMap,
@@ -25,6 +24,7 @@ pub(crate) struct OpenedConnection {
}
pub(crate) async fn open_websocket_connection(
connector: &CurlWebSocketConnector,
request: PreparedWebSocketRequest,
context: &ConnectOptions,
) -> Result<OpenedConnection, String> {
@@ -57,18 +57,9 @@ pub(crate) async fn open_websocket_connection(
);
}
}
let mut connection = match &context.connector {
Some(connector) => connector.connect(native),
None => {
static RUNTIME: OnceLock<Result<CurlWebSocketRuntime, String>> = OnceLock::new();
let runtime = RUNTIME
.get_or_init(|| CurlWebSocketRuntime::new().map_err(|error| error.to_string()))
.as_ref()
.map_err(Clone::clone)?;
runtime.connect(native)
}
}
.map_err(|error| error.to_string())?;
let mut connection = connector
.connect(native)
.map_err(|error| error.to_string())?;
match connection.recv().await {
Some(CurlWebSocketEvent::Handshake {
request: actual,
moli-websocket/src/types.rs
-3
View File
@@ -1,8 +1,5 @@
#[derive(Debug, Clone)]
pub struct ConnectOptions {
/// Native runtime for this network scope. None selects the standalone
/// process runtime. A closed provided connector fails without fallback.
pub connector: Option<moli_curl::websocket::CurlWebSocketConnector>,
pub origin: String,
pub user_agent: String,
pub extra_headers: Vec<(String, String)>,