iridium_stomp/

connection.rs

use futures::{SinkExt, StreamExt, future};
use std::collections::{HashMap, VecDeque};
use std::sync::Arc;
use std::sync::atomic::{AtomicU64, Ordering};
use std::time::Duration;
use thiserror::Error;
use tokio::net::TcpStream;
use tokio::sync::{Mutex, broadcast, mpsc, oneshot};
use tokio_util::codec::Framed;

use crate::codec::{StompCodec, StompItem};
use crate::frame::Frame;

/// Configuration for STOMP heartbeat intervals.
///
/// Provides a type-safe way to configure heartbeat values instead of using
/// raw strings. The `Display` implementation formats the value as required
/// by the STOMP protocol ("send_ms,receive_ms").
///
/// # Example
///
/// ```
/// use iridium_stomp::Heartbeat;
///
/// // Create a custom heartbeat configuration
/// let hb = Heartbeat::new(5000, 10000);
/// assert_eq!(hb.to_string(), "5000,10000");
///
/// // Use predefined configurations
/// assert_eq!(Heartbeat::disabled().to_string(), "0,0");
/// assert_eq!(Heartbeat::default().to_string(), "10000,10000");
/// ```
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct Heartbeat {
    /// Minimum interval (in milliseconds) between heartbeats the client can send.
    /// A value of 0 means the client cannot send heartbeats.
    pub send_ms: u32,

    /// Minimum interval (in milliseconds) between heartbeats the client wants to receive.
    /// A value of 0 means the client does not want to receive heartbeats.
    pub receive_ms: u32,
}

impl Heartbeat {
    /// Create a new heartbeat configuration with the specified intervals.
    ///
    /// # Arguments
    ///
    /// * `send_ms` - Minimum interval in milliseconds between heartbeats the client can send.
    /// * `receive_ms` - Minimum interval in milliseconds between heartbeats the client wants to receive.
    ///
    /// # Example
    ///
    /// ```
    /// use iridium_stomp::Heartbeat;
    ///
    /// let hb = Heartbeat::new(5000, 10000);
    /// assert_eq!(hb.send_ms, 5000);
    /// assert_eq!(hb.receive_ms, 10000);
    /// ```
    pub fn new(send_ms: u32, receive_ms: u32) -> Self {
        Self {
            send_ms,
            receive_ms,
        }
    }

    /// Create a heartbeat configuration that disables heartbeats entirely.
    ///
    /// This is equivalent to `Heartbeat::new(0, 0)`.
    ///
    /// # Example
    ///
    /// ```
    /// use iridium_stomp::Heartbeat;
    ///
    /// let hb = Heartbeat::disabled();
    /// assert_eq!(hb.send_ms, 0);
    /// assert_eq!(hb.receive_ms, 0);
    /// assert_eq!(hb.to_string(), "0,0");
    /// ```
    pub fn disabled() -> Self {
        Self::new(0, 0)
    }

    /// Create a heartbeat configuration from a Duration for symmetric heartbeats.
    ///
    /// Both send and receive intervals will be set to the same value.
    ///
    /// The maximum supported Duration is approximately 49.7 days (u32::MAX milliseconds,
    /// or 4,294,967,295 ms). If a larger Duration is provided, it will be clamped to
    /// u32::MAX milliseconds to prevent overflow.
    ///
    /// # Example
    ///
    /// ```
    /// use iridium_stomp::Heartbeat;
    /// use std::time::Duration;
    ///
    /// let hb = Heartbeat::from_duration(Duration::from_secs(15));
    /// assert_eq!(hb.send_ms, 15000);
    /// assert_eq!(hb.receive_ms, 15000);
    /// ```
    pub fn from_duration(interval: Duration) -> Self {
        let ms = interval.as_millis().min(u32::MAX as u128) as u32;
        Self::new(ms, ms)
    }
}

impl Default for Heartbeat {
    /// Returns the default heartbeat configuration: 10 seconds for both send and receive.
    fn default() -> Self {
        Self::new(10000, 10000)
    }
}

impl std::fmt::Display for Heartbeat {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "{},{}", self.send_ms, self.receive_ms)
    }
}

/// Internal subscription entry stored for each destination.
#[derive(Clone)]
pub(crate) struct SubscriptionEntry {
    pub(crate) id: String,
    pub(crate) sender: mpsc::Sender<Frame>,
    pub(crate) ack: String,
    pub(crate) headers: Vec<(String, String)>,
}

/// Alias for the subscription dispatch map: destination -> list of
/// `SubscriptionEntry`.
pub(crate) type Subscriptions = HashMap<String, Vec<SubscriptionEntry>>;

/// Alias for the pending map: subscription_id -> queue of (message-id, Frame).
pub(crate) type PendingMap = HashMap<String, VecDeque<(String, Frame)>>;

/// Internal type for resubscribe snapshot entries: (destination, id, ack, headers)
pub(crate) type ResubEntry = (String, String, String, Vec<(String, String)>);

/// Alias for pending receipt map: receipt-id -> oneshot sender to notify when received.
pub(crate) type PendingReceipts = HashMap<String, oneshot::Sender<()>>;

/// Errors returned by `Connection` operations.
#[derive(Error, Debug)]
pub enum ConnError {
    /// I/O-level error
    #[error("io error: {0}")]
    Io(#[from] std::io::Error),
    /// Protocol-level error
    #[error("protocol error: {0}")]
    Protocol(String),
    /// Receipt timeout error
    #[error("receipt timeout: no RECEIPT received for '{0}' within timeout")]
    ReceiptTimeout(String),
    /// Server rejected the connection (e.g., authentication failure)
    ///
    /// This error is returned when the server sends an ERROR frame in response
    /// to the CONNECT frame. Common causes include invalid credentials,
    /// unauthorized access, or broker configuration issues.
    #[error("server rejected connection: {0}")]
    ServerRejected(ServerError),
}

/// Represents an ERROR frame received from the STOMP server.
///
/// STOMP servers send ERROR frames to indicate protocol violations, authentication
/// failures, or other server-side errors. After sending an ERROR frame, the server
/// typically closes the connection.
///
/// # Example
///
/// ```ignore
/// use iridium_stomp::ReceivedFrame;
///
/// while let Some(received) = conn.next_frame().await {
///     match received {
///         ReceivedFrame::Frame(frame) => {
///             // Normal message processing
///         }
///         ReceivedFrame::Error(err) => {
///             eprintln!("Server error: {}", err.message);
///             if let Some(body) = &err.body {
///                 eprintln!("Details: {}", body);
///             }
///             break;
///         }
///     }
/// }
/// ```
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ServerError {
    /// The error message from the `message` header.
    pub message: String,

    /// The error body, if present. Contains additional error details.
    pub body: Option<String>,

    /// The receipt-id if this error is in response to a specific frame.
    pub receipt_id: Option<String>,

    /// The original ERROR frame for access to additional headers.
    pub frame: Frame,
}

impl ServerError {
    /// Create a `ServerError` from an ERROR frame.
    ///
    /// This is primarily used internally but is public for testing and
    /// advanced use cases where you need to construct a `ServerError` manually.
    pub fn from_frame(frame: Frame) -> Self {
        let message = frame
            .get_header("message")
            .unwrap_or("unknown error")
            .to_string();

        let body = if frame.body.is_empty() {
            None
        } else {
            String::from_utf8(frame.body.clone()).ok()
        };

        let receipt_id = frame.get_header("receipt-id").map(|s| s.to_string());

        Self {
            message,
            body,
            receipt_id,
            frame,
        }
    }
}

impl std::fmt::Display for ServerError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "STOMP server error: {}", self.message)?;
        if let Some(body) = &self.body {
            write!(f, " - {}", body)?;
        }
        Ok(())
    }
}

impl std::error::Error for ServerError {}

/// The result of receiving a frame from the server.
///
/// STOMP servers can send either normal frames (MESSAGE, RECEIPT, etc.) or
/// ERROR frames indicating a problem. This enum allows callers to handle
/// both cases with pattern matching.
///
/// # Example
///
/// ```ignore
/// use iridium_stomp::ReceivedFrame;
///
/// match conn.next_frame().await {
///     Some(ReceivedFrame::Frame(frame)) => {
///         println!("Got frame: {}", frame.command);
///     }
///     Some(ReceivedFrame::Error(err)) => {
///         eprintln!("Server error: {}", err);
///     }
///     None => {
///         println!("Connection closed");
///     }
/// }
/// ```
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum ReceivedFrame {
    /// A normal STOMP frame (MESSAGE, RECEIPT, etc.)
    Frame(Frame),
    /// An ERROR frame from the server
    Error(ServerError),
}

impl ReceivedFrame {
    /// Returns `true` if this is an error frame.
    pub fn is_error(&self) -> bool {
        matches!(self, ReceivedFrame::Error(_))
    }

    /// Returns `true` if this is a normal frame.
    pub fn is_frame(&self) -> bool {
        matches!(self, ReceivedFrame::Frame(_))
    }

    /// Returns the frame if this is a normal frame, or `None` if it's an error.
    pub fn into_frame(self) -> Option<Frame> {
        match self {
            ReceivedFrame::Frame(f) => Some(f),
            ReceivedFrame::Error(_) => None,
        }
    }

    /// Returns the error if this is an error frame, or `None` if it's a normal frame.
    pub fn into_error(self) -> Option<ServerError> {
        match self {
            ReceivedFrame::Frame(_) => None,
            ReceivedFrame::Error(e) => Some(e),
        }
    }
}

/// Subscription acknowledgement modes as defined by STOMP 1.2.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum AckMode {
    Auto,
    Client,
    ClientIndividual,
}

