nautilus_interactive_brokers/common/
shared_client.rs

1// -------------------------------------------------------------------------------------------------
2//  Copyright (C) 2015-2026 Nautech Systems Pty Ltd. All rights reserved.
3//  https://nautechsystems.io
4//
5//  Licensed under the GNU Lesser General Public License Version 3.0 (the "License");
6//  You may not use this file except in compliance with the License.
7//  You may obtain a copy of the License at https://www.gnu.org/licenses/lgpl-3.0.en.html
8//
9//  Unless required by applicable law or agreed to in writing, software
10//  distributed under the License is distributed on an "AS IS" BASIS,
11//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12//  See the License for the specific language governing permissions and
13//  limitations under the License.
14// -------------------------------------------------------------------------------------------------
15
16//! Shared IB API client connection per (host, port, client_id).
17//!
18//! Data, execution, and historical clients use a single TCP connection per logical
19//! connection to avoid client ID conflicts and redundant connections (parity with
20//! Python's get_cached_ib_client).
21
22use std::{
23    collections::HashMap,
24    fmt::Debug,
25    ops::Deref,
26    sync::{Arc, LazyLock},
27    time::Duration,
28};
29
30use anyhow::Context;
31use ibapi::client::Client;
32use parking_lot::Mutex;
33
34/// Key for the connection registry: (host, port, client_id).
35#[derive(Clone, Debug, Eq, Hash, PartialEq)]
36struct ConnectionKey(String, u16, i32);
37
38/// Registry entry: shared client and its ref count.
39type RegistryMap = HashMap<ConnectionKey, (Arc<Client>, u32)>;
40
41/// Global registry: one shared client per (host, port, client_id) with ref count.
42static REGISTRY: LazyLock<Arc<Mutex<RegistryMap>>> =
43    LazyLock::new(|| Arc::new(Mutex::new(HashMap::new())));
44
45/// Handle to a shared IB client; when dropped, ref count is decremented and the
46/// connection is removed from the registry when the count reaches zero.
47pub struct SharedClientHandle {
48    client: Arc<Client>,
49    registry: Arc<Mutex<RegistryMap>>,
50    key: ConnectionKey,
51}
52
53impl Debug for SharedClientHandle {
54    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
55        f.debug_struct(stringify!(SharedClientHandle))
56            .field("key", &self.key)
57            .finish_non_exhaustive()
58    }
59}
60
61impl SharedClientHandle {
62    fn new(client: Arc<Client>, registry: Arc<Mutex<RegistryMap>>, key: ConnectionKey) -> Self {
63        Self {
64            client,
65            registry,
66            key,
67        }
68    }
69
70    /// Returns a reference to the underlying `Arc<Client>` for call sites that need it.
71    pub fn as_arc(&self) -> &Arc<Client> {
72        &self.client
73    }
74}
75
76impl Deref for SharedClientHandle {
77    type Target = Client;
78
79    fn deref(&self) -> &Self::Target {
80        self.client.as_ref()
81    }
82}
83
84impl Drop for SharedClientHandle {
85    fn drop(&mut self) {
86        let mut guard = self.registry.lock();
87        if let Some((_, ref_count)) = guard.get_mut(&self.key) {
88            *ref_count = ref_count.saturating_sub(1);
89            if *ref_count == 0 {
90                guard.remove(&self.key);
91                tracing::debug!(
92                    "Shared IB client removed from registry (host={}, port={}, client_id={})",
93                    self.key.0,
94                    self.key.1,
95                    self.key.2
96                );
97            }
98        }
99    }
100}
101
102/// Returns a handle to the shared IB client for the given (host, port, client_id).
103/// If a connection already exists, its ref count is incremented and the same client
104/// is returned. Otherwise a new connection is established and registered.
105///
106/// # Errors
107///
108/// Returns an error if connecting to IB Gateway/TWS fails.
109pub async fn get_or_connect(
110    host: &str,
111    port: u16,
112    client_id: i32,
113    connection_timeout_secs: u64,
114) -> anyhow::Result<SharedClientHandle> {
115    let key = ConnectionKey(host.to_string(), port, client_id);
116    let registry = Arc::clone(&REGISTRY);
117
118    log::debug!(
119        "Acquiring shared IB client (host={}, port={}, client_id={}, timeout_secs={})",
120        host,
121        port,
122        client_id,
123        connection_timeout_secs
124    );
125
126    let (reuse_client, ref_count_val) = {
127        let mut guard = registry.lock();
128
129        if let Some((client, ref_count)) = guard.get_mut(&key) {
130            if client.is_connected() {
131                *ref_count += 1;
132                let ref_count_val = *ref_count;
133                let client = Arc::clone(client);
134                (Some(client), ref_count_val)
135            } else {
136                tracing::debug!(
137                    "Removing disconnected shared IB client before reconnect (host={}, port={}, client_id={})",
138                    host,
139                    port,
140                    client_id
141                );
142                guard.remove(&key);
143                (None, 0)
144            }
145        } else {
146            (None, 0)
147        }
148    };
149
150    if let Some(client) = reuse_client {
151        log::debug!(
152            "Reusing shared IB client (host={}, port={}, client_id={}, ref_count={})",
153            host,
154            port,
155            client_id,
156            ref_count_val
157        );
158        return Ok(SharedClientHandle::new(client, registry, key));
159    }
160
161    let address = format!("{host}:{port}");
162    let connect_timeout = Duration::from_secs(connection_timeout_secs);
163    log::debug!(
164        "No shared IB client found, establishing new connection to {} with timeout {:?}",
165        address,
166        connect_timeout
167    );
168    let client = tokio::time::timeout(connect_timeout, Client::connect(&address, client_id))
169        .await
170        .map_err(|_| {
171            anyhow::anyhow!(
172                "Timed out connecting to IB Gateway/TWS after {}s",
173                connection_timeout_secs
174            )
175        })?
176        .context("Failed to connect to IB Gateway/TWS")?;
177    let client = Arc::new(client);
178
179    {
180        let mut guard = registry.lock();
181        log::debug!(
182            "Registering shared IB client in registry (host={}, port={}, client_id={})",
183            host,
184            port,
185            client_id
186        );
187        guard.insert(key.clone(), (Arc::clone(&client), 1));
188    }
189
190    tracing::debug!(
191        "Registered new shared IB client (host={}, port={}, client_id={})",
192        host,
193        port,
194        client_id
195    );
196
197    Ok(SharedClientHandle::new(client, registry, key))
198}
nautilus_interactive_brokers/common/shared_client.rs

nautilus_interactive_brokers/common/
shared_client.rs
