nautilus_data/engine/
registry.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
16use super::{
17    AHashSet, ClientId, DataClientAdapter, DataEngine, FAILED, Venue, check_key_in_map,
18    check_key_not_in_map, check_predicate_true,
19};
20
21impl DataEngine {
22    #[must_use]
23    pub(crate) fn is_external_client(&self, client_id: ClientId) -> bool {
24        self.cache.borrow().is_external_data_client(&client_id)
25    }
26
27    /// Registers the `client` with the engine with an optional venue `routing`.
28    ///
29    ///
30    /// # Panics
31    ///
32    /// Panics if a client with the same client ID has already been registered.
33    pub fn register_client(&mut self, client: DataClientAdapter, routing: Option<Venue>) {
34        let client_id = client.client_id();
35
36        check_key_not_in_map(&client_id, &self.clients, "client_id", "clients").expect(FAILED);
37
38        if let Some(routing) = routing {
39            self.routing_map.insert(routing, client_id);
40            log::debug!("Set client {client_id} routing for {routing}");
41        }
42
43        if client.venue.is_none() && self.default_client_id.is_none() {
44            self.default_client_id = Some(client_id);
45            log::debug!("Registered client {client_id} for default routing");
46        }
47
48        self.clients.insert(client_id, client);
49        log::debug!("Registered client {client_id}");
50    }
51
52    /// Deregisters the client for the `client_id`.
53    ///
54    /// # Panics
55    ///
56    /// Panics if the client ID has not been registered.
57    pub fn deregister_client(&mut self, client_id: &ClientId) {
58        check_key_in_map(client_id, &self.clients, "client_id", "clients").expect(FAILED);
59
60        if self.default_client_id.as_ref() == Some(client_id) {
61            self.default_client_id = None;
62        }
63
64        self.clients.shift_remove(client_id);
65        log::info!("Deregistered client {client_id}");
66    }
67
68    /// Registers the data `client` with the engine as the default routing client.
69    ///
70    /// When a specific venue routing cannot be found, this client will receive messages.
71    ///
72    /// # Warnings
73    ///
74    /// Any existing default routing client will be overwritten.
75    ///
76    /// # Panics
77    ///
78    /// Panics if a default client has already been registered.
79    pub fn register_default_client(&mut self, client: DataClientAdapter) {
80        check_predicate_true(
81            self.default_client_id.is_none(),
82            "default client already registered",
83        )
84        .expect(FAILED);
85
86        let client_id = client.client_id();
87        self.clients.insert(client_id, client);
88        self.default_client_id = Some(client_id);
89        log::debug!("Registered default client {client_id}");
90    }
91
92    /// Marks an already-registered client as the default for fallback routing.
93    ///
94    /// # Errors
95    ///
96    /// Returns an error if no client is registered with the given ID, or a different
97    /// client is already the default.
98    pub fn set_default_client(&mut self, client_id: ClientId) -> anyhow::Result<()> {
99        if self.default_client_id.is_some_and(|id| id != client_id) {
100            anyhow::bail!("default client already registered");
101        }
102
103        if !self.clients.contains_key(&client_id) {
104            anyhow::bail!("No client registered with ID {client_id}");
105        }
106
107        self.default_client_id = Some(client_id);
108        log::debug!("Set client {client_id} as default");
109        Ok(())
110    }
111
112    /// Sets routing for a specific venue to a given client ID.
113    ///
114    /// # Errors
115    ///
116    /// Returns an error if the client ID is not registered, or the venue is already routed to a
117    /// different client.
118    pub fn register_venue_routing(
119        &mut self,
120        client_id: ClientId,
121        venue: Venue,
122    ) -> anyhow::Result<()> {
123        if !self.clients.contains_key(&client_id) {
124            anyhow::bail!("No client registered with ID {client_id}");
125        }
126
127        if let Some(existing_client_id) = self.routing_map.get(&venue)
128            && *existing_client_id != client_id
129        {
130            anyhow::bail!(
131                "Venue {venue} already routed to {existing_client_id}, \
132                 cannot re-route to {client_id}"
133            );
134        }
135
136        self.routing_map.insert(venue, client_id);
137        log::debug!("Set client {client_id} routing for {venue}");
138        Ok(())
139    }
140
141    /// Returns connection status for each registered client.
142    #[must_use]
143    pub fn client_connection_status(&self) -> Vec<(ClientId, bool)> {
144        self.get_clients()
145            .into_iter()
146            .map(|client| (client.client_id(), client.is_connected()))
147            .collect()
148    }
149
150    /// Returns a list of all registered client IDs, including the default client if set.
151    #[must_use]
152    pub fn registered_clients(&self) -> Vec<ClientId> {
153        self.get_clients()
154            .into_iter()
155            .map(|client| client.client_id())
156            .collect()
157    }
158
159    pub(crate) fn collect_subscriptions<F, T>(&self, get_subs: F) -> Vec<T>
160    where
161        F: Fn(&DataClientAdapter) -> &AHashSet<T>,
162        T: Clone,
163    {
164        self.get_clients()
165            .into_iter()
166            .flat_map(get_subs)
167            .cloned()
168            .collect()
169    }
170
171    #[must_use]
172    pub fn get_clients(&self) -> Vec<&DataClientAdapter> {
173        self.clients.values().collect()
174    }
175
176    #[must_use]
177    pub fn get_clients_mut(&mut self) -> Vec<&mut DataClientAdapter> {
178        self.clients.values_mut().collect()
179    }
180
181    pub fn get_client(
182        &mut self,
183        client_id: Option<&ClientId>,
184        venue: Option<&Venue>,
185    ) -> Option<&mut DataClientAdapter> {
186        if let Some(client_id) = client_id {
187            return self.clients.get_mut(client_id);
188        }
189
190        if let Some(v) = venue
191            && let Some(client_id) = self.routing_map.get(v)
192        {
193            return self.clients.get_mut(client_id);
194        }
195
196        self.get_default_client()
197    }
198
199    /// Resolves the client for a subscribe/unsubscribe command.
200    ///
201    /// When `BACKTEST` is registered, all commands route through it regardless of
202    /// the command's `client_id` or `venue`. Request paths skip this override.
203    pub(super) fn get_command_client(
204        &mut self,
205        client_id: Option<&ClientId>,
206        venue: Option<&Venue>,
207    ) -> Option<&mut DataClientAdapter> {
208        let backtest_id = ClientId::new("BACKTEST");
209        if self.clients.contains_key(&backtest_id) {
210            return self.clients.get_mut(&backtest_id);
211        }
212
213        self.get_client(client_id, venue)
214    }
215
216    fn get_default_client(&mut self) -> Option<&mut DataClientAdapter> {
217        match self.default_client_id {
218            Some(id) => self.clients.get_mut(&id),
219            None => None,
220        }
221    }
222}
nautilus_data/engine/registry.rs

nautilus_data/engine/
registry.rs
