simple-someip 0.5.3

mod error;
mod inner;
mod service_registry;
mod session;
mod socket_manager;

pub use error::Error;

use crate::e2e::{E2ECheckStatus, E2EKey, E2EProfile, E2ERegistry};
use crate::{protocol, protocol::Message, traits::PayloadWireFormat};
use inner::{ControlMessage, Inner};
use std::net::{Ipv4Addr, SocketAddr, SocketAddrV4};
use std::sync::{Arc, Mutex, RwLock};
use tokio::sync::{mpsc, oneshot};
use tracing::info;

/// Handle to a pending SOME/IP request-response transaction.
/// Resolves when the inner loop receives a matching unicast reply.
/// Does not borrow `Client`.
pub struct PendingResponse<P> {
    receiver: oneshot::Receiver<Result<P, Error>>,
}

impl<P> std::fmt::Debug for PendingResponse<P> {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("PendingResponse").finish_non_exhaustive()
    }
}

impl<P> PendingResponse<P> {
    /// Await the response payload.
    ///
    /// # Errors
    ///
    /// Returns the same errors as the request itself (e.g. deserialization failure).
    ///
    /// # Panics
    ///
    /// Panics if the inner loop dropped the response channel.
    pub async fn response(self) -> Result<P, Error> {
        self.receiver
            .await
            .expect("inner loop dropped response channel")
    }
}

/// A discovery message together with its source address and SOME/IP header.
pub struct DiscoveryMessage<P: PayloadWireFormat> {
    /// The network address this discovery message was received from.
    pub source: SocketAddr,
    /// The SOME/IP header (contains `request_id` = `client_id` + `session_id`).
    pub someip_header: protocol::Header,
    /// The parsed SD header payload.
    pub sd_header: P::SdHeader,
}

impl<P: PayloadWireFormat> std::fmt::Debug for DiscoveryMessage<P> {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("DiscoveryMessage")
            .field("source", &self.source)
            .field("someip_header", &self.someip_header)
            .field("sd_header", &self.sd_header)
            .finish()
    }
}

/// An update received from the SOME/IP client event loop.
pub enum ClientUpdate<P: PayloadWireFormat> {
    /// Discovery message received.
    DiscoveryUpdated(DiscoveryMessage<P>),
    /// A remote sender has rebooted (detected via SD session tracking).
    SenderRebooted(SocketAddr),
    /// Unicast message received.
    ///
    /// When E2E is configured for this message's key, `e2e_status` contains
    /// the check result and the payload has its E2E header stripped.
    /// When no E2E is configured, `e2e_status` is `None`.
    Unicast {
        /// The received SOME/IP message.
        message: Message<P>,
        /// E2E check status, if E2E was configured for this message.
        e2e_status: Option<E2ECheckStatus>,
    },
    /// The client encountered an error.
    Error(Error),
}

impl<P: PayloadWireFormat> std::fmt::Debug for ClientUpdate<P> {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::DiscoveryUpdated(msg) => f.debug_tuple("DiscoveryUpdated").field(msg).finish(),
            Self::SenderRebooted(addr) => f.debug_tuple("SenderRebooted").field(addr).finish(),
            Self::Unicast {
                message,
                e2e_status,
            } => f
                .debug_struct("Unicast")
                .field("message", message)
                .field("e2e_status", e2e_status)
                .finish(),
            Self::Error(err) => f.debug_tuple("Error").field(err).finish(),
        }
    }
}

/// Stream of updates from the SOME/IP client event loop.
///
/// Returned by [`Client::new`]. Call [`recv`](Self::recv) to receive
/// discovery, unicast, and error updates.
pub struct ClientUpdates<MessageDefinitions: PayloadWireFormat> {
    update_receiver: mpsc::UnboundedReceiver<ClientUpdate<MessageDefinitions>>,
}

impl<MessageDefinitions: PayloadWireFormat> std::fmt::Debug for ClientUpdates<MessageDefinitions> {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("ClientUpdates").finish_non_exhaustive()
    }
}

impl<MessageDefinitions: PayloadWireFormat> ClientUpdates<MessageDefinitions> {
    /// Waits for the next update from the client event loop.
    ///
    /// Returns `None` when the inner loop has exited (all `Client` handles
    /// dropped and the event loop finished draining).
    pub async fn recv(&mut self) -> Option<ClientUpdate<MessageDefinitions>> {
        self.update_receiver.recv().await
    }
}