impl AckMode {
    fn as_str(&self) -> &'static str {
        match self {
            AckMode::Auto => "auto",
            AckMode::Client => "client",
            AckMode::ClientIndividual => "client-individual",
        }
    }
}

/// Options for customizing the STOMP CONNECT frame.
///
/// Use this struct with `Connection::connect_with_options()` to set custom
/// headers, specify supported STOMP versions, or configure broker-specific
/// options like `client-id` for durable subscriptions.
///
/// # Validation
///
/// This struct performs minimal validation. Values are passed to the broker
/// as-is, and invalid configurations will be rejected by the broker at
/// connection time. Empty strings are technically accepted but may cause
/// broker-specific errors.
///
/// # Custom Headers
///
/// Custom headers added via `header()` cannot override critical STOMP headers
/// (`accept-version`, `host`, `login`, `passcode`, `heart-beat`, `client-id`).
/// Such headers are silently ignored. Use the dedicated builder methods to
/// set these values.
///
/// # Example
///
/// ```ignore
/// use iridium_stomp::{Connection, ConnectOptions};
///
/// let options = ConnectOptions::default()
///     .client_id("my-durable-client")
///     .host("my-vhost")
///     .header("custom-header", "value");
///
/// let conn = Connection::connect_with_options(
///     "localhost:61613",
///     "guest",
///     "guest",
///     "10000,10000",
///     options,
/// ).await?;
/// ```
#[derive(Clone, Default)]
pub struct ConnectOptions {
    /// STOMP version(s) to accept (e.g., "1.2" or "1.0,1.1,1.2").
    /// Defaults to "1.2" if not set.
    pub accept_version: Option<String>,

    /// Client ID for durable subscriptions (required by ActiveMQ, etc.).
    pub client_id: Option<String>,

    /// Virtual host header value. Defaults to "/" if not set.
    pub host: Option<String>,

    /// Additional custom headers to include in the CONNECT frame.
    /// Note: Headers that would override critical STOMP headers are ignored.
    pub headers: Vec<(String, String)>,

    /// Optional channel to receive heartbeat notifications.
    /// When set, the connection will send a `()` on this channel each time
    /// a heartbeat is received from the server.
    pub heartbeat_tx: Option<mpsc::Sender<()>>,
}

impl std::fmt::Debug for ConnectOptions {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("ConnectOptions")
            .field("accept_version", &self.accept_version)
            .field("client_id", &self.client_id)
            .field("host", &self.host)
            .field("headers", &self.headers)
            .field(
                "heartbeat_tx",
                &self.heartbeat_tx.as_ref().map(|_| "Some(...)"),
            )
            .finish()
    }
}

impl ConnectOptions {
    /// Create a new `ConnectOptions` with default values.
    pub fn new() -> Self {
        Self::default()
    }

    /// Set the STOMP version(s) to accept (builder style).
    ///
    /// Examples: "1.2", "1.1,1.2", "1.0,1.1,1.2"
    pub fn accept_version(mut self, version: impl Into<String>) -> Self {
        self.accept_version = Some(version.into());
        self
    }

    /// Set the client ID for durable subscriptions (builder style).
    ///
    /// Required by some brokers (e.g., ActiveMQ) for durable topic subscriptions.
    pub fn client_id(mut self, id: impl Into<String>) -> Self {
        self.client_id = Some(id.into());
        self
    }

    /// Set the virtual host (builder style).
    ///
    /// Defaults to "/" if not set.
    pub fn host(mut self, host: impl Into<String>) -> Self {
        self.host = Some(host.into());
        self
    }

    /// Add a custom header to the CONNECT frame (builder style).
    pub fn header(mut self, key: impl Into<String>, value: impl Into<String>) -> Self {
        self.headers.push((key.into(), value.into()));
        self
    }

    /// Set a channel to receive heartbeat notifications (builder style).
    ///
    /// When set, the connection will send a `()` on this channel each time
    /// a heartbeat is received from the server. This is useful for CLI tools
    /// or monitoring applications that want to display heartbeat status.
    ///
    /// # Note
    ///
    /// Notifications are sent using `try_send()` to avoid blocking the
    /// connection's background task. If the channel buffer is full,
    /// notifications will be silently dropped. Use a sufficiently sized
    /// channel buffer (e.g., 16) to avoid missing notifications.
    ///
    /// # Example
    ///
    /// ```ignore
    /// use tokio::sync::mpsc;
    /// use iridium_stomp::ConnectOptions;
    ///
    /// let (tx, mut rx) = mpsc::channel(16);
    /// let options = ConnectOptions::default()
    ///     .with_heartbeat_notify(tx);
    ///
    /// // In another task:
    /// while rx.recv().await.is_some() {
    ///     println!("Heartbeat received!");
    /// }
    /// ```
    pub fn with_heartbeat_notify(mut self, tx: mpsc::Sender<()>) -> Self {
        self.heartbeat_tx = Some(tx);
        self
    }
}

/// Parse the STOMP `heart-beat` header value (format: "cx,cy").
///
/// Parameters
/// - `header`: header string from the server or client (for example
///   "10000,10000"). The values represent milliseconds.
///
/// Returns a tuple `(cx, cy)` where each value is the heartbeat interval in
/// milliseconds. Missing or invalid fields default to `0`.
pub fn parse_heartbeat_header(header: &str) -> (u64, u64) {
    let mut parts = header.split(',');
    let cx = parts
        .next()
        .and_then(|s| s.trim().parse::<u64>().ok())
        .unwrap_or(0);
    let cy = parts
        .next()
        .and_then(|s| s.trim().parse::<u64>().ok())
        .unwrap_or(0);
    (cx, cy)
}

/// Negotiate heartbeat intervals between client and server.
///
/// Parameters
/// - `client_out`: client's desired outgoing heartbeat interval in
///   milliseconds (how often the client will send heartbeats).
/// - `client_in`: client's desired incoming heartbeat interval in
///   milliseconds (how often the client expects to receive heartbeats).
/// - `server_out`: server's advertised outgoing interval in milliseconds.
/// - `server_in`: server's advertised incoming interval in milliseconds.
///
/// Returns `(outgoing, incoming)` where each element is `Some(Duration)` if
/// heartbeats are enabled in that direction, or `None` if disabled. The
/// negotiated interval uses the STOMP rule of taking the maximum of the
/// corresponding client and server values.
pub fn negotiate_heartbeats(
    client_out: u64,
    client_in: u64,
    server_out: u64,
    server_in: u64,
) -> (Option<Duration>, Option<Duration>) {
    let negotiated_out_ms = std::cmp::max(client_out, server_in);
    let negotiated_in_ms = std::cmp::max(client_in, server_out);

    let outgoing = if negotiated_out_ms == 0 {
        None
    } else {
        Some(Duration::from_millis(negotiated_out_ms))
    };
    let incoming = if negotiated_in_ms == 0 {
        None
    } else {
        Some(Duration::from_millis(negotiated_in_ms))
    };
    (outgoing, incoming)
}

/// Extract the destination from an ERROR frame.
///
/// Tries multiple strategies:
/// 1. Check for a `destination` header (some brokers include it)
/// 2. Parse the error message/body for `/topic/...` or `/queue/...` patterns
///
/// Returns `None` if no destination can be identified.
fn extract_destination_from_error(frame: &Frame) -> Option<String> {
    // Strategy 1: Check for destination header
    if let Some(dest) = frame.get_header("destination") {
        return Some(dest.to_string());
    }

    // Strategy 2: Look for destination pattern in message header or body
    let message = frame.get_header("message").unwrap_or("");
    let body = String::from_utf8_lossy(&frame.body);

    // Combine message and body for searching
    let text = format!("{} {}", message, body);

    // Look for /topic/ or /queue/ patterns
    for prefix in ["/topic/", "/queue/"] {
        if let Some(start) = text.find(prefix) {
            // Extract until whitespace, comma, quote, or end of string
            let rest = &text[start..];
            let end = rest
                .find(|c: char| c.is_whitespace() || c == ',' || c == '"' || c == '\'')
                .unwrap_or(rest.len());
            if end > prefix.len() {
                return Some(rest[..end].to_string());
            }
        }
    }

    None
}

/// Extract subscription ID from an ERROR frame message.
///
/// Looks for patterns like "subscription 1" or "subscription sub-1" in the
/// error message or body. Artemis uses this format for subscription errors.
fn extract_subscription_id_from_error(frame: &Frame) -> Option<String> {
    let message = frame.get_header("message").unwrap_or("");
    let body = String::from_utf8_lossy(&frame.body);
    let text = format!("{} {}", message, body);

    // Look for "subscription X" pattern (Artemis format)
    if let Some(idx) = text.to_lowercase().find("subscription ") {
        let rest = &text[idx + 13..]; // "subscription " is 13 chars
        // Extract the subscription ID (could be numeric or alphanumeric like "sub-1")
        let end = rest
            .find(|c: char| c.is_whitespace() || c == ',' || c == '"' || c == '\'')
            .unwrap_or(rest.len());
        if end > 0 {
            return Some(rest[..end].to_string());
        }
    }

    None
}

/// Look up a destination by subscription ID in the subscriptions map.
async fn lookup_destination_by_sub_id(
    sub_id: &str,
    subscriptions: &Arc<Mutex<Subscriptions>>,
) -> Option<String> {
    let map = subscriptions.lock().await;
    for (dest, entries) in map.iter() {
        for entry in entries {
            if entry.id == sub_id {
                return Some(dest.clone());
            }
        }
    }
    None
}

