nautilus_live/node/
builder.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//! Builder for constructing [`LiveNode`] instances.
17
18use std::{cell::RefCell, collections::HashMap, fmt::Debug, rc::Rc, time::Duration};
19
20use ahash::{AHashMap, AHashSet};
21use nautilus_common::{
22    cache::{CacheConfig, database::CacheDatabaseFactory},
23    clients::ExecutionClient,
24    clock::Clock,
25    enums::Environment,
26    factories::{
27        ClientConfig, DataClientFactory, ExecutionClientFactory, SimulatedExecutionClientFactory,
28    },
29    logging::logger::LoggerConfig,
30    msgbus::{
31        BusMessage, MessageBusBackingFactory, MessageBusConfig, MessageBusExternalEgress,
32        MessageBusExternalIngress, external_egress_from_backing, external_io_from_backing,
33    },
34};
35use nautilus_core::UUID4;
36use nautilus_data::client::DataClientAdapter;
37use nautilus_execution::engine::ExecutionEngine;
38use nautilus_model::identifiers::{TraderId, Venue};
39use nautilus_portfolio::config::PortfolioConfig;
40#[cfg(feature = "streaming")]
41use nautilus_system::config::StreamingConfig;
42#[cfg(feature = "python")]
43use nautilus_system::trader::Trader;
44use nautilus_system::{
45    clock_factory::ClockFactory,
46    event_store::{EventStoreFactory, KernelEventStore},
47    kernel::{NautilusKernel, NautilusKernelDependencies},
48};
49use nautilus_trading::ImportableControllerConfig;
50
51use super::{
52    LiveNode, LiveNodeHandle,
53    config::{
54        LiveDataEngineConfig, LiveExecutionEngineConfig, LiveNodeConfig, LiveRiskEngineConfig,
55        RoutingConfig, validate_live_environment,
56    },
57};
58use crate::{
59    execution::{
60        client::LiveExecutionClient,
61        manager::{ExecutionManager, ExecutionManagerConfig},
62    },
63    runner::AsyncRunner,
64    socket::SocketReconnectRegistry,
65};
66
67#[derive(Debug)]
68enum ExecutionClientFactoryEntry {
69    Adapter(Box<dyn ExecutionClientFactory>),
70    Simulated(Box<dyn SimulatedExecutionClientFactory>),
71}
72
73pub(crate) struct ExternalMessageBusIngress(Box<dyn MessageBusExternalIngress>);
74
75impl Debug for ExternalMessageBusIngress {
76    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
77        f.debug_struct(stringify!(ExternalMessageBusIngress))
78            .finish_non_exhaustive()
79    }
80}
81
82/// Builder for constructing a [`LiveNode`] with a fluent API.
83///
84/// Provides configuration options specific to live nodes, including client factory
85/// registration, timeout settings, and optional event-store injection for run-lifecycle
86/// audit and replay (see [`Self::with_event_store`]).
87#[cfg_attr(
88    feature = "python",
89    pyo3::pyclass(module = "nautilus_trader.live", unsendable)
90)]
91pub struct LiveNodeBuilder {
92    name: String,
93    config: LiveNodeConfig,
94    data_client_factories: HashMap<String, Box<dyn DataClientFactory>>,
95    exec_client_factories: HashMap<String, ExecutionClientFactoryEntry>,
96    data_client_configs: HashMap<String, Box<dyn ClientConfig>>,
97    exec_client_configs: HashMap<String, Box<dyn ClientConfig>>,
98    data_client_routing: HashMap<String, RoutingConfig>,
99    exec_client_routing: HashMap<String, RoutingConfig>,
100    event_store_factory: Option<EventStoreFactory>,
101    clock_factory: Option<ClockFactory>,
102    cache_database_factory: Option<Box<dyn CacheDatabaseFactory>>,
103    external_msgbus_factory: Option<Box<dyn MessageBusBackingFactory>>,
104    external_msgbus_egress: Option<Box<dyn MessageBusExternalEgress>>,
105    external_msgbus_ingress: Option<ExternalMessageBusIngress>,
106}
107
108impl Debug for LiveNodeBuilder {
109    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
110        f.debug_struct(stringify!(LiveNodeBuilder))
111            .field("name", &self.name)
112            .field("config", &self.config)
113            .field("data_client_factories", &self.data_client_factories.keys())
114            .field("exec_client_factories", &self.exec_client_factories.keys())
115            .field("data_client_configs", &self.data_client_configs.keys())
116            .field("exec_client_configs", &self.exec_client_configs.keys())
117            .field("event_store_factory", &self.event_store_factory.is_some())
118            .field("clock_factory", &self.clock_factory.is_some())
119            .field(
120                "cache_database_factory",
121                &self.cache_database_factory.is_some(),
122            )
123            .field(
124                "external_msgbus_factory",
125                &self.external_msgbus_factory.is_some(),
126            )
127            .field(
128                "external_msgbus_egress",
129                &self.external_msgbus_egress.is_some(),
130            )
131            .field(
132                "external_msgbus_ingress",
133                &self.external_msgbus_ingress.is_some(),
134            )
135            .finish_non_exhaustive()
136    }
137}
138
139impl LiveNodeBuilder {
140    /// Creates a new [`LiveNodeBuilder`] with required parameters.
141    ///
142    /// # Errors
143    ///
144    /// Returns an error if `environment` is invalid (BACKTEST).
145    pub fn new(trader_id: TraderId, environment: Environment) -> anyhow::Result<Self> {
146        validate_live_environment(environment)?;
147
148        let config = LiveNodeConfig {
149            environment,
150            trader_id,
151            ..Default::default()
152        };
153
154        Ok(Self {
155            name: "LiveNode".to_string(),
156            config,
157            data_client_factories: HashMap::new(),
158            exec_client_factories: HashMap::new(),
159            data_client_configs: HashMap::new(),
160            exec_client_configs: HashMap::new(),
161            data_client_routing: HashMap::new(),
162            exec_client_routing: HashMap::new(),
163            event_store_factory: None,
164            clock_factory: None,
165            cache_database_factory: None,
166            external_msgbus_factory: None,
167            external_msgbus_egress: None,
168            external_msgbus_ingress: None,
169        })
170    }
171
172    /// Creates a new [`LiveNodeBuilder`] from an existing [`LiveNodeConfig`].
173    ///
174    /// # Errors
175    ///
176    /// Returns an error if the config's environment is invalid (BACKTEST).
177    pub fn from_config(config: LiveNodeConfig) -> anyhow::Result<Self> {
178        validate_live_environment(config.environment)?;
179
180        Ok(Self {
181            name: "LiveNode".to_string(),
182            config,
183            data_client_factories: HashMap::new(),
184            exec_client_factories: HashMap::new(),
185            data_client_configs: HashMap::new(),
186            exec_client_configs: HashMap::new(),
187            data_client_routing: HashMap::new(),
188            exec_client_routing: HashMap::new(),
189            event_store_factory: None,
190            clock_factory: None,
191            cache_database_factory: None,
192            external_msgbus_factory: None,
193            external_msgbus_egress: None,
194            external_msgbus_ingress: None,
195        })
196    }
197
198    /// Returns the name for the node.
199    #[must_use]
200    pub fn name(&self) -> &str {
201        &self.name
202    }
203
204    /// Set the name for the node.
205    #[must_use]
206    pub fn with_name(mut self, name: impl Into<String>) -> Self {
207        self.name = name.into();
208        self
209    }
210
211    /// Set the instance ID for the node.
212    #[must_use]
213    pub const fn with_instance_id(mut self, instance_id: UUID4) -> Self {
214        self.config.instance_id = Some(instance_id);
215        self
216    }
217
218    /// Configure whether to load state on startup.
219    #[must_use]
220    pub const fn with_load_state(mut self, load_state: bool) -> Self {
221        self.config.load_state = load_state;
222        self
223    }
224
225    /// Configure whether to save state on shutdown.
226    #[must_use]
227    pub const fn with_save_state(mut self, save_state: bool) -> Self {
228        self.config.save_state = save_state;
229        self
230    }
231
232    /// Set the importable controller configuration for the node.
233    ///
234    /// The controller is instantiated and registered with the trader during
235    /// [`LiveNodeBuilder::build`], enabling runtime strategy/actor management
236    /// (create, start, stop, remove) without restarting the node. This mirrors
237    /// the `controller` field on [`LiveNodeConfig`] used by the config-based
238    /// [`LiveNode::build`] path, so a builder that also registers client
239    /// factories can host a controller in a single node.
240    #[must_use]
241    pub fn with_controller(mut self, controller: ImportableControllerConfig) -> Self {
242        self.config.controller = Some(controller);
243        self
244    }
245
246    /// Set the connection timeout in seconds.
247    #[must_use]
248    pub const fn with_timeout_connection(mut self, timeout_secs: u64) -> Self {
249        self.config.timeout_connection = Duration::from_secs(timeout_secs);
250        self
251    }
252
253    /// Set the reconciliation timeout in seconds.
254    #[must_use]
255    pub const fn with_timeout_reconciliation(mut self, timeout_secs: u64) -> Self {
256        self.config.timeout_reconciliation = Duration::from_secs(timeout_secs);
257        self
258    }
259
260    /// Configure whether to run startup reconciliation.
261    #[must_use]
262    pub fn with_reconciliation(mut self, reconciliation: bool) -> Self {
263        self.config.exec_engine.reconciliation = reconciliation;
264        self
265    }
266
267    /// Set the reconciliation lookback in minutes.
268    #[must_use]
269    pub fn with_reconciliation_lookback_mins(mut self, mins: u32) -> Self {
270        self.config.exec_engine.reconciliation_lookback_mins = Some(mins);
271        self
272    }
273
274    /// Set the portfolio initialization timeout in seconds.
275    #[must_use]
276    pub const fn with_timeout_portfolio(mut self, timeout_secs: u64) -> Self {
277        self.config.timeout_portfolio = Duration::from_secs(timeout_secs);
278        self
279    }
280
281    /// Set the disconnection timeout in seconds.
282    #[must_use]
283    pub const fn with_timeout_disconnection_secs(mut self, timeout_secs: u64) -> Self {
284        self.config.timeout_disconnection = Duration::from_secs(timeout_secs);
285        self
286    }
287
288    /// Set the post-stop delay in seconds.
289    #[must_use]
290    pub const fn with_delay_post_stop_secs(mut self, delay_secs: u64) -> Self {
291        self.config.delay_post_stop = Duration::from_secs(delay_secs);
292        self
293    }
294
295    /// Set the shutdown timeout in seconds.
296    #[must_use]
297    pub const fn with_delay_shutdown_secs(mut self, delay_secs: u64) -> Self {
298        self.config.timeout_shutdown = Duration::from_secs(delay_secs);
299        self
300    }
301
302    /// Inject a caller-supplied clock factory for the kernel and component clocks.
303    #[must_use]
304    pub fn with_clock_factory<F>(mut self, factory: F) -> Self
305    where
306        F: Fn() -> Rc<RefCell<dyn Clock>> + 'static,
307    {
308        self.clock_factory = Some(ClockFactory::new(factory));
309        self
310    }
311
312    /// Set the cache configuration.
313    #[must_use]
314    pub fn with_cache_config(mut self, config: CacheConfig) -> Self {
315        self.config.cache = Some(config);
316        self
317    }
318
319    /// Install the cache database backing from a factory.
320    ///
321    /// The node constructs and owns the adapter when it starts, so the `load_state` and
322    /// `save_state` settings on [`LiveNodeConfig`] take effect.
323    #[must_use]
324    pub fn with_cache_database_factory(mut self, factory: Box<dyn CacheDatabaseFactory>) -> Self {
325        self.cache_database_factory = Some(factory);
326        self
327    }
328
329    /// Set the message bus configuration.
330    ///
331    /// External streams are consumed when an ingress implementation is injected with
332    /// [`Self::with_external_ingress`] or built from [`Self::with_external_msgbus_factory`].
333    #[must_use]
334    pub fn with_msgbus_config(mut self, config: MessageBusConfig) -> Self {
335        self.config.msgbus = Some(config);
336        self
337    }
338
339    /// Set the portfolio configuration.
340    #[must_use]
341    pub fn with_portfolio_config(mut self, config: PortfolioConfig) -> Self {
342        self.config.portfolio = Some(config);
343        self
344    }
345
346    /// Set the streaming configuration.
347    ///
348    /// The node streams data and events to local Feather files and, with a catalog configured,
349    /// promotes them into it. Stop flushes the writer and dispose closes it.
350    #[cfg(feature = "streaming")]
351    #[must_use]
352    pub fn with_streaming_config(mut self, config: StreamingConfig) -> Self {
353        self.config.streaming = Some(config);
354        self
355    }
356
357    /// Set the data engine configuration.
358    ///
359    /// The Rust live runtime currently supports only the default `qsize`.
360    /// `build()` returns an error for other values.
361    #[must_use]
362    pub fn with_data_engine_config(mut self, config: LiveDataEngineConfig) -> Self {
363        self.config.data_engine = config;
364        self
365    }
366
367    /// Set the risk engine configuration.
368    ///
369    /// The Rust live runtime currently supports only the default `qsize`.
370    /// `build()` returns an error for other values.
371    #[must_use]
372    pub fn with_risk_engine_config(mut self, config: LiveRiskEngineConfig) -> Self {
373        self.config.risk_engine = config;
374        self
375    }
376
377    /// Set the execution engine configuration.
378    ///
379    /// The Rust live runtime currently supports only the default `qsize`.
380    /// `build()` returns an error for other values.
381    #[must_use]
382    pub fn with_exec_engine_config(mut self, config: LiveExecutionEngineConfig) -> Self {
383        self.config.exec_engine = config;
384        self
385    }
386
387    /// Inject an event-store implementation to drive run-lifecycle capture.
388    ///
389    /// The factory receives the kernel's instance id and clock so the returned
390    /// `KernelEventStore` shares the same time source the kernel uses to stamp run
391    /// lifecycle entries. The concrete implementation lives outside this crate;
392    /// callers typically build it from
393    /// [`LiveNodeConfig::event_store`](crate::config::LiveNodeConfig::event_store)
394    /// inside the closure.
395    #[must_use]
396    pub fn with_event_store<F>(mut self, factory: F) -> Self
397    where
398        F: FnOnce(UUID4, Rc<RefCell<dyn Clock>>) -> anyhow::Result<Box<dyn KernelEventStore>>
399            + 'static,
400    {
401        self.event_store_factory = Some(Box::new(factory));
402        self
403    }
404
405    /// Inject external message bus egress for serialized message bus publications.
406    #[must_use]
407    pub fn with_external_msgbus_egress(
408        mut self,
409        external_egress: Box<dyn MessageBusExternalEgress>,
410    ) -> Self {
411        self.external_msgbus_egress = Some(external_egress);
412        self
413    }
414
415    /// Build and inject external message bus egress and configured ingress from a factory.
416    #[must_use]
417    pub fn with_external_msgbus_factory(
418        mut self,
419        factory: Box<dyn MessageBusBackingFactory>,
420    ) -> Self {
421        self.external_msgbus_factory = Some(factory);
422        self
423    }
424
425    /// Inject external message bus ingress for serialized inbound publications.
426    #[must_use]
427    pub fn with_external_ingress(
428        mut self,
429        external_ingress: Box<dyn MessageBusExternalIngress>,
430    ) -> Self {
431        self.external_msgbus_ingress = Some(ExternalMessageBusIngress(external_ingress));
432        self
433    }
434
435    /// Set the logging configuration.
436    #[must_use]
437    pub fn with_logging(mut self, logging: LoggerConfig) -> Self {
438        self.config.logging = logging;
439        self
440    }
441
442    /// Adds a data client factory with configuration.
443    ///
444    /// # Errors
445    ///
446    /// Returns an error if a client with the same name is already registered.
447    pub fn add_data_client(
448        self,
449        name: Option<String>,
450        factory: Box<dyn DataClientFactory>,
451        config: Box<dyn ClientConfig>,
452    ) -> anyhow::Result<Self> {
453        self.add_data_client_with_routing(name, factory, config, RoutingConfig::default())
454    }
455
456    #[cfg(feature = "python")]
457    pub(crate) fn has_data_client(&self, name: &str) -> bool {
458        self.data_client_factories.contains_key(name)
459    }
460
461    #[cfg(feature = "python")]
462    pub(crate) fn has_exec_client(&self, name: &str) -> bool {
463        self.exec_client_factories.contains_key(name)
464    }
465
466    /// Adds a data client factory with configuration and explicit routing.
467    ///
468    /// # Errors
469    ///
470    /// Returns an error if a client with the same name is already registered.
471    pub fn add_data_client_with_routing(
472        mut self,
473        name: Option<String>,
474        factory: Box<dyn DataClientFactory>,
475        config: Box<dyn ClientConfig>,
476        routing: RoutingConfig,
477    ) -> anyhow::Result<Self> {
478        let name = name.unwrap_or_else(|| factory.name().to_string());
479
480        if self.data_client_factories.contains_key(&name) {
481            anyhow::bail!("Data client '{name}' is already registered");
482        }
483
484        self.data_client_factories.insert(name.clone(), factory);
485        self.data_client_configs.insert(name.clone(), config);
486        self.data_client_routing.insert(name, routing);
487        Ok(self)
488    }
489
490    /// Adds an execution client factory with configuration.
491    ///
492    /// Equivalent to [`Self::add_exec_client_with_routing`] with default (empty)
493    /// routing.
494    ///
495    /// # Errors
496    ///
497    /// Returns an error if a client with the same name is already registered.
498    pub fn add_exec_client(
499        self,
500        name: Option<String>,
501        factory: Box<dyn ExecutionClientFactory>,
502        config: Box<dyn ClientConfig>,
503    ) -> anyhow::Result<Self> {
504        self.add_exec_client_with_routing(name, factory, config, RoutingConfig::default())
505    }
506
507    /// Adds an execution client factory with configuration and explicit routing.
508    ///
509    /// Explicit venue routes take precedence over automatic routes. The only client for a
510    /// venue routes it automatically. Multiple clients for that venue require an explicit
511    /// venue route or a default client.
512    ///
513    /// # Errors
514    ///
515    /// Returns an error if a client with the same name is already registered.
516    pub fn add_exec_client_with_routing(
517        mut self,
518        name: Option<String>,
519        factory: Box<dyn ExecutionClientFactory>,
520        config: Box<dyn ClientConfig>,
521        routing: RoutingConfig,
522    ) -> anyhow::Result<Self> {
523        let name = name.unwrap_or_else(|| factory.name().to_string());
524
525        if self.exec_client_factories.contains_key(&name) {
526            anyhow::bail!("Execution client '{name}' is already registered");
527        }
528
529        self.exec_client_factories
530            .insert(name.clone(), ExecutionClientFactoryEntry::Adapter(factory));
531        self.exec_client_configs.insert(name.clone(), config);
532        self.exec_client_routing.insert(name, routing);
533        Ok(self)
534    }
535
536    /// Add a simulated execution client factory.
537    ///
538    /// This path is for sync-core clients such as the sandbox matching engine, which owns cache
539    /// mutation. Live venue adapters should use [`Self::add_exec_client`].
540    ///
541    /// # Errors
542    ///
543    /// Returns an error if a client with the same name is already registered.
544    pub fn add_simulated_exec_client(
545        mut self,
546        name: Option<String>,
547        factory: Box<dyn SimulatedExecutionClientFactory>,
548        config: Box<dyn ClientConfig>,
549    ) -> anyhow::Result<Self> {
550        let name = name.unwrap_or_else(|| factory.name().to_string());
551
552        if self.exec_client_factories.contains_key(&name) {
553            anyhow::bail!("Execution client '{name}' is already registered");
554        }
555
556        self.exec_client_factories.insert(
557            name.clone(),
558            ExecutionClientFactoryEntry::Simulated(factory),
559        );
560        self.exec_client_configs.insert(name, config);
561        Ok(self)
562    }
563
564    /// Build the [`LiveNode`] with the configured settings.
565    ///
566    /// This will:
567    /// 1. Build the underlying kernel.
568    /// 2. Create clients using factories.
569    /// 3. Register clients with engines.
570    ///
571    /// # Errors
572    ///
573    /// Returns an error if node construction fails, including conflicting execution routes
574    /// or multiple execution clients for a venue without an explicit route or default client.
575    /// Also returns an error if another live node exists or is being built on this thread.
576    pub fn build(mut self) -> anyhow::Result<LiveNode> {
577        self.build_in_place()
578    }
579
580    pub(crate) fn build_in_place(&mut self) -> anyhow::Result<LiveNode> {
581        let thread_owner = LiveNode::acquire_thread()?;
582
583        log::info!(
584            "Building LiveNode with {} data clients and {} execution clients",
585            self.data_client_factories.len(),
586            self.exec_client_factories.len()
587        );
588
589        self.config.validate_runtime_support()?;
590
591        if self.config.event_store.is_some() && self.event_store_factory.is_none() {
592            anyhow::bail!(
593                "LiveNodeConfig.event_store is set but no factory was registered; \
594                 call LiveNodeBuilder::with_event_store(...) to install one"
595            );
596        }
597
598        if self.external_msgbus_factory.is_some()
599            && (self.external_msgbus_egress.is_some() || self.external_msgbus_ingress.is_some())
600        {
601            anyhow::bail!(
602                "external message bus factory cannot be combined with injected egress or ingress"
603            );
604        }
605
606        let handle = LiveNodeHandle::new();
607        let runner = AsyncRunner::new();
608        runner.bind_senders_for_node(handle.clone());
609
610        let socket_registry = SocketReconnectRegistry::default();
611
612        let kernel = NautilusKernel::new_with_dependencies(
613            self.name.clone(),
614            self.config.clone(),
615            NautilusKernelDependencies::default()
616                .with_clock_factory(self.clock_factory.clone())
617                .with_event_store_factory(self.event_store_factory.take()),
618        )?;
619
620        #[cfg(feature = "python")]
621        if let Some(controller) = self.config.controller.as_ref() {
622            Trader::add_controller_from_importable_config(&kernel.trader, controller)?;
623        }
624
625        #[cfg(not(feature = "python"))]
626        if let Some(controller) = self.config.controller.as_ref() {
627            anyhow::bail!(
628                "LiveNodeConfig.controller for importable controller '{}' requires the python feature",
629                controller.controller_path
630            );
631        }
632
633        let (external_egress, external_ingress) = self.create_external_msgbus(&kernel)?;
634
635        if let Some(external_egress) = external_egress {
636            let config = self.config.msgbus.clone().unwrap_or_default();
637            nautilus_common::msgbus::get_message_bus()
638                .borrow_mut()
639                .set_external_egress_config(external_egress, &config)?;
640        }
641
642        for (name, factory) in &self.data_client_factories {
643            if let Some(config) = self.data_client_configs.get(name) {
644                log::debug!("Creating data client {name}");
645
646                let client = socket_registry.scope(|| {
647                    factory.create(name, config.as_ref(), kernel.cache().into(), kernel.clock())
648                })?;
649
650                let client_id = client.client_id();
651                let venue = client.venue();
652                socket_registry.register_client(client_id);
653
654                let adapter = DataClientAdapter::new(
655                    client_id, venue, true, // handles_order_book_deltas
656                    true, // handles_order_book_snapshots
657                    client,
658                );
659
660                let routing = self
661                    .data_client_routing
662                    .get(name)
663                    .cloned()
664                    .unwrap_or_default();
665
666                {
667                    let mut data_engine = kernel.data_engine.borrow_mut();
668                    data_engine.register_client(adapter, venue);
669
670                    if routing.default {
671                        data_engine.set_default_client(client_id)?;
672                    }
673
674                    if let Some(venues) = &routing.venues {
675                        for venue_str in venues {
676                            data_engine.register_venue_routing(
677                                client_id,
678                                Venue::new(venue_str.as_str()),
679                            )?;
680                        }
681                    }
682                }
683
684                log::info!("Registered DataClient-{client_id}");
685            } else {
686                log::warn!("No config found for data client factory {name}");
687            }
688        }
689
690        let mut exec_clients = Vec::new();
691        let mut venue_candidates = AHashMap::<Venue, Vec<_>>::new();
692        let mut venues_explicit = AHashSet::new();
693        let mut has_default_client = false;
694        let mut instrument_venues = AHashSet::new();
695        let mut self_settling_venues = AHashSet::new();
696
697        for (name, factory) in &self.exec_client_factories {
698            if let Some(config) = self.exec_client_configs.get(name) {
699                log::debug!("Creating execution client {name}");
700
701                let client = socket_registry.scope(|| match factory {
702                    ExecutionClientFactoryEntry::Adapter(factory) => factory.create(
703                        self.config.trader_id,
704                        name,
705                        config.as_ref(),
706                        kernel.cache().into(),
707                        kernel.clock(),
708                    ),
709                    ExecutionClientFactoryEntry::Simulated(factory) => {
710                        factory.create(self.config.trader_id, name, config.as_ref(), kernel.cache())
711                    }
712                })?;
713
714                let client = LiveExecutionClient::new(client);
715                let client_id = client.client_id();
716                let venue = client.venue();
717                socket_registry.register_client(client_id);
718
719                let routing = self
720                    .exec_client_routing
721                    .get(name)
722                    .cloned()
723                    .unwrap_or_default();
724
725                {
726                    let mut exec_engine = kernel.exec_engine.borrow_mut();
727                    exec_engine.register_client(Box::new(client.clone()))?;
728
729                    if routing.default {
730                        exec_engine.set_default_client(client_id)?;
731                        has_default_client = true;
732                    }
733
734                    if let Some(venues) = &routing.venues {
735                        for venue_str in venues {
736                            let route_venue = Venue::new(venue_str.as_str());
737                            exec_engine.register_venue_routing(client_id, route_venue)?;
738                            venues_explicit.insert(route_venue);
739                            instrument_venues.insert(route_venue);
740
741                            if client.settles_contract_expirations() {
742                                self_settling_venues.insert(route_venue);
743                            }
744                        }
745                    }
746                }
747
748                venue_candidates.entry(venue).or_default().push(client_id);
749                instrument_venues.insert(venue);
750
751                if client.settles_contract_expirations() {
752                    self_settling_venues.insert(venue);
753                }
754
755                exec_clients.push(client);
756
757                log::info!("Registered ExecutionClient-{client_id}");
758            } else {
759                log::warn!("No config found for execution client factory {name}");
760            }
761        }
762
763        {
764            let mut exec_engine = kernel.exec_engine.borrow_mut();
765
766            for (venue, candidates) in venue_candidates {
767                if venues_explicit.contains(&venue) {
768                    continue;
769                }
770
771                if let [client_id] = candidates.as_slice() {
772                    exec_engine.register_venue_routing(*client_id, venue)?;
773                } else if !has_default_client {
774                    anyhow::bail!(
775                        "Multiple execution clients for venue {venue}: configure an explicit venue route or default client"
776                    );
777                }
778            }
779        }
780
781        for venue in instrument_venues {
782            ExecutionEngine::subscribe_venue_instruments(&kernel.exec_engine, venue);
783
784            if !self_settling_venues.contains(&venue) {
785                ExecutionEngine::subscribe_venue_instrument_closes(&kernel.exec_engine, venue);
786            }
787        }
788
789        let exec_manager_config = ExecutionManagerConfig::from(&self.config.exec_engine)
790            .with_trader_id(self.config.trader_id);
791
792        let mut exec_manager = ExecutionManager::new(
793            kernel.clock.clone(),
794            kernel.cache.clone(),
795            exec_manager_config,
796        )?;
797
798        for client in &exec_clients {
799            exec_manager.set_position_reconciliation_tolerance(
800                client.account_id(),
801                client.position_reconciliation_tolerance(),
802            );
803        }
804
805        let mut node = LiveNode::new_from_builder(
806            kernel,
807            runner,
808            self.config.clone(),
809            handle,
810            exec_manager,
811            exec_clients,
812            socket_registry,
813            None,
814            external_ingress,
815            thread_owner,
816        );
817        node.load_configured_plugins()?;
818        node.cache_database_factory = self.cache_database_factory.take();
819
820        log::info!("Built successfully");
821
822        Ok(node)
823    }
824
825    #[expect(
826        clippy::type_complexity,
827        reason = "external backing returns its paired egress and ingress"
828    )]
829    fn create_external_msgbus(
830        &mut self,
831        kernel: &NautilusKernel,
832    ) -> anyhow::Result<(
833        Option<Box<dyn MessageBusExternalEgress>>,
834        Option<ExternalMessageBusIngress>,
835    )> {
836        let Some(factory) = self.external_msgbus_factory.as_ref() else {
837            return Ok((
838                self.external_msgbus_egress.take(),
839                self.external_msgbus_ingress.take(),
840            ));
841        };
842
843        let config = self.config.msgbus.clone().unwrap_or_default();
844        let has_external_streams = config
845            .external_streams
846            .as_ref()
847            .is_some_and(|streams| !streams.is_empty());
848        config.validate()?;
849        let backing = factory.create(self.config.trader_id, kernel.instance_id, config)?;
850
851        if has_external_streams {
852            let (external_egress, external_ingress) = external_io_from_backing(backing);
853            Ok((
854                Some(external_egress),
855                Some(ExternalMessageBusIngress(external_ingress)),
856            ))
857        } else {
858            Ok((Some(external_egress_from_backing(backing)), None))
859        }
860    }
861}
862
863impl ExternalMessageBusIngress {
864    pub(crate) fn is_closed(&self) -> bool {
865        self.0.is_closed()
866    }
867
868    pub(crate) fn take_receiver(
869        &mut self,
870    ) -> anyhow::Result<tokio::sync::mpsc::Receiver<BusMessage>> {
871        self.0.take_receiver()
872    }
873
874    pub(crate) fn close(&mut self) {
875        self.0.close();
876    }
877}
878
879#[cfg(test)]
880mod tests {
881    use std::{cell::RefCell, collections::HashMap, rc::Rc};
882
883    use nautilus_common::{
884        cache::CacheView,
885        clients::ExecutionClient,
886        clock::Clock,
887        enums::Environment,
888        factories::{ClientConfig, ExecutionClientFactory},
889        messages::execution::{SubmitOrder, TradingCommand},
890        msgbus::{
891            self, MessagingSwitchboard, stubs::get_typed_into_message_saving_handler, switchboard,
892        },
893    };
894    use nautilus_core::{UUID4, UnixNanos};
895    use nautilus_execution::engine::stubs::StubExecutionClient;
896    use nautilus_model::{
897        accounts::{AccountAny, CashAccount},
898        data::InstrumentClose,
899        enums::{AccountType, InstrumentCloseType, OmsType, OrderSide, OrderType},
900        events::{AccountState, OrderDeniedReason, OrderEventAny},
901        identifiers::{AccountId, ClientId, ClientOrderId, TraderId, Venue},
902        instruments::{
903            Instrument, InstrumentAny,
904            stubs::{audusd_sim, binary_option},
905        },
906        orders::{Order, OrderTestBuilder},
907        stubs::TestDefault,
908        types::{AccountBalance, Money, Price, Quantity},
909    };
910    use nautilus_trading::ImportableControllerConfig;
911    use rstest::rstest;
912    use ustr::Ustr;
913
914    use super::LiveNodeBuilder;
915    use crate::node::config::RoutingConfig;
916
917    #[rstest]
918    fn test_with_controller_sets_config_controller() {
919        let controller = ImportableControllerConfig {
920            controller_path: "module:Controller".to_string(),
921            config_path: "module:ControllerConfig".to_string(),
922            config: HashMap::new(),
923        };
924
925        let builder = LiveNodeBuilder::new(TraderId::from("TRADER-001"), Environment::Live)
926            .unwrap()
927            .with_controller(controller);
928
929        assert!(builder.config.controller.is_some());
930    }
931
932    #[rstest]
933    #[case::single(1, false, false, false)]
934    #[case::explicit(2, true, false, false)]
935    #[case::default(2, false, true, false)]
936    #[case::explicit_over_default(2, true, true, false)]
937    #[case::single_empty_venues(1, false, false, true)]
938    #[case::default_empty_venues(2, false, true, true)]
939    fn test_execution_client_routing_and_instruments(
940        #[case] count: usize,
941        #[case] explicit: bool,
942        #[case] default: bool,
943        #[case] empty_venues: bool,
944        #[values(false, true)] reverse: bool,
945    ) {
946        let clients: Vec<_> = (0..count)
947            .map(|i| {
948                StubExecutionClient::new(
949                    ClientId::new(format!("CLIENT-{i}")),
950                    AccountId::new(format!("ACCOUNT-{i}")),
951                    Venue::from("SIM"),
952                    OmsType::Netting,
953                    None,
954                )
955            })
956            .collect();
957
958        let mut builder =
959            LiveNodeBuilder::new(TraderId::test_default(), Environment::Live).unwrap();
960        let mut indices: Vec<_> = (0..count).collect();
961
962        if reverse {
963            indices.reverse();
964        }
965
966        for i in indices {
967            builder = builder
968                .add_exec_client_with_routing(
969                    Some(format!("client-{i}")),
970                    Box::new(RoutingClientFactory(clients[i].clone())),
971                    Box::new(RoutingClientConfig),
972                    RoutingConfig {
973                        default: default && i == 0,
974                        venues: if empty_venues {
975                            Some(vec![])
976                        } else {
977                            (explicit && i == 1).then(|| vec!["SIM".to_string()])
978                        },
979                    },
980                )
981                .unwrap();
982        }
983
984        let node = builder.build().unwrap();
985        let instrument = InstrumentAny::CurrencyPair(audusd_sim());
986        msgbus::publish_instrument(
987            switchboard::get_instrument_topic(instrument.id()),
988            &instrument,
989        );
990        let order = OrderTestBuilder::new(OrderType::Market)
991            .instrument_id(instrument.id())
992            .quantity(Quantity::from(1))
993            .build();
994        node.kernel()
995            .cache
996            .borrow_mut()
997            .add_instrument(instrument.clone())
998            .unwrap();
999        node.kernel()
1000            .cache
1001            .borrow_mut()
1002            .add_order(order.clone(), None, None, false)
1003            .unwrap();
1004        let engine = node.kernel().exec_engine.borrow();
1005        engine.execute(TradingCommand::SubmitOrder(SubmitOrder::from_order(
1006            &order,
1007            TraderId::test_default(),
1008            None,
1009            None,
1010            UUID4::new(),
1011            UnixNanos::default(),
1012        )));
1013        let routed = engine.get_clients_for_orders(std::slice::from_ref(&order));
1014        let explicit_index = count - 1 - usize::from(explicit);
1015        let explicit_order = OrderTestBuilder::new(OrderType::Market)
1016            .instrument_id(instrument.id())
1017            .client_order_id(ClientOrderId::from("O-EXPLICIT"))
1018            .quantity(Quantity::from(2))
1019            .build();
1020        node.kernel()
1021            .cache
1022            .borrow_mut()
1023            .add_order(explicit_order.clone(), None, None, false)
1024            .unwrap();
1025        engine.execute(TradingCommand::SubmitOrder(SubmitOrder::from_order(
1026            &explicit_order,
1027            TraderId::test_default(),
1028            Some(clients[explicit_index].client_id()),
1029            None,
1030            UUID4::new(),
1031            UnixNanos::default(),
1032        )));
1033
1034        assert_eq!(engine.client_ids().len(), count);
1035        assert_eq!(routed.len(), 1);
1036        assert_eq!(
1037            routed[0].client_id(),
1038            clients[usize::from(explicit)].client_id()
1039        );
1040
1041        for (i, client) in clients.iter().enumerate() {
1042            let mut expected_orders = if i == usize::from(explicit) {
1043                vec![order.client_order_id()]
1044            } else {
1045                vec![]
1046            };
1047
1048            if i == explicit_index {
1049                expected_orders.push(explicit_order.client_order_id());
1050            }
1051
1052            assert_eq!(*client.submitted_order_ids().borrow(), expected_orders);
1053            assert_eq!(
1054                *client.received_instruments().borrow(),
1055                vec![instrument.clone()]
1056            );
1057        }
1058    }
1059
1060    #[rstest]
1061    fn test_execution_client_routing_selects_risk_check_account() {
1062        let clients: Vec<_> = (0..2)
1063            .map(|i| {
1064                StubExecutionClient::new(
1065                    ClientId::new(format!("CLIENT-{i}")),
1066                    AccountId::new(format!("ACCOUNT-{i}")),
1067                    Venue::from("SIM"),
1068                    OmsType::Netting,
1069                    None,
1070                )
1071            })
1072            .collect();
1073
1074        let mut builder =
1075            LiveNodeBuilder::new(TraderId::test_default(), Environment::Live).unwrap();
1076
1077        for (i, client) in clients.iter().enumerate() {
1078            builder = builder
1079                .add_exec_client_with_routing(
1080                    Some(format!("client-{i}")),
1081                    Box::new(RoutingClientFactory(client.clone())),
1082                    Box::new(RoutingClientConfig),
1083                    RoutingConfig {
1084                        default: i == 0,
1085                        venues: None,
1086                    },
1087                )
1088                .unwrap();
1089        }
1090
1091        let node = builder.build().unwrap();
1092        let instrument = InstrumentAny::CurrencyPair(audusd_sim());
1093        let order = OrderTestBuilder::new(OrderType::Limit)
1094            .instrument_id(instrument.id())
1095            .side(OrderSide::Buy)
1096            .quantity(Quantity::from(100))
1097            .price(Price::from("1.00000"))
1098            .build();
1099        let explicit_order = OrderTestBuilder::new(OrderType::Limit)
1100            .instrument_id(instrument.id())
1101            .client_order_id(ClientOrderId::from("O-EXPLICIT"))
1102            .side(OrderSide::Buy)
1103            .quantity(Quantity::from(200))
1104            .price(Price::from("1.00000"))
1105            .build();
1106        {
1107            let mut cache = node.kernel().cache.borrow_mut();
1108            cache.add_instrument(instrument).unwrap();
1109            cache
1110                .add_account(cash_account(clients[0].account_id(), "1000000 USD"))
1111                .unwrap();
1112            cache
1113                .add_account(cash_account(clients[1].account_id(), "150 USD"))
1114                .unwrap();
1115            cache.add_order(order.clone(), None, None, false).unwrap();
1116            cache
1117                .add_order(explicit_order.clone(), None, None, false)
1118                .unwrap();
1119        }
1120
1121        let (command_handler, commands) =
1122            get_typed_into_message_saving_handler::<TradingCommand>(None);
1123        msgbus::register_trading_command_endpoint(
1124            MessagingSwitchboard::exec_engine_queue_execute(),
1125            command_handler,
1126        );
1127        let (event_handler, events) = get_typed_into_message_saving_handler::<OrderEventAny>(None);
1128        msgbus::register_order_event_endpoint(
1129            MessagingSwitchboard::exec_engine_process(),
1130            event_handler,
1131        );
1132
1133        let mut risk_engine = node.kernel().risk_engine.borrow_mut();
1134        risk_engine.execute(TradingCommand::SubmitOrder(SubmitOrder::from_order(
1135            &order,
1136            TraderId::test_default(),
1137            None,
1138            None,
1139            UUID4::new(),
1140            UnixNanos::default(),
1141        )));
1142        risk_engine.execute(TradingCommand::SubmitOrder(SubmitOrder::from_order(
1143            &explicit_order,
1144            TraderId::test_default(),
1145            Some(clients[1].client_id()),
1146            None,
1147            UUID4::new(),
1148            UnixNanos::default(),
1149        )));
1150
1151        let forwarded: Vec<_> = commands
1152            .get_messages()
1153            .iter()
1154            .map(|command| match command {
1155                TradingCommand::SubmitOrder(cmd) => (cmd.client_order_id, cmd.client_id),
1156                other => panic!("Unexpected command {other:?}"),
1157            })
1158            .collect();
1159
1160        let events = events.get_messages();
1161        assert_eq!(forwarded, vec![(order.client_order_id(), None)]);
1162        assert_eq!(events.len(), 1);
1163
1164        let OrderEventAny::Denied(denied) = &events[0] else {
1165            panic!("Expected OrderDenied")
1166        };
1167
1168        assert_eq!(denied.client_order_id, explicit_order.client_order_id());
1169        assert_eq!(
1170            denied.reason,
1171            Ustr::from(
1172                &OrderDeniedReason::NotionalExceedsFreeBalance {
1173                    free_balance: Money::from("150 USD"),
1174                    notional: Money::from("200 USD"),
1175                }
1176                .to_string()
1177            )
1178        );
1179    }
1180
1181    #[rstest]
1182    fn test_execution_client_keeps_native_route_with_extra_venue(
1183        #[values(false, true)] other_native_client: bool,
1184    ) {
1185        let client = StubExecutionClient::new(
1186            ClientId::from("CLIENT"),
1187            AccountId::from("CLIENT-001"),
1188            Venue::from("SIM"),
1189            OmsType::Netting,
1190            None,
1191        )
1192        .with_handles_all_order_venues();
1193
1194        let other_client = StubExecutionClient::new(
1195            ClientId::from("OTHER_CLIENT"),
1196            AccountId::from("OTHER_CLIENT-002"),
1197            Venue::from("OTHER"),
1198            OmsType::Netting,
1199            None,
1200        );
1201
1202        let mut builder = LiveNodeBuilder::new(TraderId::test_default(), Environment::Live)
1203            .unwrap()
1204            .add_exec_client_with_routing(
1205                Some("client".to_string()),
1206                Box::new(RoutingClientFactory(client.clone())),
1207                Box::new(RoutingClientConfig),
1208                RoutingConfig {
1209                    default: false,
1210                    venues: Some(vec!["OTHER".to_string(), "OTHER".to_string()]),
1211                },
1212            )
1213            .unwrap();
1214
1215        if other_native_client {
1216            builder = builder
1217                .add_exec_client_with_routing(
1218                    Some("other-client".to_string()),
1219                    Box::new(RoutingClientFactory(other_client.clone())),
1220                    Box::new(RoutingClientConfig),
1221                    RoutingConfig::default(),
1222                )
1223                .unwrap();
1224        }
1225
1226        let node = builder.build().unwrap();
1227
1228        let native = audusd_sim();
1229        let mut other = native.clone();
1230        other.id.venue = Venue::from("OTHER");
1231        let instruments = [
1232            InstrumentAny::CurrencyPair(native),
1233            InstrumentAny::CurrencyPair(other),
1234        ];
1235
1236        for instrument in &instruments {
1237            let order = OrderTestBuilder::new(OrderType::Market)
1238                .instrument_id(instrument.id())
1239                .client_order_id(ClientOrderId::new(format!("O-{}", instrument.id().venue)))
1240                .quantity(Quantity::from(1))
1241                .build();
1242            let engine = node.kernel().exec_engine.borrow();
1243            node.kernel()
1244                .cache
1245                .borrow_mut()
1246                .add_instrument(instrument.clone())
1247                .unwrap();
1248            node.kernel()
1249                .cache
1250                .borrow_mut()
1251                .add_order(order.clone(), None, None, false)
1252                .unwrap();
1253            engine.execute(TradingCommand::SubmitOrder(SubmitOrder::from_order(
1254                &order,
1255                TraderId::test_default(),
1256                None,
1257                None,
1258                UUID4::new(),
1259                UnixNanos::default(),
1260            )));
1261            let routed = engine.get_clients_for_orders(&[order]);
1262            assert_eq!(
1263                routed
1264                    .iter()
1265                    .map(|client| client.client_id())
1266                    .collect::<Vec<_>>(),
1267                vec![client.client_id()]
1268            );
1269            drop(engine);
1270            msgbus::publish_instrument(
1271                switchboard::get_instrument_topic(instrument.id()),
1272                instrument,
1273            );
1274        }
1275
1276        assert_eq!(*client.received_instruments().borrow(), instruments);
1277        assert_eq!(
1278            *client.submitted_order_ids().borrow(),
1279            vec![ClientOrderId::from("O-SIM"), ClientOrderId::from("O-OTHER")]
1280        );
1281        assert_eq!(*other_client.submitted_order_ids().borrow(), vec![]);
1282
1283        let expected_other = if other_native_client {
1284            vec![instruments[1].clone()]
1285        } else {
1286            vec![]
1287        };
1288
1289        assert_eq!(
1290            *other_client.received_instruments().borrow(),
1291            expected_other
1292        );
1293    }
1294
1295    #[rstest]
1296    #[case::ambiguous(
1297        false,
1298        false,
1299        false,
1300        "Multiple execution clients for venue SIM: configure an explicit venue route or default client"
1301    )]
1302    #[case::duplicate_id(true, false, false, "Client already registered with ID CLIENT-0")]
1303    #[case::duplicate_default(false, true, false, "default client already registered")]
1304    #[case::duplicate_route(false, false, true, "cannot re-route")]
1305    fn test_execution_client_routing_rejects_ambiguity(
1306        #[case] duplicate_id: bool,
1307        #[case] default: bool,
1308        #[case] explicit: bool,
1309        #[case] expected: &str,
1310    ) {
1311        let mut builder =
1312            LiveNodeBuilder::new(TraderId::test_default(), Environment::Live).unwrap();
1313
1314        for i in 0..2 {
1315            let id = if duplicate_id { 0 } else { i };
1316
1317            let client = StubExecutionClient::new(
1318                ClientId::new(format!("CLIENT-{id}")),
1319                AccountId::new(format!("ACCOUNT-{i}")),
1320                Venue::from("SIM"),
1321                OmsType::Netting,
1322                None,
1323            );
1324            builder = builder
1325                .add_exec_client_with_routing(
1326                    Some(format!("client-{i}")),
1327                    Box::new(RoutingClientFactory(client)),
1328                    Box::new(RoutingClientConfig),
1329                    RoutingConfig {
1330                        default,
1331                        venues: explicit.then(|| vec!["SIM".to_string()]),
1332                    },
1333                )
1334                .unwrap();
1335        }
1336
1337        let error = builder.build().unwrap_err().to_string();
1338        assert!(error.contains(expected), "Unexpected error: {error}");
1339    }
1340
1341    fn cash_account(account_id: AccountId, free: &str) -> AccountAny {
1342        let state = AccountState::new(
1343            account_id,
1344            AccountType::Cash,
1345            vec![AccountBalance::new(
1346                Money::from(free),
1347                Money::from("0 USD"),
1348                Money::from(free),
1349            )],
1350            vec![],
1351            true,
1352            UUID4::new(),
1353            UnixNanos::default(),
1354            UnixNanos::default(),
1355            None,
1356        );
1357        AccountAny::Cash(CashAccount::new(state, false, false))
1358    }
1359
1360    #[rstest]
1361    #[case::venue(false, true)]
1362    #[case::self_settling_venue(true, false)]
1363    fn test_execution_engine_settles_instrument_closes_unless_venue_settles_itself(
1364        #[case] self_settling: bool,
1365        #[case] expected_settled: bool,
1366    ) {
1367        let instrument = InstrumentAny::BinaryOption(binary_option());
1368
1369        let mut client = StubExecutionClient::new(
1370            ClientId::from("CLIENT"),
1371            AccountId::from("CLIENT-001"),
1372            instrument.id().venue,
1373            OmsType::Netting,
1374            None,
1375        );
1376
1377        if self_settling {
1378            client = client.with_settles_contract_expirations();
1379        }
1380
1381        let node = LiveNodeBuilder::new(TraderId::test_default(), Environment::Live)
1382            .unwrap()
1383            .add_exec_client_with_routing(
1384                Some("client".to_string()),
1385                Box::new(RoutingClientFactory(client)),
1386                Box::new(RoutingClientConfig),
1387                RoutingConfig::default(),
1388            )
1389            .unwrap()
1390            .build()
1391            .unwrap();
1392        node.kernel()
1393            .cache
1394            .borrow_mut()
1395            .add_instrument(instrument.clone())
1396            .unwrap();
1397
1398        let close = InstrumentClose::new(
1399            instrument.id(),
1400            Price::from("1.000"),
1401            InstrumentCloseType::ContractExpired,
1402            UnixNanos::from(300),
1403            UnixNanos::from(301),
1404        );
1405
1406        msgbus::publish_any(
1407            switchboard::get_instrument_close_topic(instrument.id()),
1408            &close,
1409        );
1410
1411        assert_eq!(
1412            node.kernel()
1413                .cache
1414                .borrow()
1415                .instrument_close(&instrument.id()),
1416            expected_settled.then_some(&close)
1417        );
1418    }
1419
1420    #[derive(Debug)]
1421    struct RoutingClientFactory(StubExecutionClient);
1422
1423    impl ExecutionClientFactory for RoutingClientFactory {
1424        fn create(
1425            &self,
1426            _trader_id: TraderId,
1427            _name: &str,
1428            _config: &dyn ClientConfig,
1429            _cache: CacheView,
1430            _clock: Rc<RefCell<dyn Clock>>,
1431        ) -> anyhow::Result<Box<dyn ExecutionClient>> {
1432            Ok(Box::new(self.0.clone()))
1433        }
1434
1435        fn name(&self) -> &'static str {
1436            "routing"
1437        }
1438
1439        fn config_type(&self) -> &'static str {
1440            "RoutingClientConfig"
1441        }
1442    }
1443
1444    #[derive(Debug)]
1445    struct RoutingClientConfig;
1446
1447    impl ClientConfig for RoutingClientConfig {
1448        fn as_any(&self) -> &dyn std::any::Any {
1449            self
1450        }
1451    }
1452}
nautilus_live/node/builder.rs

nautilus_live/node/
builder.rs
