diff --git a/pact_mock_server/src/hyper_server.rs b/pact_mock_server/src/hyper_server.rs index 8305280e..a76cdd1e 100644 --- a/pact_mock_server/src/hyper_server.rs +++ b/pact_mock_server/src/hyper_server.rs @@ -4,7 +4,7 @@ use std::collections::HashMap; use std::fmt; use std::fmt::{Display, Formatter}; use std::net::SocketAddr; -#[cfg(feature = "tls")] use std::sync::Arc; +use std::sync::{Arc, Mutex}; use std::time::Duration; #[allow(unused_imports)] use anyhow::anyhow; @@ -72,7 +72,8 @@ pub(crate) async fn create_and_bind( server_id: String, pact: V4Pact, addr: SocketAddr, - config: MockServerConfig + config: MockServerConfig, + matches: Arc>> ) -> anyhow::Result<(SocketAddr, oneshot::Sender<()>, mpsc::Receiver, JoinHandle<()>)> { let listener = TcpListener::bind(addr).await?; let local_addr = listener.local_addr()?; @@ -88,6 +89,7 @@ pub(crate) async fn create_and_bind( let server_id = server_id.clone(); let pact = pact.clone(); let config = config.clone(); + let matches = matches.clone(); select! { connection = listener.accept() => { @@ -106,9 +108,10 @@ pub(crate) async fn create_and_bind( let event_send = ev.clone(); let config = config.clone(); let server_id = sid.clone(); + let matches = matches.clone(); LOG_ID.scope(server_id, async move { handle_mock_request_error( - handle_request(req, pact.clone(), event_send.clone(), &local_addr, &config).await + handle_request(req, pact.clone(), event_send.clone(), matches.clone(), &local_addr, &config).await ) }) }) @@ -162,7 +165,8 @@ pub(crate) async fn create_and_bind_https( server_id: String, pact: V4Pact, addr: SocketAddr, - config: MockServerConfig + config: MockServerConfig, + matches: Arc>> ) -> anyhow::Result<(SocketAddr, oneshot::Sender<()>, mpsc::Receiver, JoinHandle<()>)> { if CryptoProvider::get_default().is_none() { warn!("No TLS cryptographic provider has been configured, defaulting to the standard FIPS provider"); @@ -197,6 +201,7 @@ pub(crate) async fn create_and_bind_https( let server_id = server_id.clone(); let pact = pact.clone(); let config = config.clone(); + let matches = matches.clone(); select! { connection = listener.accept() => { @@ -219,9 +224,10 @@ pub(crate) async fn create_and_bind_https( let event_send = ev.clone(); let config = config.clone(); let server_id = sid.clone(); + let matches = matches.clone(); LOG_ID.scope(server_id, async move { handle_mock_request_error( - handle_request(req, pact.clone(), event_send.clone(), &local_addr, &config).await + handle_request(req, pact.clone(), event_send.clone(), matches.clone(), &local_addr, &config).await ) }) }) @@ -278,6 +284,7 @@ async fn handle_request( req: Request, pact: V4Pact, event_send: Sender, + matches: Arc>>, local_addr: &SocketAddr, config: &MockServerConfig ) -> Result>, InteractionError> { @@ -322,8 +329,13 @@ async fn handle_request( let match_result = match_request(&pact_request, &pact).await; - if let Err(_) = event_send.send(MockServerEvent::RequestMatch(match_result.clone())).await { - error!("Failed to send RequestMatch event"); + // Record the match result synchronously before returning the response. This ensures that + // pactffi_mock_server_matched() / pactffi_mock_server_mismatches() always sees the correct + // state immediately after the client receives the HTTP response, with no race against the + // async event loop. + { + let mut guard = matches.lock().unwrap(); + guard.push(match_result.clone()); } match_result_to_hyper_response(&pact_request, &match_result, local_addr, config).await @@ -569,11 +581,13 @@ mod tests { #[tokio::test] async fn can_fetch_results_on_current_thread() { + let matches = Arc::new(Mutex::new(vec![])); let (_addr, shutdown, mut events, handle) = create_and_bind( "can_fetch_results_on_current_thread".to_string(), RequestResponsePact::default().as_v4_pact().unwrap(), ([0, 0, 0, 0], 0u16).into(), - MockServerConfig::default() + MockServerConfig::default(), + matches ).await.unwrap(); shutdown.send(()).unwrap(); @@ -605,11 +619,13 @@ mod tests { interactions: vec![ RequestResponseInteraction::default() ], .. RequestResponsePact::default() }; + let matches = Arc::new(Mutex::new(vec![])); let (addr, shutdown, mut events, handle) = create_and_bind( "can_fetch_results_on_current_thread".to_string(), pact.as_v4_pact().unwrap(), ([127, 0, 0, 1], 0u16).into(), - MockServerConfig::default() + MockServerConfig::default(), + matches.clone() ).await.unwrap(); let client = reqwest::ClientBuilder::new() @@ -625,17 +641,18 @@ mod tests { shutdown.send(()).unwrap(); let _ = handle.await; - // Should be at least 3 events - expect!(events.len()).to(be_greater_or_equal_to(3)); + // Match results are now recorded directly in the mutex, not via the event channel. + // Events should contain RequestReceived plus the shutdown event (and possibly a + // ConnectionFailed on Linux once the server shuts down). + expect!(events.len()).to(be_greater_or_equal_to(2)); assert_eq!(events.recv().await.unwrap(), MockServerEvent::RequestReceived("/".to_string())); - if let MockServerEvent::RequestMatch(_) = events.recv().await.unwrap() { - // expected - } else { - panic!("Was expected a request match event"); - } // For some reason, a http2 connection returns an error once the server is shutdown on Linux let mut events_list = vec![]; events.recv_many(&mut events_list, 2).await; assert_eq!(events_list.last().unwrap(), &MockServerEvent::ServerShutdown); + + // Verify the match result was recorded synchronously in the mutex + let guard = matches.lock().unwrap(); + assert_eq!(guard.len(), 1); } } diff --git a/pact_mock_server/src/mock_server.rs b/pact_mock_server/src/mock_server.rs index 15ed60e9..7fe35d45 100644 --- a/pact_mock_server/src/mock_server.rs +++ b/pact_mock_server/src/mock_server.rs @@ -210,8 +210,6 @@ pub enum MockServerEvent { ConnectionFailed(String), /// Request received with path RequestReceived(String), - /// Result of matching a request - RequestMatch(MatchResult), /// Server is shutting down ServerShutdown } @@ -295,7 +293,8 @@ impl MockServer { }; trace!(%server_id, %address, "Starting mock server"); - let (address, shutdown_send, event_recv, _task_handle) = create_and_bind(server_id.clone(), pact.clone(), address, config.clone()).await?; + let matches = Arc::new(Mutex::new(vec![])); + let (address, shutdown_send, event_recv, _task_handle) = create_and_bind(server_id.clone(), pact.clone(), address, config.clone(), matches.clone()).await?; trace!(%server_id, %address, "Mock server started"); let mut mock_server = MockServer { @@ -303,7 +302,7 @@ impl MockServer { scheme: Default::default(), address, pact, - matches: Default::default(), + matches, shutdown_tx: RefCell::new(Some(shutdown_send)), config: config.clone(), metrics: Default::default(), @@ -333,7 +332,8 @@ impl MockServer { }; trace!(%server_id, %address, "Starting TLS mock server"); - let (address, shutdown_send, event_recv, _task_handle) = create_and_bind_https(server_id.clone(), pact.clone(), address, config.clone()).await?; + let matches = Arc::new(Mutex::new(vec![])); + let (address, shutdown_send, event_recv, _task_handle) = create_and_bind_https(server_id.clone(), pact.clone(), address, config.clone(), matches.clone()).await?; trace!(%server_id, %address, "TLS mock server started"); let mut mock_server = MockServer { @@ -341,7 +341,7 @@ impl MockServer { scheme: MockServerScheme::HTTPS, address, pact, - matches: Default::default(), + matches, shutdown_tx: RefCell::new(Some(shutdown_send)), config: config.clone(), metrics: Default::default(), @@ -383,7 +383,6 @@ impl MockServer { fn start_event_loop(&mut self, mut event_recv: Receiver) { let server_id = self.id.clone(); let metrics = self.metrics.clone(); - let matches = self.matches.clone(); let (sender, receiver) = mpsc::channel(); self.event_loop_rx = Some(receiver); @@ -392,7 +391,6 @@ impl MockServer { let mut total_events = 0; let metrics = metrics.clone(); - let matches = matches.clone(); while let Some(event) = event_recv.recv().await { trace!(%server_id, ?event, "Received event"); total_events += 1; @@ -403,10 +401,6 @@ impl MockServer { let mut guard = metrics.lock().unwrap(); guard.add_path(path); } - MockServerEvent::RequestMatch(result) => { - let mut guard = matches.lock().unwrap(); - guard.push(result.clone()); - } MockServerEvent::ServerShutdown => { trace!(%server_id, total_events, "Exiting mock server event loop"); break;