/// High-level connection object that manages a single TCP/STOMP connection.
///
/// The `Connection` spawns a background task that maintains the TCP transport,
/// sends/receives STOMP frames using `StompCodec`, negotiates heartbeats, and
/// performs simple reconnect logic with exponential backoff.
#[derive(Clone)]
pub struct Connection {
    outbound_tx: mpsc::Sender<StompItem>,
    /// The inbound receiver is shared behind a mutex so the `Connection`
    /// handle may be cloned and callers can call `next_frame` concurrently.
    inbound_rx: Arc<Mutex<mpsc::Receiver<Frame>>>,
    shutdown_tx: broadcast::Sender<()>,
    /// Map of destination -> list of (subscription id, sender) for dispatching
    /// inbound MESSAGE frames to subscribers.
    subscriptions: Arc<Mutex<Subscriptions>>,
    /// Monotonic counter used to allocate subscription ids.
    sub_id_counter: Arc<AtomicU64>,
    /// Pending messages awaiting ACK/NACK from the application.
    ///
    /// Organized by subscription id. For `client` ack mode the ACK is
    /// cumulative: acknowledging message `M` for subscription `S` acknowledges
    /// all messages previously delivered for `S` up to and including `M`.
    /// For `client-individual` the ACK/NACK applies only to the single
    /// message.
    pending: Arc<Mutex<PendingMap>>,
    /// Pending receipt confirmations.
    ///
    /// When a frame is sent with a `receipt` header, the receipt-id is stored
    /// here with a oneshot sender. When the server responds with a RECEIPT
    /// frame, the sender is notified.
    pending_receipts: Arc<Mutex<PendingReceipts>>,
}

impl Connection {
    /// Heartbeat value that disables heartbeats entirely.
    ///
    /// Use this when you don't want the client or server to send heartbeats.
    /// Note that some brokers may still require heartbeats for long-lived connections.
    ///
    /// # Example
    ///
    /// ```ignore
    /// let conn = Connection::connect(
    ///     "localhost:61613",
    ///     "guest",
    ///     "guest",
    ///     Connection::NO_HEARTBEAT,
    /// ).await?;
    /// ```
    pub const NO_HEARTBEAT: &'static str = "0,0";

    /// Default heartbeat value: 10 seconds for both send and receive.
    ///
    /// This is a reasonable default for most applications. The actual heartbeat
    /// interval will be negotiated with the server (taking the maximum of client
    /// and server preferences).
    ///
    /// # Example
    ///
    /// ```ignore
    /// let conn = Connection::connect(
    ///     "localhost:61613",
    ///     "guest",
    ///     "guest",
    ///     Connection::DEFAULT_HEARTBEAT,
    /// ).await?;
    /// ```
    pub const DEFAULT_HEARTBEAT: &'static str = "10000,10000";

    /// Establish a connection to the STOMP server at `addr` with the given
    /// credentials and heartbeat header string (e.g. "10000,10000").
    ///
    /// This is a convenience wrapper around `connect_with_options()` that uses
    /// default options (STOMP 1.2, host="/", no client-id).
    ///
    /// Parameters
    /// - `addr`: TCP address (host:port) of the STOMP server.
    /// - `login`: login username for STOMP `CONNECT`.
    /// - `passcode`: passcode for STOMP `CONNECT`.
    /// - `client_hb`: client's `heart-beat` header value ("cx,cy" in
    ///   milliseconds) that will be sent in the `CONNECT` frame.
    ///
    /// Returns a `Connection` which provides `send_frame`, `next_frame`, and
    /// `close` helpers. The detailed connection handling (I/O, heartbeats,
    /// reconnects) runs on a background task spawned by this method.
    pub async fn connect(
        addr: &str,
        login: &str,
        passcode: &str,
        client_hb: &str,
    ) -> Result<Self, ConnError> {
        Self::connect_with_options(addr, login, passcode, client_hb, ConnectOptions::default())
            .await
    }

