ibapi 4.1.0

//! Asynchronous transport implementation

mod io;
mod shutdown;
/// The frame reader itself, for tests that drive it over an in-memory cursor
/// rather than a socket. Production callers reach it through `AsyncIo`.
#[cfg(test)]
pub(crate) use io::read_framed_message;
/// The stream traits, for test fixtures that stand in for a socket.
#[cfg(test)]
pub(crate) use io::{AsyncIo, AsyncReconnect};
pub(crate) use io::{AsyncStream, AsyncTcpSocket};
pub(crate) use shutdown::ShutdownSignal;

use std::collections::HashMap;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, OnceLock};

use async_trait::async_trait;
use futures::Stream;
use log::{debug, error, info, warn};
use tokio::sync::{broadcast, mpsc, RwLock};
use tokio::task;
use tokio::time::Duration;
use tokio_stream::wrappers::errors::BroadcastStreamRecvError;
use tokio_stream::wrappers::BroadcastStream;

use crate::client::id_generator::ClientIdManager;
use crate::connection::r#async::AsyncConnection;
use crate::messages::{shared_channel_configuration, transport_reconnect_notice, IncomingMessages, OutgoingMessages, ResponseMessage};
use crate::Error;

use super::common::{log_orphan, report_unroutable_frame};
use super::routing::{
    classify_error, determine_routing, order_routing_strategy, order_update_notice, DecodedError, ErrorDisposition, OrderRoutingStrategy,
    RoutingDecision,
};
use super::RoutedItem;

/// Default capacity for broadcast channels. Subscription-data channels take
/// the per-client override from `ClientBuilder::channel_capacity`; the notice
/// fan-out channels always use this default. When a consumer falls behind by
/// more than the capacity, the channel evicts the oldest frames and a data
/// subscription receives a `SUBSCRIPTION_LAG_CODE` notice naming the count.
pub(crate) const BROADCAST_CHANNEL_CAPACITY: usize = 1024;

/// Cleanup signal for removing channels when subscriptions are dropped
#[derive(Debug, Clone)]
pub enum CleanupSignal {
    Request(i32),
    Order(i32),
    Shared(OutgoingMessages),
    OrderUpdateStream,
}

/// Asynchronous message bus trait
#[async_trait]
pub trait AsyncMessageBus: Send + Sync {
    async fn send_request(&self, request_id: i32, message: Vec<u8>) -> Result<AsyncInternalSubscription, Error>;

    async fn send_order_request(&self, order_id: i32, message: Vec<u8>) -> Result<AsyncInternalSubscription, Error>;

    async fn send_shared_request(&self, message_type: OutgoingMessages, message: Vec<u8>) -> Result<AsyncInternalSubscription, Error>;

    async fn send_message(&self, message: Vec<u8>) -> Result<(), Error>;

    #[allow(dead_code)]
    async fn cancel_subscription(&self, request_id: i32, message: Vec<u8>) -> Result<(), Error>;
    #[allow(dead_code)]
    async fn cancel_order_subscription(&self, order_id: i32, message: Vec<u8>) -> Result<(), Error>;

    async fn create_order_update_subscription(&self) -> Result<AsyncInternalSubscription, Error>;

    fn notice_subscribe(&self) -> crate::subscriptions::notice_stream::async_impl::NoticeStream;

    async fn ensure_shutdown(&self);

    fn request_shutdown_sync(&self);

    fn is_connected(&self) -> bool;
}

/// Internal subscription for async implementation.
///
/// Holds a `BroadcastStream<RoutedItem>` for poll-based consumption plus a
/// `template_receiver` kept solely so `Clone` can `resubscribe()` to produce an
/// independent stream. We cannot store the `Sender` instead — that would keep
/// the channel alive past the external sender's drop, breaking the
/// "channel closes when senders drop" termination contract.
pub struct AsyncInternalSubscription {
    /// Held only for `Clone` via `resubscribe()`. Never polled directly.
    template_receiver: broadcast::Receiver<RoutedItem>,
    stream: BroadcastStream<RoutedItem>,
    cleanup_sender: Option<mpsc::UnboundedSender<CleanupSignal>>,
    cleanup_signal: Option<CleanupSignal>,
}

impl Clone for AsyncInternalSubscription {
    fn clone(&self) -> Self {
        // For clones, both template and stream start at the current tail of
        // the broadcast channel — clones see future messages, not the
        // original's history.
        let new_template = self.template_receiver.resubscribe();
        let new_polling = self.template_receiver.resubscribe();
        Self {
            template_receiver: new_template,
            stream: BroadcastStream::new(new_polling),
            cleanup_sender: self.cleanup_sender.clone(),
            // Each clone sends its own cleanup signal on drop; stale ones
            // no-op against a registration that still has live receivers.
            cleanup_signal: self.cleanup_signal.clone(),
        }
    }
}

