Continue LiveNode in Rust

Author: cjdsellers

crates/live/src/node.rs

@@ -17,9 +17,9 @@
 #![allow(dead_code)]
 #![allow(unused_variables)]
 
-use std::collections::HashMap;
+use std::{cell::RefCell, collections::HashMap, rc::Rc};
 
-use nautilus_common::enums::Environment;
+use nautilus_common::{clock::LiveClock, component::Component, enums::Environment};
 use nautilus_core::UUID4;
 use nautilus_model::identifiers::TraderId;
 use nautilus_system::{
@@ -28,17 +28,15 @@ use nautilus_system::{
     kernel::NautilusKernel,
 };
 
-use crate::{
-    config::LiveNodeConfig,
-    runner::{AsyncRunner, Runner},
-};
+use crate::{config::LiveNodeConfig, runner::AsyncRunner};
 
 /// High-level abstraction for a live Nautilus system node.
 ///
 /// Provides a simplified interface for running live systems
 /// with automatic client management and lifecycle handling.
 #[derive(Debug)]
 pub struct LiveNode {
+    clock: Rc<RefCell<LiveClock>>,
     kernel: NautilusKernel,
     runner: AsyncRunner,
     config: LiveNodeConfig,
@@ -80,12 +78,14 @@ impl LiveNode {
             }
         }
 
+        let clock = Rc::new(RefCell::new(LiveClock::new()));
         let kernel = NautilusKernel::new(name, config.clone())?;
-        let runner = AsyncRunner::new();
+        let runner = AsyncRunner::new(clock.clone());
 
         log::info!("LiveNode built successfully with kernel config");
 
         Ok(Self {
+            clock,
             kernel,
             runner,
             config,
@@ -182,6 +182,28 @@ impl LiveNode {
     pub const fn is_running(&self) -> bool {
         self.is_running
     }
+
+    /// Adds an actor to the trader.
+    ///
+    /// This method provides a high-level interface for adding actors to the underlying
+    /// trader without requiring direct access to the kernel. Actors should be added
+    /// after the node is built but before starting the node.
+    ///
+    /// # Errors
+    ///
+    /// Returns an error if:
+    /// - The trader is not in a valid state for adding components.
+    /// - An actor with the same ID is already registered.
+    /// - The node is currently running.
+    pub fn add_actor(&mut self, actor: Box<dyn Component>) -> anyhow::Result<()> {
+        if self.is_running {
+            anyhow::bail!(
+                "Cannot add actor while node is running. Add actors before calling start()."
+            );
+        }
+
+        self.kernel.trader.add_actor(actor)
+    }
 }
 
 /// Builder for constructing a [`LiveNode`] with a fluent API.
@@ -357,10 +379,12 @@ impl LiveNodeBuilder {
         log::info!("LiveNode built successfully");
 
         // Create kernel directly with the config
+        let clock = Rc::new(RefCell::new(LiveClock::new()));
         let kernel = NautilusKernel::new("LiveNode".to_string(), self.config.clone())?;
-        let runner = AsyncRunner::new();
+        let runner = AsyncRunner::new(clock.clone());
 
         Ok(LiveNode {
+            clock,
             kernel,
             runner,
             config: self.config,

crates/live/src/runner.rs

@@ -19,7 +19,7 @@ use futures::StreamExt;
 use nautilus_common::{
     clock::{Clock, LiveClock},
     messages::DataEvent,
-    runner::{DataQueue, RunnerEvent, get_data_cmd_queue, set_data_evt_queue, set_global_clock},
+    runner::{DataQueue, RunnerEvent, get_data_cmd_queue, set_data_evt_queue},
     runtime::get_runtime,
 };
 use nautilus_data::engine::DataEngine;
@@ -43,7 +43,6 @@ impl DataQueue for AsyncDataQueue {
 
 // TODO: Use message bus instead of direct reference to DataEngine
 pub trait Runner {
-    fn new() -> Self;
     fn run(&mut self, data_engine: &mut DataEngine);
 }
 
@@ -60,17 +59,16 @@ impl Debug for AsyncRunner {
     }
 }
 
-impl Runner for AsyncRunner {
-    fn new() -> Self {
+impl AsyncRunner {
+    pub fn new(clock: Rc<RefCell<LiveClock>>) -> Self {
         let (data_tx, data_rx) = tokio::sync::mpsc::unbounded_channel::<DataEvent>();
         set_data_evt_queue(Rc::new(RefCell::new(AsyncDataQueue(data_tx))));
 
-        let clock = Rc::new(RefCell::new(LiveClock::new()));
-        set_global_clock(clock.clone());
-
         Self { clock, data_rx }
     }
+}
 
+impl Runner for AsyncRunner {
     fn run(&mut self, data_engine: &mut DataEngine) {
         let mut time_event_stream = self.clock.borrow().get_event_stream();
         let data_cmd_queue = get_data_cmd_queue();