    /// Establish a connection to the STOMP server with custom options.
    ///
    /// Use this method when you need to set a custom `client-id` (for durable
    /// subscriptions), specify a virtual host, negotiate different STOMP
    /// versions, or add custom CONNECT headers.
    ///
    /// Parameters
    /// - `addr`: TCP address (host:port) of the STOMP server.
    /// - `login`: login username for STOMP `CONNECT`.
    /// - `passcode`: passcode for STOMP `CONNECT`.
    /// - `client_hb`: client's `heart-beat` header value ("cx,cy" in
    ///   milliseconds) that will be sent in the `CONNECT` frame.
    /// - `options`: custom connection options (version, host, client-id, etc.).
    ///
    /// # Errors
    ///
    /// Returns an error if:
    /// - The TCP connection cannot be established (`ConnError::Io`)
    /// - The server rejects the connection, e.g., due to invalid credentials
    ///   (`ConnError::ServerRejected`)
    /// - The server closes the connection without responding (`ConnError::Protocol`)
    ///
    /// # Example
    ///
    /// ```ignore
    /// use iridium_stomp::{Connection, ConnectOptions};
    ///
    /// // Connect with a client-id for durable subscriptions
    /// let options = ConnectOptions::default()
    ///     .client_id("my-app-instance-1");
    ///
    /// let conn = Connection::connect_with_options(
    ///     "localhost:61613",
    ///     "guest",
    ///     "guest",
    ///     "10000,10000",
    ///     options,
    /// ).await?;
    /// ```
    pub async fn connect_with_options(
        addr: &str,
        login: &str,
        passcode: &str,
        client_hb: &str,
        options: ConnectOptions,
    ) -> Result<Self, ConnError> {
        let (out_tx, mut out_rx) = mpsc::channel::<StompItem>(32);
        let (in_tx, in_rx) = mpsc::channel::<Frame>(32);
        let subscriptions: Arc<Mutex<Subscriptions>> = Arc::new(Mutex::new(HashMap::new()));
        let sub_id_counter = Arc::new(AtomicU64::new(1));
        let (shutdown_tx, _) = broadcast::channel::<()>(1);
        let pending: Arc<Mutex<PendingMap>> = Arc::new(Mutex::new(HashMap::new()));
        let pending_clone = pending.clone();
        let pending_receipts: Arc<Mutex<PendingReceipts>> = Arc::new(Mutex::new(HashMap::new()));
        let pending_receipts_clone = pending_receipts.clone();

        let addr = addr.to_string();
        let login = login.to_string();
        let passcode = passcode.to_string();
        let client_hb = client_hb.to_string();

        // Extract options into owned values for the spawned task
        let accept_version = options.accept_version.unwrap_or_else(|| "1.2".to_string());
        let host = options.host.unwrap_or_else(|| "/".to_string());
        let client_id = options.client_id;
        let custom_headers = options.headers;
        let heartbeat_notify_tx = options.heartbeat_tx;

        // Perform initial connection and STOMP handshake before spawning background task.
        // This ensures authentication errors are returned to the caller immediately.
        let stream = TcpStream::connect(&addr).await?;
        let mut framed = Framed::new(stream, StompCodec::new());

        // Build and send CONNECT frame
        let connect = Self::build_connect_frame(
            &accept_version,
            &host,
            &login,
            &passcode,
            &client_hb,
            &client_id,
            &custom_headers,
        );

        framed
            .send(StompItem::Frame(connect))
            .await
            .map_err(|e| ConnError::Io(std::io::Error::other(e)))?;

        // Wait for CONNECTED or ERROR response
        let server_heartbeat = Self::await_connected_response(&mut framed).await?;

        // Calculate heartbeat intervals
        let (cx, cy) = parse_heartbeat_header(&client_hb);
        let (sx, sy) = parse_heartbeat_header(&server_heartbeat);
        let (send_interval, recv_interval) = negotiate_heartbeats(cx, cy, sx, sy);

        // Now spawn background task for ongoing I/O and reconnection
        let shutdown_tx_clone = shutdown_tx.clone();
        let subscriptions_clone = subscriptions.clone();

        tokio::spawn(async move {
            let mut backoff_secs: u64 = 1;

            // Use the already-established connection for the first iteration
            let mut current_framed = Some(framed);
            let mut current_send_interval = send_interval;
            let mut current_recv_interval = recv_interval;

            // Track subscription errors across reconnections. If a subscription
            // receives too many consecutive errors, we remove it to prevent
            // error loops (e.g., Artemis sending repeated permission errors).
            let mut subscription_errors: HashMap<String, u32> = HashMap::new();
            // Track subscription IDs that have been abandoned so we can ignore
            // subsequent errors for them.
            let mut abandoned_sub_ids: std::collections::HashSet<String> =
                std::collections::HashSet::new();
            const SUBSCRIPTION_ERROR_THRESHOLD: u32 = 3;

            loop {
                let mut shutdown_sub = shutdown_tx_clone.subscribe();

                // Check for shutdown before attempting connection
                tokio::select! {
                    biased;
                    _ = shutdown_sub.recv() => break,
                    _ = future::ready(()) => {},
                }

                // Either use existing connection or establish new one (reconnect)
                let framed = if let Some(f) = current_framed.take() {
                    f
                } else {
                    // Reconnection attempt
                    match TcpStream::connect(&addr).await {
                        Ok(stream) => {
                            let mut framed = Framed::new(stream, StompCodec::new());

                            let connect = Self::build_connect_frame(
                                &accept_version,
                                &host,
                                &login,
                                &passcode,
                                &client_hb,
                                &client_id,
                                &custom_headers,
                            );

                            if framed.send(StompItem::Frame(connect)).await.is_err() {
                                tokio::time::sleep(Duration::from_secs(backoff_secs)).await;
                                backoff_secs = (backoff_secs * 2).min(30);
                                continue;
                            }

                            // Wait for CONNECTED (on reconnect, silently retry on ERROR)
                            match Self::await_connected_response(&mut framed).await {
                                Ok(server_hb) => {
                                    let (cx, cy) = parse_heartbeat_header(&client_hb);
                                    let (sx, sy) = parse_heartbeat_header(&server_hb);
                                    let (si, ri) = negotiate_heartbeats(cx, cy, sx, sy);
                                    current_send_interval = si;
                                    current_recv_interval = ri;
                                    framed
                                }
                                Err(_) => {
                                    // Reconnect failed (auth error or other), retry with backoff
                                    tokio::time::sleep(Duration::from_secs(backoff_secs)).await;
                                    backoff_secs = (backoff_secs * 2).min(30);
                                    continue;
                                }
                            }
                        }
                        Err(_) => {
                            tokio::time::sleep(Duration::from_secs(backoff_secs)).await;
                            backoff_secs = (backoff_secs * 2).min(30);
                            continue;
                        }
                    }
                };

                let (send_interval, recv_interval) = (current_send_interval, current_recv_interval);

                let last_received = Arc::new(AtomicU64::new(current_millis()));
                let writer_last_sent = Arc::new(AtomicU64::new(current_millis()));

                let (mut sink, mut stream) = framed.split();
                let in_tx = in_tx.clone();
                let subscriptions = subscriptions_clone.clone();

                // Clear pending message map on reconnect — messages that were
                // outstanding before the disconnect are considered lost and
                // will be redelivered by the server as appropriate.
                {
                    let mut p = pending_clone.lock().await;
                    p.clear();
                }

                // Resubscribe any existing subscriptions after reconnect.
                // We snapshot the subscription entries while holding the lock
                // and then issue SUBSCRIBE frames using the sink.
                let subs_snapshot: Vec<ResubEntry> = {
                    let map = subscriptions.lock().await;
                    let mut v: Vec<ResubEntry> = Vec::new();
                    for (dest, vec) in map.iter() {
                        for entry in vec.iter() {
                            v.push((
                                dest.clone(),
                                entry.id.clone(),
                                entry.ack.clone(),
                                entry.headers.clone(),
                            ));
                        }
                    }
                    v
                };

                for (dest, id, ack, headers) in subs_snapshot {
                    let mut sf = Frame::new("SUBSCRIBE");
                    sf = sf
                        .header("id", &id)
                        .header("destination", &dest)
                        .header("ack", &ack);
                    for (k, v) in headers {
                        sf = sf.header(&k, &v);
                    }
                    let _ = sink.send(StompItem::Frame(sf)).await;
                }

                let mut hb_tick = match send_interval {
                    Some(d) => tokio::time::interval(d),
                    None => tokio::time::interval(Duration::from_secs(86400)),
                };
                let watchdog_half = recv_interval.map(|d| d / 2);

                let conn_start = tokio::time::Instant::now();

                'conn: loop {
                    tokio::select! {
                        _ = shutdown_sub.recv() => { let _ = sink.close().await; break 'conn; }
                        maybe = out_rx.recv() => {
                            match maybe {
                                Some(item) => if sink.send(item).await.is_err() { break 'conn } else { writer_last_sent.store(current_millis(), Ordering::SeqCst); }
                                None => break 'conn,
                            }
                        }
                        item = stream.next() => {
                            match item {
                                Some(Ok(StompItem::Heartbeat)) => {
                                    last_received.store(current_millis(), Ordering::SeqCst);
                                    if let Some(ref tx) = heartbeat_notify_tx {
                                        let _ = tx.try_send(());
                                    }
                                }
                                Some(Ok(StompItem::Frame(f))) => {
                                    last_received.store(current_millis(), Ordering::SeqCst);
                                    // Dispatch MESSAGE frames to any matching subscribers.
                                    if f.command == "MESSAGE" {
                                        // try to find destination, subscription and message-id headers
                                        let mut dest_opt: Option<String> = None;
                                        let mut sub_opt: Option<String> = None;
                                        let mut msg_id_opt: Option<String> = None;
                                        for (k, v) in &f.headers {
                                            let kl = k.to_lowercase();
                                            if kl == "destination" {
                                                dest_opt = Some(v.clone());
                                            } else if kl == "subscription" {
                                                sub_opt = Some(v.clone());
                                            } else if kl == "message-id" {
                                                msg_id_opt = Some(v.clone());
                                            }
                                        }

                                        // Determine whether we need to track this message as pending
                                        let mut need_pending = false;
                                        if let Some(sub_id) = &sub_opt {
                                            let map = subscriptions.lock().await;
                                            for (_dest, vec) in map.iter() {
                                                for entry in vec.iter() {
                                                    if &entry.id == sub_id && entry.ack != "auto" {
                                                        need_pending = true;
                                                    }
                                                }
                                            }
                                        } else if let Some(dest) = &dest_opt {
                                            let map = subscriptions.lock().await;
                                            if let Some(vec) = map.get(dest) {
                                                for entry in vec.iter() {
                                                    if entry.ack != "auto" {
                                                        need_pending = true;
                                                        break;
                                                    }
                                                }
                                            }
                                        }

                                        // If required, add to pending map (per-subscription) before
                                        // delivery so ACK/NACK requests from the application can
                                        // reference the message. We require a `message-id` header
                                        // to track messages; if missing, we cannot support ACK/NACK.
                                        if let Some(msg_id) = msg_id_opt.clone().filter(|_| need_pending) {
                                            // If the server provided a subscription id in the
                                            // MESSAGE, store pending under that subscription.
                                            if let Some(sub_id) = &sub_opt {
                                                let mut p = pending_clone.lock().await;
                                                let q = p
                                                    .entry(sub_id.clone())
                                                    .or_insert_with(VecDeque::new);
                                                q.push_back((msg_id.clone(), f.clone()));
                                            } else if let Some(dest) = &dest_opt {
                                                // Destination-based delivery: add the message to
                                                // the pending queue for each matching
                                                // subscription on that destination.
                                                let map = subscriptions.lock().await;
                                                if let Some(vec) = map.get(dest) {
                                                    let mut p = pending_clone.lock().await;
                                                    for entry in vec.iter() {
                                                        let q = p
                                                            .entry(entry.id.clone())
                                                            .or_insert_with(VecDeque::new);
                                                        q.push_back((msg_id.clone(), f.clone()));
                                                    }
                                                }
                                            }
                                        }

                                        // Deliver to subscribers.
                                        if let Some(sub_id) = sub_opt {
                                            let mut map = subscriptions.lock().await;
                                            for (_dest, vec) in map.iter_mut() {
                                                vec.retain(|entry| {
                                                    if entry.id == sub_id {
                                                        let _ = entry.sender.try_send(f.clone());
                                                        true
                                                    } else {
                                                        true
                                                    }
                                                });
                                            }
                                        } else if let Some(dest) = dest_opt {
                                            let mut map = subscriptions.lock().await;
                                            if let Some(vec) = map.get_mut(&dest) {
                                                vec.retain(|entry| entry.sender.try_send(f.clone()).is_ok());
                                            }
                                        }
                                    } else if f.command == "RECEIPT" {
                                        // Handle RECEIPT frame: notify any waiting callers
                                        if let Some(receipt_id) = f.get_header("receipt-id") {
                                            let mut receipts = pending_receipts_clone.lock().await;
                                            if let Some(sender) = receipts.remove(receipt_id) {
                                                let _ = sender.send(());
                                            }
                                        }
                                        // Don't forward RECEIPT frames to inbound channel
                                        continue;
                                    } else if f.command == "ERROR" {
                                        // Track subscription-related errors. If we see repeated
                                        // errors for the same destination, remove the subscription
                                        // to prevent error loops.
                                        //
                                        // First, check if this error is for an already-abandoned
                                        // subscription (Artemis keeps sending errors after we abandon).
                                        let sub_id = extract_subscription_id_from_error(&f);
                                        if let Some(ref id) = sub_id
                                            && abandoned_sub_ids.contains(id)
                                        {
                                            // Skip this error - subscription already abandoned
                                            continue;
                                        }

                                        // Try to identify the destination:
                                        // 1. Extract directly from ERROR frame
                                        // 2. Look up by subscription ID (Artemis uses "subscription N")
                                        let dest = if let Some(d) = extract_destination_from_error(&f)
                                        {
                                            Some(d)
                                        } else if let Some(ref id) = sub_id {
                                            lookup_destination_by_sub_id(id, &subscriptions).await
                                        } else {
                                            None
                                        };

                                        if let Some(dest) = dest {
                                            let count = {
                                                let c = subscription_errors
                                                    .entry(dest.clone())
                                                    .or_insert(0);
                                                *c += 1;
                                                *c
                                            };

                                            if count >= SUBSCRIPTION_ERROR_THRESHOLD {
                                                // Remove the subscription from auto-resubscribe
                                                let mut map = subscriptions.lock().await;
                                                if map.remove(&dest).is_some() {
                                                    // Track the subscription ID as abandoned
                                                    if let Some(id) = sub_id {
                                                        abandoned_sub_ids.insert(id);
                                                    }
                                                    // Send abandonment notification
                                                    let msg = format!(
                                                        "Subscription abandoned: {} errors for {}",
                                                        count, dest
                                                    );
                                                    let abandon_frame = Frame::new("ERROR")
                                                        .header("message", &msg)
                                                        .header("destination", &dest)
                                                        .header("x-abandoned", "true");
                                                    let _ = in_tx.send(abandon_frame).await;
                                                }
                                            }
                                        }
                                    }

                                    let _ = in_tx.send(f).await;
                                }
                                Some(Err(_)) | None => break 'conn,
                            }
                        }
                        _ = hb_tick.tick() => {
                            if let Some(dur) = send_interval {
                                let last = writer_last_sent.load(Ordering::SeqCst);
                                if current_millis().saturating_sub(last) >= dur.as_millis() as u64 {
                                    if sink.send(StompItem::Heartbeat).await.is_err() { break 'conn; }
                                    writer_last_sent.store(current_millis(), Ordering::SeqCst);
                                }
                            }
                        }
                        _ = async { if let Some(interval) = watchdog_half { tokio::time::sleep(interval).await } else { future::pending::<()>().await } } => {
                            if let Some(recv_dur) = recv_interval {
                                let last = last_received.load(Ordering::SeqCst);
                                if current_millis().saturating_sub(last) > (recv_dur.as_millis() as u64 * 2) {
                                    let _ = sink.close().await; break 'conn;
                                }
                            }
                        }
                    }
                }

                if shutdown_sub.try_recv().is_ok() {
                    break;
                }
                if conn_start.elapsed() >= Duration::from_secs(backoff_secs.max(5)) {
                    // Connection was stable — reset backoff
                    backoff_secs = 1;
                } else {
                    // Connection died quickly — increase backoff
                    backoff_secs = (backoff_secs * 2).min(30);
                }
                tokio::time::sleep(Duration::from_secs(backoff_secs)).await;
            }
        });

        Ok(Connection {
            outbound_tx: out_tx,
            inbound_rx: Arc::new(Mutex::new(in_rx)),
            shutdown_tx,
            subscriptions,
            sub_id_counter,
            pending,
            pending_receipts,
        })
    }

    /// Build a CONNECT frame with all specified headers.
    fn build_connect_frame(
        accept_version: &str,
        host: &str,
        login: &str,
        passcode: &str,
        heartbeat: &str,
        client_id: &Option<String>,
        custom_headers: &[(String, String)],
    ) -> Frame {
        let mut connect = Frame::new("CONNECT")
            .header("accept-version", accept_version)
            .header("host", host)
            .header("login", login)
            .header("passcode", passcode)
            .header("heart-beat", heartbeat);

        if let Some(id) = client_id {
            connect = connect.header("client-id", id);
        }

        // Reserved headers that custom_headers cannot override
        let reserved = [
            "accept-version",
            "host",
            "login",
            "passcode",
            "heart-beat",
            "client-id",
        ];

        for (k, v) in custom_headers {
            if !reserved.contains(&k.to_lowercase().as_str()) {
                connect = connect.header(k, v);
            }
        }

        connect
    }

    /// Wait for CONNECTED or ERROR response from the server.
    ///
    /// Returns the server's heartbeat header value on success, or an error
    /// if the server sends an ERROR frame or closes the connection.
    async fn await_connected_response(
        framed: &mut Framed<TcpStream, StompCodec>,
    ) -> Result<String, ConnError> {
        loop {
            match framed.next().await {
                Some(Ok(StompItem::Frame(f))) => {
                    if f.command == "CONNECTED" {
                        // Extract heartbeat from server
                        let server_hb = f.get_header("heart-beat").unwrap_or("0,0").to_string();
                        return Ok(server_hb);
                    } else if f.command == "ERROR" {
                        // Server rejected connection (e.g., invalid credentials)
                        return Err(ConnError::ServerRejected(ServerError::from_frame(f)));
                    }
                    // Ignore other frames during CONNECT phase
                }
                Some(Ok(StompItem::Heartbeat)) => {
                    // Ignore heartbeats during handshake
                    continue;
                }
                Some(Err(e)) => {
                    return Err(ConnError::Io(e));
                }
                None => {
                    return Err(ConnError::Protocol(
                        "connection closed before CONNECTED received".to_string(),
                    ));
                }
            }
        }
    }

    pub async fn send_frame(&self, frame: Frame) -> Result<(), ConnError> {
        // Send a frame to the background writer task.
        //
        // Parameters
        // - `frame`: ownership of the `Frame` to send. The frame is converted
        //   into a `StompItem::Frame` and sent over the internal mpsc channel.
        self.outbound_tx
            .send(StompItem::Frame(frame))
            .await
            .map_err(|_| ConnError::Protocol("send channel closed".into()))
    }

    /// Generate a unique receipt ID.
    fn generate_receipt_id() -> String {
        static RECEIPT_COUNTER: AtomicU64 = AtomicU64::new(1);
        format!("rcpt-{}", RECEIPT_COUNTER.fetch_add(1, Ordering::SeqCst))
    }

    /// Send a frame with a receipt request and return the receipt ID.
    ///
    /// This method adds a unique `receipt` header to the frame and registers
    /// the receipt ID for tracking. Use `wait_for_receipt()` to wait for the
    /// server's RECEIPT response.
    ///
    /// # Parameters
    /// - `frame`: the frame to send. A `receipt` header will be added.
    ///
    /// # Returns
    /// The generated receipt ID that can be used with `wait_for_receipt()`.
    ///
    /// # Example
    /// ```ignore
    /// let receipt_id = conn.send_frame_with_receipt(frame).await?;
    /// conn.wait_for_receipt(&receipt_id, Duration::from_secs(5)).await?;
    /// ```
    pub async fn send_frame_with_receipt(&self, frame: Frame) -> Result<String, ConnError> {
        let receipt_id = Self::generate_receipt_id();

        // Create the oneshot channel for notification
        let (tx, _rx) = oneshot::channel();

        // Register the pending receipt
        {
            let mut receipts = self.pending_receipts.lock().await;
            receipts.insert(receipt_id.clone(), tx);
        }

        // Add receipt header and send the frame
        let frame_with_receipt = frame.receipt(&receipt_id);
        self.send_frame(frame_with_receipt).await?;

        Ok(receipt_id)
    }

    /// Wait for a receipt confirmation from the server.
    ///
    /// This method blocks until the server sends a RECEIPT frame with the
    /// matching receipt-id, or until the timeout expires.
    ///
    /// # Parameters
    /// - `receipt_id`: the receipt ID returned by `send_frame_with_receipt()`.
    /// - `timeout`: maximum time to wait for the receipt.
    ///
    /// # Returns
    /// `Ok(())` if the receipt was received, or `Err(ConnError::ReceiptTimeout)`
    /// if the timeout expired.
    ///
    /// # Example
    /// ```ignore
    /// let receipt_id = conn.send_frame_with_receipt(frame).await?;
    /// conn.wait_for_receipt(&receipt_id, Duration::from_secs(5)).await?;
    /// println!("Message confirmed!");
    /// ```
    pub async fn wait_for_receipt(
        &self,
        receipt_id: &str,
        timeout: Duration,
    ) -> Result<(), ConnError> {
        // Get the receiver for this receipt
        let rx = {
            let mut receipts = self.pending_receipts.lock().await;
            // Re-create the oneshot channel and swap out the sender
            let (tx, rx) = oneshot::channel();
            if let Some(old_tx) = receipts.insert(receipt_id.to_string(), tx) {
                // Drop the old sender - this is expected if called after send_frame_with_receipt
                drop(old_tx);
            }
            rx
        };

        // Wait for the receipt with timeout
        match tokio::time::timeout(timeout, rx).await {
            Ok(Ok(())) => Ok(()),
            Ok(Err(_)) => {
                // Channel was closed without receiving - connection likely dropped
                Err(ConnError::Protocol(
                    "receipt channel closed unexpectedly".into(),
                ))
            }
            Err(_) => {
                // Timeout expired - clean up the pending receipt
                let mut receipts = self.pending_receipts.lock().await;
                receipts.remove(receipt_id);
                Err(ConnError::ReceiptTimeout(receipt_id.to_string()))
            }
        }
    }

    /// Send a frame and wait for server confirmation via RECEIPT.
    ///
    /// This is a convenience method that combines `send_frame_with_receipt()`
    /// and `wait_for_receipt()`. Use this when you want to ensure a frame
    /// was processed by the server before continuing.
    ///
    /// # Parameters
    /// - `frame`: the frame to send.
    /// - `timeout`: maximum time to wait for the receipt.
    ///
    /// # Returns
    /// `Ok(())` if the frame was sent and receipt confirmed, or an error if
    /// sending failed or the receipt timed out.
    ///
    /// # Example
    /// ```ignore
    /// let frame = Frame::new("SEND")
    ///     .header("destination", "/queue/orders")
    ///     .set_body(b"order data".to_vec());
    ///
    /// conn.send_frame_confirmed(frame, Duration::from_secs(5)).await?;
    /// println!("Order sent and confirmed!");
    /// ```
    pub async fn send_frame_confirmed(
        &self,
        frame: Frame,
        timeout: Duration,
    ) -> Result<(), ConnError> {
        let receipt_id = Self::generate_receipt_id();

        // Create the oneshot channel for notification
        let (tx, rx) = oneshot::channel();

        // Register the pending receipt before sending
        {
            let mut receipts = self.pending_receipts.lock().await;
            receipts.insert(receipt_id.clone(), tx);
        }

        // Add receipt header and send the frame
        let frame_with_receipt = frame.receipt(&receipt_id);
        self.send_frame(frame_with_receipt).await?;

        // Wait for the receipt with timeout
        match tokio::time::timeout(timeout, rx).await {
            Ok(Ok(())) => Ok(()),
            Ok(Err(_)) => Err(ConnError::Protocol(
                "receipt channel closed unexpectedly".into(),
            )),
            Err(_) => {
                // Timeout expired - clean up
                let mut receipts = self.pending_receipts.lock().await;
                receipts.remove(&receipt_id);
                Err(ConnError::ReceiptTimeout(receipt_id))
            }
        }
    }

    /// Subscribe to a destination.
    ///
    /// Parameters
    /// - `destination`: the STOMP destination to subscribe to (e.g. "/queue/foo").
    /// - `ack`: acknowledgement mode to request from the server.
    ///
    /// Returns a tuple `(subscription_id, receiver)` where `subscription_id` is
    /// the opaque id assigned locally for this subscription and `receiver` is a
    /// `mpsc::Receiver<Frame>` which will yield incoming MESSAGE frames for the
    /// destination. The caller should read from the receiver to handle messages.
    /// Subscribe to a destination using optional extra headers.
    ///
    /// This variant accepts additional headers which are stored locally and
    /// re-sent on reconnect. Use `subscribe` as a convenience wrapper when no
    /// extra headers are needed.
    pub async fn subscribe_with_headers(
        &self,
        destination: &str,
        ack: AckMode,
        extra_headers: Vec<(String, String)>,
    ) -> Result<crate::subscription::Subscription, ConnError> {
        let id = self
            .sub_id_counter
            .fetch_add(1, Ordering::SeqCst)
            .to_string();
        let (tx, rx) = mpsc::channel::<Frame>(16);
        {
            let mut map = self.subscriptions.lock().await;
            map.entry(destination.to_string())
                .or_insert_with(Vec::new)
                .push(SubscriptionEntry {
                    id: id.clone(),
                    sender: tx.clone(),
                    ack: ack.as_str().to_string(),
                    headers: extra_headers.clone(),
                });
        }

        let mut f = Frame::new("SUBSCRIBE");
        f = f
            .header("id", &id)
            .header("destination", destination)
            .header("ack", ack.as_str());
        for (k, v) in &extra_headers {
            f = f.header(k, v);
        }
        self.outbound_tx
            .send(StompItem::Frame(f))
            .await
            .map_err(|_| ConnError::Protocol("send channel closed".into()))?;

        Ok(crate::subscription::Subscription::new(
            id,
            destination.to_string(),
            rx,
            self.clone(),
        ))
    }

    /// Convenience wrapper without extra headers.
    pub async fn subscribe(
        &self,
        destination: &str,
        ack: AckMode,
    ) -> Result<crate::subscription::Subscription, ConnError> {
        self.subscribe_with_headers(destination, ack, Vec::new())
            .await
    }

    /// Subscribe with a typed `SubscriptionOptions` structure.
    ///
    /// `SubscriptionOptions.headers` are forwarded to the broker and persisted
    /// for automatic resubscribe after reconnect. If `durable_queue` is set,
    /// it will be used as the actual destination instead of `destination`.
    pub async fn subscribe_with_options(
        &self,
        destination: &str,
        ack: AckMode,
        options: crate::subscription::SubscriptionOptions,
    ) -> Result<crate::subscription::Subscription, ConnError> {
        let dest = options
            .durable_queue
            .as_deref()
            .unwrap_or(destination)
            .to_string();
        self.subscribe_with_headers(&dest, ack, options.headers)
            .await
    }

    /// Unsubscribe a previously created subscription by its local subscription id.
    pub async fn unsubscribe(&self, subscription_id: &str) -> Result<(), ConnError> {
        let mut found = false;
        {
            let mut map = self.subscriptions.lock().await;
            let mut remove_keys: Vec<String> = Vec::new();
            for (dest, vec) in map.iter_mut() {
                if let Some(pos) = vec.iter().position(|entry| entry.id == subscription_id) {
                    vec.remove(pos);
                    found = true;
                }
                if vec.is_empty() {
                    remove_keys.push(dest.clone());
                }
            }
            for k in remove_keys {
                map.remove(&k);
            }
        }

        if !found {
            return Err(ConnError::Protocol("subscription id not found".into()));
        }

        let mut f = Frame::new("UNSUBSCRIBE");
        f = f.header("id", subscription_id);
        self.outbound_tx
            .send(StompItem::Frame(f))
            .await
            .map_err(|_| ConnError::Protocol("send channel closed".into()))?;

        Ok(())
    }

    /// Acknowledge a message previously received in `client` or
    /// `client-individual` ack modes.
    ///
    /// STOMP ack semantics:
    /// - `auto`: server considers message delivered immediately; the client
    ///   should not ack.
    /// - `client`: cumulative acknowledgements. ACKing message `M` for
    ///   subscription `S` acknowledges all messages delivered to `S` up to
    ///   and including `M`.
    /// - `client-individual`: only the named message is acknowledged.
    ///
    /// Parameters
    /// - `subscription_id`: the local subscription id returned by
    ///   `Connection::subscribe`. This disambiguates which subscription's
    ///   pending queue to advance for cumulative ACKs.
    /// - `message_id`: the `message-id` header value from the received
    ///   MESSAGE frame to acknowledge.
    ///
    /// Behavior
    /// - The pending queue for `subscription_id` is searched for `message_id`.
    ///   If the subscription used `client` ack mode, all pending messages up to
    ///   and including the matched message are removed. If the subscription
    ///   used `client-individual`, only the matched message is removed.
    /// - An `ACK` frame is sent to the server with `id=<message_id>` and
    ///   `subscription=<subscription_id>` headers.
    #[allow(clippy::collapsible_if, clippy::collapsible_else_if)]
    pub async fn ack(&self, subscription_id: &str, message_id: &str) -> Result<(), ConnError> {
        // Remove from the local pending queue according to subscription ack mode.
        let mut removed_any = false;
        {
            let mut p = self.pending.lock().await;
            if let Some(queue) = p.get_mut(subscription_id) {
                if let Some(pos) = queue.iter().position(|(mid, _)| mid == message_id) {
                    // Determine ack mode for this subscription (default to client).
                    let mut ack_mode = "client".to_string();
                    {
                        let map = self.subscriptions.lock().await;
                        'outer: for (_dest, vec) in map.iter() {
                            for entry in vec.iter() {
                                if entry.id == subscription_id {
                                    ack_mode = entry.ack.clone();
                                    break 'outer;
                                }
                            }
                        }
                    }

                    if ack_mode == "client" {
                        // cumulative: remove up to and including pos
                        for _ in 0..=pos {
                            queue.pop_front();
                            removed_any = true;
                        }
                    } else if queue.remove(pos).is_some() {
                        // client-individual: remove only the specific message
                        removed_any = true;
                    }

                    if queue.is_empty() {
                        p.remove(subscription_id);
                    }
                }
            }
        }

        // Send ACK to server (include subscription header for clarity)
        let mut f = Frame::new("ACK");
        f = f
            .header("id", message_id)
            .header("subscription", subscription_id);
        self.outbound_tx
            .send(StompItem::Frame(f))
            .await
            .map_err(|_| ConnError::Protocol("send channel closed".into()))?;

        // If message wasn't found locally, still send ACK to server; server
        // may ignore or treat it as no-op.
        let _ = removed_any;
        Ok(())
    }

    /// Negative-acknowledge a message (NACK).
    ///
    /// Parameters
    /// - `subscription_id`: the local subscription id the message was delivered under.
    /// - `message_id`: the `message-id` header value from the received MESSAGE.
    ///
    /// Behavior
    /// - Removes the message from the local pending queue (cumulatively if the
    ///   subscription used `client` ack mode, otherwise only the single
    ///   message). Sends a `NACK` frame to the server with `id` and
    ///   `subscription` headers.
    #[allow(clippy::collapsible_if, clippy::collapsible_else_if)]
    pub async fn nack(&self, subscription_id: &str, message_id: &str) -> Result<(), ConnError> {
        // Mirror ack removal semantics for pending map.
        let mut removed_any = false;
        {
            let mut p = self.pending.lock().await;
            if let Some(queue) = p.get_mut(subscription_id) {
                if let Some(pos) = queue.iter().position(|(mid, _)| mid == message_id) {
                    let mut ack_mode = "client".to_string();
                    {
                        let map = self.subscriptions.lock().await;
                        'outer2: for (_dest, vec) in map.iter() {
                            for entry in vec.iter() {
                                if entry.id == subscription_id {
                                    ack_mode = entry.ack.clone();
                                    break 'outer2;
                                }
                            }
                        }
                    }

                    if ack_mode == "client" {
                        for _ in 0..=pos {
                            queue.pop_front();
                            removed_any = true;
                        }
                    } else if queue.remove(pos).is_some() {
                        removed_any = true;
                    }

                    if queue.is_empty() {
                        p.remove(subscription_id);
                    }
                }
            }
        }

        let mut f = Frame::new("NACK");
        f = f
            .header("id", message_id)
            .header("subscription", subscription_id);
        self.outbound_tx
            .send(StompItem::Frame(f))
            .await
            .map_err(|_| ConnError::Protocol("send channel closed".into()))?;

        let _ = removed_any;
        Ok(())
    }