impl AsyncInternalSubscription {
    /// Construct an internal subscription wrapping a broadcast receiver.
    ///
    /// **Receiver positioning matters.** The receiver you pass is what feeds
    /// the stream — its position in the broadcast channel determines what
    /// the subscription sees. Pass the receiver paired with the messages
    /// you want consumed (typically the `rx` returned alongside the `tx`).
    /// **Do not** pass `rx.resubscribe()` unless you specifically want to
    /// skip messages already queued in `rx`; `resubscribe()` positions a
    /// fresh receiver at the channel's *current tail*, which silently
    /// hides already-queued items and causes "test ran to completion but
    /// the assertion never fired" failures.
    #[cfg(test)]
    pub(crate) fn new(receiver: broadcast::Receiver<RoutedItem>) -> Self {
        let template = receiver.resubscribe();
        Self {
            template_receiver: template,
            stream: BroadcastStream::new(receiver),
            cleanup_sender: None,
            cleanup_signal: None,
        }
    }

    pub(crate) fn with_cleanup(
        receiver: broadcast::Receiver<RoutedItem>,
        cleanup_sender: mpsc::UnboundedSender<CleanupSignal>,
        cleanup_signal: CleanupSignal,
    ) -> Self {
        let template = receiver.resubscribe();
        Self {
            template_receiver: template,
            stream: BroadcastStream::new(receiver),
            cleanup_sender: Some(cleanup_sender),
            cleanup_signal: Some(cleanup_signal),
        }
    }

    /// Poll the underlying broadcast stream, converting a `Lagged` error into
    /// an in-band [`SUBSCRIPTION_LAG_CODE`](crate::messages::SUBSCRIPTION_LAG_CODE)
    /// notice (which also `warn!`s). The single place lag is handled — every
    /// consumer polls through here, so none can reintroduce a silent swallow.
    pub(crate) fn poll_next_routed(&mut self, cx: &mut std::task::Context<'_>) -> std::task::Poll<Option<RoutedItem>> {
        use std::task::Poll;
        match std::pin::Pin::new(&mut self.stream).poll_next(cx) {
            Poll::Ready(Some(Ok(item))) => Poll::Ready(Some(item)),
            Poll::Ready(Some(Err(BroadcastStreamRecvError::Lagged(skipped)))) => {
                Poll::Ready(Some(RoutedItem::Notice(crate::messages::subscription_lag_notice(skipped))))
            }
            Poll::Ready(None) => Poll::Ready(None),
            Poll::Pending => Poll::Pending,
        }
    }

    pub async fn next(&mut self) -> Option<Result<ResponseMessage, Error>> {
        loop {
            let item = std::future::poll_fn(|cx| self.poll_next_routed(cx)).await?;
            // A lag notice maps to `None` here (`into_legacy` drops notices);
            // its `warn!` already fired inside `poll_next_routed`.
            if let Some(legacy) = item.into_legacy() {
                return Some(legacy);
            }
        }
    }

    /// Receive the next typed envelope (Response / Notice / Error) without
    /// the legacy projection. Kept because `src/transport/async_tests.rs`
    /// stub fixtures drive the channel directly via this helper.
    #[cfg(test)]
    pub(crate) async fn next_routed(&mut self) -> Option<RoutedItem> {
        std::future::poll_fn(|cx| self.poll_next_routed(cx)).await
    }

    /// Non-blocking poll for "is anything immediately available?". Returns
    /// `None` if the stream is pending or closed. Used by test fixtures that
    /// assert no cross-talk between subscriptions.
    #[cfg(test)]
    pub(crate) fn try_next_routed(&mut self) -> Option<RoutedItem> {
        use futures::FutureExt;
        std::future::poll_fn(|cx| self.poll_next_routed(cx)).now_or_never()?
    }

    /// Send the cleanup signal, detaching this subscription's receivers first.
    fn send_cleanup_signal(&mut self) {
        let (Some(sender), Some(signal)) = (self.cleanup_sender.take(), self.cleanup_signal.take()) else {
            return;
        };
        self.detach_receivers();
        let _ = sender.send(signal);
    }

    /// Drop this subscription's receivers by swapping in detached ones.
    ///
    /// Must run before the cleanup signal is sent: the cleanup task decides
    /// whether a registration is dead via `Sender::receiver_count()`, and the
    /// struct's own receivers would otherwise outlive `drop(&mut self)` just
    /// long enough for a concurrently processed signal to count them as live
    /// and skip the removal — leaking the registration.
    fn detach_receivers(&mut self) {
        let (detached_sender, detached) = broadcast::channel(1);
        self.template_receiver = detached;
        self.stream = BroadcastStream::new(detached_sender.subscribe());
    }
}

