lightstreamer_rs/client/

implementation.rs

use crate::subscription::{ItemUpdate, Snapshot, Subscription, SubscriptionMode};

use crate::client::Transport;
pub(crate) use crate::client::listener::ClientListener;
use crate::client::message_listener::ClientMessageListener;
use crate::client::model::{ClientStatus, DisconnectionType, LogType};
use crate::client::request::SubscriptionRequest;
use crate::client::utils::get_subscription_by_id;
use crate::connection::{ConnectionDetails, ConnectionOptions};
use crate::connection::{ConnectionManager, ConnectionState, HeartbeatConfig, ReconnectionConfig};
use crate::utils::{LightstreamerError, clean_message, parse_arguments};
use cookie::Cookie;
use futures_util::{SinkExt, StreamExt};
use std::collections::HashMap;
use std::fmt::{self, Debug, Formatter};
use std::sync::Arc;
use std::time::Duration;
use tokio::sync::mpsc::channel;
use tokio::sync::{
    Mutex, Notify,
    mpsc::{Receiver, Sender},
};
use tokio::time::Instant as TokioInstant;
use tokio_tungstenite::{
    connect_async,
    tungstenite::{
        Message,
        http::{HeaderName, HeaderValue, Request},
    },
};
use tracing::{Level, debug, error, info, instrument, trace, warn};
use url::Url;

/// Facade class for the management of the communication to Lightstreamer Server. Used to provide
/// configuration settings, event handlers, operations for the control of the connection lifecycle,
/// Subscription handling and to send messages.
///
/// An instance of `LightstreamerClient` handles the communication with Lightstreamer Server on a
/// specified endpoint. Hence, it hosts one "Session"; or, more precisely, a sequence of Sessions,
/// since any Session may fail and be recovered, or it can be interrupted on purpose. So, normally,
/// a single instance of `LightstreamerClient` is needed.
///
/// However, multiple instances of `LightstreamerClient` can be used, toward the same or multiple
/// endpoints.
///
/// You can listen to the events generated by a session by registering an event listener, such as
/// `ClientListener` or `SubscriptionListener`. These listeners allow you to handle various events,
/// such as session creation, connection status, subscription updates, and server messages. However,
/// you should be aware that the event notifications are dispatched by a single thread, the so-called
/// event thread. This means that if the operations of a listener are slow or blocking, they will
/// delay the processing of the other listeners and affect the performance of your application.
/// Therefore, you should delegate any slow or blocking operations to a dedicated thread, and keep
/// the listener methods as fast and simple as possible. Note that even if you create multiple
/// instances of `LightstreamerClient`, they will all use a single event thread, that is shared
/// among them.
///
/// # Parameters
///
/// * `server_address`: the address of the Lightstreamer Server to which this `LightstreamerClient`
///   will connect to. It is possible to specify it later by using `None` here. See
///   `ConnectionDetails.setServerAddress()` for details.
/// * `adapter_set`: the name of the Adapter Set mounted on Lightstreamer Server to be used to handle
///   all requests in the Session associated with this `LightstreamerClient`. It is possible not to
///   specify it at all or to specify it later by using `None` here. See `ConnectionDetails.setAdapterSet()`
///   for details.
///
/// # Raises
///
/// * `IllegalArgumentException`: if a not valid address is passed. See `ConnectionDetails.setServerAddress()`
///   for details.
pub struct LightstreamerClient {
    /// The address of the Lightstreamer Server to which this `LightstreamerClient` will connect.
    server_address: Option<String>,
    /// The name of the Adapter Set mounted on Lightstreamer Server to be used to handle all
    /// requests in the Session associated with this `LightstreamerClient`.
    adapter_set: Option<String>,
    /// Data object that contains the details needed to open a connection to a Lightstreamer Server.
    /// This instance is set up by the `LightstreamerClient` object at its own creation. Properties
    /// of this object can be overwritten by values received from a Lightstreamer Server.
    pub connection_details: ConnectionDetails,
    /// Data object that contains options and policies for the connection to the server. This instance
    /// is set up by the `LightstreamerClient` object at its own creation. Properties of this object
    /// can be overwritten by values received from a Lightstreamer Server.
    pub connection_options: ConnectionOptions,
    /// A list of listeners that will receive events from the `LightstreamerClient` instance.
    listeners: Vec<Box<dyn ClientListener>>,
    /// A list containing all the `Subscription` instances that are currently "active" on this
    /// `LightstreamerClient`.
    subscriptions: Vec<Subscription>,
    /// The current status of the client.
    status: ClientStatus,
    /// Logging Type to be used
    logging: LogType,
    /// The sender that can be used to subscribe/unsubscribe
    pub subscription_sender: Sender<SubscriptionRequest>,
    /// The receiver used for subscribe/unsubsribe
    subscription_receiver: Receiver<SubscriptionRequest>,
    /// Connection manager for automatic reconnection
    connection_manager: Option<Arc<ConnectionManager>>,
    /// Configuration for reconnection behavior
    reconnection_config: ReconnectionConfig,
    /// Configuration for heartbeat monitoring
    heartbeat_config: HeartbeatConfig,
    /// Whether automatic reconnection is enabled
    auto_reconnect_enabled: bool,
}

impl Debug for LightstreamerClient {
    fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
        f.debug_struct("LightstreamerClient")
            .field("server_address", &self.server_address)
            .field("adapter_set", &self.adapter_set)
            .field("connection_details", &self.connection_details)
            .field("connection_options", &self.connection_options)
            .field("listeners", &self.listeners)
            .field("subscriptions", &self.subscriptions)
            .finish()
    }
}

impl LightstreamerClient {
    /// A constant string representing the name of the library.
    pub const LIB_NAME: &'static str = "rust_client";

    /// A constant string representing the version of the library.
    pub const LIB_VERSION: &'static str = "0.1.0";

    //
    // Constants for WebSocket connection.
    //
    /// WebSocket key used in the WebSocket handshake process.
    /// This is a base64-encoded value that is used to establish the WebSocket connection.
    pub const SEC_WEBSOCKET_KEY: &'static str = "PNDUibe9ex7PnsrLbt0N4w==";

    /// WebSocket protocol identifier for the TLCP protocol used by Lightstreamer.
    /// This identifies the specific subprotocol used over the WebSocket connection.
    pub const SEC_WEBSOCKET_PROTOCOL: &'static str = "TLCP-2.4.0.lightstreamer.com";

    /// WebSocket version used for the connection.
    /// Version 13 is the standard version used in modern WebSocket implementations.
    pub const SEC_WEBSOCKET_VERSION: &'static str = "13";

    /// WebSocket upgrade header value.
    /// Used to indicate that the client wishes to upgrade from HTTP to WebSocket protocol.
    pub const SEC_WEBSOCKET_UPGRADE: &'static str = "websocket";

    /// A constant string representing the version of the TLCP protocol used by the library.
    pub const TLCP_VERSION: &'static str = "TLCP-2.4.0";

    /// Grace period (milliseconds) added on top of the keepalive interval before
    /// the connection is considered dead by the liveness watchdog.
    ///
    /// The server guarantees a message (data or `probe`) at least every keepalive
    /// interval; the grace absorbs network jitter and scheduling delays.
    pub const KEEPALIVE_GRACE_MS: u64 = 5_000;

    /// Static method that can be used to share cookies between connections to the Server (performed by
    /// this library) and connections to other sites that are performed by the application. With this
    /// method, cookies received by the application can be added (or replaced if already present) to
    /// the cookie set used by the library to access the Server. Obviously, only cookies whose domain
    /// is compatible with the Server domain will be used internally.
    ///
    /// This method should be invoked before calling the `LightstreamerClient.connect()` method.
    /// However it can be invoked at any time; it will affect the internal cookie set immediately
    /// and the sending of cookies on the next HTTP request or WebSocket establishment.
    ///
    /// # Parameters
    ///
    /// * `uri`: the URI from which the supplied cookies were received. It cannot be `None`.
    /// * `cookies`: an instance of `http.cookies.SimpleCookie`.
    ///
    /// See also `getCookies()`
    pub fn add_cookies(_uri: &str, _cookies: &Cookie) {
        // Implementation for add_cookies
        unimplemented!("Implement mechanism to add cookies to LightstreamerClient");
    }

    /// Adds a listener that will receive events from the `LightstreamerClient` instance.
    ///
    /// The same listener can be added to several different `LightstreamerClient` instances.
    ///
    /// A listener can be added at any time. A call to add a listener already present will be ignored.
    ///
    /// # Parameters
    ///
    /// * `listener`: An object that will receive the events as documented in the `ClientListener`
    ///   interface.
    ///
    /// See also `removeListener()`
    pub fn add_listener(&mut self, listener: Box<dyn ClientListener>) {
        self.listeners.push(listener);
    }