/// A SOME/IP client that handles service discovery and message exchange.
///
/// `Client` is cheaply [`Clone`]-able. All clones share the same underlying
/// event loop and can be used concurrently from different tasks.
#[derive(Clone)]
pub struct Client<MessageDefinitions: PayloadWireFormat> {
    interface: Arc<RwLock<Ipv4Addr>>,
    control_sender: mpsc::Sender<inner::ControlMessage<MessageDefinitions>>,
    e2e_registry: Arc<Mutex<E2ERegistry>>,
}

impl<MessageDefinitions: PayloadWireFormat> std::fmt::Debug for Client<MessageDefinitions> {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("Client")
            .field(
                "interface",
                &*self.interface.read().expect("interface lock poisoned"),
            )
            .finish_non_exhaustive()
    }
}

impl<MessageDefinitions> Client<MessageDefinitions>
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<MessageDefinitions>) {
        let e2e_registry = Arc::new(Mutex::new(E2ERegistry::new()));
        let (control_sender, update_receiver) = Inner::spawn(interface, Arc::clone(&e2e_registry));

        let client = Self {
            interface: Arc::new(RwLock::new(interface)),
            control_sender,
            e2e_registry,
        };
        let updates = ClientUpdates { update_receiver };
        (client, updates)
    }

    /// Returns the current network interface address.
    ///
    /// # Panics
    ///
    /// Panics if the interface lock is poisoned.
    #[must_use]
    pub fn interface(&self) -> Ipv4Addr {
        *self.interface.read().expect("interface lock poisoned")
    }

    /// Changes the network interface and rebinds sockets.
    ///
    /// # Errors
    ///
    /// Returns an error if rebinding sockets on the new interface fails.
    ///
    /// # Panics
    ///
    /// Panics if the internal control channel or interface lock is poisoned/closed.
    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.interface.write().expect("interface lock poisoned") = interface;
        Ok(())
    }

    /// Binds the SD multicast discovery socket.
    ///
    /// # Errors
    ///
    /// Returns an error if binding the multicast socket fails.
    ///
    /// # Panics
    ///
    /// Panics if the internal control channel is closed.
    pub async fn bind_discovery(&self) -> Result<(), Error> {
        let (response, message) = ControlMessage::bind_discovery();
        self.control_sender.send(message).await.unwrap();
        response.await.unwrap()
    }

    /// Unbinds the SD multicast discovery socket.
    ///
    /// # Errors
    ///
    /// Returns an error if unbinding the multicast socket fails.
    ///
    /// # Panics
    ///
    /// Panics if the internal control channel is closed.
    pub async fn unbind_discovery(&self) -> Result<(), Error> {
        let (response, message) = ControlMessage::unbind_discovery();
        self.control_sender.send(message).await.unwrap();
        response.await.unwrap()
    }

    /// Subscribes to an event group on a known service.
    ///
    /// # Errors
    ///
    /// Returns an error if the service is not found or subscription fails.
    ///
    /// # Panics
    ///
    /// Panics if the internal control channel is closed.
    pub async fn subscribe(
        &self,
        service_id: u16,
        instance_id: u16,
        major_version: u8,
        ttl: u32,
        event_group_id: u16,
        client_port: u16,
    ) -> Result<(), Error> {
        let (response, message) = ControlMessage::subscribe(
            service_id,
            instance_id,
            major_version,
            ttl,
            event_group_id,
            client_port,
        );
        self.control_sender.send(message).await.unwrap();
        response.await.unwrap()
    }

    /// Like [`subscribe`](Self::subscribe) but does not wait for the
    /// subscription result.
    ///
    /// This still awaits enqueueing the control message on the internal
    /// channel, so it may block if that bounded channel is full. Useful for
    /// periodic renewals where waiting for subscription processing is
    /// unnecessary.
    pub async fn subscribe_no_wait(
        &self,
        service_id: u16,
        instance_id: u16,
        major_version: u8,
        ttl: u32,
        event_group_id: u16,
        client_port: u16,
    ) {
        let (response, message) = ControlMessage::subscribe(
            service_id,
            instance_id,
            major_version,
            ttl,
            event_group_id,
            client_port,
        );
        let _ = self.control_sender.send(message).await;
        // Consume the response in the background so the inner loop doesn't
        // warn about a dropped receiver.
        tokio::spawn(async move {
            let _ = response.await;
        });
    }

    /// Sends an SD message to a specific target address.
    ///
    /// # Errors
    ///
    /// Returns an error if sending the SD message fails.
    ///
    /// # Panics
    ///
    /// Panics if the internal control channel is closed.
    pub async fn send_sd_message(
        &self,
        target: SocketAddrV4,
        sd_header: <MessageDefinitions as PayloadWireFormat>::SdHeader,
    ) -> Result<(), Error> {
        let (response, message) = ControlMessage::send_sd(target, sd_header);
        self.control_sender.send(message).await.unwrap();
        response.await.unwrap()
    }

    /// Start periodic SD announcements on the client's discovery socket.
    ///
    /// Spawns a background task that sends the given SD header to the
    /// multicast group at a regular interval. Use this to bundle
    /// `FindService` + `OfferService` entries from a single SD identity
    /// when the application acts as both client and server.
    ///
    /// The announcements are sent via the client's SD socket, ensuring
    /// they share the same source address as the client's `Subscribe` and
    /// `FindService` messages.
    ///
    /// Returns a [`tokio::task::JoinHandle`] that can be used to abort the
    /// background task. The task uses a weak reference to the client's
    /// control channel, so it exits automatically when all `Client` handles
    /// are dropped (via `shut_down()` or going out of scope).
    ///
    /// # Arguments
    ///
    /// * `sd_header` — The SD header to send (entries + options).
    /// * `interval` — How often to send (e.g. every 1 second). Values below
    ///   100ms are clamped to 100ms to prevent tight loops.
    pub fn start_sd_announcements(
        &self,
        sd_header: <MessageDefinitions as PayloadWireFormat>::SdHeader,
        interval: std::time::Duration,
    ) -> tokio::task::JoinHandle<()>
    where
        <MessageDefinitions as PayloadWireFormat>::SdHeader: Send + 'static,
    {
        use crate::protocol::sd;

        // Use a WeakSender so this task does NOT keep the control channel
        // alive. When all strong Client handles are dropped (shut_down),
        // the weak sender will fail to upgrade and the task exits cleanly.
        let weak_sender = self.control_sender.downgrade();
        let target = SocketAddrV4::new(sd::MULTICAST_IP, sd::MULTICAST_PORT);
        let interval = interval.max(std::time::Duration::from_millis(100));

        tokio::spawn(async move {
            let mut tick = tokio::time::interval(interval);
            tick.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
            // Consume the immediate first tick so we don't send before
            // the caller has finished setting up (e.g. subscribing).
            tick.tick().await;
            let mut count = 0u64;
            loop {
                tick.tick().await;

                // Upgrade the weak sender — fails if the client has shut down.
                let Some(sender) = weak_sender.upgrade() else {
                    tracing::info!("Client shut down, stopping SD announcements");
                    break;
                };

                let (response, message) = ControlMessage::send_sd(target, sd_header.clone());

                let send_ok = sender.send(message).await.is_ok();
                // Drop the strong sender immediately so it doesn't keep
                // the channel alive across the response await.
                drop(sender);

                if !send_ok {
                    tracing::warn!("SD announcement channel closed, stopping");
                    break;
                }

                match response.await {
                    Ok(Ok(())) => {
                        count += 1;
                        if count == 1 {
                            tracing::info!("Sent first client SD announcement");
                        } else {
                            tracing::trace!("Sent {count} client SD announcements");
                        }
                    }
                    Ok(Err(e)) => {
                        tracing::error!("Failed to send SD announcement: {e:?}");
                    }
                    Err(_) => {
                        tracing::warn!("SD announcement response dropped, stopping");
                        break;
                    }
                }
            }
        })
    }

    /// Registers a service endpoint in the client's endpoint registry.
    ///
    /// `local_port` controls which source port is used when sending to this
    /// endpoint via [`send_to_service`](Self::send_to_service). Pass `0` to
    /// use an ephemeral (OS-assigned) port.
    ///
    /// Service-discovery (SD) automatically populates endpoints with
    /// `local_port = 0`. If your configuration requires a specific source
    /// port, you must call `add_endpoint` explicitly — even if SD has already
    /// registered the service — so that the correct `local_port` is stored.
    ///
    /// # Errors
    ///
    /// Returns an error if registering the endpoint fails.
    ///
    /// # Panics
    ///
    /// Panics if the internal control channel is closed.
    pub async fn add_endpoint(
        &self,
        service_id: u16,
        instance_id: u16,
        addr: SocketAddrV4,
        local_port: u16,
    ) -> 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()
    }

    /// Removes a service endpoint from the client's endpoint registry.
    ///
    /// # Errors
    ///
    /// Returns an error if removing the endpoint fails.
    ///
    /// # Panics
    ///
    /// Panics if the internal control channel is closed.
    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()
    }

    /// Sends a message to a service and returns a handle to await the response.
    ///
    /// Call `.response()` on the returned handle to await the reply payload.
    ///
    /// # Errors
    ///
    /// 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.
    pub async fn send_to_service(
        &self,
        service_id: u16,
        instance_id: u16,
        message: crate::protocol::Message<MessageDefinitions>,
    ) -> Result<PendingResponse<MessageDefinitions>, 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()?;
        Ok(PendingResponse {
            receiver: response_rx,
        })
    }

    /// Sends a request to a service and awaits the response in one call.
    ///
    /// Unlike [`send_to_service`](Self::send_to_service), this method does not
    /// require manually driving [`ClientUpdates::recv`] — the inner event loop
    /// resolves the response independently.
    ///
    /// # Errors
    ///
    /// 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.
    pub async fn request(
        &self,
        service_id: u16,
        instance_id: u16,
        message: crate::protocol::Message<MessageDefinitions>,
    ) -> Result<MessageDefinitions, 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()?;
        response_rx
            .await
            .expect("inner loop dropped response channel")
    }

    /// Register an E2E profile for the given key.
    ///
    /// Once registered, incoming messages matching `key` will have their E2E
    /// header checked and stripped, and outgoing messages will have E2E
    /// protection applied automatically.
    ///
    /// # Panics
    ///
    /// Panics if the E2E registry mutex is poisoned.
    pub fn register_e2e(&self, key: E2EKey, profile: E2EProfile) {
        self.e2e_registry
            .lock()
            .expect("e2e registry lock poisoned")
            .register(key, profile);
    }

    /// Remove E2E configuration for the given key.
    ///
    /// # Panics
    ///
    /// Panics if the E2E registry mutex is poisoned.
    pub fn unregister_e2e(&self, key: &E2EKey) {
        self.e2e_registry
            .lock()
            .expect("e2e registry lock poisoned")
            .unregister(key);
    }

    /// Shuts down the client by dropping the control channel.
    ///
    /// The inner event loop will exit once all `Client` clones are dropped.
    /// Remaining updates can be drained via [`ClientUpdates::recv`].
    pub fn shut_down(self) {
        drop(self.control_sender);
        info!("Shutting Down SOME/IP client");
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::protocol::sd::test_support::{TestPayload, empty_sd_header};
    use crate::traits::WireFormat;
    use std::format;

    type TestClient = Client<TestPayload>;

    #[tokio::test]
    async fn test_client_new_and_interface() {
        let (client, _updates) = TestClient::new(Ipv4Addr::LOCALHOST);
        assert_eq!(client.interface(), Ipv4Addr::LOCALHOST);
        client.shut_down();
    }

    #[tokio::test]
    async fn test_client_debug() {
        let (client, _updates) = TestClient::new(Ipv4Addr::LOCALHOST);
        let debug_str = format!("{client:?}");
        assert!(debug_str.contains("Client"));
        assert!(debug_str.contains("127.0.0.1"));
        client.shut_down();
    }

    #[tokio::test]
    async fn test_client_update_debug() {
        use std::net::SocketAddr;

        // DiscoveryUpdated
        let sd_header = empty_sd_header();
        let someip_header = crate::protocol::Header::new_sd(1, sd_header.required_size());
        let discovery_msg = DiscoveryMessage {
            source: SocketAddr::new(Ipv4Addr::LOCALHOST.into(), 30490),
            someip_header,
            sd_header,
        };
        let update: ClientUpdate<TestPayload> = ClientUpdate::DiscoveryUpdated(discovery_msg);
        let debug_str = format!("{update:?}");
        assert!(debug_str.contains("DiscoveryUpdated"));

        // SenderRebooted
        let update: ClientUpdate<TestPayload> =
            ClientUpdate::SenderRebooted(SocketAddr::new(Ipv4Addr::LOCALHOST.into(), 30490));
        let debug_str = format!("{update:?}");
        assert!(debug_str.contains("SenderRebooted"));

        // Unicast
        let msg = crate::protocol::Message::new_sd(1, &empty_sd_header());
        let update: ClientUpdate<TestPayload> = ClientUpdate::Unicast {
            message: msg,
            e2e_status: None,
        };
        let debug_str = format!("{update:?}");
        assert!(debug_str.contains("Unicast"));

        // Error
        let update: ClientUpdate<TestPayload> = ClientUpdate::Error(Error::ServiceNotFound);
        let debug_str = format!("{update:?}");
        assert!(debug_str.contains("Error"));
    }

    #[tokio::test]
    async fn test_subscribe_unknown_service_returns_error() {
        let (client, _updates) = TestClient::new(Ipv4Addr::LOCALHOST);
        let result = client.subscribe(0xFFFF, 0xFFFF, 1, 3, 0x01, 0).await;
        assert!(
            matches!(result, Err(Error::ServiceNotFound)),
            "expected ServiceNotFound, got {result:?}"
        );
        client.shut_down();
    }

    #[tokio::test]
    async fn test_subscribe_no_wait_unknown_service_does_not_panic() {
        let (client, _updates) = TestClient::new(Ipv4Addr::LOCALHOST);
        // 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).
        client
            .subscribe_no_wait(0xFFFF, 0xFFFF, 1, 3, 0x01, 0)
            .await;
        client.shut_down();
    }

    #[tokio::test]
    async fn test_bind_discovery_and_unbind() {
        let (client, _updates) = TestClient::new(Ipv4Addr::LOCALHOST);
        client.bind_discovery().await.unwrap();
        client.unbind_discovery().await.unwrap();
        client.shut_down();
    }

    #[tokio::test]
    async fn test_set_interface() {
        let (client, _updates) = TestClient::new(Ipv4Addr::LOCALHOST);
        let new_addr = Ipv4Addr::LOCALHOST;
        client.set_interface(new_addr).await.unwrap();
        assert_eq!(client.interface(), new_addr);
        client.shut_down();
    }

    #[tokio::test]
    async fn test_add_endpoint_succeeds() {
        let (client, _updates) = TestClient::new(Ipv4Addr::LOCALHOST);
        let addr = SocketAddrV4::new(Ipv4Addr::new(192, 168, 1, 1), 30000);
        client.add_endpoint(0x1234, 0x0001, addr, 0).await.unwrap();
        client.shut_down();
    }

    #[tokio::test]
    async fn test_send_to_service_unknown_returns_error() {
        let (client, _updates) = TestClient::new(Ipv4Addr::LOCALHOST);
        let msg = crate::protocol::Message::new_sd(1, &empty_sd_header());
        let result = client.send_to_service(0xFFFF, 0xFFFF, msg).await;
        assert!(
            matches!(result, Err(Error::ServiceNotFound)),
            "expected ServiceNotFound, got {result:?}"
        );
        client.shut_down();
    }

    #[tokio::test]
    async fn test_remove_endpoint_succeeds() {
        let (client, _updates) = TestClient::new(Ipv4Addr::LOCALHOST);
        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();
        client.shut_down();
    }

    #[test]
    fn test_pending_response_debug() {
        let (_tx, rx) = oneshot::channel::<Result<TestPayload, Error>>();
        let pending = PendingResponse { receiver: rx };
        let s = format!("{pending:?}");
        assert!(s.contains("PendingResponse"));
    }

    #[tokio::test]
    async fn test_pending_response_resolves_ok() {
        let (tx, rx) = oneshot::channel::<Result<TestPayload, Error>>();
        let pending = PendingResponse { receiver: rx };
        let payload = TestPayload {
            header: empty_sd_header(),
        };
        tx.send(Ok(payload.clone())).unwrap();
        let result = pending.response().await;
        assert_eq!(result.unwrap(), payload);
    }

    #[tokio::test]
    async fn test_pending_response_resolves_err() {
        let (tx, rx) = oneshot::channel::<Result<TestPayload, Error>>();
        let pending = PendingResponse { receiver: rx };
        tx.send(Err(Error::ServiceNotFound)).unwrap();
        let result = pending.response().await;
        assert!(
            matches!(result, Err(Error::ServiceNotFound)),
            "expected ServiceNotFound, got {result:?}"
        );
    }

    #[tokio::test]
    async fn test_send_sd_message() {
        let (client, _updates) = TestClient::new(Ipv4Addr::LOCALHOST);
        // Bind discovery first so the send path uses the existing socket
        client.bind_discovery().await.unwrap();
        let target = SocketAddrV4::new(Ipv4Addr::LOCALHOST, 30490);
        let sd_header = empty_sd_header();
        client.send_sd_message(target, sd_header).await.unwrap();
        client.shut_down();
    }

    #[tokio::test]
    async fn test_send_to_service_success_returns_pending_response() {
        let (client, _updates) = TestClient::new(Ipv4Addr::LOCALHOST);
        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());
        // send_to_service succeeds (send completes), returning a PendingResponse
        let pending = client.send_to_service(0x1234, 0x0001, msg).await;
        assert!(pending.is_ok());
        client.shut_down();
    }

    #[tokio::test]
    async fn test_recv_returns_none_after_shutdown() {
        let (client, mut updates) = TestClient::new(Ipv4Addr::LOCALHOST);
        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;
        assert!(result.is_ok());
        assert!(result.unwrap().is_none());
    }

    #[tokio::test]
    async fn test_register_and_unregister_e2e() {
        let (client, _updates) = TestClient::new(Ipv4Addr::LOCALHOST);
        let key = E2EKey {
            service_id: 0x1234,
            method_or_event_id: 0x0001,
        };
        let profile = E2EProfile::Profile4(crate::e2e::Profile4Config::new(42, 10));
        client.register_e2e(key, profile);
        client.unregister_e2e(&key);
        client.shut_down();
    }

    #[tokio::test]
    async fn test_client_is_clone() {
        let (client, _updates) = TestClient::new(Ipv4Addr::LOCALHOST);
        let client2 = client.clone();
        assert_eq!(client.interface(), client2.interface());
        client.shut_down();
    }

    #[tokio::test]
    async fn test_client_updates_debug() {
        let (_client, updates) = TestClient::new(Ipv4Addr::LOCALHOST);
        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 msg = crate::protocol::Message::new_sd(1, &empty_sd_header());
        let result = client.request(0xFFFF, 0xFFFF, msg).await;
        assert!(
            matches!(result, Err(Error::ServiceNotFound)),
            "expected ServiceNotFound, got {result:?}"
        );
        client.shut_down();
    }

    #[tokio::test]
    async fn test_start_sd_announcements_does_not_panic() {
        let (client, _updates) = TestClient::new(Ipv4Addr::LOCALHOST);
        client.bind_discovery().await.unwrap();

        let sd_header = empty_sd_header();
        let handle =
            client.start_sd_announcements(sd_header, std::time::Duration::from_millis(100));

        // Let the task fire at least once (may fail to send on loopback, that's OK).
        tokio::time::sleep(std::time::Duration::from_millis(250)).await;

        handle.abort();
        let result = handle.await;
        let err = result.unwrap_err();
        assert!(
            err.is_cancelled(),
            "task should have been cancelled, not panicked"
        );

        client.shut_down();
    }

    #[tokio::test]
    async fn test_start_sd_announcements_without_discovery_bound() {
        let (client, _updates) = TestClient::new(Ipv4Addr::LOCALHOST);
        // Don't bind discovery — the task should handle the error gracefully.
        let sd_header = empty_sd_header();
        let handle =
            client.start_sd_announcements(sd_header, std::time::Duration::from_millis(100));

        tokio::time::sleep(std::time::Duration::from_millis(250)).await;

        handle.abort();
        let result = handle.await;
        let err = result.unwrap_err();
        assert!(
            err.is_cancelled(),
            "task should have been cancelled, not panicked"
        );

        client.shut_down();
    }

    #[tokio::test]
    async fn test_start_sd_announcements_abort_stops_task() {
        let (client, _updates) = TestClient::new(Ipv4Addr::LOCALHOST);
        client.bind_discovery().await.unwrap();

        let sd_header = empty_sd_header();
        let handle =
            client.start_sd_announcements(sd_header, std::time::Duration::from_millis(100));

        handle.abort();
        let result = handle.await;
        let err = result.unwrap_err();
        assert!(
            err.is_cancelled(),
            "task should have been cancelled, not panicked"
        );

        client.shut_down();
    }

    #[tokio::test]
    async fn test_start_sd_announcements_stops_on_shutdown() {
        let (client, _updates) = TestClient::new(Ipv4Addr::LOCALHOST);
        client.bind_discovery().await.unwrap();

        let sd_header = empty_sd_header();
        let handle =
            client.start_sd_announcements(sd_header, std::time::Duration::from_millis(100));

        // Shut down the client — the weak sender should fail to upgrade
        // and the task should exit cleanly without needing abort().
        client.shut_down();

        let join_result = tokio::time::timeout(std::time::Duration::from_secs(2), handle)
            .await
            .expect("task should have exited within timeout");
        // Verify clean exit — not a panic
        assert!(
            join_result.is_ok() || join_result.as_ref().unwrap_err().is_cancelled(),
            "task should have exited cleanly, not panicked"
        );
    }
}