/// Send cleanup signal when subscription is dropped
impl Drop for AsyncInternalSubscription {
    fn drop(&mut self) {
        self.send_cleanup_signal();
    }
}

type BroadcastSender = broadcast::Sender<RoutedItem>;

/// Remove `id`'s registration only if its channel has no receivers left —
/// i.e. every subscription and clone feeding off it is gone. A stale drop
/// signal that finds a live replacement under the same key is a no-op; the
/// replacement's own drop signal performs the eventual removal. The count is
/// authoritative because a dropping subscription detaches its receivers
/// before signalling (`AsyncInternalSubscription::detach_receivers`).
async fn remove_if_dead(channels: &RwLock<HashMap<i32, BroadcastSender>>, id: i32, kind: &str) {
    let mut channels = channels.write().await;
    let removed = match channels.entry(id) {
        std::collections::hash_map::Entry::Occupied(entry) if entry.get().receiver_count() == 0 => {
            entry.remove();
            true
        }
        _ => false,
    };
    debug!("cleanup {kind} channel {id}: removed={removed}");
}

/// Asynchronous TCP message bus implementation
pub struct AsyncTcpMessageBus<S: AsyncStream = AsyncTcpSocket> {
    connection: Arc<AsyncConnection<S>>,
    /// Maps request IDs to their response channels
    request_channels: Arc<RwLock<HashMap<i32, BroadcastSender>>>,
    /// Maps IncomingMessages to broadcast senders (like sync does)
    shared_channel_senders: Arc<RwLock<HashMap<IncomingMessages, Vec<BroadcastSender>>>>,
    /// Maps OutgoingMessages to receivers for client subscription
    shared_channel_receivers: Arc<RwLock<HashMap<OutgoingMessages, broadcast::Receiver<RoutedItem>>>>,
    /// Maps order IDs to their response channels
    order_channels: Arc<RwLock<HashMap<i32, BroadcastSender>>>,
    /// Maps execution IDs to their response channels (for commission reports)
    execution_channels: Arc<RwLock<HashMap<String, BroadcastSender>>>,
    /// Optional channel for order update stream
    order_update_stream: Arc<RwLock<Option<BroadcastSender>>>,
    /// Capacity of every broadcast channel this bus creates. Default
    /// `BROADCAST_CHANNEL_CAPACITY`; see `ClientBuilder::channel_capacity`.
    channel_capacity: usize,
    /// Channel for cleanup signals
    cleanup_sender: mpsc::UnboundedSender<CleanupSignal>,
    /// Handle to the message processing task
    process_task: Arc<RwLock<Option<task::JoinHandle<()>>>>,
    /// Latching shutdown flag, shared with the connection so a reconnect in
    /// progress sees the request.
    shutdown: Arc<ShutdownSignal>,
    /// The client's order-ID generator, raised from the NextValidId frame the
    /// reconnect handshake re-receives. Installed once via
    /// [`Self::set_order_ids`] before the processing task starts; absent in
    /// bus-only test fixtures, which never reconnect a client.
    order_ids: OnceLock<Arc<ClientIdManager>>,
    connected: Arc<AtomicBool>,
}

impl<S: AsyncStream> Drop for AsyncTcpMessageBus<S> {
    fn drop(&mut self) {
        debug!("dropping async tcp message bus");
        // Latch the shutdown flag; the message loop and any reconnect in
        // progress observe it on their next check.
        self.shutdown.request();
    }
}

impl<S: AsyncStream> AsyncTcpMessageBus<S> {
    /// Create a bus with the default channel capacity. Production goes
    /// through `with_channel_capacity` (the builder owns the default), so
    /// this exists for test fixtures that don't care about capacity.
    #[cfg(test)]
    pub fn new(connection: AsyncConnection<S>) -> Result<Self, Error> {
        Self::with_channel_capacity(connection, BROADCAST_CHANNEL_CAPACITY)
    }