    /// Packs s string with the necessary parameters for a subscription request.
    ///
    /// # Parameters
    ///
    /// * `subscription`: The subscription for which to get the parameters.
    /// * `request_id`: The request ID to use in the parameters.
    ///
    fn get_subscription_params(
        subscription: &Subscription,
        request_id: usize,
    ) -> Result<String, LightstreamerError> {
        let ls_req_id = request_id.to_string();
        let ls_sub_id = subscription.id.to_string();
        let ls_mode = subscription.get_mode().to_string();
        let ls_group = match subscription.get_item_group() {
            Some(item_group) => item_group.to_string(),
            None => match subscription.get_items() {
                Some(items) => items.join(" "),
                None => {
                    return Err(LightstreamerError::invalid_argument(
                        "No item group or items found in subscription.",
                    ));
                }
            },
        };
        let ls_schema = match subscription.get_field_schema() {
            Some(field_schema) => field_schema.to_string(),
            None => match subscription.get_fields() {
                Some(fields) => fields.join(" "),
                None => {
                    return Err(LightstreamerError::invalid_argument(
                        "No field schema or fields found in subscription.",
                    ));
                }
            },
        };
        let ls_data_adapter = match subscription.get_data_adapter() {
            Some(data_adapter) => data_adapter.to_string(),
            None => "".to_string(),
        };
        let ls_snapshot = subscription
            .get_requested_snapshot()
            .unwrap_or_default()
            .to_string();
        //
        // Prepare the subscription request.
        //
        let mut params: Vec<(&str, &str)> = vec![
            ("LS_data_adapter", &ls_data_adapter),
            ("LS_reqId", &ls_req_id),
            ("LS_op", "add"),
            ("LS_subId", &ls_sub_id),
            ("LS_mode", &ls_mode),
            ("LS_group", &ls_group),
            ("LS_schema", &ls_schema),
            ("LS_ack", "false"),
        ];
        // Remove the data adapter parameter if not specified.
        if ls_data_adapter.is_empty() {
            params.remove(0);
        }
        if !ls_snapshot.is_empty() {
            params.push(("LS_snapshot", &ls_snapshot));
        }

        Ok(serde_urlencoded::to_string(&params)?)
    }

    fn get_unsubscription_params(
        subscription_id: usize,
        request_id: usize,
    ) -> Result<String, LightstreamerError> {
        let ls_req_id = request_id.to_string();
        let ls_sub_id = subscription_id.to_string();
        //
        // Prepare the unsubscription request.
        //
        let params: Vec<(&str, &str)> = vec![
            ("LS_reqId", &ls_req_id),
            ("LS_op", "delete"),
            ("LS_subId", &ls_sub_id),
        ];

        Ok(serde_urlencoded::to_string(&params)?)
    }

    /// Operation method that requests to open a Session against the configured Lightstreamer Server.
    ///
    /// When `connect()` is called, unless a single transport was forced through `ConnectionOptions.setForcedTransport()`,
    /// the so called "Stream-Sense" mechanism is started: if the client does not receive any answer
    /// for some seconds from the streaming connection, then it will automatically open a polling
    /// connection.
    ///
    /// A polling connection may also be opened if the environment is not suitable for a streaming
    /// connection.
    ///
    /// Note that as "polling connection" we mean a loop of polling requests, each of which requires
    /// opening a synchronous (i.e. not streaming) connection to Lightstreamer Server.
    ///
    /// Note that the request to connect is accomplished by the client in a separate thread; this
    /// means that an invocation to `getStatus()` right after `connect()` might not reflect the change
    /// yet.
    ///
    /// When the request to connect is finally being executed, if the current status of the client
    /// is not `DISCONNECTED`, then nothing will be done.
    ///
    /// # Raises
    ///
    /// * `IllegalStateException`: if no server address was configured.
    ///
    /// See also `getStatus()`
    ///
    /// See also `disconnect()`
    ///
    /// See also `ClientListener.onStatusChange()`
    ///
    /// See also `ConnectionDetails.setServerAddress()`
    #[instrument(level = "trace")]
    pub async fn connect(
        client: Arc<Mutex<Self>>,
        shutdown_signal: Arc<Notify>,
    ) -> Result<(), LightstreamerError> {
        // If auto-reconnect is enabled, use ConnectionManager
        let auto_reconnect_enabled = {
            let client_guard = client.lock().await;
            client_guard.auto_reconnect_enabled
        };

        if auto_reconnect_enabled {
            let (reconnection_config, heartbeat_config) = {
                let client_guard = client.lock().await;
                (
                    client_guard.reconnection_config.clone(),
                    client_guard.heartbeat_config.clone(),
                )
            };

            let weak_client = Arc::downgrade(&client);
            let connection_manager: Arc<ConnectionManager> =
                ConnectionManager::new(weak_client, reconnection_config, heartbeat_config);

            {
                let mut client_guard = client.lock().await;
                client_guard.connection_manager = Some(connection_manager.clone());
            }

            // Use direct connection method
            return client.lock().await.connect_direct(shutdown_signal).await;
        }

        // Fallback to direct connection without auto-reconnect
        let mut client_guard = client.lock().await;
        client_guard.connect_direct(shutdown_signal).await
    }

