nautilus_common/live/
dispatch.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//! Callback context propagation for live command and event dispatch.
17
18use std::thread::{self, ThreadId};
19
20use crate::{
21    actor::{ChainContext, SendChainContext, collect_command_contexts},
22    runner::{TradingCommandMessage, dispatch_scoped_trading_command},
23};
24
25/// A live message carrying an owner-thread callback context through a send-safe channel.
26///
27/// Conversion with `From<T>` creates independent ingress without capturing a callback root.
28/// External-thread sends and sends without an active root are also independent ingress.
29/// Rooted messages must be processed on
30/// their owner's thread. Foreign-thread destruction releases the root at the next owner
31/// channel boundary, callback quiescence check, or dispatcher teardown.
32#[derive(Debug)]
33pub struct DispatchMessage<T> {
34    message: Option<T>,
35    context: Option<SendChainContext>,
36}
37
38impl<T> DispatchMessage<T> {
39    /// Wraps a message, capturing callback ancestry only when called on `owner`.
40    ///
41    /// `owner` is the runtime thread that processes messages from this channel.
42    ///
43    /// # Panics
44    ///
45    /// Panics if the owner exhausts its channel context IDs.
46    #[must_use]
47    pub fn new(message: T, owner: ThreadId) -> Self {
48        let context = (owner == thread::current().id())
49            .then(SendChainContext::capture)
50            .flatten();
51
52        Self {
53            message: Some(message),
54            context,
55        }
56    }
57
58    /// Processes the message under its originating callback context.
59    ///
60    /// # Panics
61    ///
62    /// Panics if a rooted message is processed outside its owner's thread.
63    pub fn dispatch<R>(mut self, run: impl FnOnce(T) -> R) -> R {
64        let context = self.take_context();
65        context.with_chain(|| run(self.message.take().expect("message is present")))
66    }
67
68    /// Returns whether this message retains an originating callback root.
69    #[must_use]
70    pub fn is_rooted(&self) -> bool {
71        self.context.is_some()
72    }
73
74    fn take_context(&self) -> ChainContext {
75        collect_command_contexts();
76        self.context
77            .as_ref()
78            .map_or_else(ChainContext::independent, |context| {
79                context.take().expect("channel context is present")
80            })
81    }
82}
83
84impl DispatchMessage<TradingCommandMessage> {
85    /// Dispatches the command and its deferred children under their captured callback roots.
86    ///
87    /// Calls `before` immediately before each endpoint dispatch, including child commands.
88    ///
89    /// # Panics
90    ///
91    /// Panics if:
92    /// - A rooted message is processed outside its owner thread.
93    /// - The observer or a command handler panics.
94    pub fn dispatch_trading(mut self, before: impl FnMut(&TradingCommandMessage)) {
95        let context = self.take_context();
96        dispatch_scoped_trading_command(
97            self.message.take().expect("message is present"),
98            context,
99            before,
100        );
101    }
102}
103
104impl<T: std::fmt::Display> std::fmt::Display for DispatchMessage<T> {
105    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
106        self.message.as_ref().expect("message is present").fmt(f)
107    }
108}
109
110impl<T> From<T> for DispatchMessage<T> {
111    fn from(message: T) -> Self {
112        Self {
113            message: Some(message),
114            context: None,
115        }
116    }
117}
118
119impl<T> Drop for DispatchMessage<T> {
120    fn drop(&mut self) {
121        if self.message.is_none() {
122            return;
123        }
124
125        let context = self
126            .context
127            .as_ref()
128            .filter(|context| context.is_owner())
129            .and_then(SendChainContext::take)
130            .unwrap_or_else(ChainContext::independent);
131        context.with_chain(|| drop(self.message.take()));
132    }
133}
nautilus_common/live/dispatch.rs

nautilus_common/live/
dispatch.rs