    /// Create a new async TCP message bus whose broadcast channels hold up to
    /// `channel_capacity` frames per subscription before evicting the oldest.
    pub fn with_channel_capacity(connection: AsyncConnection<S>, channel_capacity: usize) -> Result<Self, Error> {
        let (cleanup_sender, cleanup_receiver) = mpsc::unbounded_channel();

        // Pre-create broadcast channels for all shared channels (like sync does)
        let mut shared_channel_senders = HashMap::new();
        let mut shared_channel_receivers = HashMap::new();

        for mapping in shared_channel_configuration::CHANNEL_MAPPINGS {
            let (sender, receiver) = broadcast::channel(channel_capacity);
            shared_channel_receivers.insert(mapping.request, receiver);

            // Map each response type to the sender (multiple response types can share same sender)
            for response_type in mapping.responses {
                shared_channel_senders.entry(*response_type).or_insert_with(Vec::new).push(sender.clone());
            }
        }

        let shutdown = connection.shutdown_signal();

        let message_bus = Self {
            connection: Arc::new(connection),
            request_channels: Arc::new(RwLock::new(HashMap::new())),
            shared_channel_senders: Arc::new(RwLock::new(shared_channel_senders)),
            shared_channel_receivers: Arc::new(RwLock::new(shared_channel_receivers)),
            order_channels: Arc::new(RwLock::new(HashMap::new())),
            execution_channels: Arc::new(RwLock::new(HashMap::new())),
            order_update_stream: Arc::new(RwLock::new(None)),
            channel_capacity,
            cleanup_sender,
            process_task: Arc::new(RwLock::new(None)),
            shutdown,
            order_ids: OnceLock::new(),
            connected: Arc::new(AtomicBool::new(true)),
        };

        // Start cleanup task
        let request_channels = message_bus.request_channels.clone();
        let order_channels = message_bus.order_channels.clone();
        let order_update_stream = message_bus.order_update_stream.clone();

        // A signal can be processed arbitrarily long after the drop that sent
        // it — including after a newer subscription registered under the same
        // key — so removal is gated on the registration being dead. See
        // `remove_if_dead` and `AsyncInternalSubscription::detach_receivers`.
        task::spawn(async move {
            let mut receiver = cleanup_receiver;
            while let Some(signal) = receiver.recv().await {
                match signal {
                    CleanupSignal::Request(request_id) => remove_if_dead(&request_channels, request_id, "request").await,
                    CleanupSignal::Order(order_id) => remove_if_dead(&order_channels, order_id, "order").await,
                    CleanupSignal::Shared(message_type) => {
                        // Shared channels are persistent and should not be removed
                        // They are created at initialization and reused across multiple requests
                        debug!("Subscription for shared channel {:?} ended (channel remains active)", message_type);
                    }
                    CleanupSignal::OrderUpdateStream => {
                        let mut stream = order_update_stream.write().await;
                        let removed = stream.as_ref().is_some_and(|sender| sender.receiver_count() == 0);
                        if removed {
                            *stream = None;
                        }
                        debug!("cleanup order update stream: removed={removed}");
                    }
                }
            }
        });

        Ok(message_bus)
    }

    /// Installs the client's order-ID generator so a successful reconnect
    /// re-seeds it from the handshake's NextValidId. Called exactly once,
    /// before [`Self::process_messages`] starts the processing task.
    pub(crate) fn set_order_ids(&self, order_ids: Arc<ClientIdManager>) {
        self.order_ids.set(order_ids).expect("order-id generator installed twice");
    }