    /// Helper to send a transaction frame (BEGIN, COMMIT, or ABORT).
    async fn send_transaction_frame(
        &self,
        command: &str,
        transaction_id: &str,
    ) -> Result<(), ConnError> {
        let f = Frame::new(command).header("transaction", transaction_id);
        self.outbound_tx
            .send(StompItem::Frame(f))
            .await
            .map_err(|_| ConnError::Protocol("send channel closed".into()))
    }

    /// Begin a transaction.
    ///
    /// Parameters
    /// - `transaction_id`: unique identifier for the transaction. The caller is
    ///   responsible for ensuring uniqueness within the connection.
    ///
    /// Behavior
    /// - Sends a `BEGIN` frame to the server with `transaction:<transaction_id>`
    ///   header. Subsequent `SEND`, `ACK`, and `NACK` frames may include this
    ///   transaction id to group them into the transaction. The transaction must
    ///   be finalized with either `commit` or `abort`.
    pub async fn begin(&self, transaction_id: &str) -> Result<(), ConnError> {
        self.send_transaction_frame("BEGIN", transaction_id).await
    }

    /// Commit a transaction.
    ///
    /// Parameters
    /// - `transaction_id`: the transaction identifier previously passed to `begin`.
    ///
    /// Behavior
    /// - Sends a `COMMIT` frame to the server with `transaction:<transaction_id>`
    ///   header. All operations within the transaction are applied atomically.
    pub async fn commit(&self, transaction_id: &str) -> Result<(), ConnError> {
        self.send_transaction_frame("COMMIT", transaction_id).await
    }

