diff --git a/README.md b/README.md index 1f827a39..00dded6e 100644 --- a/README.md +++ b/README.md @@ -66,9 +66,17 @@ use std::net::Ipv4Addr; #[tokio::main] async fn main() { - // Client::new returns a (Client, ClientUpdates) pair. - // Client is Clone-able and can be shared across tasks. - let (client, mut updates) = Client::::new(Ipv4Addr::new(192, 168, 1, 100)); + // Client::new returns a Clone-able handle, an update stream, and + // the run-loop future. The future must be actively driven — either + // spawned on the runtime as shown below, or awaited alongside your + // own work in a `tokio::select!`. If the future is never polled, + // Client method calls that send commands over the control channel + // will hang indefinitely waiting on their oneshot response. + // `Error::Shutdown` is returned only once the run-loop future has + // been dropped or its task cancelled. + let (client, mut updates, run) = + Client::::new(Ipv4Addr::new(192, 168, 1, 100)); + let _run_task = tokio::spawn(run); // Bind the SD multicast socket to discover services client.bind_discovery().await.unwrap(); @@ -95,12 +103,17 @@ use std::net::Ipv4Addr; async fn main() -> Result<(), Box> { let config = ServerConfig::new(Ipv4Addr::new(192, 168, 1, 200), 30500, 0x1234, 1); let mut server = Server::new(config).await?; - server.start_announcing()?; + let announce_handle = tokio::spawn(server.announcement_loop()?); let publisher = server.publisher(); - tokio::spawn(async move { server.run().await }); + let run_handle = tokio::spawn(async move { server.run().await }); // Publish events to subscribers... + + tokio::select! { + res = announce_handle => eprintln!("announcement loop exited unexpectedly: {res:?}"), + res = run_handle => eprintln!("server run loop exited: {res:?}"), + } Ok(()) } ``` diff --git a/examples/client_server/src/main.rs b/examples/client_server/src/main.rs index 82771d04..1cc716de 100644 --- a/examples/client_server/src/main.rs +++ b/examples/client_server/src/main.rs @@ -10,7 +10,7 @@ //! This ensures remote nodes see a single coherent network identity for //! multicast announcements. //! -//! The server's built-in `start_announcing()` is NOT used — instead, the +//! The server's built-in `announcement_loop()` is NOT used — instead, the //! client's `start_sd_announcements()` handles periodic multicast //! announcements. The server's `run()` loop still handles unicast SD //! traffic (e.g. `SubscribeAck`/`SubscribeNack` responses) on its own @@ -106,7 +106,8 @@ async fn main() -> Result<(), Box> { // ── Create the client (handles discovery, subscriptions, SD socket) ── - let (client, mut updates) = simple_someip::Client::::new(interface); + let (client, mut updates, run) = simple_someip::Client::::new(interface); + let _run_handle = tokio::spawn(run); client.bind_discovery().await?; info!("Client discovery bound"); @@ -125,13 +126,13 @@ async fn main() -> Result<(), Box> { let mut server = Server::new(config).await?; info!("Server bound on port {MY_SERVER_PORT}"); - // NOTE: We intentionally do NOT call server.start_announcing(). + // NOTE: We intentionally do NOT spawn server.announcement_loop(). // The client's start_sd_announcements handles all SD traffic. let _publisher = server.publisher(); // Spawn the server event loop (handles incoming subscriptions). - tokio::spawn(async move { + let _server_handle = tokio::spawn(async move { if let Err(e) = server.run().await { error!("Server error: {e}"); } diff --git a/examples/discovery_client/src/main.rs b/examples/discovery_client/src/main.rs index 41b90fc4..ae866bf7 100644 --- a/examples/discovery_client/src/main.rs +++ b/examples/discovery_client/src/main.rs @@ -287,7 +287,8 @@ async fn main() -> Result<(), Error> { info!("Starting discovery client on interface {interface}"); - let (client, mut updates) = simple_someip::Client::::new(interface); + let (client, mut updates, run) = simple_someip::Client::::new(interface); + let _run_handle = tokio::spawn(run); client.bind_discovery().await.unwrap(); let mut state = DiscoveryState::new(); diff --git a/src/client/error.rs b/src/client/error.rs index 97063c09..8746c4b6 100644 --- a/src/client/error.rs +++ b/src/client/error.rs @@ -42,14 +42,26 @@ pub enum Error { /// A fixed-capacity internal structure is full. The argument is a /// lowercase `snake_case` tag naming the resource; grep the crate for /// the tag to find the compile-time constant that governs it. Current - /// tags: `"unicast_sockets"` (→ `UNICAST_SOCKETS_CAP`), `"udp_buffer"` - /// (→ `crate::UDP_BUFFER_SIZE`). + /// tags: + /// - `"unicast_sockets"` → `UNICAST_SOCKETS_CAP` + /// - `"udp_buffer"` → `crate::UDP_BUFFER_SIZE` + /// - `"pending_responses"` → `PENDING_RESPONSES_CAP` #[error("internal capacity exceeded: {0}")] Capacity(&'static str), /// An error surfaced by the pluggable transport backend (see /// [`crate::transport::TransportError`]). #[error(transparent)] Transport(#[from] crate::transport::TransportError), + /// The client's internal run-loop future has exited — either because + /// the caller dropped it before or during polling, the executor + /// cancelled its task, or it returned. All public `Client` methods + /// that enqueue a control message or await its response return + /// this variant when the control channel is closed, rather than + /// panicking on `.unwrap()` of the send / recv result. Treat it as + /// a caller-side lifecycle error: the `Client` handle has outlived + /// its driver and further calls on it cannot make progress. + #[error("client run loop is no longer running")] + Shutdown, } #[cfg(test)] diff --git a/src/client/inner.rs b/src/client/inner.rs index 5db723aa..8d35a845 100644 --- a/src/client/inner.rs +++ b/src/client/inner.rs @@ -344,13 +344,24 @@ impl Inner where PayloadDefinitions: PayloadWireFormat + Clone + std::fmt::Debug + 'static, { - pub fn spawn( + /// Construct an `Inner` and return the control/update channels plus + /// the run-loop future. The caller must drive the future on a Tokio + /// runtime (e.g. via `tokio::spawn`). + /// + /// The future is bounded `Send + 'static` because every in-repo + /// consumer spawns it on a multithreaded Tokio runtime and because the + /// concrete captured state (tokio mpsc, `TokioSocket`, E2E registry) + /// is already Send. A bare-metal consumer whose transport produces + /// `!Send` state needs a cfg-gated alternative constructor; none + /// exists yet — it's planned alongside the bare-metal port. + pub fn build( interface: Ipv4Addr, e2e_registry: Arc>, multicast_loopback: bool, ) -> ( Sender>, mpsc::UnboundedReceiver>, + impl core::future::Future + Send + 'static, ) { info!("Initializing SOME/IP Client"); let (control_sender, control_receiver) = mpsc::channel(4); @@ -374,8 +385,7 @@ where multicast_loopback, phantom: std::marker::PhantomData, }; - inner.run(); - (control_sender, update_receiver) + (control_sender, update_receiver, inner.run_future()) } async fn bind_discovery(&mut self) -> Result<(), Error> { @@ -878,8 +888,8 @@ where } #[allow(clippy::too_many_lines)] - fn run(mut self) { - tokio::spawn(async move { + fn run_future(mut self) -> impl core::future::Future + Send + 'static { + async move { info!("SOME/IP Client processing loop started"); loop { let Self { @@ -896,23 +906,20 @@ where } = &mut self; select! { () = tokio::time::sleep(std::time::Duration::from_millis(125)) => {} - // Receive a control message - ctrl = control_receiver.recv() => { + // Receive a control message only when the request queue + // has spare capacity, so we apply backpressure on the + // control channel instead of dropping the message — + // which would cancel any embedded oneshot senders and + // surface to callers as `RecvError` (mapped to + // `Error::Shutdown`), conflating overload with shutdown. + ctrl = control_receiver.recv(), if request_queue.len() < REQUEST_QUEUE_CAP => { if let Some(ctrl) = ctrl { debug!("Received control message: {:?}", ctrl); - if let Err(rejected) = request_queue.push_back(ctrl) { - // Queue full: rather than silently drop the - // rejected ControlMessage (which would - // cancel its oneshot senders and panic any - // caller awaiting with `.unwrap()`), reply - // on each sender with - // `Err(Error::Capacity("request_queue"))`. - warn!( - "request_queue at capacity ({}); rejecting control message", - REQUEST_QUEUE_CAP - ); - rejected.reject_with_capacity("request_queue"); - } + let push_result = request_queue.push_back(ctrl); + debug_assert!( + push_result.is_ok(), + "request_queue had capacity before recv but push_back failed" + ); } else { // The sender has been dropped, so we should exit *run = false; @@ -920,115 +927,115 @@ where } // Receive a discovery message discovery = Inner::receive_discovery(discovery_socket) => { - trace!("Received discovery message: {:?}", discovery); - match discovery { - Ok((source, someip_header, sd_header)) => { - // Extract session ID from SOME/IP request_id (lower 16 bits) - let session_id = (someip_header.request_id() & 0xFFFF) as u16; - let sd_payload = PayloadDefinitions::new_sd_payload(&sd_header); - // Extract reboot flag from the SD payload flags - let reboot_flag = sd_payload - .sd_flags() - .map_or(crate::protocol::sd::RebootFlag::Continuous, |f| { - f.reboot() - }); - - // Track sender session/reboot state for every SD entry - // that identifies a service instance, not only - // offer/stop-offer entries. This ensures reboot - // detection works for all SD traffic (FindService, - // Subscribe, SubscribeAck, etc.). - let mut rebooted = false; - for (svc_id, inst_id) in sd_payload.service_instances() { - let verdict = session_tracker.check( - source, - TransportKind::Multicast, - svc_id, - inst_id, - session_id, - reboot_flag, - ); - if verdict == SessionVerdict::Reboot { - rebooted = true; - } + trace!("Received discovery message: {:?}", discovery); + match discovery { + Ok((source, someip_header, sd_header)) => { + // Extract session ID from SOME/IP request_id (lower 16 bits) + let session_id = (someip_header.request_id() & 0xFFFF) as u16; + let sd_payload = PayloadDefinitions::new_sd_payload(&sd_header); + // Extract reboot flag from the SD payload flags + let reboot_flag = sd_payload + .sd_flags() + .map_or(crate::protocol::sd::RebootFlag::Continuous, |f| { + f.reboot() + }); + + // Track sender session/reboot state for every SD entry + // that identifies a service instance, not only + // offer/stop-offer entries. This ensures reboot + // detection works for all SD traffic (FindService, + // Subscribe, SubscribeAck, etc.). + let mut rebooted = false; + for (svc_id, inst_id) in sd_payload.service_instances() { + let verdict = session_tracker.check( + source, + TransportKind::Multicast, + svc_id, + inst_id, + session_id, + reboot_flag, + ); + if verdict == SessionVerdict::Reboot { + rebooted = true; } + } - // Auto-populate service registry from offer/stop-offer - // SD entries. - for ep in sd_payload.offered_endpoints() { - let id = ServiceInstanceId { - service_id: ep.service_id, - instance_id: ep.instance_id, - }; - if ep.is_offer { - if let Some(addr) = ep.addr { - service_registry.insert( - id, - ServiceEndpointInfo { - addr, - local_port: 0, - major_version: ep.major_version, - minor_version: ep.minor_version, - }, - ); - trace!( - "Registry: added 0x{:04X}.0x{:04X} -> {}", - ep.service_id, ep.instance_id, addr, - ); - } - } else { - service_registry.remove(id); + // Auto-populate service registry from offer/stop-offer + // SD entries. + for ep in sd_payload.offered_endpoints() { + let id = ServiceInstanceId { + service_id: ep.service_id, + instance_id: ep.instance_id, + }; + if ep.is_offer { + if let Some(addr) = ep.addr { + service_registry.insert( + id, + ServiceEndpointInfo { + addr, + local_port: 0, + major_version: ep.major_version, + minor_version: ep.minor_version, + }, + ); trace!( - "Registry: removed 0x{:04X}.0x{:04X}", - ep.service_id, ep.instance_id, + "Registry: added 0x{:04X}.0x{:04X} -> {}", + ep.service_id, ep.instance_id, addr, ); } + } else { + service_registry.remove(id); + trace!( + "Registry: removed 0x{:04X}.0x{:04X}", + ep.service_id, ep.instance_id, + ); } - - if rebooted { - let _ = update_sender.send(ClientUpdate::SenderRebooted(source)); - } - - let discovery_msg = DiscoveryMessage { - source, - someip_header, - sd_header, - }; - let _ = update_sender.send(ClientUpdate::DiscoveryUpdated(discovery_msg)); } - Err(err) => { - error!("Error receiving discovery message: {:?}", err); - let _ = update_sender.send(ClientUpdate::Error(err)); + + if rebooted { + let _ = update_sender.send(ClientUpdate::SenderRebooted(source)); } + + let discovery_msg = DiscoveryMessage { + source, + someip_header, + sd_header, + }; + let _ = update_sender.send(ClientUpdate::DiscoveryUpdated(discovery_msg)); } - } - unicast = Inner::receive_any_unicast(unicast_sockets) => { - trace!("Received unicast message: {:?}", unicast); - match unicast { - Ok(received) => { - let ReceivedMessage { message: received_message, e2e_status, .. } = received; - // Check if this matches a pending request-response by request_id - let request_id = received_message.header().request_id(); - if let Some(sender) = pending_responses.remove(&request_id) { - let _ = sender.send(Ok(received_message.payload().clone())); - continue; - } - // Not a response — forward as ClientUpdate::Unicast - let _ = update_sender.send(ClientUpdate::Unicast { message: received_message, e2e_status }); - } - Err(err) => { - let _ = update_sender.send(ClientUpdate::Error(err)); + Err(err) => { + error!("Error receiving discovery message: {:?}", err); + let _ = update_sender.send(ClientUpdate::Error(err)); + } + } + } + unicast = Inner::receive_any_unicast(unicast_sockets) => { + trace!("Received unicast message: {:?}", unicast); + match unicast { + Ok(received) => { + let ReceivedMessage { message: received_message, e2e_status, .. } = received; + // Check if this matches a pending request-response by request_id + let request_id = received_message.header().request_id(); + if let Some(sender) = pending_responses.remove(&request_id) { + let _ = sender.send(Ok(received_message.payload().clone())); + continue; } + // Not a response — forward as ClientUpdate::Unicast + let _ = update_sender.send(ClientUpdate::Unicast { message: received_message, e2e_status }); + } + Err(err) => { + let _ = update_sender.send(ClientUpdate::Error(err)); } } - } - if !*run { - info!("SOME/IP Client processing loop exiting"); - break; - } - self.handle_control_message().await; + } } - }); + if !*run { + info!("SOME/IP Client processing loop exiting"); + break; + } + self.handle_control_message().await; + } + } } } @@ -1366,12 +1373,13 @@ mod tests { } #[tokio::test] - async fn test_inner_spawn_and_shutdown() { - let (control_sender, mut update_receiver) = Inner::::spawn( + async fn test_inner_build_and_shutdown() { + let (control_sender, mut update_receiver, run_fut) = Inner::::build( Ipv4Addr::LOCALHOST, Arc::new(Mutex::new(E2ERegistry::new())), false, ); + let _ = tokio::spawn(run_fut); // Drop control sender to trigger loop exit drop(control_sender); // The update receiver should eventually return None when the inner loop exits @@ -1401,11 +1409,12 @@ mod tests { #[tokio::test] async fn test_dropped_receiver_bind_discovery_continues() { - let (control_sender, _update_receiver) = Inner::::spawn( + let (control_sender, _update_receiver, run_fut) = Inner::::build( Ipv4Addr::LOCALHOST, Arc::new(Mutex::new(E2ERegistry::new())), false, ); + let _ = tokio::spawn(run_fut); let (rx, msg) = TestControl::bind_discovery(); drop(rx); @@ -1417,11 +1426,12 @@ mod tests { #[tokio::test] async fn test_dropped_receiver_unbind_discovery_continues() { - let (control_sender, _update_receiver) = Inner::::spawn( + let (control_sender, _update_receiver, run_fut) = Inner::::build( Ipv4Addr::LOCALHOST, Arc::new(Mutex::new(E2ERegistry::new())), false, ); + let _ = tokio::spawn(run_fut); let (rx, msg) = TestControl::unbind_discovery(); drop(rx); @@ -1433,11 +1443,12 @@ mod tests { #[tokio::test] async fn test_dropped_receiver_set_interface_continues() { - let (control_sender, _update_receiver) = Inner::::spawn( + let (control_sender, _update_receiver, run_fut) = Inner::::build( Ipv4Addr::LOCALHOST, Arc::new(Mutex::new(E2ERegistry::new())), false, ); + let _ = tokio::spawn(run_fut); // SetInterface(LOCALHOST) on a fresh inner goes straight to // bind_discovery + send response (interface already matches). @@ -1451,11 +1462,12 @@ mod tests { #[tokio::test] async fn test_dropped_receiver_send_sd_continues() { - let (control_sender, _update_receiver) = Inner::::spawn( + let (control_sender, _update_receiver, run_fut) = Inner::::build( Ipv4Addr::LOCALHOST, Arc::new(Mutex::new(E2ERegistry::new())), false, ); + let _ = tokio::spawn(run_fut); // Bind discovery first so the SendSD path has a socket to use let (rx, msg) = TestControl::bind_discovery(); @@ -1480,11 +1492,12 @@ mod tests { #[tokio::test] async fn test_queued_messages_all_complete() { - let (control_sender, _update_receiver) = Inner::::spawn( + let (control_sender, _update_receiver, run_fut) = Inner::::build( Ipv4Addr::LOCALHOST, Arc::new(Mutex::new(E2ERegistry::new())), false, ); + let _ = tokio::spawn(run_fut); // Bind discovery so SetInterface will take the multi-step path: // iteration 1: unbind discovery, re-queue SetInterface @@ -1550,11 +1563,12 @@ mod tests { #[tokio::test] async fn test_dropped_receiver_add_endpoint_continues() { - let (control_sender, _update_receiver) = Inner::::spawn( + let (control_sender, _update_receiver, run_fut) = Inner::::build( Ipv4Addr::LOCALHOST, Arc::new(Mutex::new(E2ERegistry::new())), false, ); + let _ = tokio::spawn(run_fut); let addr = SocketAddrV4::new(Ipv4Addr::LOCALHOST, 5000); let (rx, msg) = TestControl::add_endpoint(0x1234, 0x0001, addr, 0); @@ -1567,11 +1581,12 @@ mod tests { #[tokio::test] async fn test_dropped_receiver_remove_endpoint_continues() { - let (control_sender, _update_receiver) = Inner::::spawn( + let (control_sender, _update_receiver, run_fut) = Inner::::build( Ipv4Addr::LOCALHOST, Arc::new(Mutex::new(E2ERegistry::new())), false, ); + let _ = tokio::spawn(run_fut); let (rx, msg) = TestControl::remove_endpoint(0x1234, 0x0001); drop(rx); @@ -1583,11 +1598,12 @@ mod tests { #[tokio::test] async fn test_dropped_receiver_send_to_service_send_complete_continues() { - let (control_sender, _update_receiver) = Inner::::spawn( + let (control_sender, _update_receiver, run_fut) = Inner::::build( Ipv4Addr::LOCALHOST, Arc::new(Mutex::new(E2ERegistry::new())), false, ); + let _ = tokio::spawn(run_fut); // Add an endpoint first so SendToService doesn't fail with ServiceNotFound let addr = SocketAddrV4::new(Ipv4Addr::LOCALHOST, 5000); @@ -1609,11 +1625,12 @@ mod tests { async fn test_bind_discovery_with_loopback() { // Spawn inner with multicast_loopback=true so bind_discovery exercises // the loopback-enabled branch of SocketManager::bind_discovery. - let (control_sender, _update_receiver) = Inner::::spawn( + let (control_sender, _update_receiver, run_fut) = Inner::::build( Ipv4Addr::LOCALHOST, Arc::new(Mutex::new(E2ERegistry::new())), true, ); + let _ = tokio::spawn(run_fut); let (rx, msg) = TestControl::bind_discovery(); control_sender.send(msg).await.unwrap(); @@ -1623,11 +1640,12 @@ mod tests { #[tokio::test] async fn test_bind_discovery_idempotent() { // Binding discovery twice should succeed (early return on already-bound) - let (control_sender, _update_receiver) = Inner::::spawn( + let (control_sender, _update_receiver, run_fut) = Inner::::build( Ipv4Addr::LOCALHOST, Arc::new(Mutex::new(E2ERegistry::new())), false, ); + let _ = tokio::spawn(run_fut); let (rx, msg) = TestControl::bind_discovery(); control_sender.send(msg).await.unwrap(); @@ -1642,11 +1660,12 @@ mod tests { #[tokio::test] async fn test_send_sd_auto_binds_discovery() { // SendSD without a bound discovery socket should auto-bind and succeed - let (control_sender, _update_receiver) = Inner::::spawn( + let (control_sender, _update_receiver, run_fut) = Inner::::build( Ipv4Addr::LOCALHOST, Arc::new(Mutex::new(E2ERegistry::new())), false, ); + let _ = tokio::spawn(run_fut); let target = SocketAddrV4::new(Ipv4Addr::LOCALHOST, 30490); let sd_header = empty_sd_header(); @@ -1662,11 +1681,12 @@ mod tests { #[tokio::test] async fn test_send_to_service_auto_binds_unicast() { // SendToService with no unicast sockets should auto-bind ephemeral - let (control_sender, _update_receiver) = Inner::::spawn( + let (control_sender, _update_receiver, run_fut) = Inner::::build( Ipv4Addr::LOCALHOST, Arc::new(Mutex::new(E2ERegistry::new())), false, ); + let _ = tokio::spawn(run_fut); let addr = SocketAddrV4::new(Ipv4Addr::LOCALHOST, 5000); let (rx, msg) = TestControl::add_endpoint(0x1234, 0x0001, addr, 0); @@ -1686,11 +1706,12 @@ mod tests { #[tokio::test] async fn test_subscribe_with_endpoint_sends_sd() { // Subscribe with a known endpoint and bound discovery should send the SD message - let (control_sender, _update_receiver) = Inner::::spawn( + let (control_sender, _update_receiver, run_fut) = Inner::::build( Ipv4Addr::LOCALHOST, Arc::new(Mutex::new(E2ERegistry::new())), false, ); + let _ = tokio::spawn(run_fut); // Bind discovery first let (rx, msg) = TestControl::bind_discovery(); @@ -1716,11 +1737,12 @@ mod tests { #[tokio::test] async fn test_subscribe_auto_binds_discovery() { // Subscribe without discovery bound should auto-bind and succeed - let (control_sender, _update_receiver) = Inner::::spawn( + let (control_sender, _update_receiver, run_fut) = Inner::::build( Ipv4Addr::LOCALHOST, Arc::new(Mutex::new(E2ERegistry::new())), false, ); + let _ = tokio::spawn(run_fut); // Add endpoint but do NOT bind discovery let addr = SocketAddrV4::new(Ipv4Addr::LOCALHOST, 5000); @@ -1740,11 +1762,12 @@ mod tests { #[tokio::test] async fn test_subscribe_unknown_service_returns_error() { - let (control_sender, _update_receiver) = Inner::::spawn( + let (control_sender, _update_receiver, run_fut) = Inner::::build( Ipv4Addr::LOCALHOST, Arc::new(Mutex::new(E2ERegistry::new())), false, ); + let _ = tokio::spawn(run_fut); let (rx, msg) = TestControl::subscribe(0xFFFF, 0xFFFF, 1, 3, 0x01, 0); control_sender.send(msg).await.unwrap(); @@ -1758,11 +1781,12 @@ mod tests { #[tokio::test] async fn test_send_to_service_reuses_existing_unicast_socket() { // When a unicast socket already exists, SendToService should reuse it - let (control_sender, _update_receiver) = Inner::::spawn( + let (control_sender, _update_receiver, run_fut) = Inner::::build( Ipv4Addr::LOCALHOST, Arc::new(Mutex::new(E2ERegistry::new())), false, ); + let _ = tokio::spawn(run_fut); let addr = SocketAddrV4::new(Ipv4Addr::LOCALHOST, 5000); let (rx, msg) = TestControl::add_endpoint(0x1234, 0x0001, addr, 0); @@ -1792,11 +1816,12 @@ mod tests { #[tokio::test] async fn test_dropped_receiver_subscribe_service_not_found_continues() { // Subscribe with no endpoint → ServiceNotFound response is dropped - let (control_sender, _update_receiver) = Inner::::spawn( + let (control_sender, _update_receiver, run_fut) = Inner::::build( Ipv4Addr::LOCALHOST, Arc::new(Mutex::new(E2ERegistry::new())), false, ); + let _ = tokio::spawn(run_fut); let (rx, msg) = TestControl::subscribe(0x1234, 0x0001, 1, 3, 0x01, 0); drop(rx); @@ -1809,11 +1834,12 @@ mod tests { #[tokio::test] async fn test_set_interface_changes_interface() { // SetInterface to a different address exercises the interface!=current path - let (control_sender, _update_receiver) = Inner::::spawn( + let (control_sender, _update_receiver, run_fut) = Inner::::build( Ipv4Addr::LOCALHOST, Arc::new(Mutex::new(E2ERegistry::new())), false, ); + let _ = tokio::spawn(run_fut); // Change to a different loopback-range address (127.0.0.2). // Binding discovery on 127.0.0.2 should succeed on most systems. @@ -1833,11 +1859,12 @@ mod tests { #[tokio::test] async fn test_set_interface_with_discovery_bound_changes_interface() { // SetInterface when discovery is already bound: unbind → change → rebind - let (control_sender, _update_receiver) = Inner::::spawn( + let (control_sender, _update_receiver, run_fut) = Inner::::build( Ipv4Addr::LOCALHOST, Arc::new(Mutex::new(E2ERegistry::new())), false, ); + let _ = tokio::spawn(run_fut); // Bind discovery on LOCALHOST first let (rx, msg) = TestControl::bind_discovery(); @@ -1863,11 +1890,12 @@ mod tests { async fn test_subscribe_specific_port_reuse() { // Subscribe twice with the same specific client_port exercises the // bind_unicast port-reuse path (port != 0 && already bound). - let (control_sender, _update_receiver) = Inner::::spawn( + let (control_sender, _update_receiver, run_fut) = Inner::::build( Ipv4Addr::LOCALHOST, Arc::new(Mutex::new(E2ERegistry::new())), false, ); + let _ = tokio::spawn(run_fut); // Add endpoint and bind discovery let (rx, msg) = TestControl::bind_discovery(); @@ -1909,11 +1937,12 @@ mod tests { use std::vec; use tokio::net::UdpSocket; - let (control_sender, _update_receiver) = Inner::::spawn( + let (control_sender, _update_receiver, run_fut) = Inner::::build( Ipv4Addr::LOCALHOST, Arc::new(Mutex::new(E2ERegistry::new())), false, ); + let _ = tokio::spawn(run_fut); let raw = UdpSocket::bind("127.0.0.1:0").await.unwrap(); let target = SocketAddrV4::new(Ipv4Addr::LOCALHOST, raw.local_addr().unwrap().port()); diff --git a/src/client/mod.rs b/src/client/mod.rs index 223ab5b9..88fb2bb0 100644 --- a/src/client/mod.rs +++ b/src/client/mod.rs @@ -62,15 +62,17 @@ impl