    /// Direct connection method without automatic reconnection
    #[instrument(level = "trace")]
    pub async fn connect_direct(
        &mut self,
        shutdown_signal: Arc<Notify>,
    ) -> Result<(), LightstreamerError> {
        // Check if the server address is configured.
        if self.server_address.is_none() {
            return Err(LightstreamerError::invalid_state(
                "No server address was configured.",
            ));
        }
        //
        // Only WebSocket streaming transport is currently supported.
        //
        let forced_transport = self.connection_options.get_forced_transport();
        if forced_transport.is_none_or(|t| *t != Transport::WsStreaming) {
            return Err(LightstreamerError::invalid_state(
                "Only WebSocket streaming transport is currently supported.",
            ));
        }
        //
        // Convert the HTTP URL to a WebSocket URL.
        //
        let http_url = self
            .connection_details
            .get_server_address()
            .ok_or_else(|| LightstreamerError::invalid_state("Server address is not set."))?;
        let mut url = Url::parse(http_url).map_err(|e| {
            LightstreamerError::invalid_state(format!("Failed to parse server address URL: {}", e))
        })?;
        match url.scheme() {
            "http" => url.set_scheme("ws").map_err(|()| {
                LightstreamerError::invalid_state("Failed to set scheme to ws for WebSocket URL.")
            })?,
            "https" => url.set_scheme("wss").map_err(|()| {
                LightstreamerError::invalid_state("Failed to set scheme to wss for WebSocket URL.")
            })?,
            invalid_scheme => {
                return Err(LightstreamerError::invalid_state(format!(
                    "Unsupported scheme '{}' found when converting HTTP URL to WebSocket URL.",
                    invalid_scheme
                )));
            }
        }
        let ws_url = url.as_str();

        // Build the WebSocket request with the necessary headers.
        let request = Request::builder()
            .uri(ws_url)
            .header(
                HeaderName::from_static("connection"),
                HeaderValue::from_static("Upgrade"),
            )
            .header(
                HeaderName::from_static("host"),
                HeaderValue::from_str(url.host_str().unwrap_or("localhost")).map_err(|err| {
                    LightstreamerError::invalid_state(format!(
                        "Invalid header value for header with name 'host': {}",
                        err
                    ))
                })?,
            )
            .header(
                HeaderName::from_static("sec-websocket-key"),
                HeaderValue::from_static(Self::SEC_WEBSOCKET_KEY),
            )
            .header(
                HeaderName::from_static("sec-websocket-protocol"),
                HeaderValue::from_static(Self::SEC_WEBSOCKET_PROTOCOL),
            )
            .header(
                HeaderName::from_static("sec-websocket-version"),
                HeaderValue::from_static(Self::SEC_WEBSOCKET_VERSION),
            )
            .header(
                HeaderName::from_static("upgrade"),
                HeaderValue::from_static(Self::SEC_WEBSOCKET_UPGRADE),
            )
            .body(())?;

        // Connect to the Lightstreamer server using WebSocket.
        let ws_stream = match connect_async(request).await {
            Ok((ws_stream, response)) => {
                if let Some(server_header) = response.headers().get("server") {
                    self.make_log(
                        Level::INFO,
                        &format!(
                            "Connected to Lightstreamer server: {}",
                            server_header.to_str().unwrap_or("")
                        ),
                    );
                } else {
                    self.make_log(Level::INFO, "Connected to Lightstreamer server");
                }
                ws_stream
            }
            Err(err) => {
                return Err(LightstreamerError::connection(format!(
                    "Failed to connect to Lightstreamer server with WebSocket: {}",
                    err
                )));
            }
        };

        // Split the WebSocket stream into a write and a read stream.
        let (mut write_stream, mut read_stream) = ws_stream.split();

        //
        // Initiate communication with the server by sending a 'wsok' message.
        //
        write_stream.send(Message::Text("wsok".into())).await?;

        //
        // Start reading and processing messages from the server.
        //
        let mut is_connected = false;
        let mut request_id: usize = 0;
        let mut session_id: Option<String> = None;
        let mut subscription_id: usize = 0;
        let mut subscription_item_updates: HashMap<usize, HashMap<usize, ItemUpdate>> =
            HashMap::new();

        //
        // Liveness enforcement and reverse heartbeat.
        //
        // The server guarantees that some message (data or `probe`) is sent at
        // least every keepalive interval. If nothing arrives within the
        // effective keepalive window plus a grace period, the connection is
        // considered dead (half-open TCP) and an error is returned so the
        // caller can reconnect. The window starts from the configured
        // keepalive interval (if any) and is widened to the server-declared
        // keepalive from the `conok` message once known.
        //
        let configured_keepalive_ms = self.connection_options.get_keepalive_interval();
        let reverse_heartbeat_ms = self.connection_options.get_reverse_heartbeat_interval();
        let mut keepalive_timeout_ms: u64 = configured_keepalive_ms;
        let mut last_message_at = TokioInstant::now();

        let mut heartbeat_timer = if reverse_heartbeat_ms > 0 {
            let mut timer = tokio::time::interval(Duration::from_millis(reverse_heartbeat_ms));
            timer.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
            Some(timer)
        } else {
            None
        };

        loop {
            let liveness_deadline = last_message_at
                + Duration::from_millis(
                    keepalive_timeout_ms.saturating_add(Self::KEEPALIVE_GRACE_MS),
                );
            tokio::select! {
                message = read_stream.next() => {
                    last_message_at = TokioInstant::now();
                    match message {
                        Some(Ok(Message::Text(text))) => {
                            // Messages could include multiple submessages separated by /r/n.
                            // Split the message into submessages and process each one separately.
                            let submessages: Vec<&str> = text.split("\r\n")
                                .filter(|&line| !line.trim().is_empty()) // Filter out empty lines.
                                .collect();
                            for submessage in submessages {
                                let clean_text = clean_message(submessage);
                                let submessage_fields: Vec<&str> = clean_text.split(",").collect();
                                match *submessage_fields.first().unwrap_or(&"") {
                                    //
                                    // Errors from server.
                                    //
                                    //
                                    // Session-level error: the session is not usable — fail the
                                    // connection so the caller can reconnect. (The previous
                                    // `break` only exited the submessage loop, leaving a dead
                                    // session looping forever.)
                                    //
                                    "conerr" => {
                                        self.make_log( Level::ERROR, &format!("Received connection error from Lightstreamer server: {}", clean_text) );
                                        return Err(LightstreamerError::connection(format!(
                                            "connection error from server: {}",
                                            clean_text
                                        )));
                                    },
                                    //
                                    // Request-level error: the session survives — log and continue.
                                    //
                                    "reqerr" => {
                                        self.make_log( Level::ERROR, &format!("Received request error from Lightstreamer server: {}", clean_text) );
                                    },
                                    //
                                    // Session created successfully.
                                    //
                                    "conok" => {
                                        is_connected = true;
                                        // CONOK,<session-ID>,<request-limit>,<keep-alive>,<control-link>
                                        // Widen the liveness window to the server-declared keepalive
                                        // so enforcement works even when no keepalive was configured.
                                        if let Some(server_keepalive_ms) = submessage_fields
                                            .get(3)
                                            .and_then(|v| v.trim().parse::<u64>().ok())
                                            .filter(|v| *v > 0)
                                        {
                                            keepalive_timeout_ms = widen_keepalive_timeout(
                                                keepalive_timeout_ms,
                                                server_keepalive_ms,
                                            );
                                            self.make_log( Level::DEBUG, &format!("Server-declared keepalive interval: {} ms", server_keepalive_ms) );
                                        }
                                        if let Some(conok_session_id) = submessage_fields.get(1) {
                                            session_id = Some((*conok_session_id).to_string());
                                            self.make_log( Level::DEBUG, &format!("Session creation confirmed by server: {}", clean_text) );
                                            self.make_log( Level::DEBUG, &format!("Session created with ID: {:?}", conok_session_id) );
                                            //
                                            // Subscribe to the desired items.
                                            //
                                            while let Some(subscription) = self.subscriptions.get_mut(subscription_id) {
                                                //
                                                // Gather all the necessary subscription parameters.
                                                //
                                                subscription_id += 1;
                                                request_id += 1;
                                                subscription.id = subscription_id;
                                                subscription.id_sender.try_send(subscription_id)?;

                                                let encoded_params = match Self::get_subscription_params(subscription, request_id)
                                                {
                                                    Ok(params) => params,
                                                    Err(err) => {
                                                        return Err(err);
                                                    },
                                                };

                                                write_stream
                                                    .send(Message::Text(format!("control\r\n{}", encoded_params).into()))
                                                    .await?;
                                                debug!("Sent subscription request: '{}'", encoded_params);
                                            }
                                        } else {
                                            return Err(LightstreamerError::protocol(
                                                "Session ID not found in 'conok' message from server",
                                            ));
                                        }
                                    },
                                    //
                                    // Notifications from server.
                                    //
                                    "conf" | "cons" | "clientip" | "servname" | "prog" | "sync" | "eos" => {
                                        self.make_log( Level::INFO, &format!("Received notification from server: {}", clean_text) );
                                        // Don't do anything with these notifications for now.
                                    },
                                    "probe" => {
                                        self.make_log( Level::DEBUG, &format!("Received probe message from server: {}", clean_text ) );
                                    },
                                    "reqok" => {
                                        self.make_log( Level::DEBUG, &format!("Received reqok message from server: '{}'", clean_text ) );
                                    },
                                    //
                                    // Subscription confirmation from server.
                                    //
                                    "subok" => {
                                        self.make_log( Level::INFO, &format!("Subscription confirmed by server: '{}'", clean_text) );
                                    },
                                    //
                                    // Usubscription confirmation from server.
                                    //
                                    "unsub" => {
                                        self.make_log( Level::INFO, &format!("Unsubscription confirmed by server: '{}'", clean_text) );
                                    },
                                    //
                                    // Data updates from server.
                                    //
                                    "u" => {
                                        // Parse arguments from the received message.
                                        let arguments = parse_arguments(&clean_text);
                                        //
                                        // Extract the subscription from the first argument.
                                        //
                                        let subscription_index = arguments.get(1).unwrap_or(&"").parse::<usize>().unwrap_or(0);
                                        let subscription = match get_subscription_by_id(self.get_subscriptions(), subscription_index) {
                                            Some(subscription) => subscription,
                                            None => {
                                                self.make_log( Level::WARN, &format!("Subscription not found for index: {}", subscription_index) );
                                                continue;

                                            }
                                        };
                                        //
                                        // Extract the item from the second argument.
                                        //
                                        let item_index = arguments.get(2).unwrap_or(&"").parse::<usize>().unwrap_or(0);
                                        let item = match subscription.get_items() {
                                            Some(items) => items.get(item_index-1),
                                            None => {
                                                self.make_log( Level::WARN, &format!("No items found in subscription: {:?}", subscription) );
                                                continue;
                                            }
                                        };
                                        //
                                        // Determine if the update is a snapshot or real-time update based on the subscription parameters.
                                        //
                                        let is_snapshot = match subscription.get_requested_snapshot() {
                                            Some(ls_snapshot) => {
                                                match ls_snapshot {
                                                    Snapshot::No => false,
                                                    Snapshot::Yes => {
                                                        match subscription.get_mode() {
                                                            SubscriptionMode::Merge => {
                                                                if arguments.len() == 4 && arguments[3] == "$" {
                                                                    // EOS notification received
                                                                    true
                                                                } else {
                                                                    // If item doesn't exist in item_updates yet, the first update
                                                                    // is always a snapshot.
                                                                    if let Some(item_updates) = subscription_item_updates.get(&(subscription_index)) {
                                                                        if item_updates.get(&(item_index)).is_some() {
                                                                            // Item update already exists in item_updates, so it's not a snapshot.
                                                                            false
                                                                        } else {
                                                                            // Item update doesn't exist in item_updates, so the first update is always a snapshot.
                                                                            true
                                                                        }
                                                                    } else {
                                                                        // Item updates not found for subscription, so the first update is always a snapshot.
                                                                        true
                                                                    }
                                                                }
                                                            },
                                                            SubscriptionMode::Distinct | SubscriptionMode::Command => {
                                                                !subscription.is_subscribed()
                                                            },
                                                            _ => false,
                                                        }
                                                    },
                                                    _ => false,
                                                }
                                            },
                                            None => false,
                                        };

                                        // Extract the field values from the third argument.
                                        let field_values: Vec<&str> = arguments.get(3).unwrap_or(&"").split('|').collect();

                                        //
                                        // Get fields from subscription and create a HashMap of field names and values.
                                        //
                                        let subscription_fields = subscription.get_fields();
                                        let mut field_map: HashMap<String, Option<String>> = subscription_fields
                                            .map(|fields| fields.iter().map(|field_name| (field_name.to_string(), None)).collect())
                                            .unwrap_or_default();

                                        let mut field_index = 0;
                                        for value in field_values {
                                            match value {
                                                "" => {
                                                    // An empty value means the field is unchanged compared to the previous update of the same field.
                                                    if let Some(field_name) = subscription_fields.and_then(|fields| fields.get(field_index)) {
                                                        field_map.insert(field_name.to_string(), None);
                                                    }
                                                    field_index += 1;
                                                }
                                                "#" | "$" => {
                                                    // A value corresponding to a hash sign "#" or dollar sign "$" means the field is null or empty.
                                                    if let Some(field_name) = subscription_fields.and_then(|fields| fields.get(field_index)) {
                                                        field_map.insert(field_name.to_string(), Some("".to_string()));
                                                    }
                                                    field_index += 1;
                                                }
                                                value if value.starts_with('^') => {
                                                    let command = value.chars().nth(1).unwrap_or(' ');
                                                    match command {
                                                        '0'..='9' => {
                                                            let count = value[1..].parse().unwrap_or(0);
                                                            for i in 0..count {
                                                                if let Some(field_name) = subscription_fields.and_then(|fields| fields.get(field_index + i)) {
                                                                    field_map.insert(field_name.to_string(), None);
                                                                }
                                                            }
                                                            field_index += count;
                                                        }
                                                        'P' | 'T' => {
                                                            let diff_value = serde_urlencoded::from_str(&value[2..]).unwrap_or_else(|_| value[2..].to_string());
                                                            if let Some(field_name) = subscription_fields.and_then(|fields| fields.get(field_index))
                                                                && let Some(prev_value) = field_map.get(field_name).and_then(|v| v.as_ref()) {
                                                                    let new_value = match command {
                                                                        'P' => {
                                                                            // Apply JSON Patch
                                                                            let patch: serde_json::Value = serde_json::from_str(&diff_value).unwrap_or(serde_json::Value::Null);
                                                                            let mut prev_json: serde_json::Value = serde_json::from_str(prev_value).unwrap_or(serde_json::Value::Null);
                                                                            let patch_operations: Vec<json_patch::PatchOperation> = serde_json::from_value(patch).unwrap_or_default();
                                                                            if let Err(e) = json_patch::patch(&mut prev_json, &patch_operations) {
                                                                                tracing::warn!("Failed to apply JSON patch: {}", e);
                                                                            }
                                                                            prev_json.to_string()
                                                                        }
                                                                        'T' => {
                                                                            // Apply TLCP-diff
                                                                            //tlcp_diff::apply_diff(prev_value, &diff_value).unwrap_or_else(|_| prev_value.to_string())
                                                                            unimplemented!("Implement TLCP-diff");
                                                                        }
                                                                        _ => unreachable!(),
                                                                    };
                                                                    field_map.insert(field_name.to_string(), Some(new_value.to_string()));
                                                                }
                                                            field_index += 1;
                                                        }
                                                        _ => {
                                                            let decoded_value = serde_urlencoded::from_str(value).unwrap_or_else(|_| value.to_string());
                                                            if let Some(field_name) = subscription_fields.and_then(|fields| fields.get(field_index)) {
                                                                field_map.insert(field_name.to_string(), Some(decoded_value));
                                                            }
                                                            field_index += 1;
                                                        }
                                                    }
                                                }
                                                value if value.starts_with('{') => {
                                                    // in this case it is a json payload that we will let the consumer handle. In this case, it is important
                                                    // to preserve casing for parsing.
                                                    let original_json = parse_arguments(submessage).get(3).unwrap_or(&"").split('|').collect::<Vec<&str>>();
                                                    let mut payload = "";
                                                    for json in original_json.iter()
                                                    {
                                                        if json.is_empty() || *json == "#"
                                                        {
                                                            continue;
                                                        }

                                                        payload = json;
                                                    }

                                                    if let Some(field_name) = subscription_fields.and_then(|fields| fields.get(field_index)) {
                                                        field_map.insert(field_name.to_string(), Some(payload.to_string()));
                                                    }
                                                    field_index += 1;
                                                }
                                                _ => {
                                                    let decoded_value = serde_urlencoded::from_str(value).unwrap_or_else(|_| value.to_string());
                                                    if let Some(field_name) = subscription_fields.and_then(|fields| fields.get(field_index)) {
                                                        field_map.insert(field_name.to_string(), Some(decoded_value));
                                                    }
                                                    field_index += 1;
                                                }
                                            }
                                        }

                                        // Store only item_update's changed fields.
                                        let changed_fields: HashMap<String, String> = field_map.iter()
                                            .filter_map(|(k, v)| v.as_ref().map(|v| (k.clone(), v.clone())))
                                            .collect();

                                        //
                                        // Take the proper item_update from item_updates and update it with changed fields.
                                        // If the item_update doesn't exist yet, create a new one.
                                        //
                                        let current_item_update: ItemUpdate;
                                        match subscription_item_updates.get_mut(&(subscription_index)) {
                                            Some(item_updates) => match item_updates.get_mut(&(item_index)) {
                                                Some(item_update) => {
                                                    //
                                                    // Iterate changed_fields and update existing item_update.fields assigning the new values.
                                                    //
                                                    for (field_name, new_value) in &changed_fields {
                                                        if item_update.fields.contains_key(field_name) {
                                                            item_update.fields.insert((*field_name).clone(), Some(new_value.clone()));
                                                        }
                                                    }
                                                    item_update.changed_fields = changed_fields.clone();
                                                    item_update.is_snapshot = is_snapshot;
                                                    current_item_update = item_update.clone();
                                                },
                                                None => {
                                                    // Create a new item_update and add it to item_updates.
                                                    let item_update = ItemUpdate {
                                                        item_name: item.cloned(),
                                                        item_pos: item_index,
                                                        fields: field_map.clone(),
                                                        changed_fields: changed_fields.clone(),
                                                        is_snapshot,
                                                    };
                                                    current_item_update = item_update.clone();
                                                    item_updates.insert(item_index, item_update);
                                                }
                                            },
                                            None => {
                                                // Create a new item_update and add it to item_updates.
                                                let item_update = ItemUpdate {
                                                    item_name: item.cloned(),
                                                    item_pos: item_index,
                                                    fields: field_map,
                                                    changed_fields,
                                                    is_snapshot,
                                                };
                                                current_item_update = item_update.clone();
                                                let mut item_updates = HashMap::new();
                                                item_updates.insert(item_index, item_update);
                                                subscription_item_updates.insert(subscription_index, item_updates);
                                            }
                                        };

                                        // Get mutable subscription listeners directly.
                                        let subscription_listeners = subscription.get_listeners();

                                        // Iterate subscription listeners and call on_item_update for each listener.
                                        for listener in subscription_listeners {
                                            listener.on_item_update(&current_item_update);
                                        }
                                    }
                                    //
                                    // Connection confirmation from server.
                                    //
                                    "wsok" => {
                                        self.make_log( Level::INFO, &format!("Connection confirmed by server: '{}'", clean_text) );
                                        //
                                        // Request session creation.
                                        //
                                        let ls_adapter_set = match self.connection_details.get_adapter_set() {
                                            Some(adapter_set) => adapter_set,
                                            None => {
                                                return Err(LightstreamerError::invalid_state(
                                                    "No adapter set found in connection details.",
                                                ));
                                            },
                                        };
                                        let ls_send_sync = self.connection_options.get_send_sync().to_string();
                                        let ls_keepalive_millis = configured_keepalive_ms.to_string();
                                        let ls_inactivity_millis = reverse_heartbeat_ms.to_string();
                                        let mut params: Vec<(&str, &str)> = vec![
                                            ("LS_adapter_set", ls_adapter_set),
                                            ("LS_cid", "mgQkwtwdysogQz2BJ4Ji kOj2Bg"),
                                            ("LS_send_sync", &ls_send_sync),
                                        ];
                                        if configured_keepalive_ms > 0 {
                                            params.push(("LS_keepalive_millis", &ls_keepalive_millis));
                                        }
                                        if reverse_heartbeat_ms > 0 {
                                            params.push(("LS_inactivity_millis", &ls_inactivity_millis));
                                        }
                                        if let Some(user) = &self.connection_details.get_user() {
                                            params.push(("LS_user", user));
                                        }
                                        if let Some(password) = &self.connection_details.get_password() {
                                            params.push(("LS_password", password));
                                        }
                                        params.push(("LS_protocol", Self::TLCP_VERSION));
                                        let encoded_params = serde_urlencoded::to_string(&params)?;
                                        write_stream
                                            .send(Message::Text(format!("create_session\r\n{}\n", encoded_params).into()))
                                            .await?;
                                        self.make_log( Level::DEBUG, &format!("Sent create session request: '{}'", encoded_params) );
                                    },
                                    unexpected_message => {
                                        return Err(LightstreamerError::protocol(format!(
                                            "Unexpected message received from server: '{:?}' (full message: '{}')",
                                            unexpected_message, clean_text
                                        )));
                                    },
                                }
                            }
                        },
                        Some(Ok(non_text_message)) => {
                            return Err(LightstreamerError::protocol(format!(
                                "Unexpected non-text message from server: {:?}",
                                non_text_message
                            )));
                        },
                        Some(Err(err)) => {
                            return Err(LightstreamerError::protocol(format!(
                                "Error reading message from server: {}",
                                err
                            )));
                        },
                        None => {
                            self.make_log( Level::DEBUG, "No more messages from server" );
                            break;
                        },
                    }
                },
                Some(subscription_request) = self.subscription_receiver.recv() => {
                    request_id += 1;
                    // Process subscription requests.
                    if let Some(subscription) = subscription_request.subscription
                    {
                        self.subscriptions.push(subscription);

                        // if we are not connected yet, we will subscribe later
                        if !is_connected {
                            continue;
                        }

                        subscription_id += 1;
                        // SAFETY: We just pushed a subscription, so last_mut() must return Some.
                        // Using debug_assert to catch invariant violations during development.
                        let sub = self.subscriptions.last_mut();
                        debug_assert!(sub.is_some(), "subscriptions.last_mut() returned None after push()");
                        if let Some(sub) = sub {
                            sub.id = subscription_id;
                            sub.id_sender.try_send(subscription_id)?;
                        }

                        // SAFETY: We just pushed a subscription, so last() must return Some.
                        // Extract encoded_params in a separate scope to avoid borrow across await.
                        let encoded_params = {
                            let last_sub = self.subscriptions.last();
                            debug_assert!(last_sub.is_some(), "subscriptions.last() returned None after push()");
                            let Some(last_sub) = last_sub else {
                                return Err(LightstreamerError::invalid_state(
                                    "subscriptions.last() returned None immediately after push()"
                                ));
                            };
                            Self::get_subscription_params(last_sub, request_id)?
                        };

                        write_stream
                            .send(Message::Text(format!("control\r\n{}", encoded_params).into()))
                            .await?;

                        self.make_log( Level::INFO, &format!("Sent subscription request: '{}'", encoded_params) );
                    }
                    // Process unsubscription requests.
                    else if let Some(unsubscription_id) = subscription_request.subscription_id
                    {
                        let encoded_params = match Self::get_unsubscription_params(unsubscription_id, request_id)
                        {
                            Ok(params) => params,
                            Err(err) => {
                                return Err(err);
                            },
                        };

                        write_stream
                            .send(Message::Text(format!("control\r\n{}", encoded_params).into()))
                            .await?;

                        self.make_log( Level::INFO, &format!("Sent unsubscription request: '{}'", encoded_params) );

                        self.subscriptions.retain(|s| s.id != unsubscription_id);

                        if self.subscriptions.is_empty()
                        {
                            self.make_log( Level::INFO, "No more subscriptions, disconnecting" );
                            shutdown_signal.notify_one();
                        }
                    }
                },
                _ = shutdown_signal.notified() => {
                    self.make_log( Level::INFO, "Received shutdown signal" );
                    break;
                },
                //
                // Liveness watchdog: nothing received within the keepalive
                // window plus grace — the connection is half-open / dead.
                //
                _ = tokio::time::sleep_until(liveness_deadline), if keepalive_timeout_ms > 0 => {
                    let idle_ms = keepalive_timeout_ms.saturating_add(Self::KEEPALIVE_GRACE_MS);
                    self.make_log(
                        Level::ERROR,
                        &format!(
                            "No message received from server within {} ms (keepalive + grace); connection considered dead",
                            idle_ms
                        ),
                    );
                    return Err(LightstreamerError::connection(format!(
                        "no message received from server within {} ms (keepalive + grace)",
                        idle_ms
                    )));
                },
                //
                // Reverse heartbeat: prove client liveness to the server and
                // surface write errors promptly on a dead connection.
                //
                _ = async {
                    match heartbeat_timer.as_mut() {
                        Some(timer) => { timer.tick().await; },
                        None => std::future::pending::<()>().await,
                    }
                }, if is_connected && session_id.is_some() => {
                    // TLCP requires every request to carry at least one
                    // parameter — an empty body is rejected with
                    // "error,67,Empty request unexpected".
                    if let Some(id) = session_id.as_deref() {
                        let params = serde_urlencoded::to_string([("LS_session", id)])?;
                        write_stream
                            .send(Message::Text(format!("heartbeat\r\n{}", params).into()))
                            .await?;
                        self.make_log(Level::DEBUG, "Sent reverse heartbeat to server");
                    }
                },
            }
        }

        Ok(())
    }