    /// Abort a transaction.
    ///
    /// Parameters
    /// - `transaction_id`: the transaction identifier previously passed to `begin`.
    ///
    /// Behavior
    /// - Sends an `ABORT` frame to the server with `transaction:<transaction_id>`
    ///   header. All operations within the transaction are discarded.
    pub async fn abort(&self, transaction_id: &str) -> Result<(), ConnError> {
        self.send_transaction_frame("ABORT", transaction_id).await
    }

    /// Receive the next frame from the server.
    ///
    /// Returns `Some(ReceivedFrame::Frame(..))` for normal frames (MESSAGE, etc.),
    /// `Some(ReceivedFrame::Error(..))` for ERROR frames, or `None` if the
    /// connection has been closed.
    ///
    /// # Example
    ///
    /// ```ignore
    /// use iridium_stomp::ReceivedFrame;
    ///
    /// while let Some(received) = conn.next_frame().await {
    ///     match received {
    ///         ReceivedFrame::Frame(frame) => {
    ///             println!("Got {}: {:?}", frame.command, frame.body);
    ///         }
    ///         ReceivedFrame::Error(err) => {
    ///             eprintln!("Server error: {}", err);
    ///             break;
    ///         }
    ///     }
    /// }
    /// ```
    pub async fn next_frame(&self) -> Option<ReceivedFrame> {
        let mut rx = self.inbound_rx.lock().await;
        let frame = rx.recv().await?;

        // Convert ERROR frames to ServerError for better ergonomics
        if frame.command == "ERROR" {
            Some(ReceivedFrame::Error(ServerError::from_frame(frame)))
        } else {
            Some(ReceivedFrame::Frame(frame))
        }
    }