PendingResponse

{ /// /// # Errors /// - /// Returns the same errors as the request itself (e.g. deserialization failure). - /// - /// # Panics - /// - /// Panics if the inner loop dropped the response channel. + /// Returns the same errors as the request itself (e.g. deserialization + /// failure). Returns [`Error::Capacity`] with tag `"pending_responses"` + /// if the inner loop's response-tracking map was full when the request + /// was sent — the UDP send still went out, but the reply (if any) + /// arrives on [`ClientUpdates`] rather than this oneshot. + /// Returns [`Error::Shutdown`] only if the client's run-loop future + /// exits before the response is delivered — the caller's + /// `PendingResponse` handle outlived its driver. Reserving `Shutdown` + /// for actual lifecycle failure keeps `RecvError` unambiguous. pub async fn response(self) -> Result { - self.receiver - .await - .expect("inner loop dropped response channel") + self.receiver.await.map_err(|_| Error::Shutdown)? } } @@ -183,13 +185,38 @@ impl Client where MessageDefinitions: PayloadWireFormat + Clone + std::fmt::Debug + 'static, { - /// Creates a new client bound to the given network interface and spawns its event loop. - /// - /// Returns a `(Client, ClientUpdates)` pair. The `Client` handle is - /// [`Clone`]-able and can be shared across tasks. `ClientUpdates` receives - /// discovery, unicast, and error updates from the event loop. - #[must_use] - pub fn new(interface: Ipv4Addr) -> (Self, ClientUpdates) { + /// Creates a new client bound to the given network interface and returns its run-loop future to be driven by the caller. + /// + /// Returns a `(Client, ClientUpdates, run_future)` triple. The `Client` + /// handle is [`Clone`]-able and can be shared across tasks. + /// `ClientUpdates` receives discovery, unicast, and error updates from + /// the event loop. `run_future` is the event loop itself — the caller + /// must drive it to completion (typically via `tokio::spawn`) for the + /// client to process any messages. + /// + /// The future is bounded `Send + 'static` because every in-repo + /// consumer spawns it on a multithreaded executor. Bare-metal + /// consumers whose transport produces `!Send` state will get a + /// cfg-gated alternative constructor alongside the bare-metal port. + /// + /// ```no_run + /// # use simple_someip::{Client, RawPayload}; + /// # use std::net::Ipv4Addr; + /// # async fn demo() { + /// let (client, mut updates, run) = Client::::new(Ipv4Addr::LOCALHOST); + /// let _run_task = tokio::spawn(run); + /// // ...interact with `client` and `updates`... + /// # let _ = (client, updates); + /// # } + /// ``` + #[must_use = "the returned run-loop future must be spawned (e.g. tokio::spawn) for the client to make progress"] + pub fn new( + interface: Ipv4Addr, + ) -> ( + Self, + ClientUpdates, + impl core::future::Future + Send + 'static, + ) { Self::new_with_loopback(interface, false) } @@ -216,14 +243,18 @@ where /// Consumers of [`ClientUpdates`] that need to ignore self-sent SD should /// filter on source address (the sender's IP/port is included on the /// update). - #[must_use] + #[must_use = "the returned run-loop future must be spawned (e.g. tokio::spawn) for the client to make progress"] pub fn new_with_loopback( interface: Ipv4Addr, multicast_loopback: bool, - ) -> (Self, ClientUpdates) { + ) -> ( + Self, + ClientUpdates, + impl core::future::Future + Send + 'static, + ) { let e2e_registry = Arc::new(Mutex::new(E2ERegistry::new())); - let (control_sender, update_receiver) = - Inner::spawn(interface, Arc::clone(&e2e_registry), multicast_loopback); + let (control_sender, update_receiver, run_future) = + Inner::build(interface, Arc::clone(&e2e_registry), multicast_loopback); let client = Self { interface: Arc::new(RwLock::new(interface)), @@ -231,7 +262,7 @@ where e2e_registry, }; let updates = ClientUpdates { update_receiver }; - (client, updates) + (client, updates, run_future) } /// Returns the current network interface address. @@ -250,13 +281,21 @@ where /// /// Returns an error if rebinding sockets on the new interface fails. /// + /// Returns [`Error::Shutdown`] if the client's run-loop future has + /// exited before this call — the control-channel send cannot + /// complete without its receiver. + /// /// # Panics /// - /// Panics if the internal control channel or interface lock is poisoned/closed. + /// Panics if the interface lock is poisoned (indicates prior panic + /// while the lock was held). pub async fn set_interface(&self, interface: Ipv4Addr) -> Result<(), Error> { let (response, message) = ControlMessage::set_interface(interface); - self.control_sender.send(message).await.unwrap(); - response.await.unwrap()?; + self.control_sender + .send(message) + .await + .map_err(|_| Error::Shutdown)?; + response.await.map_err(|_| Error::Shutdown)??; *self.interface.write().expect("interface lock poisoned") = interface; Ok(()) } @@ -266,14 +305,17 @@ where /// # Errors /// /// Returns an error if binding the multicast socket fails. - /// - /// # Panics - /// - /// Panics if the internal control channel is closed. + /// Returns [`Error::Shutdown`] if the client's run-loop future has + /// exited before this call (dropped, cancelled, or otherwise gone) + /// — the `Client` handle has outlived its driver and further + /// control-channel sends cannot make progress. pub async fn bind_discovery(&self) -> Result<(), Error> { let (response, message) = ControlMessage::bind_discovery(); - self.control_sender.send(message).await.unwrap(); - response.await.unwrap() + self.control_sender + .send(message) + .await + .map_err(|_| Error::Shutdown)?; + response.await.map_err(|_| Error::Shutdown)? } /// Unbinds the SD multicast discovery socket. @@ -281,14 +323,17 @@ where /// # Errors /// /// Returns an error if unbinding the multicast socket fails. - /// - /// # Panics - /// - /// Panics if the internal control channel is closed. + /// Returns [`Error::Shutdown`] if the client's run-loop future has + /// exited before this call (dropped, cancelled, or otherwise gone) + /// — the `Client` handle has outlived its driver and further + /// control-channel sends cannot make progress. pub async fn unbind_discovery(&self) -> Result<(), Error> { let (response, message) = ControlMessage::unbind_discovery(); - self.control_sender.send(message).await.unwrap(); - response.await.unwrap() + self.control_sender + .send(message) + .await + .map_err(|_| Error::Shutdown)?; + response.await.map_err(|_| Error::Shutdown)? } /// Subscribes to an event group on a known service. @@ -296,10 +341,10 @@ where /// # Errors /// /// Returns an error if the service is not found or subscription fails. - /// - /// # Panics - /// - /// Panics if the internal control channel is closed. + /// Returns [`Error::Shutdown`] if the client's run-loop future has + /// exited before this call (dropped, cancelled, or otherwise gone) + /// — the `Client` handle has outlived its driver and further + /// control-channel sends cannot make progress. pub async fn subscribe( &self, service_id: u16, @@ -317,8 +362,11 @@ where event_group_id, client_port, ); - self.control_sender.send(message).await.unwrap(); - response.await.unwrap() + self.control_sender + .send(message) + .await + .map_err(|_| Error::Shutdown)?; + response.await.map_err(|_| Error::Shutdown)? } /// Like [`subscribe`](Self::subscribe) but does not wait for the @@ -400,18 +448,21 @@ where /// # Errors /// /// Returns an error if sending the SD message fails. - /// - /// # Panics - /// - /// Panics if the internal control channel is closed. + /// Returns [`Error::Shutdown`] if the client's run-loop future has + /// exited before this call (dropped, cancelled, or otherwise gone) + /// — the `Client` handle has outlived its driver and further + /// control-channel sends cannot make progress. pub async fn send_sd_message( &self, target: SocketAddrV4, sd_header: ::SdHeader, ) -> Result<(), Error> { let (response, message) = ControlMessage::send_sd(target, sd_header); - self.control_sender.send(message).await.unwrap(); - response.await.unwrap() + self.control_sender + .send(message) + .await + .map_err(|_| Error::Shutdown)?; + response.await.map_err(|_| Error::Shutdown)? } /// Start periodic SD announcements on the client's discovery socket. @@ -544,10 +595,10 @@ where /// # Errors /// /// Returns an error if registering the endpoint fails. - /// - /// # Panics - /// - /// Panics if the internal control channel is closed. + /// Returns [`Error::Shutdown`] if the client's run-loop future has + /// exited before this call (dropped, cancelled, or otherwise gone) + /// — the `Client` handle has outlived its driver and further + /// control-channel sends cannot make progress. pub async fn add_endpoint( &self, service_id: u16, @@ -557,8 +608,11 @@ where ) -> Result<(), Error> { let (response, message) = ControlMessage::add_endpoint(service_id, instance_id, addr, local_port); - self.control_sender.send(message).await.unwrap(); - response.await.unwrap() + self.control_sender + .send(message) + .await + .map_err(|_| Error::Shutdown)?; + response.await.map_err(|_| Error::Shutdown)? } /// Removes a service endpoint from the client's endpoint registry. @@ -566,14 +620,17 @@ where /// # Errors /// /// Returns an error if removing the endpoint fails. - /// - /// # Panics - /// - /// Panics if the internal control channel is closed. + /// Returns [`Error::Shutdown`] if the client's run-loop future has + /// exited before this call (dropped, cancelled, or otherwise gone) + /// — the `Client` handle has outlived its driver and further + /// control-channel sends cannot make progress. pub async fn remove_endpoint(&self, service_id: u16, instance_id: u16) -> Result<(), Error> { let (response, message) = ControlMessage::remove_endpoint(service_id, instance_id); - self.control_sender.send(message).await.unwrap(); - response.await.unwrap() + self.control_sender + .send(message) + .await + .map_err(|_| Error::Shutdown)?; + response.await.map_err(|_| Error::Shutdown)? } /// Sends a message to a service and returns a handle to await the response. @@ -596,10 +653,10 @@ where /// /// Returns an error if the service is not found, unicast binding fails, /// or the UDP send fails. - /// - /// # Panics - /// - /// Panics if the internal control channel is closed. + /// Returns [`Error::Shutdown`] if the client's run-loop future has + /// exited before this call (dropped, cancelled, or otherwise gone) + /// — the `Client` handle has outlived its driver and further + /// control-channel sends cannot make progress. pub async fn send_to_service( &self, service_id: u16, @@ -608,8 +665,11 @@ where ) -> Result, Error> { let (send_rx, response_rx, ctrl_msg) = ControlMessage::send_to_service(service_id, instance_id, message); - self.control_sender.send(ctrl_msg).await.unwrap(); - send_rx.await.unwrap()?; + self.control_sender + .send(ctrl_msg) + .await + .map_err(|_| Error::Shutdown)?; + send_rx.await.map_err(|_| Error::Shutdown)??; Ok(PendingResponse { receiver: response_rx, }) @@ -625,10 +685,15 @@ where /// /// Returns an error if the service is not found, unicast binding fails, /// the UDP send fails, or the response payload fails to deserialize. - /// - /// # Panics - /// - /// Panics if the internal control channel is closed. + /// Returns [`Error::Capacity`] with tag `"pending_responses"` if the + /// inner loop's response-tracking map was full when this request was + /// sent — the UDP send still went out, but the reply cannot be + /// routed back to this caller's oneshot (it arrives on + /// [`ClientUpdates`] instead). + /// Returns [`Error::Shutdown`] only if the client's run-loop future + /// has exited before this call (dropped, cancelled, or otherwise + /// gone) — the `Client` handle has outlived its driver and further + /// control-channel sends cannot make progress. pub async fn request( &self, service_id: u16, @@ -637,11 +702,12 @@ where ) -> Result { let (send_rx, response_rx, ctrl_msg) = ControlMessage::send_to_service(service_id, instance_id, message); - self.control_sender.send(ctrl_msg).await.unwrap(); - send_rx.await.unwrap()?; - response_rx + self.control_sender + .send(ctrl_msg) .await - .expect("inner loop dropped response channel") + .map_err(|_| Error::Shutdown)?; + send_rx.await.map_err(|_| Error::Shutdown)??; + response_rx.await.map_err(|_| Error::Shutdown)? } /// Register an E2E profile for the given key. @@ -693,14 +759,16 @@ mod tests { #[tokio::test] async fn test_client_new_and_interface() { - let (client, _updates) = TestClient::new(Ipv4Addr::LOCALHOST); + let (client, _updates, run_fut) = TestClient::new(Ipv4Addr::LOCALHOST); + let _ = tokio::spawn(run_fut); assert_eq!(client.interface(), Ipv4Addr::LOCALHOST); client.shut_down(); } #[tokio::test] async fn test_client_debug() { - let (client, _updates) = TestClient::new(Ipv4Addr::LOCALHOST); + let (client, _updates, run_fut) = TestClient::new(Ipv4Addr::LOCALHOST); + let _ = tokio::spawn(run_fut); let debug_str = format!("{client:?}"); assert!(debug_str.contains("Client")); assert!(debug_str.contains("127.0.0.1")); @@ -746,7 +814,8 @@ mod tests { #[tokio::test] async fn test_subscribe_unknown_service_returns_error() { - let (client, _updates) = TestClient::new(Ipv4Addr::LOCALHOST); + let (client, _updates, run_fut) = TestClient::new(Ipv4Addr::LOCALHOST); + let _ = tokio::spawn(run_fut); let result = client.subscribe(0xFFFF, 0xFFFF, 1, 3, 0x01, 0).await; assert!( matches!(result, Err(Error::ServiceNotFound)), @@ -757,7 +826,8 @@ mod tests { #[tokio::test] async fn test_subscribe_no_wait_unknown_service_does_not_panic() { - let (client, _updates) = TestClient::new(Ipv4Addr::LOCALHOST); + let (client, _updates, run_fut) = TestClient::new(Ipv4Addr::LOCALHOST); + let _ = tokio::spawn(run_fut); // subscribe_no_wait is fire-and-forget — it should not panic even // when the service is unknown (the inner loop sends ServiceNotFound // on the dropped response channel, which is harmless). @@ -769,7 +839,8 @@ mod tests { #[tokio::test] async fn test_bind_discovery_and_unbind() { - let (client, _updates) = TestClient::new(Ipv4Addr::LOCALHOST); + let (client, _updates, run_fut) = TestClient::new(Ipv4Addr::LOCALHOST); + let _ = tokio::spawn(run_fut); client.bind_discovery().await.unwrap(); client.unbind_discovery().await.unwrap(); client.shut_down(); @@ -777,7 +848,8 @@ mod tests { #[tokio::test] async fn test_set_interface() { - let (client, _updates) = TestClient::new(Ipv4Addr::LOCALHOST); + let (client, _updates, run_fut) = TestClient::new(Ipv4Addr::LOCALHOST); + let _ = tokio::spawn(run_fut); let new_addr = Ipv4Addr::LOCALHOST; client.set_interface(new_addr).await.unwrap(); assert_eq!(client.interface(), new_addr); @@ -786,7 +858,8 @@ mod tests { #[tokio::test] async fn test_add_endpoint_succeeds() { - let (client, _updates) = TestClient::new(Ipv4Addr::LOCALHOST); + let (client, _updates, run_fut) = TestClient::new(Ipv4Addr::LOCALHOST); + let _ = tokio::spawn(run_fut); let addr = SocketAddrV4::new(Ipv4Addr::new(192, 168, 1, 1), 30000); client.add_endpoint(0x1234, 0x0001, addr, 0).await.unwrap(); client.shut_down(); @@ -794,7 +867,8 @@ mod tests { #[tokio::test] async fn test_send_to_service_unknown_returns_error() { - let (client, _updates) = TestClient::new(Ipv4Addr::LOCALHOST); + let (client, _updates, run_fut) = TestClient::new(Ipv4Addr::LOCALHOST); + let _ = tokio::spawn(run_fut); let msg = crate::protocol::Message::new_sd(1, &empty_sd_header()); let result = client.send_to_service(0xFFFF, 0xFFFF, msg).await; assert!( @@ -806,7 +880,8 @@ mod tests { #[tokio::test] async fn test_remove_endpoint_succeeds() { - let (client, _updates) = TestClient::new(Ipv4Addr::LOCALHOST); + let (client, _updates, run_fut) = TestClient::new(Ipv4Addr::LOCALHOST); + let _ = tokio::spawn(run_fut); let addr = SocketAddrV4::new(Ipv4Addr::new(192, 168, 1, 1), 30000); client.add_endpoint(0x1234, 0x0001, addr, 0).await.unwrap(); client.remove_endpoint(0x1234, 0x0001).await.unwrap(); @@ -847,7 +922,8 @@ mod tests { #[tokio::test] async fn test_send_sd_message() { - let (client, _updates) = TestClient::new(Ipv4Addr::LOCALHOST); + let (client, _updates, run_fut) = TestClient::new(Ipv4Addr::LOCALHOST); + let _ = tokio::spawn(run_fut); // Bind discovery first so the send path uses the existing socket client.bind_discovery().await.unwrap(); let target = SocketAddrV4::new(Ipv4Addr::LOCALHOST, 30490); @@ -858,7 +934,8 @@ mod tests { #[tokio::test] async fn test_send_to_service_success_returns_pending_response() { - let (client, _updates) = TestClient::new(Ipv4Addr::LOCALHOST); + let (client, _updates, run_fut) = TestClient::new(Ipv4Addr::LOCALHOST); + let _ = tokio::spawn(run_fut); let addr = SocketAddrV4::new(Ipv4Addr::LOCALHOST, 30000); client.add_endpoint(0x1234, 0x0001, addr, 0).await.unwrap(); let msg = crate::protocol::Message::new_sd(1, &empty_sd_header()); @@ -870,7 +947,8 @@ mod tests { #[tokio::test] async fn test_recv_returns_none_after_shutdown() { - let (client, mut updates) = TestClient::new(Ipv4Addr::LOCALHOST); + let (client, mut updates, run_fut) = TestClient::new(Ipv4Addr::LOCALHOST); + let _ = tokio::spawn(run_fut); client.shut_down(); // Now the inner loop should exit; recv() should return None let result = tokio::time::timeout(std::time::Duration::from_secs(2), updates.recv()).await; @@ -880,7 +958,8 @@ mod tests { #[tokio::test] async fn test_register_and_unregister_e2e() { - let (client, _updates) = TestClient::new(Ipv4Addr::LOCALHOST); + let (client, _updates, run_fut) = TestClient::new(Ipv4Addr::LOCALHOST); + let _ = tokio::spawn(run_fut); let key = E2EKey { service_id: 0x1234, method_or_event_id: 0x0001, @@ -893,7 +972,8 @@ mod tests { #[tokio::test] async fn test_client_is_clone() { - let (client, _updates) = TestClient::new(Ipv4Addr::LOCALHOST); + let (client, _updates, run_fut) = TestClient::new(Ipv4Addr::LOCALHOST); + let _ = tokio::spawn(run_fut); let client2 = client.clone(); assert_eq!(client.interface(), client2.interface()); client.shut_down(); @@ -901,14 +981,16 @@ mod tests { #[tokio::test] async fn test_client_updates_debug() { - let (_client, updates) = TestClient::new(Ipv4Addr::LOCALHOST); + let (_client, updates, run_fut) = TestClient::new(Ipv4Addr::LOCALHOST); + let _ = tokio::spawn(run_fut); let debug_str = format!("{updates:?}"); assert!(debug_str.contains("ClientUpdates")); } #[tokio::test] async fn test_request_unknown_service_returns_error() { - let (client, _updates) = TestClient::new(Ipv4Addr::LOCALHOST); + let (client, _updates, run_fut) = TestClient::new(Ipv4Addr::LOCALHOST); + let _ = tokio::spawn(run_fut); let msg = crate::protocol::Message::new_sd(1, &empty_sd_header()); let result = client.request(0xFFFF, 0xFFFF, msg).await; assert!( @@ -920,7 +1002,8 @@ mod tests { #[tokio::test] async fn test_start_sd_announcements_does_not_panic() { - let (client, _updates) = TestClient::new(Ipv4Addr::LOCALHOST); + let (client, _updates, run_fut) = TestClient::new(Ipv4Addr::LOCALHOST); + let _ = tokio::spawn(run_fut); client.bind_discovery().await.unwrap(); let sd_header = empty_sd_header(); @@ -943,7 +1026,8 @@ mod tests { #[tokio::test] async fn test_start_sd_announcements_without_discovery_bound() { - let (client, _updates) = TestClient::new(Ipv4Addr::LOCALHOST); + let (client, _updates, run_fut) = TestClient::new(Ipv4Addr::LOCALHOST); + let _ = tokio::spawn(run_fut); // Don't bind discovery — the task should handle the error gracefully. let sd_header = empty_sd_header(); let handle = @@ -964,7 +1048,8 @@ mod tests { #[tokio::test] async fn test_start_sd_announcements_abort_stops_task() { - let (client, _updates) = TestClient::new(Ipv4Addr::LOCALHOST); + let (client, _updates, run_fut) = TestClient::new(Ipv4Addr::LOCALHOST); + let _ = tokio::spawn(run_fut); client.bind_discovery().await.unwrap(); let sd_header = empty_sd_header(); @@ -990,7 +1075,9 @@ mod tests { // session counter has not wrapped on a freshly-bound socket). This // verifies the announcer calls `set_reboot_flag` on each tick rather // than using the stale caller-supplied value. - let (client, mut updates) = TestClient::new_with_loopback(Ipv4Addr::LOCALHOST, true); + let (client, mut updates, run_fut) = + TestClient::new_with_loopback(Ipv4Addr::LOCALHOST, true); + tokio::spawn(run_fut); client.bind_discovery().await.unwrap(); // Caller bakes in Continuous — the announcer must override this. @@ -1039,7 +1126,8 @@ mod tests { // past 0xFFFF would regress to `RecentlyRebooted` on the next // `reboot_flag()` call after unbind — falsely advertising a reboot // to peers on the next manually-built SD header. - let (client, _updates) = TestClient::new(Ipv4Addr::LOCALHOST); + let (client, _updates, run_fut) = TestClient::new(Ipv4Addr::LOCALHOST); + tokio::spawn(run_fut); // No discovery bound. Fallback should reflect persisted state. // Default (unwrapped) → RecentlyRebooted. @@ -1072,7 +1160,8 @@ mod tests { #[tokio::test] async fn test_reboot_flag_defaults_to_recently_rebooted() { - let (client, _updates) = TestClient::new(Ipv4Addr::LOCALHOST); + let (client, _updates, run_fut) = TestClient::new(Ipv4Addr::LOCALHOST); + tokio::spawn(run_fut); // Discovery not bound — should fall back to RecentlyRebooted. assert_eq!( client.reboot_flag().await, @@ -1089,7 +1178,8 @@ mod tests { #[tokio::test] async fn test_start_sd_announcements_stops_on_shutdown() { - let (client, _updates) = TestClient::new(Ipv4Addr::LOCALHOST); + let (client, _updates, run_fut) = TestClient::new(Ipv4Addr::LOCALHOST); + let _ = tokio::spawn(run_fut); client.bind_discovery().await.unwrap(); let sd_header = empty_sd_header(); @@ -1109,4 +1199,59 @@ mod tests { "task should have exited cleanly, not panicked" ); } + + /// Documents the footgun: if the caller drops `run_fut` without ever + /// polling it, the control channel's receiver goes with it and + /// subsequent `Client` method calls return [`Error::Shutdown`] + /// rather than panicking. + /// + /// This is intrinsic to the caller-driven lifecycle introduced in + /// phase 6 — the run loop is no longer owned by `Client::new`, so + /// failing to spawn it is the caller's responsibility. The test + /// pins the behavior deterministically so that any attempt to + /// silently "fix" this (e.g. internal spawn fallback) would break + /// it and force a review. + /// + /// Prior to the phase-6 API change these call sites panicked on + /// `.unwrap()` of the send `Result`; the typed error surfaced here + /// lets library consumers observe lifecycle mismatches cleanly + /// instead of bringing down the caller's task. + #[tokio::test] + async fn dropping_run_future_without_spawn_returns_shutdown_error() { + let (client, _updates, run_fut) = TestClient::new(Ipv4Addr::LOCALHOST); + // Caller explicitly discards the run loop. + drop(run_fut); + let err = client + .bind_discovery() + .await + .expect_err("must surface a typed error, not Ok or panic"); + assert!( + matches!(err, Error::Shutdown), + "expected Error::Shutdown after run-loop drop, got {err:?}", + ); + } + + /// If the run loop is cancelled mid-poll (caller-initiated timeout, + /// graceful shutdown), subsequent `Client` calls see the control + /// channel closed and surface [`Error::Shutdown`]. Same structural + /// contract as dropping the run future. + #[tokio::test] + async fn cancelling_run_future_closes_control_channel_returns_shutdown_error() { + let (client, _updates, run_fut) = TestClient::new(Ipv4Addr::LOCALHOST); + let handle = tokio::spawn(run_fut); + // Let the loop start. + tokio::time::sleep(std::time::Duration::from_millis(50)).await; + handle.abort(); + // Give the abort time to land. + tokio::time::sleep(std::time::Duration::from_millis(50)).await; + + let err = client + .bind_discovery() + .await + .expect_err("must surface a typed error, not Ok or panic"); + assert!( + matches!(err, Error::Shutdown), + "expected Error::Shutdown after run-loop cancel, got {err:?}", + ); + } } diff --git a/src/lib.rs b/src/lib.rs index 7cc0529e..96c3d150 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -66,8 +66,12 @@ //! //! #[tokio::main] //! async fn main() { -//! // Client::new returns a Clone-able handle and an update stream. -//! let (client, mut updates) = Client::::new([192, 168, 1, 100].into()); +//! // Client::new returns a Clone-able handle, an update stream, and +//! // the run-loop future. Spawn the future on the tokio runtime; +//! // the returned future depends on `tokio::select!` / `tokio::time` +//! // / tokio sockets, so it is not executor-agnostic today. +//! let (client, mut updates, run) = Client::::new([192, 168, 1, 100].into()); +//! let _run_task = tokio::spawn(run); //! client.bind_discovery().await.unwrap(); //! //! while let Some(update) = updates.recv().await { diff --git a/src/server/README.md b/src/server/README.md index cb78f052..845905ac 100644 --- a/src/server/README.md +++ b/src/server/README.md @@ -55,8 +55,9 @@ async fn main() -> Result<(), Box> { // Create the server let mut server = Server::new(config).await?; - // Start announcing the service (sends OfferService every 1s) - server.start_announcing()?; + // Start announcing the service (sends OfferService every 1s). + // Spawn the announcement loop future on the Tokio runtime. + tokio::spawn(server.announcement_loop()?); // Get event publisher for sending events let publisher = server.publisher(); @@ -152,10 +153,10 @@ Configuration for a SOME/IP service provider: Main server struct: -- `new(config: ServerConfig) -> Result` - Create new server -- `start_announcing() -> Result<()>` - Start SD announcements +- `new(config: ServerConfig) -> Result` - Create new server +- `announcement_loop() -> Result + Send + 'static, Error>` - Build the SD announcement future; caller spawns on the Tokio runtime - `publisher() -> Arc` - Get event publisher -- `run() -> Result<()>` - Run event loop (handles subscriptions) +- `run() -> Result<(), Error>` - Run event loop (handles subscriptions) - `register_e2e(key, profile)` - Register E2E protection for a message key - `unregister_e2e(key)` - Remove E2E protection for a message key diff --git a/src/server/mod.rs b/src/server/mod.rs index b1293017..2d644aa9 100644 --- a/src/server/mod.rs +++ b/src/server/mod.rs @@ -82,7 +82,7 @@ pub struct Server { e2e_registry: Arc>, /// `true` if this server was constructed via [`Server::new_passive`]. /// Passive servers have no real SD socket bound to port 30490; their - /// SD handling is managed externally. Calling [`Self::start_announcing`] + /// SD handling is managed externally. Calling [`Self::announcement_loop`] /// or [`Self::run`] on a passive server is a programming error and /// returns an [`Error::Io`] with [`std::io::ErrorKind::InvalidInput`]. is_passive: bool, @@ -209,7 +209,7 @@ impl Server { /// incoming `SubscribeEventGroup` / `FindService` messages and routes /// them to the right `EventPublisher` via /// [`EventPublisher::register_subscriber`]). Do **not** call - /// [`Server::start_announcing`] or spawn [`Server::run`] on a passive + /// [`Server::announcement_loop`] or spawn [`Server::run`] on a passive /// server — the external dispatcher owns those responsibilities. /// /// # Errors @@ -229,7 +229,7 @@ impl Server { // Bind a placeholder SD socket on an ephemeral port. Nothing will // route to it (neither multicast nor unicast on 30490), and neither - // `start_announcing` nor `run` should be called for a passive + // `announcement_loop` nor `run` should be called for a passive // server. We still allocate it so the `Server` struct shape is // identical to the full-server path. let sd_placeholder_addr = std::net::SocketAddr::new(IpAddr::V4(config.interface), 0); @@ -263,9 +263,21 @@ impl Server { }) } - /// Start announcing the service via Service Discovery + /// Build the periodic-SD-announcement future. /// - /// This sends periodic `OfferService` messages to the SD multicast group + /// Returns a future that sends an `OfferService` message to the SD + /// multicast group every second. The caller must drive the future + /// (typically via `tokio::spawn`) for announcements to fire; this + /// function does no work on its own. + /// + /// ```no_run + /// # use simple_someip::server::Server; + /// # async fn demo(server: Server) -> Result<(), simple_someip::server::Error> { + /// let announce_fut = server.announcement_loop()?; + /// tokio::spawn(announce_fut); + /// # Ok(()) + /// # } + /// ``` /// /// # Errors /// @@ -273,15 +285,15 @@ impl Server { /// called on a server constructed via [`Server::new_passive`] — passive /// servers have no real SD socket bound to port 30490, so any /// announcements would go out with an incorrect source port. - /// - /// Otherwise currently always returns `Ok(())`; SD send failures are - /// logged internally. - pub fn start_announcing(&self) -> Result<(), Error> { + #[must_use = "the returned announcement-loop future must be spawned (e.g. tokio::spawn) or awaited for the server to emit SD announcements; dropping it silently disables announcements"] + pub fn announcement_loop( + &self, + ) -> Result + Send + 'static, Error> { if self.is_passive { return Err(Error::Io(std::io::Error::new( std::io::ErrorKind::InvalidInput, format!( - "start_announcing called on passive Server for service 0x{:04X}; \ + "announcement_loop called on passive Server for service 0x{:04X}; \ announcements must be driven externally (e.g. via \ `simple_someip::Client::start_sd_announcements`)", self.config.service_id @@ -292,7 +304,7 @@ impl Server { let sd_socket = Arc::clone(&self.sd_socket); let sd_state = Arc::clone(&self.sd_state); - tokio::spawn(async move { + Ok(async move { let mut announcement_count = 0u32; loop { match sd_state.send_offer_service(&config, &sd_socket).await { @@ -319,9 +331,7 @@ impl Server { // Send announcements every 1 second tokio::time::sleep(tokio::time::Duration::from_secs(1)).await; } - }); - - Ok(()) + }) } /// Send a unicast `OfferService` to a specific address (in response to `FindService`) @@ -1333,10 +1343,19 @@ mod tests { assert_eq!(entry.service_id(), 0x5B); assert_eq!(entry.instance_id(), 1); - // Also test that start_announcing doesn't error + // Also test that announcement_loop builds a future without error. drop(server); let (server2, _) = create_test_server(0x5B, 1).await; - assert!(server2.start_announcing().is_ok()); + let fut = server2 + .announcement_loop() + .expect("announcement_loop on a regular server must build"); + // Intentionally do not poll or spawn the future: we only care + // that constructing it returned Ok. If this future were + // spawned, the announcer would loop indefinitely and emit + // multicast until explicitly aborted or the Tokio runtime + // shut down at end-of-test, which could interfere with + // parallel tests using the same multicast group. + drop(fut); } #[tokio::test] @@ -1924,11 +1943,12 @@ mod tests { } #[tokio::test] - async fn start_announcing_on_passive_returns_invalid_input() { + async fn announcement_loop_on_passive_returns_invalid_input() { let server = make_passive_server(0x005C, 0x0001).await; let err = server - .start_announcing() - .expect_err("start_announcing on a passive server must fail"); + .announcement_loop() + .err() + .expect("announcement_loop on a passive server must fail"); match err { Error::Io(io_err) => { assert_eq!(io_err.kind(), std::io::ErrorKind::InvalidInput); @@ -1971,18 +1991,114 @@ mod tests { } #[tokio::test] - async fn start_announcing_on_regular_server_still_succeeds() { - // Regression guard: the new is_passive check must not break the + async fn announcement_loop_on_regular_server_still_succeeds() { + // Regression guard: the is_passive check must not break the // standard non-passive path. let (server, _port) = create_test_server(0x005C, 0x0001).await; - server - .start_announcing() - .expect("start_announcing on a regular server must still succeed"); - // The announcer task runs forever; the test succeeds as soon as - // start_announcing returns Ok. The spawned task is cleaned up - // when the Tokio test runtime shuts down at the end of this - // test — `tokio::spawn` tasks are not aborted by dropping - // unrelated handles, they ride the runtime lifecycle. + let fut = server + .announcement_loop() + .expect("announcement_loop on a regular server must build"); + // The announcer loops forever; the test succeeds as soon as + // construction returns Ok. + // Do not poll or spawn the future: doing so would leave the + // announcer running and emitting multicast for the rest of the + // test binary's lifetime, interfering with parallel tests that + // bind the same multicast group. We only care that construction + // returned Ok, so drop the future without polling it. + drop(fut); + } + + /// Direct test that `announcement_loop` actually emits an SD + /// announcement when driven. Explicit coverage for the primary entry + /// point (avoids regressions where only the deleted shim was exercised). + #[ignore = "requires MULTICAST on loopback; consistent with the \ + #[ignore]-gated sd_state.rs tests. Runs in any environment \ + where loopback multicast is available."] + #[tokio::test] + async fn announcement_loop_sends_offer_service_when_driven() { + use crate::protocol::MessageId; + + // Use service/instance IDs not used elsewhere in this test module + // so parallel tests joined to the same SD multicast group cannot + // produce false matches. + const SID: u16 = 0xAA01; + const IID: u16 = 0xFF01; + + // Bind a receiver on the SD multicast port with loopback so we + // actually see the outgoing announcement. Use a dedicated + // receiver socket via socket2 to match the SD bind pattern. + let iface = std::net::Ipv4Addr::LOCALHOST; + let recv = { + let s = socket2::Socket::new( + socket2::Domain::IPV4, + socket2::Type::DGRAM, + Some(socket2::Protocol::UDP), + ) + .unwrap(); + s.set_reuse_address(true).unwrap(); + #[cfg(unix)] + s.set_reuse_port(true).unwrap(); + s.bind(&std::net::SocketAddr::new(IpAddr::V4(iface), sd::MULTICAST_PORT).into()) + .unwrap(); + s.set_nonblocking(true).unwrap(); + let std_s: std::net::UdpSocket = s.into(); + let rs = tokio::net::UdpSocket::from_std(std_s).unwrap(); + rs.join_multicast_v4(sd::MULTICAST_IP, iface).unwrap(); + rs + }; + + let config = ServerConfig::new(iface, 30501, SID, IID); + let server = Server::new_with_loopback(config, true).await.unwrap(); + let fut = server.announcement_loop().expect("build loop"); + let handle = tokio::spawn(fut); + + // Filter out any stray SD traffic from other parallel tests + // until we see one whose OfferService entry carries OUR sid/iid. + // Bounded by a single outer timeout so a totally-silent server + // (the regression we actually care about) still fails the test. + let mut buf = [0u8; 1500]; + let offer_fields = tokio::time::timeout(std::time::Duration::from_secs(3), async { + loop { + let (n, _src) = recv.recv_from(&mut buf).await.expect("recv failed"); + let Ok(view) = crate::protocol::MessageView::parse(&buf[..n]) else { + continue; + }; + if view.header().message_id() != MessageId::SD { + continue; + } + let Ok(sd_view) = view.sd_header() else { + continue; + }; + let Some(entry) = sd_view.entries().next() else { + continue; + }; + if !matches!(entry.entry_type(), Ok(sd::EntryType::OfferService)) { + continue; + } + if entry.service_id() != SID || entry.instance_id() != IID { + continue; + } + break ( + entry.service_id(), + entry.instance_id(), + entry.major_version(), + entry.ttl(), + ); + } + }) + .await + .expect("timed out waiting for our OfferService"); + + let (svc, inst, major, ttl) = offer_fields; + assert_eq!(svc, SID, "emitted service_id must match server config"); + assert_eq!(inst, IID, "emitted instance_id must match server config"); + assert_eq!(major, 1, "default major_version from ServerConfig::new"); + assert!( + ttl > 0, + "OfferService TTL must be non-zero (TTL=0 means StopOffering)", + ); + + handle.abort(); } #[tokio::test] @@ -2164,9 +2280,10 @@ mod tests { let server = Server::new_with_loopback(config, true) .await .expect("server must bind with loopback enabled"); - server - .start_announcing() - .expect("start_announcing should succeed on a non-passive server"); + let announce_fut = server + .announcement_loop() + .expect("announcement_loop should build on a non-passive server"); + tokio::spawn(announce_fut); // Scan the multicast group for our OfferService. The first tick // happens immediately; 2s is ample headroom for scheduler jitter. diff --git a/tests/client_server.rs b/tests/client_server.rs index ffd6d349..9e0388c2 100644 --- a/tests/client_server.rs +++ b/tests/client_server.rs @@ -56,7 +56,8 @@ async fn test_client_server_subscribe_and_receive_event() { let server_handle = tokio::spawn(async move { server.run().await }); // Create client and subscribe to the server's event group - let (client, mut updates) = TestClient::new(Ipv4Addr::LOCALHOST); + let (client, mut updates, run_fut) = TestClient::new(Ipv4Addr::LOCALHOST); + let _ = tokio::spawn(run_fut); let server_addr = SocketAddrV4::new(Ipv4Addr::LOCALHOST, server_port); client.add_endpoint(0x5B, 1, server_addr, 0).await.unwrap(); client.subscribe(0x5B, 1, 1, 3, 0x01, 0).await.unwrap(); @@ -99,7 +100,8 @@ async fn test_client_send_sd_auto_binds_discovery() { let server_handle = tokio::spawn(async move { server.run().await }); // Create client — NO bind_discovery - let (client, _updates) = TestClient::new(Ipv4Addr::LOCALHOST); + let (client, _updates, run_fut) = TestClient::new(Ipv4Addr::LOCALHOST); + let _ = tokio::spawn(run_fut); // send_sd_message should auto-bind discovery and succeed let sd_header = VecSdHeader { @@ -130,7 +132,8 @@ async fn test_client_bind_unbind_lifecycle_with_server() { let (mut server, server_port) = create_server(0x5B, 1).await; let server_handle = tokio::spawn(async move { server.run().await }); - let (client, _updates) = TestClient::new(Ipv4Addr::LOCALHOST); + let (client, _updates, run_fut) = TestClient::new(Ipv4Addr::LOCALHOST); + let _ = tokio::spawn(run_fut); // Bind discovery, subscribe, then unbind and rebind client.bind_discovery().await.unwrap(); @@ -158,7 +161,8 @@ async fn test_add_endpoint_and_send_to_service() { let publisher = server.publisher(); let server_handle = tokio::spawn(async move { server.run().await }); - let (client, mut updates) = TestClient::new(Ipv4Addr::LOCALHOST); + let (client, mut updates, run_fut) = TestClient::new(Ipv4Addr::LOCALHOST); + let _ = tokio::spawn(run_fut); client.bind_discovery().await.unwrap(); // Register the server's endpoint manually (simulating non-broadcasting service) @@ -218,7 +222,8 @@ async fn test_subscribe_auto_binds_discovery() { let server_handle = tokio::spawn(async move { server.run().await }); // Create client — do NOT bind discovery manually - let (client, mut updates) = TestClient::new(Ipv4Addr::LOCALHOST); + let (client, mut updates, run_fut) = TestClient::new(Ipv4Addr::LOCALHOST); + let _ = tokio::spawn(run_fut); let server_addr = SocketAddrV4::new(Ipv4Addr::LOCALHOST, server_port); client.add_endpoint(0x5B, 1, server_addr, 0).await.unwrap(); // Subscribe should auto-bind discovery internally @@ -260,7 +265,8 @@ async fn test_client_request_resolves_via_unicast_reply() { let publisher = server.publisher(); let server_handle = tokio::spawn(async move { server.run().await }); - let (client, mut updates) = TestClient::new(Ipv4Addr::LOCALHOST); + let (client, mut updates, run_fut) = TestClient::new(Ipv4Addr::LOCALHOST); + let _ = tokio::spawn(run_fut); let server_addr = SocketAddrV4::new(Ipv4Addr::LOCALHOST, server_port); client.add_endpoint(0x5B, 1, server_addr, 0).await.unwrap(); client.subscribe(0x5B, 1, 1, 3, 0x01, 0).await.unwrap(); @@ -321,7 +327,8 @@ async fn test_e2e_protect_on_publish_and_check_on_receive() { let server_handle = tokio::spawn(async move { server.run().await }); - let (client, mut updates) = TestClient::new(Ipv4Addr::LOCALHOST); + let (client, mut updates, run_fut) = TestClient::new(Ipv4Addr::LOCALHOST); + let _ = tokio::spawn(run_fut); // Register matching E2E profile on client client.register_e2e(key, profile); @@ -385,12 +392,14 @@ async fn test_multiple_subscribers_receive_events() { let server_addr = SocketAddrV4::new(Ipv4Addr::LOCALHOST, server_port); // Client 1 - let (client1, mut updates1) = TestClient::new(Ipv4Addr::LOCALHOST); + let (client1, mut updates1, run_fut1) = TestClient::new(Ipv4Addr::LOCALHOST); + tokio::spawn(run_fut1); client1.add_endpoint(0x5B, 1, server_addr, 0).await.unwrap(); client1.subscribe(0x5B, 1, 1, 3, 0x01, 0).await.unwrap(); // Client 2 - let (client2, mut updates2) = TestClient::new(Ipv4Addr::LOCALHOST); + let (client2, mut updates2, run_fut2) = TestClient::new(Ipv4Addr::LOCALHOST); + tokio::spawn(run_fut2); client2.add_endpoint(0x5B, 1, server_addr, 0).await.unwrap(); client2.subscribe(0x5B, 1, 1, 3, 0x01, 0).await.unwrap(); @@ -443,7 +452,8 @@ async fn test_multiple_subscribers_receive_events() { /// Verify ClientUpdates returns None after client shutdown. #[tokio::test] async fn test_updates_drain_after_shutdown() { - let (client, mut updates) = TestClient::new(Ipv4Addr::LOCALHOST); + let (client, mut updates, run_fut) = TestClient::new(Ipv4Addr::LOCALHOST); + let _ = tokio::spawn(run_fut); client.shut_down(); let result = tokio::time::timeout(std::time::Duration::from_secs(2), updates.recv()) @@ -458,7 +468,8 @@ async fn test_cloned_client_works() { let (mut server, server_port) = create_server(0x5B, 1).await; let server_handle = tokio::spawn(async move { server.run().await }); - let (client, _updates) = TestClient::new(Ipv4Addr::LOCALHOST); + let (client, _updates, run_fut) = TestClient::new(Ipv4Addr::LOCALHOST); + let _ = tokio::spawn(run_fut); let client2 = client.clone(); // Both clones can send commands @@ -478,7 +489,8 @@ async fn test_subscribe_specific_port_reuse() { let (mut server, server_port) = create_server(0x5B, 1).await; let server_handle = tokio::spawn(async move { server.run().await }); - let (client, _updates) = TestClient::new(Ipv4Addr::LOCALHOST); + let (client, _updates, run_fut) = TestClient::new(Ipv4Addr::LOCALHOST); + let _ = tokio::spawn(run_fut); let server_addr = SocketAddrV4::new(Ipv4Addr::LOCALHOST, server_port); client.add_endpoint(0x5B, 1, server_addr, 0).await.unwrap();