nautilus_live/
runner.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//! Async event loop runner for live and sandbox trading nodes.
17//!
18//! `AsyncRunner` owns five tokio mpsc channel pairs plus a shutdown
19//! signal channel. Construction creates the channels without side
20//! effects. The sender halves are placed into thread-local storage
21//! via [`AsyncRunner::bind_senders`] so that adapters and engine
22//! components can resolve them through the `get_*_sender()` accessors
23//! in `nautilus_common::runner` and `nautilus_common::live::runner`.
24//!
25//! Channel pairs:
26//!
27//! - **Time events**: timer callbacks dispatched by the clock.
28//! - **Execution events**: fills, order updates, and account state from
29//!   execution clients to the execution engine.
30//! - **Trading commands**: order actions to execution clients.
31//! - **Data events**: market data from adapters to the data engine.
32//! - **Data commands**: subscribe/unsubscribe requests to data clients.
33//!
34//! Both `AsyncRunner::run` and `LiveNode::run` use a `biased;` select with
35//! exec branches polled ahead of data branches, so a strategy action
36//! (cancel, submit) is not delayed behind a market-data backlog when the
37//! select polls receivers each iteration. The two loops use slightly
38//! different cmd/evt sub-orders because `LiveNode::run` also folds in the
39//! maintenance timer and signal handling that `AsyncRunner::run` does not
40//! see; check each `select!` block for the exact order at that site.
41//!
42//! The runner can drive the event loop in two ways:
43//!
44//! - **Standalone**: call [`AsyncRunner::run`], which binds senders and
45//!   enters a `tokio::select!` loop internally.
46//! - **Integrated**: call [`AsyncRunner::take_channels`] to extract the
47//!   receivers and run the `select!` loop directly inside `LiveNode::run`,
48//!   where it is interleaved with startup, reconciliation, and shutdown
49//!   phases.
50//!
51//! # Invariants
52//!
53//! - `bind_senders` must be called before any code that reads from TLS.
54//!   This includes adapter constructors, clock initialization, and
55//!   execution client start methods. Every path from construction to
56//!   the event loop must bind before the first TLS read.
57//! - The event loop and all TLS consumers must execute on the same
58//!   thread. Senders are cloneable and `Send`, but the `RefCell`-backed
59//!   TLS slots are not accessible from other threads.
60//! - Only one runner at a time should own the TLS slots on a given
61//!   thread. `bind_senders` overwrites any existing TLS contents on the
62//!   thread, so the last caller wins.
63
64use std::{fmt::Debug, sync::Arc};
65
66use nautilus_common::{
67    live::runner::{replace_data_event_sender, replace_exec_event_sender},
68    messages::{
69        DataEvent, ExecutionEvent, ExecutionReport, data::DataCommand, execution::TradingCommand,
70    },
71    msgbus::{self, MessagingSwitchboard},
72    runner::{
73        DataCommandSender, TimeEventSender, TradingCommandSender, replace_data_cmd_sender,
74        replace_exec_cmd_sender, replace_time_event_sender,
75    },
76    timer::TimeEventHandler,
77};
78use nautilus_model::events::OrderEventAny;
79
80/// Asynchronous implementation of `DataCommandSender` for live environments.
81#[derive(Debug)]
82pub struct AsyncDataCommandSender {
83    cmd_tx: tokio::sync::mpsc::UnboundedSender<DataCommand>,
84}
85
86impl AsyncDataCommandSender {
87    #[must_use]
88    pub const fn new(cmd_tx: tokio::sync::mpsc::UnboundedSender<DataCommand>) -> Self {
89        Self { cmd_tx }
90    }
91}
92
93impl DataCommandSender for AsyncDataCommandSender {
94    fn execute(&self, command: DataCommand) {
95        if let Err(e) = self.cmd_tx.send(command) {
96            log::error!("Failed to send data command: {e}");
97        }
98    }
99}
100
101/// Asynchronous implementation of `TimeEventSender` for live environments.
102#[derive(Debug, Clone)]
103pub struct AsyncTimeEventSender {
104    time_tx: tokio::sync::mpsc::UnboundedSender<TimeEventHandler>,
105}
106
107impl AsyncTimeEventSender {
108    #[must_use]
109    pub const fn new(time_tx: tokio::sync::mpsc::UnboundedSender<TimeEventHandler>) -> Self {
110        Self { time_tx }
111    }
112}
113
114impl TimeEventSender for AsyncTimeEventSender {
115    fn send(&self, handler: TimeEventHandler) {
116        if let Err(e) = self.time_tx.send(handler) {
117            log::error!("Failed to send time event handler: {e}");
118        }
119    }
120}
121
122/// Asynchronous implementation of `TradingCommandSender` for live environments.
123#[derive(Debug)]
124pub struct AsyncTradingCommandSender {
125    cmd_tx: tokio::sync::mpsc::UnboundedSender<TradingCommand>,
126}
127
128impl AsyncTradingCommandSender {
129    #[must_use]
130    pub const fn new(cmd_tx: tokio::sync::mpsc::UnboundedSender<TradingCommand>) -> Self {
131        Self { cmd_tx }
132    }
133}
134
135impl TradingCommandSender for AsyncTradingCommandSender {
136    fn execute(&self, command: TradingCommand) {
137        if let Err(e) = self.cmd_tx.send(command) {
138            log::error!("Failed to send trading command: {e}");
139        }
140    }
141}
142
143pub trait Runner {
144    fn run(&mut self);
145}
146
147/// Channel receivers for the async event loop.
148///
149/// These can be extracted from `AsyncRunner` via `take_channels()` to drive
150/// the event loop directly on the same thread as the msgbus endpoints.
151#[derive(Debug)]
152pub struct AsyncRunnerChannels {
153    pub time_evt_rx: tokio::sync::mpsc::UnboundedReceiver<TimeEventHandler>,
154    pub exec_evt_rx: tokio::sync::mpsc::UnboundedReceiver<ExecutionEvent>,
155    pub exec_cmd_rx: tokio::sync::mpsc::UnboundedReceiver<TradingCommand>,
156    pub data_evt_rx: tokio::sync::mpsc::UnboundedReceiver<DataEvent>,
157    pub data_cmd_rx: tokio::sync::mpsc::UnboundedReceiver<DataCommand>,
158}
159
160pub struct AsyncRunner {
161    channels: AsyncRunnerChannels,
162    time_evt_tx: tokio::sync::mpsc::UnboundedSender<TimeEventHandler>,
163    signal_rx: tokio::sync::mpsc::UnboundedReceiver<()>,
164    signal_tx: tokio::sync::mpsc::UnboundedSender<()>,
165    exec_cmd_tx: tokio::sync::mpsc::UnboundedSender<TradingCommand>,
166    exec_evt_tx: tokio::sync::mpsc::UnboundedSender<ExecutionEvent>,
167    data_cmd_tx: tokio::sync::mpsc::UnboundedSender<DataCommand>,
168    data_evt_tx: tokio::sync::mpsc::UnboundedSender<DataEvent>,
169}
170
171/// Handle for stopping the `AsyncRunner` from another context.
172#[derive(Clone, Debug)]
173pub struct AsyncRunnerHandle {
174    signal_tx: tokio::sync::mpsc::UnboundedSender<()>,
175}
176
177impl AsyncRunnerHandle {
178    /// Signals the runner to stop.
179    pub fn stop(&self) {
180        if let Err(e) = self.signal_tx.send(()) {
181            log::error!("Failed to send shutdown signal: {e}");
182        }
183    }
184}
185
186impl Default for AsyncRunner {
187    fn default() -> Self {
188        Self::new()
189    }
190}
191
192impl Debug for AsyncRunner {
193    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
194        f.debug_struct(stringify!(AsyncRunner)).finish()
195    }
196}
197
198impl AsyncRunner {
199    /// Creates a new [`AsyncRunner`] instance.
200    ///
201    /// Creates channels but does not bind senders to thread-local storage.
202    /// Call [`bind_senders`](Self::bind_senders) before creating clients that
203    /// read from TLS, and again before entering the event loop.
204    #[must_use]
205    pub fn new() -> Self {
206        use tokio::sync::mpsc::unbounded_channel; // tokio-import-ok
207
208        let (time_evt_tx, time_evt_rx) = unbounded_channel::<TimeEventHandler>();
209        let (signal_tx, signal_rx) = unbounded_channel::<()>();
210        let (exec_cmd_tx, exec_cmd_rx) = unbounded_channel::<TradingCommand>();
211        let (exec_evt_tx, exec_evt_rx) = unbounded_channel::<ExecutionEvent>();
212        let (data_cmd_tx, data_cmd_rx) = unbounded_channel::<DataCommand>();
213        let (data_evt_tx, data_evt_rx) = unbounded_channel::<DataEvent>();
214
215        Self {
216            channels: AsyncRunnerChannels {
217                time_evt_rx,
218                exec_evt_rx,
219                exec_cmd_rx,
220                data_evt_rx,
221                data_cmd_rx,
222            },
223            time_evt_tx,
224            signal_rx,
225            signal_tx,
226            exec_cmd_tx,
227            exec_evt_tx,
228            data_cmd_tx,
229            data_evt_tx,
230        }
231    }
232
233    /// Binds this runner's channel senders to thread-local storage.
234    ///
235    /// Call before creating clients that read from TLS (e.g., in the builder),
236    /// and again before entering the event loop to reclaim ownership if another
237    /// runner was constructed on this thread in the interim.
238    pub fn bind_senders(&self) {
239        replace_time_event_sender(Arc::new(AsyncTimeEventSender::new(
240            self.time_evt_tx.clone(),
241        )));
242        replace_exec_cmd_sender(Arc::new(AsyncTradingCommandSender::new(
243            self.exec_cmd_tx.clone(),
244        )));
245        replace_exec_event_sender(self.exec_evt_tx.clone());
246        replace_data_cmd_sender(Arc::new(AsyncDataCommandSender::new(
247            self.data_cmd_tx.clone(),
248        )));
249        replace_data_event_sender(self.data_evt_tx.clone());
250    }
251
252    /// Stops the runner with an internal shutdown signal.
253    pub fn stop(&self) {
254        if let Err(e) = self.signal_tx.send(()) {
255            log::error!("Failed to send shutdown signal: {e}");
256        }
257    }
258
259    /// Returns a handle that can be used to stop the runner from another context.
260    #[must_use]
261    pub fn handle(&self) -> AsyncRunnerHandle {
262        AsyncRunnerHandle {
263            signal_tx: self.signal_tx.clone(),
264        }
265    }
266
267    /// Consumes the runner and returns the channel receivers for direct event loop driving.
268    ///
269    /// This is used when the event loop needs to run on the same thread as the msgbus
270    /// endpoints (which use thread-local storage).
271    #[must_use]
272    pub fn take_channels(self) -> AsyncRunnerChannels {
273        self.channels
274    }
275
276    /// Flushes all pending data events and commands from the channels.
277    ///
278    /// Loops until both data channels are empty, processing each item
279    /// into the cache immediately. Used in `start()` where channels are
280    /// not extracted.
281    pub fn flush_pending_data(&mut self) {
282        let mut total = 0;
283
284        loop {
285            let mut progressed = false;
286
287            // Events drain before commands here even though the runtime select
288            // prefers the opposite for everything-else: `LiveNode::start()`
289            // calls this after `connect_data_clients()` to push queued
290            // `DataEvent::Instrument` items into the cache. A pending
291            // subscription command (e.g. `SubscribeBars`) processed before the
292            // matching instrument lands would be rejected by the data engine.
293            while let Ok(evt) = self.channels.data_evt_rx.try_recv() {
294                Self::handle_data_event(evt);
295                progressed = true;
296                total += 1;
297            }
298
299            while let Ok(cmd) = self.channels.data_cmd_rx.try_recv() {
300                Self::handle_data_command(cmd);
301                progressed = true;
302                total += 1;
303            }
304
305            if !progressed {
306                break;
307            }
308        }
309
310        if total > 0 {
311            log::debug!("Flushed {total} pending data events/commands");
312        }
313    }
314
315    /// Runs the async runner event loop.
316    ///
317    /// This method processes data events, time events, execution events, and signal events in an async loop.
318    /// It will run until a signal is received or the event streams are closed.
319    pub async fn run(&mut self) {
320        self.bind_senders();
321
322        log::info!("AsyncRunner starting");
323
324        loop {
325            tokio::select! {
326                biased;
327
328                Some(()) = self.signal_rx.recv() => {
329                    log::info!("AsyncRunner received signal, shutting down");
330                    return;
331                },
332                Some(handler) = self.channels.time_evt_rx.recv() => {
333                    Self::handle_time_event(handler);
334                },
335                Some(cmd) = self.channels.exec_cmd_rx.recv() => {
336                    Self::handle_exec_command(cmd);
337                },
338                Some(evt) = self.channels.exec_evt_rx.recv() => {
339                    Self::handle_exec_event(evt);
340                },
341                Some(cmd) = self.channels.data_cmd_rx.recv() => {
342                    Self::handle_data_command(cmd);
343                },
344                Some(evt) = self.channels.data_evt_rx.recv() => {
345                    Self::handle_data_event(evt);
346                },
347                else => {
348                    log::debug!("AsyncRunner all channels closed, exiting");
349                    return;
350                }
351            };
352        }
353    }
354
355    /// Handles a time event by running its callback.
356    #[inline]
357    pub fn handle_time_event(handler: TimeEventHandler) {
358        handler.run();
359    }
360
361    /// Handles a data command by sending to the `DataEngine`.
362    #[inline]
363    pub fn handle_data_command(cmd: DataCommand) {
364        msgbus::send_data_command(MessagingSwitchboard::data_engine_execute(), cmd);
365    }
366
367    /// Handles a data event by sending to the appropriate `DataEngine` endpoint.
368    #[inline]
369    pub fn handle_data_event(event: DataEvent) {
370        match event {
371            DataEvent::Data(data) => {
372                msgbus::send_data(MessagingSwitchboard::data_engine_process_data(), data);
373            }
374            DataEvent::Instrument(data) => {
375                msgbus::send_any(MessagingSwitchboard::data_engine_process(), &data);
376            }
377            DataEvent::Response(resp) => {
378                msgbus::send_data_response(MessagingSwitchboard::data_engine_response(), resp);
379            }
380            DataEvent::FundingRate(funding_rate) => {
381                msgbus::send_any(MessagingSwitchboard::data_engine_process(), &funding_rate);
382            }
383            DataEvent::InstrumentStatus(status) => {
384                msgbus::send_any(MessagingSwitchboard::data_engine_process(), &status);
385            }
386            DataEvent::OptionGreeks(greeks) => {
387                msgbus::send_any(MessagingSwitchboard::data_engine_process(), &greeks);
388            }
389            #[cfg(feature = "defi")]
390            DataEvent::DeFi(data) => {
391                msgbus::send_defi_data(MessagingSwitchboard::data_engine_process_defi_data(), data);
392            }
393        }
394    }
395
396    /// Handles an execution command by sending to the `ExecEngine`.
397    #[inline]
398    pub fn handle_exec_command(cmd: TradingCommand) {
399        msgbus::send_trading_command(MessagingSwitchboard::exec_engine_execute(), cmd);
400    }
401
402    /// Handles an execution event by sending to the appropriate engine endpoint.
403    #[inline]
404    pub fn handle_exec_event(event: ExecutionEvent) {
405        match event {
406            ExecutionEvent::Order(order_event) => {
407                msgbus::send_order_event(MessagingSwitchboard::exec_engine_process(), order_event);
408            }
409            ExecutionEvent::OrderSubmittedBatch(batch) => {
410                for submitted in batch {
411                    msgbus::send_order_event(
412                        MessagingSwitchboard::exec_engine_process(),
413                        OrderEventAny::Submitted(submitted),
414                    );
415                }
416            }
417            ExecutionEvent::OrderAcceptedBatch(batch) => {
418                for accepted in batch {
419                    msgbus::send_order_event(
420                        MessagingSwitchboard::exec_engine_process(),
421                        OrderEventAny::Accepted(accepted),
422                    );
423                }
424            }
425            ExecutionEvent::OrderCanceledBatch(batch) => {
426                for canceled in batch {
427                    msgbus::send_order_event(
428                        MessagingSwitchboard::exec_engine_process(),
429                        OrderEventAny::Canceled(canceled),
430                    );
431                }
432            }
433            ExecutionEvent::Report(report) => {
434                Self::handle_exec_report(report);
435            }
436            ExecutionEvent::Account(ref account) => {
437                msgbus::send_account_state(
438                    MessagingSwitchboard::portfolio_update_account(),
439                    account,
440                );
441            }
442        }
443    }
444
445    #[inline]
446    pub fn handle_exec_report(report: ExecutionReport) {
447        let endpoint = MessagingSwitchboard::exec_engine_reconcile_execution_report();
448        msgbus::send_execution_report(endpoint, report);
449    }
450}
451
452#[cfg(test)]
453mod tests {
454    use std::time::Duration;
455
456    use nautilus_common::{
457        live::runner::{get_data_event_sender, get_exec_event_sender},
458        messages::{
459            ExecutionEvent, ExecutionReport,
460            data::{SubscribeCommand, SubscribeCustomData},
461            execution::{CancelAllOrders, TradingCommand},
462        },
463        runner::{
464            get_data_cmd_sender, get_time_event_sender, get_trading_cmd_sender,
465            try_get_time_event_sender, try_get_trading_cmd_sender,
466        },
467        timer::{TimeEvent, TimeEventCallback, TimeEventHandler},
468    };
469    use nautilus_core::{UUID4, UnixNanos};
470    use nautilus_model::{
471        data::{Data, DataType, quote::QuoteTick},
472        enums::{
473            AccountType, LiquiditySide, OrderSide, OrderStatus, OrderType, PositionSideSpecified,
474            TimeInForce,
475        },
476        events::{
477            OrderAcceptedBatch, OrderCanceledBatch, OrderEvent, OrderEventAny, OrderSubmittedBatch,
478            account::state::AccountState,
479            order::spec::{OrderAcceptedSpec, OrderCanceledSpec, OrderSubmittedSpec},
480        },
481        identifiers::{
482            AccountId, ClientId, ClientOrderId, InstrumentId, PositionId, StrategyId, TradeId,
483            TraderId, VenueOrderId,
484        },
485        reports::{FillReport, OrderStatusReport, PositionStatusReport},
486        types::{Money, Price, Quantity},
487    };
488    use rstest::rstest;
489    use ustr::Ustr;
490
491    use super::*;
492
493    // Test fixture for creating test quotes
494    fn test_quote() -> QuoteTick {
495        QuoteTick {
496            instrument_id: InstrumentId::from("EUR/USD.SIM"),
497            bid_price: Price::from("1.10000"),
498            ask_price: Price::from("1.10001"),
499            bid_size: Quantity::from(1_000_000),
500            ask_size: Quantity::from(1_000_000),
501            ts_event: UnixNanos::default(),
502            ts_init: UnixNanos::default(),
503        }
504    }
505
506    // Test fixture to create AsyncRunner with manual channels.
507    // Sender halves are dummies (not connected to the test receivers) since
508    // these tests exercise the event loop, not TLS binding.
509    fn create_test_runner(
510        time_evt_rx: tokio::sync::mpsc::UnboundedReceiver<TimeEventHandler>,
511        data_evt_rx: tokio::sync::mpsc::UnboundedReceiver<DataEvent>,
512        data_cmd_rx: tokio::sync::mpsc::UnboundedReceiver<DataCommand>,
513        exec_evt_rx: tokio::sync::mpsc::UnboundedReceiver<ExecutionEvent>,
514        exec_cmd_rx: tokio::sync::mpsc::UnboundedReceiver<TradingCommand>,
515        signal_rx: tokio::sync::mpsc::UnboundedReceiver<()>,
516        signal_tx: tokio::sync::mpsc::UnboundedSender<()>,
517    ) -> AsyncRunner {
518        let (time_evt_tx, _) = tokio::sync::mpsc::unbounded_channel();
519        let (data_cmd_tx, _) = tokio::sync::mpsc::unbounded_channel();
520        let (data_evt_tx, _) = tokio::sync::mpsc::unbounded_channel();
521        let (exec_cmd_tx, _) = tokio::sync::mpsc::unbounded_channel();
522        let (exec_evt_tx, _) = tokio::sync::mpsc::unbounded_channel();
523
524        AsyncRunner {
525            channels: AsyncRunnerChannels {
526                time_evt_rx,
527                exec_evt_rx,
528                exec_cmd_rx,
529                data_evt_rx,
530                data_cmd_rx,
531            },
532            time_evt_tx,
533            exec_cmd_tx,
534            exec_evt_tx,
535            data_cmd_tx,
536            data_evt_tx,
537            signal_rx,
538            signal_tx,
539        }
540    }
541
542    #[rstest]
543    fn test_async_data_command_sender_creation() {
544        let (tx, _rx) = tokio::sync::mpsc::unbounded_channel();
545        let sender = AsyncDataCommandSender::new(tx);
546        assert!(format!("{sender:?}").contains("AsyncDataCommandSender"));
547    }
548
549    #[rstest]
550    fn test_async_time_event_sender_creation() {
551        let (tx, _rx) = tokio::sync::mpsc::unbounded_channel();
552        let sender = AsyncTimeEventSender::new(tx);
553        assert!(format!("{sender:?}").contains("AsyncTimeEventSender"));
554    }
555
556    #[tokio::test]
557    async fn test_async_data_command_sender_execute() {
558        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
559        let sender = AsyncDataCommandSender::new(tx);
560
561        let command = DataCommand::Subscribe(SubscribeCommand::Data(SubscribeCustomData {
562            client_id: Some(ClientId::from("TEST")),
563            venue: None,
564            data_type: DataType::new("QuoteTick", None, None),
565            command_id: UUID4::new(),
566            ts_init: UnixNanos::default(),
567            correlation_id: None,
568            params: None,
569        }));
570
571        sender.execute(command.clone());
572
573        let received = rx.recv().await.unwrap();
574        match (received, command) {
575            (
576                DataCommand::Subscribe(SubscribeCommand::Data(r)),
577                DataCommand::Subscribe(SubscribeCommand::Data(c)),
578            ) => {
579                assert_eq!(r.client_id, c.client_id);
580                assert_eq!(r.data_type, c.data_type);
581            }
582            _ => panic!("Command mismatch"),
583        }
584    }
585
586    #[tokio::test]
587    async fn test_async_time_event_sender_send() {
588        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
589        let sender = AsyncTimeEventSender::new(tx);
590
591        let event = TimeEvent::new(
592            Ustr::from("test"),
593            UUID4::new(),
594            UnixNanos::from(1),
595            UnixNanos::from(2),
596        );
597        let callback = TimeEventCallback::from(|_: TimeEvent| {});
598        let handler = TimeEventHandler::new(event, callback);
599
600        sender.send(handler);
601
602        assert!(rx.recv().await.is_some());
603    }
604
605    #[tokio::test]
606    async fn test_runner_shutdown_signal() {
607        // Create runner with manual channels to avoid global state
608        let (_data_tx, data_evt_rx) = tokio::sync::mpsc::unbounded_channel::<DataEvent>();
609        let (_cmd_tx, data_cmd_rx) = tokio::sync::mpsc::unbounded_channel::<DataCommand>();
610        let (_time_tx, time_evt_rx) = tokio::sync::mpsc::unbounded_channel::<TimeEventHandler>();
611        let (_exec_evt_tx, exec_evt_rx) = tokio::sync::mpsc::unbounded_channel::<ExecutionEvent>();
612        let (_exec_cmd_tx, exec_cmd_rx) = tokio::sync::mpsc::unbounded_channel::<TradingCommand>();
613        let (signal_tx, signal_rx) = tokio::sync::mpsc::unbounded_channel::<()>();
614
615        let mut runner = create_test_runner(
616            time_evt_rx,
617            data_evt_rx,
618            data_cmd_rx,
619            exec_evt_rx,
620            exec_cmd_rx,
621            signal_rx,
622            signal_tx.clone(),
623        );
624
625        // Start runner
626        let runner_handle = tokio::spawn(async move {
627            runner.run().await;
628        });
629
630        // Send shutdown signal
631        signal_tx.send(()).unwrap();
632
633        // Runner should stop quickly
634        let result = tokio::time::timeout(Duration::from_millis(100), runner_handle).await;
635        assert!(result.is_ok(), "Runner should stop on signal");
636    }
637
638    #[tokio::test]
639    async fn test_runner_closes_on_channel_drop() {
640        let (data_tx, data_evt_rx) = tokio::sync::mpsc::unbounded_channel::<DataEvent>();
641        let (_cmd_tx, data_cmd_rx) = tokio::sync::mpsc::unbounded_channel::<DataCommand>();
642        let (_time_tx, time_evt_rx) = tokio::sync::mpsc::unbounded_channel::<TimeEventHandler>();
643        let (_exec_evt_tx, exec_evt_rx) = tokio::sync::mpsc::unbounded_channel::<ExecutionEvent>();
644        let (_exec_cmd_tx, exec_cmd_rx) = tokio::sync::mpsc::unbounded_channel::<TradingCommand>();
645        let (signal_tx, signal_rx) = tokio::sync::mpsc::unbounded_channel::<()>();
646
647        let mut runner = create_test_runner(
648            time_evt_rx,
649            data_evt_rx,
650            data_cmd_rx,
651            exec_evt_rx,
652            exec_cmd_rx,
653            signal_rx,
654            signal_tx.clone(),
655        );
656
657        // Start runner
658        let runner_handle = tokio::spawn(async move {
659            runner.run().await;
660        });
661
662        drop(data_tx);
663
664        // Yield to let runner enter event loop before stop signal
665        tokio::task::yield_now().await;
666        signal_tx.send(()).ok();
667
668        // Runner should stop when channels close or on signal
669        let result = tokio::time::timeout(Duration::from_millis(200), runner_handle).await;
670        assert!(
671            result.is_ok(),
672            "Runner should stop when channels close or on signal"
673        );
674    }
675
676    #[tokio::test]
677    async fn test_concurrent_event_sending() {
678        let (data_evt_tx, data_evt_rx) = tokio::sync::mpsc::unbounded_channel::<DataEvent>();
679        let (_data_cmd_tx, data_cmd_rx) = tokio::sync::mpsc::unbounded_channel::<DataCommand>();
680        let (_time_evt_tx, time_evt_rx) =
681            tokio::sync::mpsc::unbounded_channel::<TimeEventHandler>();
682        let (_exec_evt_tx, exec_evt_rx) = tokio::sync::mpsc::unbounded_channel::<ExecutionEvent>();
683        let (_exec_cmd_tx, exec_cmd_rx) = tokio::sync::mpsc::unbounded_channel::<TradingCommand>();
684        let (signal_tx, signal_rx) = tokio::sync::mpsc::unbounded_channel::<()>();
685
686        // Setup runner
687        let mut runner = create_test_runner(
688            time_evt_rx,
689            data_evt_rx,
690            data_cmd_rx,
691            exec_evt_rx,
692            exec_cmd_rx,
693            signal_rx,
694            signal_tx.clone(),
695        );
696
697        // Spawn multiple concurrent senders
698        let mut handles = vec![];
699
700        for _ in 0..5 {
701            let tx_clone = data_evt_tx.clone();
702
703            let handle = tokio::spawn(async move {
704                for _ in 0..20 {
705                    let quote = test_quote();
706                    tx_clone.send(DataEvent::Data(Data::Quote(quote))).unwrap();
707                    tokio::task::yield_now().await;
708                }
709            });
710            handles.push(handle);
711        }
712
713        // Start runner in background
714        let runner_handle = tokio::spawn(async move {
715            runner.run().await;
716        });
717
718        // Wait for all senders
719        for handle in handles {
720            handle.await.unwrap();
721        }
722
723        // Yield to let runner enter event loop before stop signal
724        tokio::task::yield_now().await;
725        signal_tx.send(()).unwrap();
726
727        let _ = tokio::time::timeout(Duration::from_millis(200), runner_handle).await;
728    }
729
730    #[rstest]
731    #[case(10)]
732    #[case(100)]
733    #[case(1000)]
734    fn test_channel_send_performance(#[case] count: usize) {
735        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel::<DataEvent>();
736        let quote = test_quote();
737
738        // Send events
739        for _ in 0..count {
740            tx.send(DataEvent::Data(Data::Quote(quote))).unwrap();
741        }
742
743        // Verify all received
744        let mut received = 0;
745        while rx.try_recv().is_ok() {
746            received += 1;
747        }
748
749        assert_eq!(received, count);
750    }
751
752    #[rstest]
753    fn test_async_trading_command_sender_creation() {
754        let (tx, _rx) = tokio::sync::mpsc::unbounded_channel();
755        let sender = AsyncTradingCommandSender::new(tx);
756        assert!(format!("{sender:?}").contains("AsyncTradingCommandSender"));
757    }
758
759    #[tokio::test]
760    async fn test_async_trading_command_sender_execute() {
761        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel::<TradingCommand>();
762        let sender = AsyncTradingCommandSender::new(tx);
763
764        let command = TradingCommand::CancelAllOrders(CancelAllOrders::new(
765            TraderId::from("TRADER-001"),
766            None,
767            StrategyId::from("S-001"),
768            InstrumentId::from("EUR/USD.SIM"),
769            OrderSide::Buy,
770            UUID4::new(),
771            UnixNanos::default(),
772            None,
773            None, // correlation_id
774        ));
775
776        sender.execute(command);
777
778        let received = rx.recv().await;
779        assert!(received.is_some());
780        assert!(matches!(
781            received.unwrap(),
782            TradingCommand::CancelAllOrders(_)
783        ));
784    }
785
786    #[tokio::test]
787    async fn test_runner_processes_trading_commands() {
788        let (_data_evt_tx, data_evt_rx) = tokio::sync::mpsc::unbounded_channel::<DataEvent>();
789        let (_data_cmd_tx, data_cmd_rx) = tokio::sync::mpsc::unbounded_channel::<DataCommand>();
790        let (_time_evt_tx, time_evt_rx) =
791            tokio::sync::mpsc::unbounded_channel::<TimeEventHandler>();
792        let (_exec_evt_tx, exec_evt_rx) = tokio::sync::mpsc::unbounded_channel::<ExecutionEvent>();
793        let (exec_cmd_tx, exec_cmd_rx) = tokio::sync::mpsc::unbounded_channel::<TradingCommand>();
794        let (signal_tx, signal_rx) = tokio::sync::mpsc::unbounded_channel::<()>();
795
796        let mut runner = create_test_runner(
797            time_evt_rx,
798            data_evt_rx,
799            data_cmd_rx,
800            exec_evt_rx,
801            exec_cmd_rx,
802            signal_rx,
803            signal_tx.clone(),
804        );
805
806        let runner_handle = tokio::spawn(async move {
807            runner.run().await;
808        });
809
810        let command = TradingCommand::CancelAllOrders(CancelAllOrders::new(
811            TraderId::from("TRADER-001"),
812            None,
813            StrategyId::from("S-001"),
814            InstrumentId::from("EUR/USD.SIM"),
815            OrderSide::Buy,
816            UUID4::new(),
817            UnixNanos::default(),
818            None,
819            None, // correlation_id
820        ));
821        exec_cmd_tx.send(command).unwrap();
822
823        tokio::task::yield_now().await;
824        signal_tx.send(()).unwrap();
825
826        let result = tokio::time::timeout(Duration::from_millis(100), runner_handle).await;
827        assert!(result.is_ok(), "Runner should process command and stop");
828    }
829
830    #[tokio::test]
831    async fn test_runner_processes_multiple_trading_commands() {
832        let (_data_evt_tx, data_evt_rx) = tokio::sync::mpsc::unbounded_channel::<DataEvent>();
833        let (_data_cmd_tx, data_cmd_rx) = tokio::sync::mpsc::unbounded_channel::<DataCommand>();
834        let (_time_evt_tx, time_evt_rx) =
835            tokio::sync::mpsc::unbounded_channel::<TimeEventHandler>();
836        let (_exec_evt_tx, exec_evt_rx) = tokio::sync::mpsc::unbounded_channel::<ExecutionEvent>();
837        let (exec_cmd_tx, exec_cmd_rx) = tokio::sync::mpsc::unbounded_channel::<TradingCommand>();
838        let (signal_tx, signal_rx) = tokio::sync::mpsc::unbounded_channel::<()>();
839
840        let mut runner = create_test_runner(
841            time_evt_rx,
842            data_evt_rx,
843            data_cmd_rx,
844            exec_evt_rx,
845            exec_cmd_rx,
846            signal_rx,
847            signal_tx.clone(),
848        );
849
850        let runner_handle = tokio::spawn(async move {
851            runner.run().await;
852        });
853
854        for i in 0..10 {
855            let strategy_id = format!("S-{i:03}");
856            let command = TradingCommand::CancelAllOrders(CancelAllOrders::new(
857                TraderId::from("TRADER-001"),
858                None,
859                StrategyId::from(strategy_id.as_str()),
860                InstrumentId::from("EUR/USD.SIM"),
861                OrderSide::Buy,
862                UUID4::new(),
863                UnixNanos::default(),
864                None,
865                None, // correlation_id
866            ));
867            exec_cmd_tx.send(command).unwrap();
868        }
869
870        tokio::task::yield_now().await;
871        signal_tx.send(()).unwrap();
872
873        let result = tokio::time::timeout(Duration::from_millis(100), runner_handle).await;
874        assert!(
875            result.is_ok(),
876            "Runner should process all commands and stop"
877        );
878    }
879
880    #[tokio::test]
881    async fn test_execution_event_order_channel() {
882        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel::<ExecutionEvent>();
883
884        let event = OrderSubmittedSpec::builder()
885            .client_order_id(ClientOrderId::from("O-001"))
886            .build();
887
888        tx.send(ExecutionEvent::Order(OrderEventAny::Submitted(event)))
889            .unwrap();
890
891        let received = rx.recv().await.unwrap();
892        match received {
893            ExecutionEvent::Order(OrderEventAny::Submitted(e)) => {
894                assert_eq!(e.client_order_id(), ClientOrderId::from("O-001"));
895            }
896            _ => panic!("Expected OrderSubmitted event"),
897        }
898    }
899
900    #[tokio::test]
901    async fn test_execution_report_order_status_channel() {
902        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel::<ExecutionEvent>();
903
904        let report = OrderStatusReport::new(
905            AccountId::from("SIM-001"),
906            InstrumentId::from("EUR/USD.SIM"),
907            Some(ClientOrderId::from("O-001")),
908            VenueOrderId::from("V-001"),
909            OrderSide::Buy,
910            OrderType::Market,
911            TimeInForce::Gtc,
912            OrderStatus::Accepted,
913            Quantity::from(100_000),
914            Quantity::from(100_000),
915            UnixNanos::from(1),
916            UnixNanos::from(2),
917            UnixNanos::from(3),
918            None,
919        );
920
921        tx.send(ExecutionEvent::Report(ExecutionReport::Order(Box::new(
922            report,
923        ))))
924        .unwrap();
925
926        let received = rx.recv().await.unwrap();
927        match received {
928            ExecutionEvent::Report(ExecutionReport::Order(r)) => {
929                assert_eq!(r.venue_order_id.as_str(), "V-001");
930                assert_eq!(r.order_status, OrderStatus::Accepted);
931            }
932            _ => panic!("Expected OrderStatusReport"),
933        }
934    }
935
936    #[tokio::test]
937    async fn test_execution_report_fill() {
938        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel::<ExecutionEvent>();
939
940        let report = FillReport::new(
941            AccountId::from("SIM-001"),
942            InstrumentId::from("EUR/USD.SIM"),
943            VenueOrderId::from("V-001"),
944            TradeId::from("T-001"),
945            OrderSide::Buy,
946            Quantity::from(100_000),
947            Price::from("1.10000"),
948            Money::from("10 USD"),
949            LiquiditySide::Taker,
950            Some(ClientOrderId::from("O-001")),
951            None,
952            UnixNanos::from(1),
953            UnixNanos::from(2),
954            None,
955        );
956
957        tx.send(ExecutionEvent::Report(ExecutionReport::Fill(Box::new(
958            report,
959        ))))
960        .unwrap();
961
962        let received = rx.recv().await.unwrap();
963        match received {
964            ExecutionEvent::Report(ExecutionReport::Fill(r)) => {
965                assert_eq!(r.venue_order_id.as_str(), "V-001");
966                assert_eq!(r.trade_id.to_string(), "T-001");
967            }
968            _ => panic!("Expected FillReport"),
969        }
970    }
971
972    #[tokio::test]
973    async fn test_execution_report_position() {
974        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel::<ExecutionEvent>();
975
976        let report = PositionStatusReport::new(
977            AccountId::from("SIM-001"),
978            InstrumentId::from("EUR/USD.SIM"),
979            PositionSideSpecified::Long,
980            Quantity::from(100_000),
981            UnixNanos::from(1),
982            UnixNanos::from(2),
983            None,
984            Some(PositionId::from("P-001")),
985            None,
986        );
987
988        tx.send(ExecutionEvent::Report(ExecutionReport::Position(Box::new(
989            report,
990        ))))
991        .unwrap();
992
993        let received = rx.recv().await.unwrap();
994        match received {
995            ExecutionEvent::Report(ExecutionReport::Position(r)) => {
996                assert_eq!(r.venue_position_id.unwrap().as_str(), "P-001");
997            }
998            _ => panic!("Expected PositionStatusReport"),
999        }
1000    }
1001
1002    #[tokio::test]
1003    async fn test_execution_event_account() {
1004        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel::<ExecutionEvent>();
1005
1006        let account_state = AccountState::new(
1007            AccountId::from("SIM-001"),
1008            AccountType::Cash,
1009            vec![],
1010            vec![],
1011            true,
1012            UUID4::new(),
1013            UnixNanos::from(1),
1014            UnixNanos::from(2),
1015            None,
1016        );
1017
1018        tx.send(ExecutionEvent::Account(account_state)).unwrap();
1019
1020        let received = rx.recv().await.unwrap();
1021        match received {
1022            ExecutionEvent::Account(r) => {
1023                assert_eq!(r.account_id.as_str(), "SIM-001");
1024            }
1025            _ => panic!("Expected AccountState"),
1026        }
1027    }
1028
1029    #[tokio::test]
1030    async fn test_runner_stop_method() {
1031        let (_data_tx, data_evt_rx) = tokio::sync::mpsc::unbounded_channel::<DataEvent>();
1032        let (_cmd_tx, data_cmd_rx) = tokio::sync::mpsc::unbounded_channel::<DataCommand>();
1033        let (_time_tx, time_evt_rx) = tokio::sync::mpsc::unbounded_channel::<TimeEventHandler>();
1034        let (_exec_evt_tx, exec_evt_rx) = tokio::sync::mpsc::unbounded_channel::<ExecutionEvent>();
1035        let (_exec_cmd_tx, exec_cmd_rx) = tokio::sync::mpsc::unbounded_channel::<TradingCommand>();
1036        let (signal_tx, signal_rx) = tokio::sync::mpsc::unbounded_channel::<()>();
1037
1038        let mut runner = create_test_runner(
1039            time_evt_rx,
1040            data_evt_rx,
1041            data_cmd_rx,
1042            exec_evt_rx,
1043            exec_cmd_rx,
1044            signal_rx,
1045            signal_tx.clone(),
1046        );
1047
1048        let runner_handle = tokio::spawn(async move {
1049            runner.run().await;
1050        });
1051
1052        // Use stop via signal_tx directly
1053        signal_tx.send(()).unwrap();
1054
1055        let result = tokio::time::timeout(Duration::from_millis(100), runner_handle).await;
1056        assert!(result.is_ok(), "Runner should stop when stop() is called");
1057    }
1058
1059    #[tokio::test]
1060    async fn test_all_event_types_integration() {
1061        let (data_evt_tx, data_evt_rx) = tokio::sync::mpsc::unbounded_channel::<DataEvent>();
1062        let (data_cmd_tx, data_cmd_rx) = tokio::sync::mpsc::unbounded_channel::<DataCommand>();
1063        let (time_evt_tx, time_evt_rx) = tokio::sync::mpsc::unbounded_channel::<TimeEventHandler>();
1064        let (exec_evt_tx, exec_evt_rx) = tokio::sync::mpsc::unbounded_channel::<ExecutionEvent>();
1065        let (_exec_cmd_tx, exec_cmd_rx) = tokio::sync::mpsc::unbounded_channel::<TradingCommand>();
1066        let (signal_tx, signal_rx) = tokio::sync::mpsc::unbounded_channel::<()>();
1067
1068        let mut runner = create_test_runner(
1069            time_evt_rx,
1070            data_evt_rx,
1071            data_cmd_rx,
1072            exec_evt_rx,
1073            exec_cmd_rx,
1074            signal_rx,
1075            signal_tx.clone(),
1076        );
1077
1078        let runner_handle = tokio::spawn(async move {
1079            runner.run().await;
1080        });
1081
1082        // Send data event
1083        let quote = test_quote();
1084        data_evt_tx
1085            .send(DataEvent::Data(Data::Quote(quote)))
1086            .unwrap();
1087
1088        // Send data command
1089        let command = DataCommand::Subscribe(SubscribeCommand::Data(SubscribeCustomData {
1090            client_id: Some(ClientId::from("TEST")),
1091            venue: None,
1092            data_type: DataType::new("QuoteTick", None, None),
1093            command_id: UUID4::new(),
1094            ts_init: UnixNanos::default(),
1095            correlation_id: None,
1096            params: None,
1097        }));
1098        data_cmd_tx.send(command).unwrap();
1099
1100        // Send time event
1101        let event = TimeEvent::new(
1102            Ustr::from("test"),
1103            UUID4::new(),
1104            UnixNanos::from(1),
1105            UnixNanos::from(2),
1106        );
1107        let callback = TimeEventCallback::from(|_: TimeEvent| {});
1108        let handler = TimeEventHandler::new(event, callback);
1109        time_evt_tx.send(handler).unwrap();
1110
1111        // Send execution order event
1112        let order_event = OrderSubmittedSpec::builder()
1113            .client_order_id(ClientOrderId::from("O-001"))
1114            .build();
1115        exec_evt_tx
1116            .send(ExecutionEvent::Order(OrderEventAny::Submitted(order_event)))
1117            .unwrap();
1118
1119        // Send execution report (OrderStatus)
1120        let order_status = OrderStatusReport::new(
1121            AccountId::from("SIM-001"),
1122            InstrumentId::from("EUR/USD.SIM"),
1123            Some(ClientOrderId::from("O-001")),
1124            VenueOrderId::from("V-001"),
1125            OrderSide::Buy,
1126            OrderType::Market,
1127            TimeInForce::Gtc,
1128            OrderStatus::Accepted,
1129            Quantity::from(100_000),
1130            Quantity::from(100_000),
1131            UnixNanos::from(1),
1132            UnixNanos::from(2),
1133            UnixNanos::from(3),
1134            None,
1135        );
1136        exec_evt_tx
1137            .send(ExecutionEvent::Report(ExecutionReport::Order(Box::new(
1138                order_status,
1139            ))))
1140            .unwrap();
1141
1142        // Send execution report (Fill)
1143        let fill = FillReport::new(
1144            AccountId::from("SIM-001"),
1145            InstrumentId::from("EUR/USD.SIM"),
1146            VenueOrderId::from("V-001"),
1147            TradeId::from("T-001"),
1148            OrderSide::Buy,
1149            Quantity::from(100_000),
1150            Price::from("1.10000"),
1151            Money::from("10 USD"),
1152            LiquiditySide::Taker,
1153            Some(ClientOrderId::from("O-001")),
1154            None,
1155            UnixNanos::from(1),
1156            UnixNanos::from(2),
1157            None,
1158        );
1159        exec_evt_tx
1160            .send(ExecutionEvent::Report(ExecutionReport::Fill(Box::new(
1161                fill,
1162            ))))
1163            .unwrap();
1164
1165        // Send execution report (Position)
1166        let position = PositionStatusReport::new(
1167            AccountId::from("SIM-001"),
1168            InstrumentId::from("EUR/USD.SIM"),
1169            PositionSideSpecified::Long,
1170            Quantity::from(100_000),
1171            UnixNanos::from(1),
1172            UnixNanos::from(2),
1173            None,
1174            Some(PositionId::from("P-001")),
1175            None,
1176        );
1177        exec_evt_tx
1178            .send(ExecutionEvent::Report(ExecutionReport::Position(Box::new(
1179                position,
1180            ))))
1181            .unwrap();
1182
1183        // Send account event
1184        let account_state = AccountState::new(
1185            AccountId::from("SIM-001"),
1186            AccountType::Cash,
1187            vec![],
1188            vec![],
1189            true,
1190            UUID4::new(),
1191            UnixNanos::from(1),
1192            UnixNanos::from(2),
1193            None,
1194        );
1195        exec_evt_tx
1196            .send(ExecutionEvent::Account(account_state))
1197            .unwrap();
1198
1199        // Yield to let runner enter event loop before stop signal
1200        tokio::task::yield_now().await;
1201        signal_tx.send(()).unwrap();
1202
1203        let result = tokio::time::timeout(Duration::from_millis(200), runner_handle).await;
1204        assert!(
1205            result.is_ok(),
1206            "Runner should process all event types and stop cleanly"
1207        );
1208    }
1209
1210    #[tokio::test]
1211    async fn test_runner_handle_stops_runner() {
1212        let (_data_tx, data_evt_rx) = tokio::sync::mpsc::unbounded_channel::<DataEvent>();
1213        let (_cmd_tx, data_cmd_rx) = tokio::sync::mpsc::unbounded_channel::<DataCommand>();
1214        let (_time_tx, time_evt_rx) = tokio::sync::mpsc::unbounded_channel::<TimeEventHandler>();
1215        let (_exec_evt_tx, exec_evt_rx) = tokio::sync::mpsc::unbounded_channel::<ExecutionEvent>();
1216        let (_exec_cmd_tx, exec_cmd_rx) = tokio::sync::mpsc::unbounded_channel::<TradingCommand>();
1217        let (signal_tx, signal_rx) = tokio::sync::mpsc::unbounded_channel::<()>();
1218
1219        let mut runner = create_test_runner(
1220            time_evt_rx,
1221            data_evt_rx,
1222            data_cmd_rx,
1223            exec_evt_rx,
1224            exec_cmd_rx,
1225            signal_rx,
1226            signal_tx.clone(),
1227        );
1228
1229        // Get handle before moving runner
1230        let handle = runner.handle();
1231
1232        let runner_handle = tokio::spawn(async move {
1233            runner.run().await;
1234        });
1235
1236        // Use handle to stop
1237        handle.stop();
1238
1239        let result = tokio::time::timeout(Duration::from_millis(100), runner_handle).await;
1240        assert!(result.is_ok(), "Runner should stop via handle");
1241    }
1242
1243    #[tokio::test]
1244    async fn test_runner_handle_is_cloneable() {
1245        let (signal_tx, _signal_rx) = tokio::sync::mpsc::unbounded_channel::<()>();
1246        let handle = AsyncRunnerHandle { signal_tx };
1247
1248        let handle2 = handle.clone();
1249
1250        // Both handles should be able to send stop signals
1251        assert!(handle.signal_tx.send(()).is_ok());
1252        assert!(handle2.signal_tx.send(()).is_ok());
1253    }
1254
1255    #[tokio::test]
1256    async fn test_runner_processes_events_before_stop() {
1257        let (data_evt_tx, data_evt_rx) = tokio::sync::mpsc::unbounded_channel::<DataEvent>();
1258        let (_cmd_tx, data_cmd_rx) = tokio::sync::mpsc::unbounded_channel::<DataCommand>();
1259        let (_time_tx, time_evt_rx) = tokio::sync::mpsc::unbounded_channel::<TimeEventHandler>();
1260        let (_exec_evt_tx, exec_evt_rx) = tokio::sync::mpsc::unbounded_channel::<ExecutionEvent>();
1261        let (_exec_cmd_tx, exec_cmd_rx) = tokio::sync::mpsc::unbounded_channel::<TradingCommand>();
1262        let (signal_tx, signal_rx) = tokio::sync::mpsc::unbounded_channel::<()>();
1263
1264        let mut runner = create_test_runner(
1265            time_evt_rx,
1266            data_evt_rx,
1267            data_cmd_rx,
1268            exec_evt_rx,
1269            exec_cmd_rx,
1270            signal_rx,
1271            signal_tx.clone(),
1272        );
1273
1274        let handle = runner.handle();
1275
1276        // Send events before starting runner
1277        for _ in 0..10 {
1278            let quote = test_quote();
1279            data_evt_tx
1280                .send(DataEvent::Data(Data::Quote(quote)))
1281                .unwrap();
1282        }
1283
1284        let runner_handle = tokio::spawn(async move {
1285            runner.run().await;
1286        });
1287
1288        // Yield to let runner enter event loop before stop signal
1289        tokio::task::yield_now().await;
1290        handle.stop();
1291
1292        let result = tokio::time::timeout(Duration::from_millis(200), runner_handle).await;
1293        assert!(result.is_ok(), "Runner should process events and stop");
1294    }
1295
1296    #[rstest]
1297    fn test_new_does_not_bind_tls() {
1298        std::thread::spawn(|| {
1299            let _runner = AsyncRunner::new();
1300            assert!(try_get_time_event_sender().is_none());
1301            assert!(try_get_trading_cmd_sender().is_none());
1302        })
1303        .join()
1304        .unwrap();
1305    }
1306
1307    #[rstest]
1308    fn test_bind_senders_routes_to_runner_channels() {
1309        std::thread::spawn(|| {
1310            let mut runner = AsyncRunner::new();
1311            runner.bind_senders();
1312
1313            get_data_cmd_sender().execute(DataCommand::Subscribe(SubscribeCommand::Data(
1314                SubscribeCustomData {
1315                    client_id: Some(ClientId::from("TEST")),
1316                    venue: None,
1317                    data_type: DataType::new("test", None, None),
1318                    command_id: UUID4::new(),
1319                    ts_init: UnixNanos::default(),
1320                    correlation_id: None,
1321                    params: None,
1322                },
1323            )));
1324            assert!(runner.channels.data_cmd_rx.try_recv().is_ok());
1325
1326            get_trading_cmd_sender().execute(TradingCommand::CancelAllOrders(
1327                CancelAllOrders::new(
1328                    TraderId::from("TRADER-001"),
1329                    None,
1330                    StrategyId::from("S-001"),
1331                    InstrumentId::from("EUR/USD.SIM"),
1332                    OrderSide::Buy,
1333                    UUID4::new(),
1334                    UnixNanos::default(),
1335                    None,
1336                    None, // correlation_id
1337                ),
1338            ));
1339            assert!(runner.channels.exec_cmd_rx.try_recv().is_ok());
1340
1341            let event = TimeEvent::new(
1342                Ustr::from("test"),
1343                UUID4::new(),
1344                UnixNanos::from(1),
1345                UnixNanos::from(2),
1346            );
1347            let callback = TimeEventCallback::from(|_: TimeEvent| {});
1348            get_time_event_sender().send(TimeEventHandler::new(event, callback));
1349            assert!(runner.channels.time_evt_rx.try_recv().is_ok());
1350
1351            get_data_event_sender()
1352                .send(DataEvent::Data(Data::Quote(test_quote())))
1353                .unwrap();
1354            assert!(runner.channels.data_evt_rx.try_recv().is_ok());
1355
1356            let account = AccountState::new(
1357                AccountId::from("SIM-001"),
1358                AccountType::Cash,
1359                vec![],
1360                vec![],
1361                true,
1362                UUID4::new(),
1363                UnixNanos::from(1),
1364                UnixNanos::from(2),
1365                None,
1366            );
1367            get_exec_event_sender()
1368                .send(ExecutionEvent::Account(account))
1369                .unwrap();
1370            assert!(runner.channels.exec_evt_rx.try_recv().is_ok());
1371        })
1372        .join()
1373        .unwrap();
1374    }
1375
1376    #[rstest]
1377    fn test_bind_senders_reclaims_tls_from_previous_runner() {
1378        std::thread::spawn(|| {
1379            let mut runner1 = AsyncRunner::new();
1380            runner1.bind_senders();
1381
1382            let mut runner2 = AsyncRunner::new();
1383            runner2.bind_senders();
1384
1385            get_data_cmd_sender().execute(DataCommand::Subscribe(SubscribeCommand::Data(
1386                SubscribeCustomData {
1387                    client_id: Some(ClientId::from("TEST")),
1388                    venue: None,
1389                    data_type: DataType::new("test", None, None),
1390                    command_id: UUID4::new(),
1391                    ts_init: UnixNanos::default(),
1392                    correlation_id: None,
1393                    params: None,
1394                },
1395            )));
1396
1397            assert!(runner2.channels.data_cmd_rx.try_recv().is_ok());
1398            assert!(runner1.channels.data_cmd_rx.try_recv().is_err());
1399        })
1400        .join()
1401        .unwrap();
1402    }
1403
1404    #[tokio::test]
1405    async fn test_execution_event_order_submitted_batch_channel() {
1406        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel::<ExecutionEvent>();
1407
1408        let events = vec![
1409            OrderSubmittedSpec::builder()
1410                .client_order_id(ClientOrderId::from("O-001"))
1411                .build(),
1412            OrderSubmittedSpec::builder()
1413                .client_order_id(ClientOrderId::from("O-002"))
1414                .build(),
1415        ];
1416
1417        let batch = OrderSubmittedBatch::new(events);
1418        tx.send(ExecutionEvent::OrderSubmittedBatch(batch)).unwrap();
1419
1420        let received = rx.recv().await.unwrap();
1421        match received {
1422            ExecutionEvent::OrderSubmittedBatch(b) => {
1423                assert_eq!(b.len(), 2);
1424                assert_eq!(b.events[0].client_order_id, ClientOrderId::from("O-001"));
1425                assert_eq!(b.events[1].client_order_id, ClientOrderId::from("O-002"));
1426            }
1427            _ => panic!("Expected OrderSubmittedBatch event"),
1428        }
1429    }
1430
1431    #[tokio::test]
1432    async fn test_execution_event_order_accepted_batch_channel() {
1433        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel::<ExecutionEvent>();
1434
1435        let events = vec![
1436            OrderAcceptedSpec::builder()
1437                .client_order_id(ClientOrderId::from("O-001"))
1438                .build(),
1439            OrderAcceptedSpec::builder()
1440                .client_order_id(ClientOrderId::from("O-002"))
1441                .build(),
1442        ];
1443
1444        let batch = OrderAcceptedBatch::new(events);
1445        tx.send(ExecutionEvent::OrderAcceptedBatch(batch)).unwrap();
1446
1447        let received = rx.recv().await.unwrap();
1448        match received {
1449            ExecutionEvent::OrderAcceptedBatch(b) => {
1450                assert_eq!(b.len(), 2);
1451                assert_eq!(b.events[0].client_order_id, ClientOrderId::from("O-001"));
1452                assert_eq!(b.events[1].client_order_id, ClientOrderId::from("O-002"));
1453            }
1454            _ => panic!("Expected OrderAcceptedBatch event"),
1455        }
1456    }
1457
1458    #[tokio::test]
1459    async fn test_execution_event_order_canceled_batch_channel() {
1460        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel::<ExecutionEvent>();
1461
1462        let events = vec![
1463            OrderCanceledSpec::builder()
1464                .client_order_id(ClientOrderId::from("O-001"))
1465                .build(),
1466            OrderCanceledSpec::builder()
1467                .client_order_id(ClientOrderId::from("O-002"))
1468                .build(),
1469        ];
1470
1471        let batch = OrderCanceledBatch::new(events);
1472        tx.send(ExecutionEvent::OrderCanceledBatch(batch)).unwrap();
1473
1474        let received = rx.recv().await.unwrap();
1475        match received {
1476            ExecutionEvent::OrderCanceledBatch(b) => {
1477                assert_eq!(b.len(), 2);
1478                assert_eq!(b.events[0].client_order_id, ClientOrderId::from("O-001"));
1479                assert_eq!(b.events[1].client_order_id, ClientOrderId::from("O-002"));
1480            }
1481            _ => panic!("Expected OrderCanceledBatch event"),
1482        }
1483    }
1484}
nautilus_live/runner.rs

nautilus_live/
runner.rs