    /// Start processing messages from TWS
    pub fn process_messages(self: Arc<Self>, _server_version: i32, _reconnect_delay: Duration) -> Result<(), Error> {
        let message_bus = self.clone();
        let shutdown = self.shutdown.clone();

        let handle = task::spawn(async move {
            loop {
                // The flag latches, so a request made while the loop was off
                // in `reconnect()` is still seen here.
                if shutdown.is_requested() {
                    debug!("Shutdown requested, stopping message processing");
                    break;
                }

                // Use select with shutdown notification instead of a polling sleep.
                // This prevents cancelling read_and_route_message mid-read, which
                // would corrupt the TCP stream (read_exact is not cancellation-safe).
                tokio::select! {
                    _ = shutdown.wait() => {
                        debug!("Shutdown notification received, stopping message processing");
                        break;
                    }
                    result = message_bus.read_and_route_message() => {
                        match result {
                            Ok(_) => continue,
                            Err(ref err) if err.is_read_timeout() => {
                                if message_bus.shutdown.is_requested() {
                                    debug!("dispatcher task exiting");
                                    break;
                                }
                                continue;
                            }
                            Err(ref err) if err.is_connection_lost() => {
                                error!("Connection error detected, attempting to reconnect: {err:?}");
                                message_bus.connected.store(false, Ordering::Relaxed);

                                match message_bus.connection.reconnect().await {
                                    Ok(_) => {
                                        // Shutdown may have been requested while the
                                        // connect was in flight; the reconnect
                                        // succeeded anyway, so check before reporting
                                        // the session as live and resetting channels.
                                        if message_bus.shutdown.is_requested() {
                                            debug!("shutdown requested during reconnect; dispatcher task exiting");
                                            break;
                                        }

                                        // The handshake re-received NextValidId; raise
                                        // the client's generator from the server's floor so
                                        // allocation never resumes below it. Only the initial
                                        // connection seeded the generator before this. Do it
                                        // before reporting the session as live so a caller
                                        // gating on `is_connected()` cannot allocate below
                                        // the new floor.
                                        if let Some(order_ids) = message_bus.order_ids.get() {
                                            let metadata = message_bus.connection.connection_metadata().await;
                                            order_ids.raise_order_id(metadata.next_order_id);
                                        }

                                        info!("Successfully reconnected to TWS/Gateway");
                                        message_bus.connected.store(true, Ordering::Relaxed);
                                        message_bus.reset_channels().await;
                                    }
                                    // Shutdown was requested while reconnecting:
                                    // not a failure, and the flag is already
                                    // set, so just exit the task.
                                    Err(Error::Shutdown) => {
                                        debug!("shutdown requested during reconnect; dispatcher task exiting");
                                        break;
                                    }
                                    Err(e) => {
                                        error!("Failed to reconnect to TWS/Gateway: {e:?}");
                                        message_bus.request_shutdown().await;
                                        break;
                                    }
                                }
                                continue;
                            }
                            Err(Error::Shutdown) => {
                                error!("Received shutdown signal, stopping message processing.");
                                break;
                            }
                            Err(err) => {
                                error!("Error processing message (shutting down): {err:?}");
                                message_bus.request_shutdown().await;
                                break;
                            }
                        }
                    }
                }
            }
        });

        // Store the task handle
        let process_task = self.process_task.clone();
        tokio::spawn(async move {
            let mut task_guard = process_task.write().await;
            *task_guard = Some(handle);
        });

        Ok(())
    }

    /// Read a message and route it to the appropriate channel
    pub(crate) async fn read_and_route_message(&self) -> Result<(), Error> {
        let message = self.connection.read_message().await?;

        // Use common routing logic
        match determine_routing(&message) {
            RoutingDecision::ByRequestId(request_id) => self.route_to_request_channel(request_id, message).await,
            RoutingDecision::ByOrderId(order_id) => self.route_to_order_channel(order_id, message).await,
            RoutingDecision::ByMessageType(message_type) => self.route_to_shared_channel(message_type, message).await,
            RoutingDecision::SharedMessage(message_type) => self.route_to_shared_channel(message_type, message).await,
            RoutingDecision::Error(payload) => self.route_error_message(payload).await,
            RoutingDecision::Shutdown => {
                debug!("Received shutdown message, calling request_shutdown");
                self.request_shutdown().await;
                Err(Error::Shutdown)
            }
        }
    }

    /// Reset all channels after reconnection
    async fn reset_channels(&self) {
        debug!("resetting message bus channels");

        for sender in self.request_channels.read().await.values() {
            let _ = sender.send(Error::ConnectionReset.into());
        }
        for sender in self.order_channels.read().await.values() {
            let _ = sender.send(Error::ConnectionReset.into());
        }
        // Shared channels too, mirroring sync's `notify_all`: an in-flight
        // open_orders/positions subscription awaits an end marker only the
        // pre-reconnect request could produce, so it would hang forever.
        // Unfiltered on purpose — `fail_one_shot_channels`' one-shot filter
        // protects live streams from *unrelated* errors, but a reset
        // terminates every stream by definition. The senders are not cleared
        // here, unlike the maps below.
        for sender in self.shared_channel_senders.read().await.values().flatten() {
            let _ = sender.send(Error::ConnectionReset.into());
        }

        self.request_channels.write().await.clear();
        self.order_channels.write().await.clear();
        self.execution_channels.write().await.clear();

        // The notice stream is the central carrier of connection-status
        // information (1100/1101/1102 land there), but TWS never frames the
        // socket reconnect itself and does not replay restoration notices on
        // the new connection — so publish the reconnect there the same way
        // `report_unroutable_frame` publishes decode failures: as a
        // synthesized notice. A connection-state consumer that recorded 1100
        // before the drop reconciles on this instead of stranding on it.
        // Published last: a consumer that resubscribes on it registers into
        // maps the clears above can no longer wipe.
        let _ = self.connection.notice_sender.send(transport_reconnect_notice());
    }

