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 nautilus_common::{
21    cache::{CacheConfig, database::CacheDatabaseFactory},
22    clients::ExecutionClient,
23    clock::Clock,
24    enums::Environment,
25    factories::{
26        ClientConfig, DataClientFactory, ExecutionClientFactory, SimulatedExecutionClientFactory,
27    },
28    logging::logger::LoggerConfig,
29    msgbus::{
30        BusMessage, MessageBusBackingFactory, MessageBusConfig, MessageBusExternalEgress,
31        MessageBusExternalIngress, external_egress_from_backing, external_io_from_backing,
32    },
33};
34use nautilus_core::UUID4;
35use nautilus_data::client::DataClientAdapter;
36use nautilus_execution::engine::ExecutionEngine;
37use nautilus_model::identifiers::{TraderId, Venue};
38use nautilus_portfolio::config::PortfolioConfig;
39#[cfg(feature = "python")]
40use nautilus_system::trader::Trader;
41use nautilus_system::{
42    clock_factory::ClockFactory,
43    config::StreamingConfig,
44    event_store::{EventStoreFactory, KernelEventStore},
45    kernel::{NautilusKernel, NautilusKernelDependencies},
46};
47use nautilus_trading::ImportableControllerConfig;
48
49use super::{
50    LiveNode,
51    config::{
52        LiveDataEngineConfig, LiveExecutionEngineConfig, LiveNodeConfig, LiveRiskEngineConfig,
53        RoutingConfig, validate_live_environment,
54    },
55};
56use crate::{
57    execution::{
58        client::LiveExecutionClient,
59        manager::{ExecutionManager, ExecutionManagerConfig},
60    },
61    runner::AsyncRunner,
62    socket::SocketReconnectRegistry,
63};
64
65#[derive(Debug)]
66enum ExecutionClientFactoryEntry {
67    Adapter(Box<dyn ExecutionClientFactory>),
68    Simulated(Box<dyn SimulatedExecutionClientFactory>),
69}
70
71pub(crate) struct ExternalMessageBusIngress(Box<dyn MessageBusExternalIngress>);
72
73impl Debug for ExternalMessageBusIngress {
74    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
75        f.debug_struct(stringify!(ExternalMessageBusIngress))
76            .finish_non_exhaustive()
77    }
78}
79
80/// Builder for constructing a [`LiveNode`] with a fluent API.
81///
82/// Provides configuration options specific to live nodes, including client factory
83/// registration, timeout settings, and optional event-store injection for run-lifecycle
84/// audit and replay (see [`Self::with_event_store`]).
85#[cfg_attr(
86    feature = "python",
87    pyo3::pyclass(module = "nautilus_trader.live", unsendable)
88)]
89pub struct LiveNodeBuilder {
90    name: String,
91    config: LiveNodeConfig,
92    data_client_factories: HashMap<String, Box<dyn DataClientFactory>>,
93    exec_client_factories: HashMap<String, ExecutionClientFactoryEntry>,
94    data_client_configs: HashMap<String, Box<dyn ClientConfig>>,
95    exec_client_configs: HashMap<String, Box<dyn ClientConfig>>,
96    data_client_routing: HashMap<String, RoutingConfig>,
97    exec_client_routing: HashMap<String, RoutingConfig>,
98    event_store_factory: Option<EventStoreFactory>,
99    clock_factory: Option<ClockFactory>,
100    cache_database_factory: Option<Box<dyn CacheDatabaseFactory>>,
101    external_msgbus_factory: Option<Box<dyn MessageBusBackingFactory>>,
102    external_msgbus_egress: Option<Box<dyn MessageBusExternalEgress>>,
103    external_msgbus_ingress: Option<ExternalMessageBusIngress>,
104}
105
106impl Debug for LiveNodeBuilder {
107    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
108        f.debug_struct(stringify!(LiveNodeBuilder))
109            .field("name", &self.name)
110            .field("config", &self.config)
111            .field("data_client_factories", &self.data_client_factories.keys())
112            .field("exec_client_factories", &self.exec_client_factories.keys())
113            .field("data_client_configs", &self.data_client_configs.keys())
114            .field("exec_client_configs", &self.exec_client_configs.keys())
115            .field("event_store_factory", &self.event_store_factory.is_some())
116            .field("clock_factory", &self.clock_factory.is_some())
117            .field(
118                "cache_database_factory",
119                &self.cache_database_factory.is_some(),
120            )
121            .field(
122                "external_msgbus_factory",
123                &self.external_msgbus_factory.is_some(),
124            )
125            .field(
126                "external_msgbus_egress",
127                &self.external_msgbus_egress.is_some(),
128            )
129            .field(
130                "external_msgbus_ingress",
131                &self.external_msgbus_ingress.is_some(),
132            )
133            .finish_non_exhaustive()
134    }
135}
136
137impl LiveNodeBuilder {
138    /// Creates a new [`LiveNodeBuilder`] with required parameters.
139    ///
140    /// # Errors
141    ///
142    /// Returns an error if `environment` is invalid (BACKTEST).
143    pub fn new(trader_id: TraderId, environment: Environment) -> anyhow::Result<Self> {
144        validate_live_environment(environment)?;
145
146        let config = LiveNodeConfig {
147            environment,
148            trader_id,
149            ..Default::default()
150        };
151
152        Ok(Self {
153            name: "LiveNode".to_string(),
154            config,
155            data_client_factories: HashMap::new(),
156            exec_client_factories: HashMap::new(),
157            data_client_configs: HashMap::new(),
158            exec_client_configs: HashMap::new(),
159            data_client_routing: HashMap::new(),
160            exec_client_routing: HashMap::new(),
161            event_store_factory: None,
162            clock_factory: None,
163            cache_database_factory: None,
164            external_msgbus_factory: None,
165            external_msgbus_egress: None,
166            external_msgbus_ingress: None,
167        })
168    }
169
170    /// Creates a new [`LiveNodeBuilder`] from an existing [`LiveNodeConfig`].
171    ///
172    /// # Errors
173    ///
174    /// Returns an error if the config's environment is invalid (BACKTEST).
175    pub fn from_config(config: LiveNodeConfig) -> anyhow::Result<Self> {
176        validate_live_environment(config.environment)?;
177
178        Ok(Self {
179            name: "LiveNode".to_string(),
180            config,
181            data_client_factories: HashMap::new(),
182            exec_client_factories: HashMap::new(),
183            data_client_configs: HashMap::new(),
184            exec_client_configs: HashMap::new(),
185            data_client_routing: HashMap::new(),
186            exec_client_routing: HashMap::new(),
187            event_store_factory: None,
188            clock_factory: None,
189            cache_database_factory: None,
190            external_msgbus_factory: None,
191            external_msgbus_egress: None,
192            external_msgbus_ingress: None,
193        })
194    }
195
196    /// Returns the name for the node.
197    #[must_use]
198    pub fn name(&self) -> &str {
199        &self.name
200    }
201
202    /// Set the name for the node.
203    #[must_use]
204    pub fn with_name(mut self, name: impl Into<String>) -> Self {
205        self.name = name.into();
206        self
207    }
208
209    /// Set the instance ID for the node.
210    #[must_use]
211    pub const fn with_instance_id(mut self, instance_id: UUID4) -> Self {
212        self.config.instance_id = Some(instance_id);
213        self
214    }
215
216    /// Configure whether to load state on startup.
217    #[must_use]
218    pub const fn with_load_state(mut self, load_state: bool) -> Self {
219        self.config.load_state = load_state;
220        self
221    }
222
223    /// Configure whether to save state on shutdown.
224    #[must_use]
225    pub const fn with_save_state(mut self, save_state: bool) -> Self {
226        self.config.save_state = save_state;
227        self
228    }
229
230    /// Set the importable controller configuration for the node.
231    ///
232    /// The controller is instantiated and registered with the trader during
233    /// [`LiveNodeBuilder::build`], enabling runtime strategy/actor management
234    /// (create, start, stop, remove) without restarting the node. This mirrors
235    /// the `controller` field on [`LiveNodeConfig`] used by the config-based
236    /// [`LiveNode::build`] path, so a builder that also registers client
237    /// factories can host a controller in a single node.
238    #[must_use]
239    pub fn with_controller(mut self, controller: ImportableControllerConfig) -> Self {
240        self.config.controller = Some(controller);
241        self
242    }
243
244    /// Set the connection timeout in seconds.
245    #[must_use]
246    pub const fn with_timeout_connection(mut self, timeout_secs: u64) -> Self {
247        self.config.timeout_connection = Duration::from_secs(timeout_secs);
248        self
249    }
250
251    /// Set the reconciliation timeout in seconds.
252    #[must_use]
253    pub const fn with_timeout_reconciliation(mut self, timeout_secs: u64) -> Self {
254        self.config.timeout_reconciliation = Duration::from_secs(timeout_secs);
255        self
256    }
257
258    /// Configure whether to run startup reconciliation.
259    #[must_use]
260    pub fn with_reconciliation(mut self, reconciliation: bool) -> Self {
261        self.config.exec_engine.reconciliation = reconciliation;
262        self
263    }
264
265    /// Set the reconciliation lookback in minutes.
266    #[must_use]
267    pub fn with_reconciliation_lookback_mins(mut self, mins: u32) -> Self {
268        self.config.exec_engine.reconciliation_lookback_mins = Some(mins);
269        self
270    }
271
272    /// Set the portfolio initialization timeout in seconds.
273    #[must_use]
274    pub const fn with_timeout_portfolio(mut self, timeout_secs: u64) -> Self {
275        self.config.timeout_portfolio = Duration::from_secs(timeout_secs);
276        self
277    }
278
279    /// Set the disconnection timeout in seconds.
280    #[must_use]
281    pub const fn with_timeout_disconnection_secs(mut self, timeout_secs: u64) -> Self {
282        self.config.timeout_disconnection = Duration::from_secs(timeout_secs);
283        self
284    }
285
286    /// Set the post-stop delay in seconds.
287    #[must_use]
288    pub const fn with_delay_post_stop_secs(mut self, delay_secs: u64) -> Self {
289        self.config.delay_post_stop = Duration::from_secs(delay_secs);
290        self
291    }
292
293    /// Set the shutdown timeout in seconds.
294    #[must_use]
295    pub const fn with_delay_shutdown_secs(mut self, delay_secs: u64) -> Self {
296        self.config.timeout_shutdown = Duration::from_secs(delay_secs);
297        self
298    }
299
300    /// Inject a caller-supplied clock factory for the kernel and component clocks.
301    #[must_use]
302    pub fn with_clock_factory<F>(mut self, factory: F) -> Self
303    where
304        F: Fn() -> Rc<RefCell<dyn Clock>> + 'static,
305    {
306        self.clock_factory = Some(ClockFactory::new(factory));
307        self
308    }
309
310    /// Set the cache configuration.
311    #[must_use]
312    pub fn with_cache_config(mut self, config: CacheConfig) -> Self {
313        self.config.cache = Some(config);
314        self
315    }
316
317    /// Install the cache database backing from a factory.
318    ///
319    /// The node constructs and owns the adapter when it starts, so the `load_state` and
320    /// `save_state` settings on [`LiveNodeConfig`] take effect.
321    #[must_use]
322    pub fn with_cache_database_factory(mut self, factory: Box<dyn CacheDatabaseFactory>) -> Self {
323        self.cache_database_factory = Some(factory);
324        self
325    }
326
327    /// Set the message bus configuration.
328    ///
329    /// External streams are consumed when an ingress implementation is injected with
330    /// [`Self::with_external_ingress`] or built from [`Self::with_external_msgbus_factory`].
331    #[must_use]
332    pub fn with_msgbus_config(mut self, config: MessageBusConfig) -> Self {
333        self.config.msgbus = Some(config);
334        self
335    }
336
337    /// Set the portfolio configuration.
338    #[must_use]
339    pub fn with_portfolio_config(mut self, config: PortfolioConfig) -> Self {
340        self.config.portfolio = Some(config);
341        self
342    }
343
344    /// Set the streaming configuration.
345    ///
346    /// The Rust live runtime does not support this setting yet.
347    /// `build()` returns an error when it is set.
348    #[must_use]
349    pub fn with_streaming_config(mut self, config: StreamingConfig) -> Self {
350        self.config.streaming = Some(config);
351        self
352    }
353
354    /// Set the data engine configuration.
355    ///
356    /// The Rust live runtime currently supports only the default `qsize`.
357    /// `build()` returns an error for other values.
358    #[must_use]
359    pub fn with_data_engine_config(mut self, config: LiveDataEngineConfig) -> Self {
360        self.config.data_engine = config;
361        self
362    }
363
364    /// Set the risk engine configuration.
365    ///
366    /// The Rust live runtime currently supports only the default `qsize`.
367    /// `build()` returns an error for other values.
368    #[must_use]
369    pub fn with_risk_engine_config(mut self, config: LiveRiskEngineConfig) -> Self {
370        self.config.risk_engine = config;
371        self
372    }
373
374    /// Set the execution engine configuration.
375    ///
376    /// The Rust live runtime currently supports only the default `qsize`.
377    /// `build()` returns an error for other values.
378    #[must_use]
379    pub fn with_exec_engine_config(mut self, config: LiveExecutionEngineConfig) -> Self {
380        self.config.exec_engine = config;
381        self
382    }
383
384    /// Inject an event-store implementation to drive run-lifecycle capture.
385    ///
386    /// The factory receives the kernel's instance id and clock so the returned
387    /// `KernelEventStore` shares the same time source the kernel uses to stamp run
388    /// lifecycle entries. The concrete implementation lives outside this crate;
389    /// callers typically build it from
390    /// [`LiveNodeConfig::event_store`](crate::config::LiveNodeConfig::event_store)
391    /// inside the closure.
392    #[must_use]
393    pub fn with_event_store<F>(mut self, factory: F) -> Self
394    where
395        F: FnOnce(UUID4, Rc<RefCell<dyn Clock>>) -> anyhow::Result<Box<dyn KernelEventStore>>
396            + 'static,
397    {
398        self.event_store_factory = Some(Box::new(factory));
399        self
400    }
401
402    /// Inject external message bus egress for serialized message bus publications.
403    #[must_use]
404    pub fn with_external_msgbus_egress(
405        mut self,
406        external_egress: Box<dyn MessageBusExternalEgress>,
407    ) -> Self {
408        self.external_msgbus_egress = Some(external_egress);
409        self
410    }
411
412    /// Build and inject external message bus egress and configured ingress from a factory.
413    #[must_use]
414    pub fn with_external_msgbus_factory(
415        mut self,
416        factory: Box<dyn MessageBusBackingFactory>,
417    ) -> Self {
418        self.external_msgbus_factory = Some(factory);
419        self
420    }
421
422    /// Inject external message bus ingress for serialized inbound publications.
423    #[must_use]
424    pub fn with_external_ingress(
425        mut self,
426        external_ingress: Box<dyn MessageBusExternalIngress>,
427    ) -> Self {
428        self.external_msgbus_ingress = Some(ExternalMessageBusIngress(external_ingress));
429        self
430    }
431
432    /// Set the logging configuration.
433    #[must_use]
434    pub fn with_logging(mut self, logging: LoggerConfig) -> Self {
435        self.config.logging = logging;
436        self
437    }
438
439    /// Adds a data client factory with configuration.
440    ///
441    /// # Errors
442    ///
443    /// Returns an error if a client with the same name is already registered.
444    pub fn add_data_client(
445        self,
446        name: Option<String>,
447        factory: Box<dyn DataClientFactory>,
448        config: Box<dyn ClientConfig>,
449    ) -> anyhow::Result<Self> {
450        self.add_data_client_with_routing(name, factory, config, RoutingConfig::default())
451    }
452
453    /// Adds a data client factory with configuration and explicit routing.
454    ///
455    /// # Errors
456    ///
457    /// Returns an error if a client with the same name is already registered.
458    pub fn add_data_client_with_routing(
459        mut self,
460        name: Option<String>,
461        factory: Box<dyn DataClientFactory>,
462        config: Box<dyn ClientConfig>,
463        routing: RoutingConfig,
464    ) -> anyhow::Result<Self> {
465        let name = name.unwrap_or_else(|| factory.name().to_string());
466
467        if self.data_client_factories.contains_key(&name) {
468            anyhow::bail!("Data client '{name}' is already registered");
469        }
470
471        self.data_client_factories.insert(name.clone(), factory);
472        self.data_client_configs.insert(name.clone(), config);
473        self.data_client_routing.insert(name, routing);
474        Ok(self)
475    }
476
477    /// Adds an execution client factory with configuration.
478    ///
479    /// Equivalent to [`Self::add_exec_client_with_routing`] with default (empty)
480    /// routing.
481    ///
482    /// # Errors
483    ///
484    /// Returns an error if a client with the same name is already registered.
485    pub fn add_exec_client(
486        self,
487        name: Option<String>,
488        factory: Box<dyn ExecutionClientFactory>,
489        config: Box<dyn ClientConfig>,
490    ) -> anyhow::Result<Self> {
491        self.add_exec_client_with_routing(name, factory, config, RoutingConfig::default())
492    }
493
494    /// Adds an execution client factory with configuration and explicit routing.
495    ///
496    /// # Errors
497    ///
498    /// Returns an error if a client with the same name is already registered.
499    pub fn add_exec_client_with_routing(
500        mut self,
501        name: Option<String>,
502        factory: Box<dyn ExecutionClientFactory>,
503        config: Box<dyn ClientConfig>,
504        routing: RoutingConfig,
505    ) -> anyhow::Result<Self> {
506        let name = name.unwrap_or_else(|| factory.name().to_string());
507
508        if self.exec_client_factories.contains_key(&name) {
509            anyhow::bail!("Execution client '{name}' is already registered");
510        }
511
512        self.exec_client_factories
513            .insert(name.clone(), ExecutionClientFactoryEntry::Adapter(factory));
514        self.exec_client_configs.insert(name.clone(), config);
515        self.exec_client_routing.insert(name, routing);
516        Ok(self)
517    }
518
519    /// Add a simulated execution client factory.
520    ///
521    /// This path is for sync-core clients such as the sandbox matching engine, which owns cache
522    /// mutation. Live venue adapters should use [`Self::add_exec_client`].
523    ///
524    /// # Errors
525    ///
526    /// Returns an error if a client with the same name is already registered.
527    pub fn add_simulated_exec_client(
528        mut self,
529        name: Option<String>,
530        factory: Box<dyn SimulatedExecutionClientFactory>,
531        config: Box<dyn ClientConfig>,
532    ) -> anyhow::Result<Self> {
533        let name = name.unwrap_or_else(|| factory.name().to_string());
534
535        if self.exec_client_factories.contains_key(&name) {
536            anyhow::bail!("Execution client '{name}' is already registered");
537        }
538
539        self.exec_client_factories.insert(
540            name.clone(),
541            ExecutionClientFactoryEntry::Simulated(factory),
542        );
543        self.exec_client_configs.insert(name, config);
544        Ok(self)
545    }
546
547    /// Build the [`LiveNode`] with the configured settings.
548    ///
549    /// This will:
550    /// 1. Build the underlying kernel.
551    /// 2. Create clients using factories.
552    /// 3. Register clients with engines.
553    ///
554    /// # Errors
555    ///
556    /// Returns an error if node construction fails.
557    pub fn build(mut self) -> anyhow::Result<LiveNode> {
558        log::info!(
559            "Building LiveNode with {} data clients and {} execution clients",
560            self.data_client_factories.len(),
561            self.exec_client_factories.len()
562        );
563
564        self.config.validate_runtime_support()?;
565
566        if self.config.event_store.is_some() && self.event_store_factory.is_none() {
567            anyhow::bail!(
568                "LiveNodeConfig.event_store is set but no factory was registered; \
569                 call LiveNodeBuilder::with_event_store(...) to install one"
570            );
571        }
572
573        if self.external_msgbus_factory.is_some()
574            && (self.external_msgbus_egress.is_some() || self.external_msgbus_ingress.is_some())
575        {
576            anyhow::bail!(
577                "external message bus factory cannot be combined with injected egress or ingress"
578            );
579        }
580
581        let runner = AsyncRunner::new();
582        runner.bind_senders();
583
584        let socket_registry = SocketReconnectRegistry::default();
585
586        let kernel = NautilusKernel::new_with_dependencies(
587            self.name.clone(),
588            self.config.clone(),
589            NautilusKernelDependencies::default()
590                .with_clock_factory(self.clock_factory.take())
591                .with_event_store_factory(self.event_store_factory.take()),
592        )?;
593        #[cfg(feature = "python")]
594        if let Some(controller) = self.config.controller.as_ref() {
595            Trader::add_controller_from_importable_config(&kernel.trader, controller)?;
596        }
597        #[cfg(not(feature = "python"))]
598        if let Some(controller) = self.config.controller.as_ref() {
599            anyhow::bail!(
600                "LiveNodeConfig.controller for importable controller '{}' requires the python feature",
601                controller.controller_path
602            );
603        }
604
605        self.install_external_msgbus_factory(&kernel)?;
606
607        if let Some(external_egress) = self.external_msgbus_egress.take() {
608            let config = self.config.msgbus.clone().unwrap_or_default();
609            nautilus_common::msgbus::get_message_bus()
610                .borrow_mut()
611                .set_external_egress_config(external_egress, &config)?;
612        }
613
614        for (name, factory) in self.data_client_factories {
615            if let Some(config) = self.data_client_configs.remove(&name) {
616                log::debug!("Creating data client {name}");
617
618                let client = socket_registry.scope(|| {
619                    factory.create(
620                        &name,
621                        config.as_ref(),
622                        kernel.cache().into(),
623                        kernel.clock(),
624                    )
625                })?;
626                let client_id = client.client_id();
627                let venue = client.venue();
628                socket_registry.register_client(client_id);
629
630                let adapter = DataClientAdapter::new(
631                    client_id, venue, true, // handles_order_book_deltas
632                    true, // handles_order_book_snapshots
633                    client,
634                );
635
636                let routing = self.data_client_routing.remove(&name).unwrap_or_default();
637
638                {
639                    let mut data_engine = kernel.data_engine.borrow_mut();
640                    data_engine.register_client(adapter, venue);
641
642                    if routing.default {
643                        data_engine.set_default_client(client_id)?;
644                    }
645
646                    if let Some(venues) = &routing.venues {
647                        for venue_str in venues {
648                            data_engine.register_venue_routing(
649                                client_id,
650                                Venue::new(venue_str.as_str()),
651                            )?;
652                        }
653                    }
654                }
655
656                log::info!("Registered DataClient-{client_id}");
657            } else {
658                log::warn!("No config found for data client factory {name}");
659            }
660        }
661
662        let mut exec_clients = Vec::new();
663
664        for (name, factory) in self.exec_client_factories {
665            if let Some(config) = self.exec_client_configs.remove(&name) {
666                log::debug!("Creating execution client {name}");
667
668                let client = socket_registry.scope(|| match factory {
669                    ExecutionClientFactoryEntry::Adapter(factory) => factory.create(
670                        self.config.trader_id,
671                        &name,
672                        config.as_ref(),
673                        kernel.cache().into(),
674                    ),
675                    ExecutionClientFactoryEntry::Simulated(factory) => factory.create(
676                        self.config.trader_id,
677                        &name,
678                        config.as_ref(),
679                        kernel.cache(),
680                    ),
681                })?;
682                let client = LiveExecutionClient::new(client);
683                let client_id = client.client_id();
684                let venue = client.venue();
685                socket_registry.register_client(client_id);
686
687                let routing = self.exec_client_routing.remove(&name).unwrap_or_default();
688
689                {
690                    let mut exec_engine = kernel.exec_engine.borrow_mut();
691                    exec_engine.register_client(Box::new(client.clone()))?;
692
693                    if routing.default {
694                        exec_engine.set_default_client(client_id)?;
695                    }
696
697                    if let Some(venues) = &routing.venues {
698                        for venue_str in venues {
699                            exec_engine.register_venue_routing(
700                                client_id,
701                                Venue::new(venue_str.as_str()),
702                            )?;
703                        }
704                    }
705                }
706                ExecutionEngine::subscribe_venue_instruments(&kernel.exec_engine, venue);
707                exec_clients.push(client);
708
709                log::info!("Registered ExecutionClient-{client_id}");
710            } else {
711                log::warn!("No config found for execution client factory {name}");
712            }
713        }
714
715        let exec_manager_config = ExecutionManagerConfig::from(&self.config.exec_engine)
716            .with_trader_id(self.config.trader_id);
717        let mut exec_manager = ExecutionManager::new(
718            kernel.clock.clone(),
719            kernel.cache.clone(),
720            exec_manager_config,
721        )?;
722
723        for client in &exec_clients {
724            exec_manager.set_position_reconciliation_tolerance(
725                client.account_id(),
726                client.position_reconciliation_tolerance(),
727            );
728        }
729
730        let node = LiveNode::new_from_builder(
731            kernel,
732            runner,
733            self.config,
734            exec_manager,
735            exec_clients,
736            socket_registry,
737            self.cache_database_factory,
738            self.external_msgbus_ingress,
739        );
740        node.load_configured_plugins()?;
741
742        log::info!("Built successfully");
743
744        Ok(node)
745    }
746
747    fn install_external_msgbus_factory(&mut self, kernel: &NautilusKernel) -> anyhow::Result<()> {
748        let Some(factory) = self.external_msgbus_factory.take() else {
749            return Ok(());
750        };
751
752        let config = self.config.msgbus.clone().unwrap_or_default();
753        let has_external_streams = config
754            .external_streams
755            .as_ref()
756            .is_some_and(|streams| !streams.is_empty());
757        config.validate()?;
758        let backing = factory.create(self.config.trader_id, kernel.instance_id, config)?;
759
760        if has_external_streams {
761            let (external_egress, external_ingress) = external_io_from_backing(backing);
762            self.external_msgbus_egress = Some(external_egress);
763            self.external_msgbus_ingress = Some(ExternalMessageBusIngress(external_ingress));
764        } else {
765            self.external_msgbus_egress = Some(external_egress_from_backing(backing));
766        }
767
768        Ok(())
769    }
770}
771
772impl ExternalMessageBusIngress {
773    pub(crate) fn is_closed(&self) -> bool {
774        self.0.is_closed()
775    }
776
777    pub(crate) fn take_receiver(
778        &mut self,
779    ) -> anyhow::Result<tokio::sync::mpsc::Receiver<BusMessage>> {
780        self.0.take_receiver()
781    }
782
783    pub(crate) fn close(&mut self) {
784        self.0.close();
785    }
786}
787
788#[cfg(test)]
789mod tests {
790    use std::collections::HashMap;
791
792    use nautilus_common::enums::Environment;
793    use nautilus_model::identifiers::TraderId;
794    use nautilus_trading::ImportableControllerConfig;
795    use rstest::rstest;
796
797    use super::LiveNodeBuilder;
798
799    #[rstest]
800    fn test_with_controller_sets_config_controller() {
801        let controller = ImportableControllerConfig {
802            controller_path: "module:Controller".to_string(),
803            config_path: "module:ControllerConfig".to_string(),
804            config: HashMap::new(),
805        };
806
807        let builder = LiveNodeBuilder::new(TraderId::from("TRADER-001"), Environment::Live)
808            .unwrap()
809            .with_controller(controller);
810
811        assert!(builder.config.controller.is_some());
812    }
813}
nautilus_live/node/builder.rs

nautilus_live/node/
builder.rs
