nautilus_common/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//! Tokio-based channel senders for live trading runtime.
17//!
18//! Stores channel senders for the current runtime thread.
19//!
20//! Data and execution event setters accept either a runtime-bound [`EventSender`] or a plain
21//! Tokio sender. Plain senders carry no callback context. Getters and clones preserve the
22//! installed mode and owner; moving a runtime-bound sender to another thread makes its sends
23//! independent ingress.
24
25use std::cell::RefCell;
26
27use crate::{
28    live::sender::{DispatchSender, EventSender},
29    messages::{DataEvent, ExecutionEvent, SystemCommand, SystemEvent},
30};
31
32/// Gets the thread-local data event sender.
33///
34/// # Panics
35///
36/// Panics if the sender is uninitialized.
37#[must_use]
38pub fn get_data_event_sender() -> EventSender<DataEvent> {
39    DATA_EVENT_SENDER.with(|sender| {
40        sender
41            .borrow()
42            .as_ref()
43            .expect("Data event sender should be initialized by runner")
44            .clone()
45    })
46}
47
48/// Attempts to get the thread-local data event sender without panicking.
49///
50/// Returns `None` if the sender is not initialized (e.g., in Python/v1 bridge environments
51/// before a runner or adapter bridge has registered a sender).
52#[must_use]
53pub fn try_get_data_event_sender() -> Option<EventSender<DataEvent>> {
54    DATA_EVENT_SENDER.with(|sender| sender.borrow().as_ref().cloned())
55}
56
57/// Sets the thread-local data event sender.
58///
59/// Can only be called once per thread.
60///
61/// # Panics
62///
63/// Panics if a sender has already been set.
64pub fn set_data_event_sender(sender: impl Into<EventSender<DataEvent>>) {
65    let sender = sender.into();
66    DATA_EVENT_SENDER.with(|s| {
67        let mut slot = s.borrow_mut();
68        assert!(slot.is_none(), "Data event sender can only be set once");
69        *slot = Some(sender);
70    });
71}
72
73/// Replaces the data event sender for the current thread.
74pub fn replace_data_event_sender(sender: impl Into<EventSender<DataEvent>>) {
75    let sender = sender.into();
76    DATA_EVENT_SENDER.with(|s| {
77        *s.borrow_mut() = Some(sender);
78    });
79}
80
81/// Gets the thread-local system event sender.
82///
83/// # Panics
84///
85/// Panics if the sender is uninitialized.
86#[must_use]
87pub fn get_system_event_sender() -> EventSender<SystemEvent> {
88    SYSTEM_EVENT_SENDER.with(|sender| {
89        sender
90            .borrow()
91            .as_ref()
92            .expect("System event sender should be initialized by runner")
93            .clone()
94    })
95}
96
97/// Attempts to get the thread-local system event sender without panicking.
98///
99/// Returns `None` if the sender is not initialized (e.g., in test environments).
100#[must_use]
101pub fn try_get_system_event_sender() -> Option<EventSender<SystemEvent>> {
102    SYSTEM_EVENT_SENDER.with(|sender| sender.borrow().as_ref().cloned())
103}
104
105/// Sets the thread-local system event sender.
106///
107/// Can only be called once per thread.
108///
109/// # Panics
110///
111/// Panics if a sender has already been set.
112pub fn set_system_event_sender(sender: impl Into<EventSender<SystemEvent>>) {
113    let sender = sender.into();
114    SYSTEM_EVENT_SENDER.with(|s| {
115        let mut slot = s.borrow_mut();
116        assert!(slot.is_none(), "System event sender can only be set once");
117        *slot = Some(sender);
118    });
119}
120
121/// Replaces the system event sender for the current thread.
122pub fn replace_system_event_sender(sender: impl Into<EventSender<SystemEvent>>) {
123    let sender = sender.into();
124    SYSTEM_EVENT_SENDER.with(|s| {
125        *s.borrow_mut() = Some(sender);
126    });
127}
128
129/// Gets the thread-local system command sender.
130///
131/// # Panics
132///
133/// Panics if the sender is uninitialized.
134#[must_use]
135pub fn get_system_command_sender() -> DispatchSender<SystemCommand> {
136    SYSTEM_COMMAND_SENDER.with(|sender| {
137        sender
138            .borrow()
139            .as_ref()
140            .expect("System command sender should be initialized by runner")
141            .clone()
142    })
143}
144
145/// Attempts to get the thread-local system command sender without panicking.
146///
147/// Returns `None` if the sender is not initialized.
148#[must_use]
149pub fn try_get_system_command_sender() -> Option<DispatchSender<SystemCommand>> {
150    SYSTEM_COMMAND_SENDER.with(|sender| sender.borrow().as_ref().cloned())
151}
152
153/// Sets the thread-local system command sender.
154///
155/// Can only be called once per thread.
156///
157/// # Panics
158///
159/// Panics if a sender has already been set.
160pub fn set_system_command_sender(sender: impl Into<DispatchSender<SystemCommand>>) {
161    let sender = sender.into();
162    SYSTEM_COMMAND_SENDER.with(|s| {
163        let mut slot = s.borrow_mut();
164        assert!(slot.is_none(), "System command sender can only be set once");
165        *slot = Some(sender);
166    });
167}
168
169/// Replaces the system command sender for the current thread.
170pub fn replace_system_command_sender(sender: impl Into<DispatchSender<SystemCommand>>) {
171    let sender = sender.into();
172    SYSTEM_COMMAND_SENDER.with(|s| {
173        *s.borrow_mut() = Some(sender);
174    });
175}
176
177/// Gets the thread-local execution event sender.
178///
179/// # Panics
180///
181/// Panics if the sender is uninitialized.
182#[must_use]
183pub fn get_exec_event_sender() -> EventSender<ExecutionEvent> {
184    EXEC_EVENT_SENDER.with(|sender| {
185        sender
186            .borrow()
187            .as_ref()
188            .expect("Execution event sender should be initialized by runner")
189            .clone()
190    })
191}
192
193/// Attempts to get the thread-local execution event sender without panicking.
194///
195/// Returns `None` if the sender is not initialized (e.g., in test environments).
196#[must_use]
197pub fn try_get_exec_event_sender() -> Option<EventSender<ExecutionEvent>> {
198    EXEC_EVENT_SENDER.with(|sender| sender.borrow().as_ref().cloned())
199}
200
201/// Sets the thread-local execution event sender.
202///
203/// Can only be called once per thread.
204///
205/// # Panics
206///
207/// Panics if a sender has already been set.
208pub fn set_exec_event_sender(sender: impl Into<EventSender<ExecutionEvent>>) {
209    let sender = sender.into();
210    EXEC_EVENT_SENDER.with(|s| {
211        let mut slot = s.borrow_mut();
212        assert!(
213            slot.is_none(),
214            "Execution event sender can only be set once"
215        );
216        *slot = Some(sender);
217    });
218}
219
220/// Replaces the execution event sender for the current thread.
221pub fn replace_exec_event_sender(sender: impl Into<EventSender<ExecutionEvent>>) {
222    let sender = sender.into();
223    EXEC_EVENT_SENDER.with(|s| {
224        *s.borrow_mut() = Some(sender);
225    });
226}
227
228thread_local! {
229    static DATA_EVENT_SENDER: RefCell<Option<EventSender<DataEvent>>> = const { RefCell::new(None) };
230    static EXEC_EVENT_SENDER: RefCell<Option<EventSender<ExecutionEvent>>> = const { RefCell::new(None) };
231    static SYSTEM_EVENT_SENDER: RefCell<Option<EventSender<SystemEvent>>> = const { RefCell::new(None) };
232    static SYSTEM_COMMAND_SENDER: RefCell<Option<DispatchSender<SystemCommand>>> = const { RefCell::new(None) };
233}
234
235#[cfg(test)]
236mod tests {
237    use std::sync::{Arc, Barrier};
238
239    use rstest::rstest;
240
241    use super::*;
242
243    #[rstest]
244    fn test_replace_data_event_sender_overwrites_previous() {
245        assert_sender_replaced(replace_data_event_sender, get_data_event_sender);
246    }
247
248    #[rstest]
249    fn test_replace_exec_event_sender_overwrites_previous() {
250        assert_sender_replaced(replace_exec_event_sender, get_exec_event_sender);
251    }
252
253    #[rstest]
254    fn test_replace_system_event_sender_overwrites_previous() {
255        assert_sender_replaced(replace_system_event_sender, get_system_event_sender);
256    }
257
258    #[rstest]
259    fn test_replace_system_command_sender_overwrites_previous() {
260        assert_sender_replaced(replace_system_command_sender, get_system_command_sender);
261    }
262
263    #[rstest]
264    fn test_event_senders_are_thread_local() {
265        assert_sender_thread_local(replace_data_event_sender, get_data_event_sender);
266        assert_sender_thread_local(replace_exec_event_sender, get_exec_event_sender);
267        assert_sender_thread_local(replace_system_event_sender, get_system_event_sender);
268        assert_sender_thread_local(replace_system_command_sender, get_system_command_sender);
269    }
270
271    #[rstest]
272    fn test_set_data_event_sender_panics_on_double_set() {
273        let result = std::thread::spawn(|| {
274            let (tx1, _rx1) = tokio::sync::mpsc::unbounded_channel();
275            let (tx2, _rx2) = tokio::sync::mpsc::unbounded_channel();
276            set_data_event_sender(tx1);
277            set_data_event_sender(tx2);
278        })
279        .join();
280        assert!(result.is_err());
281    }
282
283    #[rstest]
284    fn test_set_exec_event_sender_panics_on_double_set() {
285        let result = std::thread::spawn(|| {
286            let (tx1, _rx1) = tokio::sync::mpsc::unbounded_channel();
287            let (tx2, _rx2) = tokio::sync::mpsc::unbounded_channel();
288            set_exec_event_sender(tx1);
289            set_exec_event_sender(tx2);
290        })
291        .join();
292        assert!(result.is_err());
293    }
294
295    #[rstest]
296    fn test_set_system_event_sender_panics_on_double_set() {
297        let result = std::thread::spawn(|| {
298            let (tx1, _rx1) = tokio::sync::mpsc::unbounded_channel();
299            let (tx2, _rx2) = tokio::sync::mpsc::unbounded_channel();
300            set_system_event_sender(tx1);
301            set_system_event_sender(tx2);
302        })
303        .join();
304        assert!(result.is_err());
305    }
306
307    #[rstest]
308    fn test_set_system_command_sender_panics_on_double_set() {
309        let result = std::thread::spawn(|| {
310            let (tx1, _rx1) = tokio::sync::mpsc::unbounded_channel();
311            let (tx2, _rx2) = tokio::sync::mpsc::unbounded_channel();
312            set_system_command_sender(tx1);
313            set_system_command_sender(tx2);
314        })
315        .join();
316        assert!(result.is_err());
317    }
318
319    #[rstest]
320    fn test_try_get_exec_event_sender_returns_none_when_unset() {
321        let result = std::thread::spawn(try_get_exec_event_sender)
322            .join()
323            .unwrap();
324        assert!(result.is_none());
325    }
326
327    #[rstest]
328    fn test_try_get_system_event_sender_returns_none_when_unset() {
329        let result = std::thread::spawn(try_get_system_event_sender)
330            .join()
331            .unwrap();
332        assert!(result.is_none());
333    }
334
335    #[rstest]
336    fn test_try_get_system_command_sender_returns_none_when_unset() {
337        let result = std::thread::spawn(try_get_system_command_sender)
338            .join()
339            .unwrap();
340        assert!(result.is_none());
341    }
342
343    fn assert_sender_replaced<T: Send + 'static, S: SenderChannel<T> + Send + 'static>(
344        replace: fn(tokio::sync::mpsc::UnboundedSender<T>),
345        get: fn() -> S,
346    ) {
347        std::thread::spawn(move || {
348            let (tx1, _rx1) = tokio::sync::mpsc::unbounded_channel();
349            let (tx2, _rx2) = tokio::sync::mpsc::unbounded_channel();
350
351            replace(tx1.clone());
352            replace(tx2.clone());
353            let sender = get();
354
355            assert!(!sender.targets(&tx1));
356            assert!(sender.targets(&tx2));
357        })
358        .join()
359        .expect("sender replacement test thread should join");
360    }
361
362    fn assert_sender_thread_local<T: Send + 'static, S: SenderChannel<T> + Send + 'static>(
363        replace: fn(tokio::sync::mpsc::UnboundedSender<T>),
364        get: fn() -> S,
365    ) {
366        let barrier = Arc::new(Barrier::new(2));
367        let (tx1, _rx1) = tokio::sync::mpsc::unbounded_channel();
368        let (tx2, _rx2) = tokio::sync::mpsc::unbounded_channel();
369        let expected1 = tx1.clone();
370        let expected2 = tx2.clone();
371
372        let barrier1 = Arc::clone(&barrier);
373
374        let thread1 = std::thread::spawn(move || {
375            replace(tx1);
376            barrier1.wait();
377            assert!(get().targets(&expected1));
378        });
379
380        let thread2 = std::thread::spawn(move || {
381            replace(tx2);
382            barrier.wait();
383            assert!(get().targets(&expected2));
384        });
385
386        thread1
387            .join()
388            .expect("first sender isolation test thread should join");
389        thread2
390            .join()
391            .expect("second sender isolation test thread should join");
392    }
393
394    trait SenderChannel<T> {
395        fn targets(&self, sender: &tokio::sync::mpsc::UnboundedSender<T>) -> bool;
396    }
397
398    impl<T> SenderChannel<T> for tokio::sync::mpsc::UnboundedSender<T> {
399        fn targets(&self, sender: &Self) -> bool {
400            self.same_channel(sender)
401        }
402    }
403
404    impl<T> SenderChannel<T> for EventSender<T> {
405        fn targets(&self, sender: &tokio::sync::mpsc::UnboundedSender<T>) -> bool {
406            self.same_channel(&sender.clone().into())
407        }
408    }
409}
nautilus_common/live/runner.rs

nautilus_common/live/
runner.rs