    /// Operation method that requests to close the Session opened against the configured Lightstreamer
    /// Server (if any).
    ///
    /// When `disconnect()` is called, the "Stream-Sense" mechanism is stopped.
    ///
    /// Note that active `Subscription` instances, associated with this `LightstreamerClient` instance,
    /// are preserved to be re-subscribed to on future Sessions.
    ///
    /// Note that the request to disconnect is accomplished by the client in a separate thread; this
    /// means that an invocation to `getStatus()` right after `disconnect()` might not reflect the
    /// change yet.
    ///
    /// When the request to disconnect is finally being executed, if the status of the client is
    /// "DISCONNECTED", then nothing will be done.
    ///
    /// See also `connect()`
    #[instrument(level = "trace")]
    pub async fn disconnect(&mut self) {
        self.make_log(Level::INFO, "Disconnecting from Lightstreamer server");

        // If auto-reconnect is enabled and we have a connection manager, stop it
        if self.auto_reconnect_enabled {
            if let Some(ref manager) = self.connection_manager {
                manager.shutdown().await;
            }
            self.connection_manager = None;
        }

        // Update status to disconnected
        self.status = ClientStatus::Disconnected(DisconnectionType::WillRetry);

        // Notify listeners about status change
        for listener in &self.listeners {
            listener.on_status_change(&self.status.to_string());
        }
    }