    /// Notify all waiting subscriptions about shutdown
    async fn request_shutdown(&self) {
        debug!("shutdown requested");

        // Set the shutdown flag and mark as disconnected
        self.connected.store(false, Ordering::Relaxed);
        self.shutdown.request();

        // Clear all channels - dropping the senders will close the channels
        // and cause all receivers to get RecvError::Closed. This diverges
        // from sync's shutdown, which sends Error::Shutdown before clearing:
        // async consumers see end-of-stream, sync consumers see the error.
        {
            let mut channels = self.request_channels.write().await;
            channels.clear();
        }

        {
            let mut channels = self.order_channels.write().await;
            channels.clear();
        }

        {
            let mut channels = self.shared_channel_senders.write().await;
            channels.clear();
        }

        {
            let mut channels = self.shared_channel_receivers.write().await;
            channels.clear();
        }

        {
            let mut order_update_stream = self.order_update_stream.write().await;
            *order_update_stream = None;
        }
    }

    /// Route error message using routing decision
    async fn route_error_message(&self, payload: DecodedError) -> Result<(), Error> {
        let id_owned_by_data_request = self.request_channels.read().await.contains_key(&payload.request_id);
        let sent_to_update_stream = match order_update_notice(&payload, id_owned_by_data_request) {
            Some(notice) => self.send_order_update_item(RoutedItem::Notice(notice)).await,
            None => false,
        };
        match classify_error(payload) {
            ErrorDisposition::NoticeOnly(notice) => {
                super::common::log_unrouted_notice(&notice);
                let _ = self.connection.notice_sender.send(notice);
            }
            ErrorDisposition::NoticeAndFailOneShots(notice, error) => {
                super::common::log_unrouted_notice(&notice);
                let _ = self.connection.notice_sender.send(notice);
                self.fail_one_shot_channels(error).await;
            }
            ErrorDisposition::Route(request_id, item) => {
                self.deliver_to_request_id(request_id, item, sent_to_update_stream).await;
            }
        }
        Ok(())
    }

    /// Deliver a request-less hard error to every in-flight one-shot shared
    /// request so it fails fast rather than hanging. Streaming shared channels
    /// are excluded (see [`shared_channel_configuration::exclusive_one_shot_response_types`]).
    async fn fail_one_shot_channels(&self, error: Error) {
        let channels = self.shared_channel_senders.read().await;
        for message_type in shared_channel_configuration::exclusive_one_shot_response_types() {
            if let Some(senders) = channels.get(message_type) {
                for sender in senders {
                    let _ = sender.send(RoutedItem::Error(error.clone()));
                }
            }
        }
    }

    /// Deliver a pre-classified Notice or Error to its owning subscription.
    /// Tries the request-channel first, falls back to the order-channel for
    /// notices/errors that arrive bound to an order_id.
    async fn deliver_to_request_id(&self, request_id: i32, item: RoutedItem, sent_to_update_stream: bool) {
        {
            let channels = self.request_channels.read().await;
            if let Some(sender) = channels.get(&request_id) {
                let _ = sender.send(item);
                return;
            }
        }
        {
            let order_channels = self.order_channels.read().await;
            if let Some(sender) = order_channels.get(&request_id) {
                let _ = sender.send(item);
                return;
            }
        }
        if !sent_to_update_stream {
            log_orphan(request_id, &item);
        }
    }

    /// Route message to request-specific channel
    async fn route_to_request_channel(&self, request_id: i32, message: ResponseMessage) -> Result<(), Error> {
        let channels = self.request_channels.read().await;
        if let Some(sender) = channels.get(&request_id) {
            let _ = sender.send(message.into());
        }
        Ok(())
    }