    pub async fn close(self) {
        // Signal the background task to shutdown by broadcasting on the
        // shutdown channel. Consumers may await task termination separately
        // if needed.
        let _ = self.shutdown_tx.send(());
    }
}

fn current_millis() -> u64 {
    use std::time::{SystemTime, UNIX_EPOCH};
    SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .map(|d| d.as_millis() as u64)
        .unwrap_or(0)
}

#[cfg(test)]
mod tests {
    use super::*;
    use tokio::sync::mpsc;

    // Helper to build a MESSAGE frame with given message-id and subscription/destination headers
    fn make_message(
        message_id: &str,
        subscription: Option<&str>,
        destination: Option<&str>,
    ) -> Frame {
        let mut f = Frame::new("MESSAGE");
        f = f.header("message-id", message_id);
        if let Some(s) = subscription {
            f = f.header("subscription", s);
        }
        if let Some(d) = destination {
            f = f.header("destination", d);
        }
        f
    }

    #[tokio::test]
    async fn test_cumulative_ack_removes_prefix() {
        // setup channels
        let (out_tx, mut out_rx) = mpsc::channel::<StompItem>(8);
        let (_in_tx, in_rx) = mpsc::channel::<Frame>(8);
        let (shutdown_tx, _) = broadcast::channel::<()>(1);

        let subscriptions: Arc<Mutex<Subscriptions>> = Arc::new(Mutex::new(HashMap::new()));
        let pending: Arc<Mutex<PendingMap>> = Arc::new(Mutex::new(HashMap::new()));

        let sub_id_counter = Arc::new(AtomicU64::new(1));

        // create a subscription entry s1 with client (cumulative) ack
        let (sub_sender, _sub_rx) = mpsc::channel::<Frame>(4);
        {
            let mut map = subscriptions.lock().await;
            map.insert(
                "/queue/x".to_string(),
                vec![SubscriptionEntry {
                    id: "s1".to_string(),
                    sender: sub_sender,
                    ack: "client".to_string(),
                    headers: Vec::new(),
                }],
            );
        }

        // fill pending queue for s1: m1,m2,m3
        {
            let mut p = pending.lock().await;
            let mut q = VecDeque::new();
            q.push_back((
                "m1".to_string(),
                make_message("m1", Some("s1"), Some("/queue/x")),
            ));
            q.push_back((
                "m2".to_string(),
                make_message("m2", Some("s1"), Some("/queue/x")),
            ));
            q.push_back((
                "m3".to_string(),
                make_message("m3", Some("s1"), Some("/queue/x")),
            ));
            p.insert("s1".to_string(), q);
        }

        let conn = Connection {
            outbound_tx: out_tx,
            inbound_rx: Arc::new(Mutex::new(in_rx)),
            shutdown_tx,
            subscriptions: subscriptions.clone(),
            sub_id_counter,
            pending: pending.clone(),
            pending_receipts: Arc::new(Mutex::new(HashMap::new())),
        };

        // ack m2 cumulatively: should remove m1 and m2, leaving m3
        conn.ack("s1", "m2").await.expect("ack failed");

        // verify pending for s1 contains only m3
        {
            let p = pending.lock().await;
            let q = p.get("s1").expect("missing s1");
            assert_eq!(q.len(), 1);
            assert_eq!(q.front().unwrap().0, "m3");
        }

        // verify an ACK frame was emitted
        if let Some(item) = out_rx.recv().await {
            match item {
                StompItem::Frame(f) => assert_eq!(f.command, "ACK"),
                _ => panic!("expected frame"),
            }
        } else {
            panic!("no outbound frame sent")
        }
    }

    #[tokio::test]
    async fn test_client_individual_ack_removes_only_one() {
        // setup channels
        let (out_tx, mut out_rx) = mpsc::channel::<StompItem>(8);
        let (_in_tx, in_rx) = mpsc::channel::<Frame>(8);
        let (shutdown_tx, _) = broadcast::channel::<()>(1);

        let subscriptions: Arc<Mutex<Subscriptions>> = Arc::new(Mutex::new(HashMap::new()));
        let pending: Arc<Mutex<PendingMap>> = Arc::new(Mutex::new(HashMap::new()));

        let sub_id_counter = Arc::new(AtomicU64::new(1));

        // create a subscription entry s2 with client-individual ack
        let (sub_sender, _sub_rx) = mpsc::channel::<Frame>(4);
        {
            let mut map = subscriptions.lock().await;
            map.insert(
                "/queue/y".to_string(),
                vec![SubscriptionEntry {
                    id: "s2".to_string(),
                    sender: sub_sender,
                    ack: "client-individual".to_string(),
                    headers: Vec::new(),
                }],
            );
        }

        // fill pending queue for s2: a,b,c
        {
            let mut p = pending.lock().await;
            let mut q = VecDeque::new();
            q.push_back((
                "a".to_string(),
                make_message("a", Some("s2"), Some("/queue/y")),
            ));
            q.push_back((
                "b".to_string(),
                make_message("b", Some("s2"), Some("/queue/y")),
            ));
            q.push_back((
                "c".to_string(),
                make_message("c", Some("s2"), Some("/queue/y")),
            ));
            p.insert("s2".to_string(), q);
        }

        let conn = Connection {
            outbound_tx: out_tx,
            inbound_rx: Arc::new(Mutex::new(in_rx)),
            shutdown_tx,
            subscriptions: subscriptions.clone(),
            sub_id_counter,
            pending: pending.clone(),
            pending_receipts: Arc::new(Mutex::new(HashMap::new())),
        };

        // ack only 'b' individually
        conn.ack("s2", "b").await.expect("ack failed");

        // verify pending for s2 contains a and c
        {
            let p = pending.lock().await;
            let q = p.get("s2").expect("missing s2");
            assert_eq!(q.len(), 2);
            assert_eq!(q[0].0, "a");
            assert_eq!(q[1].0, "c");
        }

        // verify an ACK frame was emitted
        if let Some(item) = out_rx.recv().await {
            match item {
                StompItem::Frame(f) => assert_eq!(f.command, "ACK"),
                _ => panic!("expected frame"),
            }
        } else {
            panic!("no outbound frame sent")
        }
    }