    /// Static inquiry method that can be used to share cookies between connections to the Server
    /// (performed by this library) and connections to other sites that are performed by the application.
    /// With this method, cookies received from the Server can be extracted for sending through other
    /// connections, according with the URI to be accessed.
    ///
    /// See `addCookies()` for clarifications on when cookies are directly stored by the library and
    /// when not.
    ///
    /// # Parameters
    ///
    /// * `uri`: the URI to which the cookies should be sent, or `None`.
    ///
    /// # Returns
    ///
    /// A list with the various cookies that can be sent in a HTTP request for the specified URI.
    /// If a `None` URI was supplied, all available non-expired cookies will be returned.
    pub fn get_cookies(_uri: Option<&str>) -> Cookie<'_> {
        // Implementation for get_cookies
        unimplemented!()
    }

    /// Returns a list containing the `ClientListener` instances that were added to this client.
    ///
    /// # Returns
    ///
    /// A list containing the listeners that were added to this client.
    ///
    /// See also `addListener()`
    pub fn get_listeners(&self) -> &Vec<Box<dyn ClientListener>> {
        &self.listeners
    }

    /// Inquiry method that gets the current client status and transport (when applicable).
    ///
    /// # Returns
    ///
    /// The current client status. It can be one of the following values:
    ///
    /// - `"CONNECTING"`: the client is waiting for a Server's response in order to establish a connection;
    /// - `"CONNECTED:STREAM-SENSING"`: the client has received a preliminary response from the server
    ///   and is currently verifying if a streaming connection is possible;
    /// - `"CONNECTED:WS-STREAMING"`: a streaming connection over WebSocket is active;
    /// - `"CONNECTED:HTTP-STREAMING"`: a streaming connection over HTTP is active;
    /// - `"CONNECTED:WS-POLLING"`: a polling connection over WebSocket is in progress;
    /// - `"CONNECTED:HTTP-POLLING"`: a polling connection over HTTP is in progress;
    /// - `"STALLED"`: the Server has not been sending data on an active streaming connection for
    ///   longer than a configured time;
    /// - `"DISCONNECTED:WILL-RETRY"`: no connection is currently active but one will be opened
    ///   (possibly after a timeout);
    /// - `"DISCONNECTED:TRYING-RECOVERY"`: no connection is currently active, but one will be opened
    ///   as soon as possible, as an attempt to recover the current session after a connection issue;
    /// - `"DISCONNECTED"`: no connection is currently active.
    ///
    /// See also `ClientListener.onStatusChange()`
    pub fn get_status(&self) -> &ClientStatus {
        &self.status
    }

    /// Inquiry method that returns a list containing all the `Subscription` instances that are
    /// currently "active" on this `LightstreamerClient`.
    ///
    /// Internal second-level `Subscription` are not included.
    ///
    /// # Returns
    ///
    /// A list, containing all the `Subscription` currently "active" on this `LightstreamerClient`.
    /// The list can be empty.
    ///
    /// See also `subscribe()`
    pub fn get_subscriptions(&self) -> &Vec<Subscription> {
        &self.subscriptions
    }

    /// Creates a new instance of `LightstreamerClient`.
    ///
    /// The constructor initializes the client with the server address and adapter set, if provided.
    /// It sets up the connection details and options for the client. If no server address or
    /// adapter set is specified, those properties on the client will be `None`. This allows
    /// for late configuration of these details before connecting to the Lightstreamer server.
    ///
    /// # Arguments
    /// * `server_address` - An optional reference to a string slice that represents the server
    ///   address to connect to. If `None`, the server address must be set later.
    /// * `adapter_set` - An optional reference to a string slice that specifies the adapter set name.
    ///   If `None`, the adapter set must be set later.
    ///
    /// # Returns
    /// A result containing the new `LightstreamerClient` instance if successful, or an
    /// `IllegalStateException` if the initialization fails due to invalid state conditions.
    ///
    /// # Panics
    /// Does not panic under normal circumstances. However, unexpected internal errors during
    /// the creation of internal components could cause panics, which should be considered when
    /// using this function in production code.
    ///
    pub fn new(
        server_address: Option<&str>,
        adapter_set: Option<&str>,
        username: Option<&str>,
        password: Option<&str>,
    ) -> Result<LightstreamerClient, LightstreamerError> {
        let connection_details =
            ConnectionDetails::new(server_address, adapter_set, username, password)?;
        let connection_options = ConnectionOptions::default();
        let (subscription_sender, subscription_receiver) = channel(100);

        Ok(LightstreamerClient {
            server_address: server_address.map(|s| s.to_string()),
            adapter_set: adapter_set.map(|s| s.to_string()),
            connection_details,
            connection_options,
            listeners: Vec::new(),
            subscriptions: Vec::new(),
            status: ClientStatus::Disconnected(DisconnectionType::WillRetry),
            logging: LogType::StdLogs,
            subscription_sender,
            subscription_receiver,
            connection_manager: None,
            reconnection_config: ReconnectionConfig::default(),
            heartbeat_config: HeartbeatConfig::default(),
            auto_reconnect_enabled: false,
        })
    }

    /// Removes a listener from the `LightstreamerClient` instance so that it will not receive
    /// events anymore.
    ///
    /// A listener can be removed at any time.
    ///
    /// # Parameters
    ///
    /// * `listener`: The listener to be removed.
    ///
    /// See also `addListener()`
    pub fn remove_listener(&mut self, _listener: Box<dyn ClientListener>) {
        unimplemented!("Implement mechanism to remove listener from LightstreamerClient");
        //self.listeners.remove(&listener);
    }

    /// Operation method that sends a message to the Server. The message is interpreted and handled
    /// by the Metadata Adapter associated to the current Session. This operation supports in-order
    /// guaranteed message delivery with automatic batching. In other words, messages are guaranteed
    /// to arrive exactly once and respecting the original order, whatever is the underlying transport
    /// (HTTP or WebSockets). Furthermore, high frequency messages are automatically batched, if
    /// necessary, to reduce network round trips.
    ///
    /// Upon subsequent calls to the method, the sequential management of the involved messages is
    /// guaranteed. The ordering is determined by the order in which the calls to `sendMessage` are
    /// issued.
    ///
    /// If a message, for any reason, doesn't reach the Server (this is possible with the HTTP transport),
    /// it will be resent; however, this may cause the subsequent messages to be delayed. For this
    /// reason, each message can specify a "delayTimeout", which is the longest time the message,
    /// after reaching the Server, can be kept waiting if one of more preceding messages haven't
    /// been received yet. If the "delayTimeout" expires, these preceding messages will be discarded;
    /// any discarded message will be notified to the listener through `ClientMessageListener.onDiscarded()`.
    /// Note that, because of the parallel transport of the messages, if a zero or very low timeout
    /// is set for a message and the previous message was sent immediately before, it is possible
    /// that the latter gets discarded even if no communication issues occur. The Server may also
    /// enforce its own timeout on missing messages, to prevent keeping the subsequent messages for
    /// long time.
    ///
    /// Sequence identifiers can also be associated with the messages. In this case, the sequential
    /// management is restricted to all subsets of messages with the same sequence identifier associated.
    ///
    /// Notifications of the operation outcome can be received by supplying a suitable listener. The
    /// supplied listener is guaranteed to be eventually invoked; listeners associated with a sequence
    /// are guaranteed to be invoked sequentially.
    ///
    /// The "UNORDERED_MESSAGES" sequence name has a special meaning. For such a sequence, immediate
    /// processing is guaranteed, while strict ordering and even sequentialization of the processing
    /// is not enforced. Likewise, strict ordering of the notifications is not enforced. However,
    /// messages that, for any reason, should fail to reach the Server whereas subsequent messages
    /// had succeeded, might still be discarded after a server-side timeout, in order to ensure that
    /// the listener eventually gets a notification.
    ///
    /// Moreover, if "UNORDERED_MESSAGES" is used and no listener is supplied, a "fire and forget"
    /// scenario is assumed. In this case, no checks on missing, duplicated or overtaken messages
    /// are performed at all, so as to optimize the processing and allow the highest possible throughput.
    ///
    /// Since a message is handled by the Metadata Adapter associated to the current connection, a
    /// message can be sent only if a connection is currently active. If the special `enqueueWhileDisconnected`
    /// flag is specified it is possible to call the method at any time and the client will take
    /// care of sending the message as soon as a connection is available, otherwise, if the current
    /// status is "DISCONNECTED*", the message will be abandoned and the `ClientMessageListener.onAbort()`
    /// event will be fired.
    ///
    /// Note that, in any case, as soon as the status switches again to "DISCONNECTED*", any message
    /// still pending is aborted, including messages that were queued with the `enqueueWhileDisconnected`
    /// flag set to `true`.
    ///
    /// Also note that forwarding of the message to the server is made in a separate thread, hence,
    /// if a message is sent while the connection is active, it could be aborted because of a subsequent
    /// disconnection. In the same way a message sent while the connection is not active might be
    /// sent because of a subsequent connection.
    ///
    /// # Parameters
    ///
    /// * `message`: a text message, whose interpretation is entirely demanded to the Metadata Adapter
    ///   associated to the current connection.
    /// * `sequence`: an alphanumeric identifier, used to identify a subset of messages to be managed
    ///   in sequence; underscore characters are also allowed. If the "UNORDERED_MESSAGES" identifier
    ///   is supplied, the message will be processed in the special way described above. The parameter
    ///   is optional; if set to `None`, "UNORDERED_MESSAGES" is used as the sequence name.
    /// * `delay_timeout`: a timeout, expressed in milliseconds. If higher than the Server configured
    ///   timeout on missing messages, the latter will be used instead. The parameter is optional; if
    ///   a negative value is supplied, the Server configured timeout on missing messages will be applied.
    ///   This timeout is ignored for the special "UNORDERED_MESSAGES" sequence, although a server-side
    ///   timeout on missing messages still applies.
    /// * `listener`: an object suitable for receiving notifications about the processing outcome. The
    ///   parameter is optional; if not supplied, no notification will be available.
    /// * `enqueue_while_disconnected`: if this flag is set to `true`, and the client is in a disconnected
    ///   status when the provided message is handled, then the message is not aborted right away but
    ///   is queued waiting for a new session. Note that the message can still be aborted later when
    ///   a new session is established.
    pub fn send_message(
        &mut self,
        message: &str,
        sequence: Option<&str>,
        _delay_timeout: Option<u64>,
        listener: Option<Box<dyn ClientMessageListener>>,
        enqueue_while_disconnected: bool,
    ) {
        let _sequence = sequence.unwrap_or("UNORDERED_MESSAGES");

        // Handle the message based on the current connection status
        match &self.status {
            ClientStatus::Connected(_connection_type) => {
                // Send the message to the server in a separate thread
                // ...
            }
            ClientStatus::Disconnected(_disconnection_type) => {
                if enqueue_while_disconnected {
                    // Enqueue the message to be sent when a connection is available
                    // ...
                } else {
                    // Abort the message and notify the listener
                    if let Some(listener) = listener {
                        listener.on_abort(message, false);
                    }
                }
            }
            _ => {
                // Enqueue the message to be sent when a connection is available
                // ...
            }
        }
        unimplemented!("Complete mechanism to send message to LightstreamerClient.");
    }

    /// Static method that permits to configure the logging system used by the library. The logging
    /// system must respect the `LoggerProvider` interface. A custom class can be used to wrap any
    /// third-party logging system.
    ///
    /// If no logging system is specified, all the generated log is discarded.
    ///
    /// The following categories are available to be consumed:
    ///
    /// - `lightstreamer.stream`: logs socket activity on Lightstreamer Server connections; at INFO
    ///   level, socket operations are logged; at DEBUG level, read/write data exchange is logged.
    /// - `lightstreamer.protocol`: logs requests to Lightstreamer Server and Server answers; at INFO
    ///   level, requests are logged; at DEBUG level, request details and events from the Server are logged.
    /// - `lightstreamer.session`: logs Server Session lifecycle events; at INFO level, lifecycle events
    ///   are logged; at DEBUG level, lifecycle event details are logged.
    /// - `lightstreamer.subscriptions`: logs subscription requests received by the clients and the related
    ///   updates; at WARN level, alert events from the Server are logged; at INFO level, subscriptions
    ///   and unsubscriptions are logged; at DEBUG level, requests batching and update details are logged.
    /// - `lightstreamer.actions`: logs settings / API calls.
    ///
    /// # Parameters
    ///
    /// * `provider`: A `LoggerProvider` instance that will be used to generate log messages by the
    ///   library classes.
    pub fn set_logger_provider() {
        unimplemented!("Implement mechanism to set logger provider for LightstreamerClient.");
    }
    /*
    pub fn set_logger_provider(provider: LoggerProvider) {
        // Implementation for set_logger_provider
    }
    */

    /// Provides a mean to control the way TLS certificates are evaluated, with the possibility to
    /// accept untrusted ones.
    ///
    /// May be called only once before creating any `LightstreamerClient` instance.
    ///
    /// # Parameters
    ///
    /// * `factory`: an instance of `ssl.SSLContext`
    ///
    /// # Raises
    ///
    /// * `IllegalArgumentException`: if the factory is `None`
    /// * `IllegalStateException`: if a factory is already installed
    pub fn set_trust_manager_factory() {
        unimplemented!("Implement mechanism to set trust manager factory for LightstreamerClient.");
    }
    /*
    pub fn set_trust_manager_factory(factory: Option<SslContext>) -> Result<(), IllegalArgumentException> {
        if factory.is_none() {
            return Err(IllegalArgumentException::new(
                "Factory cannot be None",
            ));
        }

        // Implementation for set_trust_manager_factory
        Ok(())
    }
    */

    /// Adds a subscription to the `LightstreamerClient` instance.
    ///
    /// Active subscriptions are subscribed to through the server as soon as possible (i.e. as soon
    /// as there is a session available). Active `Subscription` are automatically persisted across different
    /// sessions as long as a related unsubscribe call is not issued.
    ///
    /// Subscriptions can be given to the `LightstreamerClient` at any time. Once done the `Subscription`
    /// immediately enters the "active" state.
    ///
    /// Once "active", a `Subscription` instance cannot be provided again to a `LightstreamerClient`
    /// unless it is first removed from the "active" state through a call to `unsubscribe()`.
    ///
    /// Also note that forwarding of the subscription to the server is made in a separate thread.
    ///
    /// A successful subscription to the server will be notified through a `SubscriptionListener.onSubscription()`
    /// event.
    ///
    /// # Parameters
    ///
    /// * `subscription_sender`: A `Sender` object that sends a `SubscriptionRequest` to the `LightstreamerClient`
    /// * `subscription`: A `Subscription` object, carrying all the information needed to process real-time
    ///   values.
    ///
    /// See also `unsubscribe()`
    ///
    /// # Errors
    ///
    /// Returns an error if the subscription channel is closed.
    pub async fn subscribe(
        subscription_sender: Sender<SubscriptionRequest>,
        subscription: Subscription,
    ) -> Result<(), LightstreamerError> {
        subscription_sender
            .send(SubscriptionRequest {
                subscription: Some(subscription),
                subscription_id: None,
            })
            .await
            .map_err(|e| LightstreamerError::channel(format!("Failed to send subscription: {}", e)))
    }

    /// If you want to be able to unsubscribe from a subscription, you need to keep track of the id
    /// of the subscription. This blocking method allows you to wait for the id of the subscription
    /// to be returned.
    ///
    /// # Parameters
    ///
    /// * `subscription_sender`: A `Sender` object that sends a `SubscriptionRequest` to the `LightstreamerClient`
    /// * `subscription`: A `Subscription` object, carrying all the information needed to process real-time
    ///
    pub async fn subscribe_get_id(
        subscription_sender: Sender<SubscriptionRequest>,
        mut subscription: Subscription,
    ) -> Result<usize, LightstreamerError> {
        // Extract the id_receiver before sending the subscription
        let mut id_receiver = subscription.id_receiver;

        // Create a new channel for the subscription we're about to send
        let (_new_sender, new_receiver) = channel(1);
        subscription.id_receiver = new_receiver;

        // Send the subscription
        LightstreamerClient::subscribe(subscription_sender, subscription).await?;

        // Wait for the ID to be updated through the channel
        match id_receiver.recv().await {
            Some(id) => Ok(id),
            None => Err(LightstreamerError::invalid_state(
                "Failed to get subscription id",
            )),
        }
    }

    /// Operation method that removes a `Subscription` that is currently in the "active" state.
    ///
    /// By bringing back a `Subscription` to the "inactive" state, the unsubscription from all its
    /// items is requested to Lightstreamer Server.
    ///
    /// Subscription can be unsubscribed from at any time. Once done the `Subscription` immediately
    /// exits the "active" state.
    ///
    /// Note that forwarding of the unsubscription to the server is made in a separate thread.
    ///
    /// The unsubscription will be notified through a `SubscriptionListener.onUnsubscription()` event.
    ///
    /// # Parameters
    ///
    /// * `subscription_sender`: A `Sender` object that sends a `SubscriptionRequest` to the `LightstreamerClient`
    /// * `subscription_id`: The id of the subscription to be unsubscribed from.
    ///   instance.
    ///
    /// # Errors
    ///
    /// Returns an error if the subscription channel is closed.
    pub async fn unsubscribe(
        subscription_sender: Sender<SubscriptionRequest>,
        subscription_id: usize,
    ) -> Result<(), LightstreamerError> {
        subscription_sender
            .send(SubscriptionRequest {
                subscription: None,
                subscription_id: Some(subscription_id),
            })
            .await
            .map_err(|e| {
                LightstreamerError::channel(format!("Failed to send unsubscription: {}", e))
            })
    }

    /// Method setting enum for the logging of this instance.
    ///
    /// Default logging type is StdLogs, corresponding to `stdout`
    ///
    /// `LightstreamerClient` has methods for logging that are compatible with the `Tracing` crate.
    /// Enabling logging for the `Tracing` crate requires implementation of a tracing subscriber
    /// and its configuration and formatting.
    ///
    /// # Parameters
    ///
    /// * `logging`: An enum declaring the logging type of this `LightstreamerClient` instance.
    pub fn set_logging_type(&mut self, logging: LogType) {
        self.logging = logging;
    }

    /// Method for logging messages
    ///
    /// Match case wraps log types. `loglevel` param ignored in StdLogs case, all output to stdout.
    ///
    /// # Parameters
    ///
    /// * `loglevel` Enum determining use of stdout or Tracing subscriber.
    pub fn make_log(&mut self, loglevel: Level, log: &str) {
        match self.logging {
            LogType::StdLogs => {
                debug!("{}", log);
            }
            LogType::TracingLogs => match loglevel {
                Level::INFO => {
                    info!(log);
                }
                Level::WARN => {
                    warn!(log);
                }
                Level::ERROR => {
                    error!(log);
                }
                Level::TRACE => {
                    trace!(log);
                }
                Level::DEBUG => {
                    debug!(log);
                }
            },
        }
    }

    /// Enables automatic reconnection with default configuration.
    pub fn enable_auto_reconnect(&mut self) {
        self.auto_reconnect_enabled = true;
        // ConnectionManager will be created during connect() to avoid circular dependency
    }

    /// Enables automatic reconnection with the specified configuration.
    ///
    /// # Parameters
    ///
    /// * `reconnection_config`: Configuration for reconnection behavior
    /// * `heartbeat_config`: Configuration for heartbeat monitoring
    pub fn enable_auto_reconnect_with_config(
        &mut self,
        reconnection_config: ReconnectionConfig,
        heartbeat_config: HeartbeatConfig,
    ) -> Result<(), LightstreamerError> {
        self.auto_reconnect_enabled = true;
        self.reconnection_config = reconnection_config;
        self.heartbeat_config = heartbeat_config;
        self.auto_reconnect_enabled = true;
        // ConnectionManager will be created during connect() to avoid circular dependency
        Ok(())
    }

    /// Disables automatic reconnection.
    pub async fn disable_auto_reconnect(&mut self) {
        self.auto_reconnect_enabled = false;
        if let Some(manager) = &self.connection_manager {
            manager.shutdown().await;
        }
        self.connection_manager = None;
    }

    /// Returns whether automatic reconnection is currently enabled.
    pub fn is_auto_reconnect_enabled(&self) -> bool {
        self.auto_reconnect_enabled
    }

    /// Gets the current reconnection configuration.
    pub fn get_reconnection_config(&self) -> &ReconnectionConfig {
        &self.reconnection_config
    }

    /// Gets the current heartbeat configuration.
    pub fn get_heartbeat_config(&self) -> &HeartbeatConfig {
        &self.heartbeat_config
    }

    /// Updates the reconnection configuration.
    pub fn set_reconnection_config(&mut self, config: ReconnectionConfig) {
        self.reconnection_config = config;
        // Note: ConnectionManager doesn't support runtime config updates
        // The configuration is set during ConnectionManager creation
    }

    /// Updates the heartbeat configuration.
    pub fn set_heartbeat_config(&mut self, config: HeartbeatConfig) {
        self.heartbeat_config = config;
        // Note: ConnectionManager doesn't support runtime config updates
        // The configuration is set during ConnectionManager creation
    }

    /// Gets the current connection state from the connection manager.
    pub async fn get_connection_state(&self) -> ConnectionState {
        if let Some(manager) = &self.connection_manager {
            manager.get_connection_state().await
        } else {
            ConnectionState::Disconnected
        }
    }

    /// Gets connection metrics from the connection manager.
    pub async fn get_connection_metrics(&self) -> crate::connection::management::ConnectionMetrics {
        if let Some(manager) = &self.connection_manager {
            manager.get_metrics().await
        } else {
            crate::connection::management::ConnectionMetrics::default()
        }
    }

    /// Forces an immediate reconnection attempt if auto-reconnect is enabled.
    pub async fn force_reconnect(&mut self) -> Result<(), LightstreamerError> {
        if !self.auto_reconnect_enabled {
            return Err(LightstreamerError::invalid_state(
                "Auto-reconnect is not enabled",
            ));
        }

        if let Some(manager) = &mut self.connection_manager {
            manager.force_reconnect().await?;
        }

        Ok(())
    }
}