    /// Route message to order-specific channel
    async fn route_to_order_channel(&self, order_id: i32, message: ResponseMessage) -> Result<(), Error> {
        let routed = self.send_order_update(&message).await;
        let strategy = order_routing_strategy(message.message_type());

        match strategy {
            OrderRoutingStrategy::OrderUpdateOnly => {}
            OrderRoutingStrategy::ExecutionData => {
                // Try order_id channel first, then request_id, storing execution_id mapping
                if let Some(actual_order_id) = message.order_id() {
                    let channels = self.order_channels.read().await;
                    if let Some(sender) = channels.get(&actual_order_id) {
                        self.store_execution_mapping(&message, sender).await;
                        let _ = sender.send(message.into());
                        return Ok(());
                    }
                }
                if let Some(req_id) = message.request_id() {
                    let channels = self.request_channels.read().await;
                    if let Some(sender) = channels.get(&req_id) {
                        self.store_execution_mapping(&message, sender).await;
                        let _ = sender.send(message.into());
                        return Ok(());
                    }
                }
                if !routed {
                    warn!("could not route ExecutionData message {:?}", message);
                }
            }
            OrderRoutingStrategy::ExecutionDataEnd => {
                if let Some(actual_order_id) = message.order_id() {
                    let channels = self.order_channels.read().await;
                    if let Some(sender) = channels.get(&actual_order_id) {
                        let _ = sender.send(message.into());
                        return Ok(());
                    }
                }
                if let Some(req_id) = message.request_id() {
                    let channels = self.request_channels.read().await;
                    if let Some(sender) = channels.get(&req_id) {
                        let _ = sender.send(message.into());
                        return Ok(());
                    }
                }
                warn!("could not route ExecutionDataEnd message {:?}", message);
            }
            OrderRoutingStrategy::OrderOrShared => {
                if let Some(actual_order_id) = message.order_id() {
                    let channels = self.order_channels.read().await;
                    if let Some(sender) = channels.get(&actual_order_id) {
                        let _ = sender.send(message.into());
                        return Ok(());
                    }
                    drop(channels);

                    let shared_channels = self.shared_channel_senders.read().await;
                    if let Some(senders) = shared_channels.get(&message.message_type()) {
                        for sender in senders {
                            let _ = sender.send(message.clone().into());
                        }
                        return Ok(());
                    }
                }
                if !routed {
                    warn!("could not route message {:?}", message);
                }
            }
            OrderRoutingStrategy::ByExecutionId => {
                if let Some(execution_id) = message.execution_id() {
                    let exec_channels = self.execution_channels.read().await;
                    if let Some(sender) = exec_channels.get(&execution_id) {
                        let _ = sender.send(message.into());
                        return Ok(());
                    }
                }
            }
            OrderRoutingStrategy::SharedOnly => {
                let shared_channels = self.shared_channel_senders.read().await;
                if let Some(senders) = shared_channels.get(&message.message_type()) {
                    for sender in senders {
                        let _ = sender.send(message.clone().into());
                    }
                    return Ok(());
                }
                if !routed {
                    warn!("could not route message {:?}", message);
                }
            }
            OrderRoutingStrategy::ByOrderId => {
                if order_id >= 0 {
                    let channels = self.order_channels.read().await;
                    if let Some(sender) = channels.get(&order_id) {
                        let _ = sender.send(message.into());
                        return Ok(());
                    }
                }
                if !routed {
                    warn!("could not route message {:?}", message);
                }
            }
        }

        Ok(())
    }

    /// Store execution_id -> sender mapping for commission report routing
    async fn store_execution_mapping(&self, message: &ResponseMessage, sender: &BroadcastSender) {
        if let Some(execution_id) = message.execution_id() {
            let mut exec_channels = self.execution_channels.write().await;
            exec_channels.insert(execution_id, sender.clone());
        }
    }

    /// Route message to shared channel
    async fn route_to_shared_channel(&self, message_type: IncomingMessages, message: ResponseMessage) -> Result<(), Error> {
        // Send order-related messages to order update stream
        match message_type {
            IncomingMessages::OpenOrder
            | IncomingMessages::OrderStatus
            | IncomingMessages::ExecutionData
            | IncomingMessages::CommissionsReport
            | IncomingMessages::CompletedOrder => {
                self.send_order_update(&message).await;
            }
            _ => {}
        }

        // Route to all senders for this message type (like sync does)
        let channels = self.shared_channel_senders.read().await;
        if let Some(senders) = channels.get(&message_type) {
            // Broadcast to all subscribers
            for sender in senders {
                if let Err(e) = sender.send(message.clone().into()) {
                    warn!("error sending to shared channel for {message_type:?}: {e}");
                }
            }
        } else {
            // Nothing claimed the frame. Silent until now, which is why a
            // desynchronized stream looked identical to an idle one.
            report_unroutable_frame(&message, &self.connection.notice_sender);
        }

        Ok(())
    }

    /// Send message to order update stream if it exists
    async fn send_order_update(&self, message: &ResponseMessage) -> bool {
        self.send_order_update_item(message.clone().into()).await
    }

    async fn send_order_update_item(&self, item: RoutedItem) -> bool {
        let order_update_stream = self.order_update_stream.read().await;
        if let Some(sender) = order_update_stream.as_ref() {
            if let Err(e) = sender.send(item) {
                warn!("error sending to order update stream: {e}");
                return false;
            }
            return true;
        }
        false
    }
}

#[async_trait]
impl<S: AsyncStream> AsyncMessageBus for AsyncTcpMessageBus<S> {
    async fn send_request(&self, request_id: i32, message: Vec<u8>) -> Result<AsyncInternalSubscription, Error> {
        let (sender, receiver) = broadcast::channel(self.channel_capacity);

        {
            let mut channels = self.request_channels.write().await;
            channels.insert(request_id, sender);
        }

        self.connection.write_message(&message).await?;

        Ok(AsyncInternalSubscription::with_cleanup(
            receiver,
            self.cleanup_sender.clone(),
            CleanupSignal::Request(request_id),
        ))
    }