    #[tokio::test]
    async fn test_subscription_receive_delivers_message() {
        // setup channels
        let (out_tx, _out_rx) = mpsc::channel::<StompItem>(8);
        let (_in_tx, in_rx) = mpsc::channel::<Frame>(8);
        let (shutdown_tx, _) = broadcast::channel::<()>(1);

        let subscriptions: Arc<Mutex<Subscriptions>> = Arc::new(Mutex::new(HashMap::new()));
        let pending: Arc<Mutex<PendingMap>> = Arc::new(Mutex::new(HashMap::new()));

        let sub_id_counter = Arc::new(AtomicU64::new(1));

        let conn = Connection {
            outbound_tx: out_tx,
            inbound_rx: Arc::new(Mutex::new(in_rx)),
            shutdown_tx,
            subscriptions: subscriptions.clone(),
            sub_id_counter,
            pending: pending.clone(),
            pending_receipts: Arc::new(Mutex::new(HashMap::new())),
        };

        // subscribe
        let subscription = conn
            .subscribe("/queue/test", AckMode::Auto)
            .await
            .expect("subscribe failed");

        // find the sender stored in the subscriptions map and push a message
        {
            let map = conn.subscriptions.lock().await;
            let vec = map.get("/queue/test").expect("missing subscription vec");
            let sender = &vec[0].sender;
            let f = make_message("m1", Some(&vec[0].id), Some("/queue/test"));
            sender.try_send(f).expect("send to subscription failed");
        }

        // consume from the subscription receiver
        let mut rx = subscription.into_receiver();
        if let Some(received) = rx.recv().await {
            assert_eq!(received.command, "MESSAGE");
            // message-id header should be present
            let mut found = false;
            for (k, _v) in &received.headers {
                if k.to_lowercase() == "message-id" {
                    found = true;
                    break;
                }
            }
            assert!(found, "message-id header missing");
        } else {
            panic!("no message received on subscription")
        }
    }

    #[tokio::test]
    async fn test_subscription_ack_removes_pending_and_sends_ack() {
        // setup channels
        let (out_tx, mut out_rx) = mpsc::channel::<StompItem>(8);
        let (_in_tx, in_rx) = mpsc::channel::<Frame>(8);
        let (shutdown_tx, _) = broadcast::channel::<()>(1);

        let subscriptions: Arc<Mutex<Subscriptions>> = Arc::new(Mutex::new(HashMap::new()));
        let pending: Arc<Mutex<PendingMap>> = Arc::new(Mutex::new(HashMap::new()));

        let sub_id_counter = Arc::new(AtomicU64::new(1));

        let conn = Connection {
            outbound_tx: out_tx,
            inbound_rx: Arc::new(Mutex::new(in_rx)),
            shutdown_tx,
            subscriptions: subscriptions.clone(),
            sub_id_counter,
            pending: pending.clone(),
            pending_receipts: Arc::new(Mutex::new(HashMap::new())),
        };

        // subscribe with client ack
        let subscription = conn
            .subscribe("/queue/ack", AckMode::Client)
            .await
            .expect("subscribe failed");

        let sub_id = subscription.id().to_string();

        // drain any initial outbound frames (SUBSCRIBE) emitted by subscribe()
        while out_rx.try_recv().is_ok() {}

        // populate pending queue for this subscription
        {
            let mut p = conn.pending.lock().await;
            let mut q = VecDeque::new();
            q.push_back((
                "mid-1".to_string(),
                make_message("mid-1", Some(&sub_id), Some("/queue/ack")),
            ));
            p.insert(sub_id.clone(), q);
        }

        // ack the message via the subscription helper
        subscription.ack("mid-1").await.expect("ack failed");

        // ensure pending queue no longer contains the message
        {
            let p = conn.pending.lock().await;
            assert!(p.get(&sub_id).is_none() || p.get(&sub_id).unwrap().is_empty());
        }

        // verify an ACK frame was emitted
        if let Some(item) = out_rx.recv().await {
            match item {
                StompItem::Frame(f) => assert_eq!(f.command, "ACK"),
                _ => panic!("expected frame"),
            }
        } else {
            panic!("no outbound frame sent")
        }
    }

    // Helper function to create a test connection and output receiver
    fn setup_test_connection() -> (Connection, mpsc::Receiver<StompItem>) {
        let (out_tx, out_rx) = mpsc::channel::<StompItem>(8);
        let (_in_tx, in_rx) = mpsc::channel::<Frame>(8);
        let (shutdown_tx, _) = broadcast::channel::<()>(1);

        let subscriptions: Arc<Mutex<Subscriptions>> = Arc::new(Mutex::new(HashMap::new()));
        let pending: Arc<Mutex<PendingMap>> = Arc::new(Mutex::new(HashMap::new()));
        let sub_id_counter = Arc::new(AtomicU64::new(1));

        let conn = Connection {
            outbound_tx: out_tx,
            inbound_rx: Arc::new(Mutex::new(in_rx)),
            shutdown_tx,
            subscriptions,
            sub_id_counter,
            pending,
            pending_receipts: Arc::new(Mutex::new(HashMap::new())),
        };

        (conn, out_rx)
    }

    // Helper function to verify a frame with a transaction header
    fn verify_transaction_frame(frame: Frame, expected_command: &str, expected_tx_id: &str) {
        assert_eq!(frame.command, expected_command);
        assert!(
            frame
                .headers
                .iter()
                .any(|(k, v)| k == "transaction" && v == expected_tx_id),
            "transaction header with id '{}' not found",
            expected_tx_id
        );
    }

    #[tokio::test]
    async fn test_begin_transaction_sends_frame() {
        let (conn, mut out_rx) = setup_test_connection();

        conn.begin("tx1").await.expect("begin failed");

        // verify BEGIN frame was emitted
        if let Some(StompItem::Frame(f)) = out_rx.recv().await {
            verify_transaction_frame(f, "BEGIN", "tx1");
        } else {
            panic!("no outbound frame sent")
        }
    }

    #[tokio::test]
    async fn test_commit_transaction_sends_frame() {
        let (conn, mut out_rx) = setup_test_connection();

        conn.commit("tx1").await.expect("commit failed");

        // verify COMMIT frame was emitted
        if let Some(StompItem::Frame(f)) = out_rx.recv().await {
            verify_transaction_frame(f, "COMMIT", "tx1");
        } else {
            panic!("no outbound frame sent")
        }
    }

    #[tokio::test]
    async fn test_abort_transaction_sends_frame() {
        let (conn, mut out_rx) = setup_test_connection();

        conn.abort("tx1").await.expect("abort failed");

        // verify ABORT frame was emitted
        if let Some(StompItem::Frame(f)) = out_rx.recv().await {
            verify_transaction_frame(f, "ABORT", "tx1");
        } else {
            panic!("no outbound frame sent")
        }
    }

    #[test]
    fn test_extract_destination_from_error_header() {
        // When ERROR frame has destination header, extract it directly
        let frame = Frame::new("ERROR")
            .header("message", "AMQ339016: Error creating STOMP subscription")
            .header("destination", "/topic/test.restricted");

        let dest = extract_destination_from_error(&frame);
        assert_eq!(dest, Some("/topic/test.restricted".to_string()));
    }

    #[test]
    fn test_extract_destination_from_error_message() {
        // When destination is in message header text
        let frame = Frame::new("ERROR").header(
            "message",
            "AMQ339016: Error creating subscription for /topic/test.restricted",
        );

        let dest = extract_destination_from_error(&frame);
        assert_eq!(dest, Some("/topic/test.restricted".to_string()));
    }

    #[test]
    fn test_extract_destination_from_error_body() {
        // When destination is in body text
        let frame = Frame::new("ERROR")
            .header("message", "AMQ339016: Error creating subscription")
            .set_body(b"User guest is not authorized for /queue/orders".to_vec());

        let dest = extract_destination_from_error(&frame);
        assert_eq!(dest, Some("/queue/orders".to_string()));
    }

    #[test]
    fn test_extract_destination_from_error_none() {
        // When no destination can be identified
        let frame = Frame::new("ERROR").header("message", "Generic error without destination info");

        let dest = extract_destination_from_error(&frame);
        assert_eq!(dest, None);
    }

    #[test]
    fn test_extract_destination_from_error_with_trailing_punct() {
        // When destination has trailing punctuation
        let frame = Frame::new("ERROR").header(
            "message",
            "Error for /topic/events, please check permissions",
        );

        let dest = extract_destination_from_error(&frame);
        assert_eq!(dest, Some("/topic/events".to_string()));
    }

    #[test]
    fn test_extract_subscription_id_from_error_artemis_format() {
        // Artemis format: "AMQ339016 Error creating subscription 1"
        let frame =
            Frame::new("ERROR").header("message", "AMQ339016 Error creating subscription 1");

        let sub_id = extract_subscription_id_from_error(&frame);
        assert_eq!(sub_id, Some("1".to_string()));
    }

    #[test]
    fn test_extract_subscription_id_from_error_numeric() {
        // Multiple digit subscription ID
        let frame = Frame::new("ERROR").header("message", "Error for subscription 123 on server");

        let sub_id = extract_subscription_id_from_error(&frame);
        assert_eq!(sub_id, Some("123".to_string()));
    }

    #[test]
    fn test_extract_subscription_id_from_error_none() {
        // No subscription ID in error
        let frame = Frame::new("ERROR").header("message", "Generic connection error");

        let sub_id = extract_subscription_id_from_error(&frame);
        assert_eq!(sub_id, None);
    }

    #[tokio::test]
    async fn test_lookup_destination_by_sub_id() {
        let subscriptions: Arc<Mutex<Subscriptions>> = Arc::new(Mutex::new(HashMap::new()));
        let (sender, _rx) = mpsc::channel::<Frame>(4);

        // Add a subscription
        {
            let mut map = subscriptions.lock().await;
            map.insert(
                "/topic/test.restricted".to_string(),
                vec![SubscriptionEntry {
                    id: "1".to_string(),
                    sender,
                    ack: "auto".to_string(),
                    headers: Vec::new(),
                }],
            );
        }

        // Should find the destination
        let dest = lookup_destination_by_sub_id("1", &subscriptions).await;
        assert_eq!(dest, Some("/topic/test.restricted".to_string()));

        // Should not find non-existent subscription
        let dest = lookup_destination_by_sub_id("999", &subscriptions).await;
        assert_eq!(dest, None);
    }
}