/// Computes the effective liveness window after the server declares its
/// keepalive interval in the `conok` message.
///
/// Returns the server-declared value when no window was configured
/// (`current_ms == 0`), otherwise the larger of the two — the window may only
/// widen, never shrink, so a server declaring a longer keepalive cannot cause
/// false-positive disconnects.
#[must_use]
#[inline]
fn widen_keepalive_timeout(current_ms: u64, server_declared_ms: u64) -> u64 {
    if current_ms == 0 {
        server_declared_ms
    } else {
        current_ms.max(server_declared_ms)
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::subscription::{Subscription, SubscriptionListener, SubscriptionMode};

    use std::fmt::Debug;
    use std::sync::{Arc, Mutex};
    use tokio::sync::Notify;

    #[derive(Debug)]
    struct MockClientListener {
        property_changes: Arc<Mutex<Vec<String>>>,
        status_changes: Arc<Mutex<Vec<String>>>,
        server_errors: Arc<Mutex<Vec<(i32, String)>>>,
    }

    impl MockClientListener {
        fn new() -> Self {
            MockClientListener {
                property_changes: Arc::new(Mutex::new(Vec::new())),
                status_changes: Arc::new(Mutex::new(Vec::new())),
                server_errors: Arc::new(Mutex::new(Vec::new())),
            }
        }

        #[allow(dead_code)]
        fn with_shared_data(
            property_changes: Arc<Mutex<Vec<String>>>,
            status_changes: Arc<Mutex<Vec<String>>>,
            server_errors: Arc<Mutex<Vec<(i32, String)>>>,
        ) -> Self {
            MockClientListener {
                property_changes,
                status_changes,
                server_errors,
            }
        }
    }

    impl ClientListener for MockClientListener {
        fn on_property_change(&self, property: &str) {
            if let Ok(mut guard) = self.property_changes.lock() {
                guard.push(property.to_string());
            }
        }

        fn on_status_change(&self, status: &str) {
            if let Ok(mut guard) = self.status_changes.lock() {
                guard.push(status.to_string());
            }
        }

        fn on_server_error(&self, code: i32, message: &str) {
            if let Ok(mut guard) = self.server_errors.lock() {
                guard.push((code, message.to_string()));
            }
        }
    }

    #[allow(dead_code)]
    struct MockSubscriptionListener;

    impl SubscriptionListener for MockSubscriptionListener {
        fn on_subscription(&mut self) {}
        fn on_unsubscription(&mut self) {}
        fn on_item_update(&self, _update: &ItemUpdate) {}
    }

    impl Debug for MockSubscriptionListener {
        fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
            write!(f, "MockSubscriptionListener")
        }
    }

    #[allow(dead_code)]
    struct LightstreamerClientTestContext {
        client: LightstreamerClient,
        property_changes: Arc<Mutex<Vec<String>>>,
        status_changes: Arc<Mutex<Vec<String>>>,
        server_errors: Arc<Mutex<Vec<(i32, String)>>>,
    }

    impl LightstreamerClientTestContext {
        #[allow(dead_code)]
        fn new() -> Result<Self, LightstreamerError> {
            let property_changes = Arc::new(Mutex::new(Vec::new()));
            let status_changes = Arc::new(Mutex::new(Vec::new()));
            let server_errors = Arc::new(Mutex::new(Vec::new()));
            let listener = MockClientListener::with_shared_data(
                Arc::clone(&property_changes),
                Arc::clone(&status_changes),
                Arc::clone(&server_errors),
            );

            let mut client = LightstreamerClient::new(
                Some("http://test.lightstreamer.com"),
                Some("DEMO"),
                None,
                None,
            )?;
            client.add_listener(Box::new(listener));

            Ok(LightstreamerClientTestContext {
                client,
                property_changes,
                status_changes,
                server_errors,
            })
        }
    }

    #[test]
    fn test_new_lightstreamer_client() -> Result<(), LightstreamerError> {
        let client = LightstreamerClient::new(
            Some("http://test.lightstreamer.com"),
            Some("DEMO"),
            None,
            None,
        )?;
        assert_eq!(
            client.server_address,
            Some("http://test.lightstreamer.com".to_string())
        );
        assert_eq!(client.adapter_set, Some("DEMO".to_string()));
        let client = LightstreamerClient::new(
            Some("http://test.lightstreamer.com"),
            Some("DEMO"),
            Some("user1"),
            Some("pass1"),
        )?;
        assert_eq!(
            client.connection_details.get_user(),
            Some(&"user1".to_string())
        );
        assert_eq!(
            client.connection_details.get_password(),
            Some(&"pass1".to_string())
        );
        let result = LightstreamerClient::new(Some("invalid-url"), Some("DEMO"), None, None);
        assert!(result.is_err());
        let client = LightstreamerClient::new(None, Some("DEMO"), None, None)?;
        assert_eq!(client.server_address, None);
        Ok(())
    }

    #[test]
    fn test_add_listener() -> Result<(), LightstreamerError> {
        let mut client = LightstreamerClient::new(
            Some("http://test.lightstreamer.com"),
            Some("DEMO"),
            None,
            None,
        )?;
        assert_eq!(client.listeners.len(), 0);
        let listener = Box::new(MockClientListener::new());
        client.add_listener(listener);
        assert_eq!(client.listeners.len(), 1);
        let listener2 = Box::new(MockClientListener::new());
        client.add_listener(listener2);
        assert_eq!(client.listeners.len(), 2);
        Ok(())
    }

    #[test]
    fn test_get_listeners() -> Result<(), LightstreamerError> {
        let mut client = LightstreamerClient::new(
            Some("http://test.lightstreamer.com"),
            Some("DEMO"),
            None,
            None,
        )?;
        assert_eq!(client.get_listeners().len(), 0);

        let listener = Box::new(MockClientListener::new());
        client.add_listener(listener);
        assert_eq!(client.get_listeners().len(), 1);
        Ok(())
    }

    #[test]
    fn test_get_status() -> Result<(), LightstreamerError> {
        let client = LightstreamerClient::new(
            Some("http://test.lightstreamer.com"),
            Some("DEMO"),
            None,
            None,
        )?;
        match client.get_status() {
            ClientStatus::Disconnected(DisconnectionType::WillRetry) => {}
            _ => panic!("Expected initial status to be DISCONNECTED:WILL-RETRY"),
        }
        Ok(())
    }

    #[test]
    fn test_get_subscriptions() -> Result<(), LightstreamerError> {
        let client = LightstreamerClient::new(
            Some("http://test.lightstreamer.com"),
            Some("DEMO"),
            None,
            None,
        )?;
        assert_eq!(client.get_subscriptions().len(), 0);
        Ok(())
    }

    #[tokio::test]
    async fn test_connect_with_no_server_address() -> Result<(), LightstreamerError> {
        let client = LightstreamerClient::new(None, Some("DEMO"), None, None)?;
        let client_arc = Arc::new(tokio::sync::Mutex::new(client));
        let shutdown_signal = Arc::new(Notify::new());
        let result = LightstreamerClient::connect(client_arc, shutdown_signal).await;
        assert!(result.is_err());
        Ok(())
    }

    #[tokio::test]
    async fn test_forced_transport_validation() -> Result<(), LightstreamerError> {
        let mut client = LightstreamerClient::new(
            Some("http://test.lightstreamer.com"),
            Some("DEMO"),
            None,
            None,
        )?;

        client.connection_options.set_forced_transport(None);
        let client_arc = Arc::new(tokio::sync::Mutex::new(client));
        let shutdown_signal = Arc::new(Notify::new());
        let result = LightstreamerClient::connect(client_arc.clone(), shutdown_signal).await;
        assert!(result.is_err());
        client_arc
            .lock()
            .await
            .connection_options
            .set_forced_transport(Some(Transport::WsStreaming));
        Ok(())
    }

    #[test]
    fn test_subscription_params_generation() -> Result<(), LightstreamerError> {
        let subscription = Subscription::new(
            SubscriptionMode::Merge,
            Some(vec!["item1".to_string(), "item2".to_string()]),
            Some(vec!["field1".to_string(), "field2".to_string()]),
        )?;

        let params_str = LightstreamerClient::get_subscription_params(&subscription, 1)?;

        assert!(params_str.contains("LS_reqId=1"));
        assert!(params_str.contains("LS_op=add"));
        assert!(params_str.contains("LS_subId="));
        assert!(params_str.contains("LS_mode=MERGE"));
        assert!(params_str.contains("LS_group="));
        assert!(params_str.contains("LS_schema="));
        Ok(())
    }

    #[test]
    fn test_unsubscription_params_generation() -> Result<(), LightstreamerError> {
        let params_str = LightstreamerClient::get_unsubscription_params(42, 123)?;

        assert!(params_str.contains("LS_reqId=123"));
        assert!(params_str.contains("LS_op=delete"));
        assert!(params_str.contains("LS_subId=42"));
        Ok(())
    }

    #[test]
    fn test_logging_functions() -> Result<(), LightstreamerError> {
        let mut client = LightstreamerClient::new(
            Some("http://test.lightstreamer.com"),
            Some("DEMO"),
            None,
            None,
        )?;

        client.set_logging_type(LogType::StdLogs);

        client.make_log(Level::INFO, "Test log message");
        client.make_log(Level::DEBUG, "Test debug message");
        client.set_logging_type(LogType::TracingLogs);
        client.make_log(Level::INFO, "Test tracing log message");
        client.make_log(Level::DEBUG, "Test tracing debug message");
        Ok(())
    }

    #[test]
    fn test_debug_implementation() -> Result<(), LightstreamerError> {
        let client = LightstreamerClient::new(
            Some("http://test.lightstreamer.com"),
            Some("DEMO"),
            None,
            None,
        )?;

        // Test that Debug implementation works without panicking
        let debug_string = format!("{:?}", client);

        // Verify it contains expected fields
        assert!(debug_string.contains("server_address"));
        assert!(debug_string.contains("adapter_set"));
        assert!(debug_string.contains("connection_details"));
        assert!(debug_string.contains("connection_options"));
        assert!(debug_string.contains("listeners"));
        assert!(debug_string.contains("subscriptions"));

        // Verify the values are included
        assert!(debug_string.contains("http://test.lightstreamer.com"));
        assert!(debug_string.contains("DEMO"));
        Ok(())
    }

    #[test]
    #[should_panic(expected = "Implement mechanism to add cookies to LightstreamerClient")]
    fn test_add_cookies() {
        // Test the static method add_cookies
        let cookie = Cookie::new("test_cookie", "test_value");
        LightstreamerClient::add_cookies("http://test.lightstreamer.com", &cookie);
    }

    #[test]
    #[should_panic(expected = "not implemented")]
    fn test_get_cookies() {
        // Test the static method get_cookies
        LightstreamerClient::get_cookies(Some("http://test.lightstreamer.com"));
    }
}

#[cfg(test)]
mod liveness_tests {
    use super::widen_keepalive_timeout;

    #[test]
    fn test_widen_keepalive_timeout_unconfigured_adopts_server_value() {
        assert_eq!(widen_keepalive_timeout(0, 5_000), 5_000);
    }

    #[test]
    fn test_widen_keepalive_timeout_server_longer_widens() {
        assert_eq!(widen_keepalive_timeout(15_000, 30_000), 30_000);
    }

    #[test]
    fn test_widen_keepalive_timeout_server_shorter_never_shrinks() {
        assert_eq!(widen_keepalive_timeout(15_000, 5_000), 15_000);
    }

    #[test]
    fn test_widen_keepalive_timeout_equal_values_unchanged() {
        assert_eq!(widen_keepalive_timeout(15_000, 15_000), 15_000);
    }
}