    async fn send_order_request(&self, order_id: i32, message: Vec<u8>) -> Result<AsyncInternalSubscription, Error> {
        let (sender, receiver) = broadcast::channel(self.channel_capacity);

        {
            let mut channels = self.order_channels.write().await;
            channels.insert(order_id, sender);
        }

        self.connection.write_message(&message).await?;

        Ok(AsyncInternalSubscription::with_cleanup(
            receiver,
            self.cleanup_sender.clone(),
            CleanupSignal::Order(order_id),
        ))
    }

    async fn send_shared_request(&self, message_type: OutgoingMessages, message: Vec<u8>) -> Result<AsyncInternalSubscription, Error> {
        let receiver = {
            let channels = self.shared_channel_receivers.read().await;
            if let Some(receiver) = channels.get(&message_type) {
                receiver.resubscribe()
            } else {
                return Err(Error::InvalidArgument(format!(
                    "No shared channel configured for message type: {:?}",
                    message_type
                )));
            }
        };

        self.connection.write_message(&message).await?;

        Ok(AsyncInternalSubscription::with_cleanup(
            receiver,
            self.cleanup_sender.clone(),
            CleanupSignal::Shared(message_type),
        ))
    }

    async fn send_message(&self, message: Vec<u8>) -> Result<(), Error> {
        self.connection.write_message(&message).await
    }

    async fn cancel_subscription(&self, request_id: i32, message: Vec<u8>) -> Result<(), Error> {
        self.connection.write_message(&message).await?;

        // Single write lock: the previous version held a read guard while
        // awaiting the write upgrade and self-deadlocked on the same task.
        let mut channels = self.request_channels.write().await;
        if let Some(sender) = channels.get(&request_id) {
            let _ = sender.send(Error::Cancelled.into());
        }
        channels.remove(&request_id);

        Ok(())
    }

    async fn cancel_order_subscription(&self, order_id: i32, message: Vec<u8>) -> Result<(), Error> {
        self.connection.write_message(&message).await?;

        let mut channels = self.order_channels.write().await;
        if let Some(sender) = channels.get(&order_id) {
            let _ = sender.send(Error::Cancelled.into());
        }
        channels.remove(&order_id);

        Ok(())
    }

    async fn create_order_update_subscription(&self) -> Result<AsyncInternalSubscription, Error> {
        let mut order_update_stream = self.order_update_stream.write().await;

        // A registration with no receivers is a dropped stream whose cleanup
        // signal has not been processed yet (see `remove_if_dead`); replace it
        // rather than refusing, so drop-then-recreate never races the cleanup task.
        if order_update_stream.as_ref().is_some_and(|sender| sender.receiver_count() > 0) {
            return Err(Error::AlreadySubscribed);
        }

        let (sender, receiver) = broadcast::channel(self.channel_capacity);

        *order_update_stream = Some(sender);

        Ok(AsyncInternalSubscription::with_cleanup(
            receiver,
            self.cleanup_sender.clone(),
            CleanupSignal::OrderUpdateStream,
        ))
    }

    fn notice_subscribe(&self) -> crate::subscriptions::notice_stream::async_impl::NoticeStream {
        crate::subscriptions::notice_stream::async_impl::NoticeStream::new(self.connection.notice_sender.subscribe())
    }

    async fn ensure_shutdown(&self) {
        debug!("ensure_shutdown called");

        // Request shutdown
        self.request_shutdown().await;

        // Wait for the processing task to finish
        let task_handle = {
            let mut task_guard = self.process_task.write().await;
            task_guard.take()
        };

        if let Some(handle) = task_handle {
            debug!("Waiting for processing task to finish");
            if let Err(e) = handle.await {
                warn!("Error joining processing task: {e}");
            }
            debug!("Processing task finished");
        }
    }

    fn request_shutdown_sync(&self) {
        debug!("sync shutdown requested");
        self.connected.store(false, Ordering::Relaxed);
        // Latching and runtime-free: safe from `Drop`.
        self.shutdown.request();
    }

    fn is_connected(&self) -> bool {
        self.connected.load(Ordering::Relaxed) && !self.shutdown.is_requested()
    }
}

#[cfg(test)]
mod memory;
#[cfg(test)]
pub(crate) use memory::MemoryStream;

#[cfg(test)]
pub(crate) mod test_listener;

#[cfg(test)]
#[path = "async_tests.rs"]
mod tests;