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

nautilus_data/engine/
registry.rs
