nautilus_common/messages/system/
queue.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
16use std::{any::Any, fmt::Display};
17
18use nautilus_core::{UUID4, UnixNanos};
19use nautilus_model::identifiers::TraderId;
20
21use crate::runner::SystemChannel;
22
23/// Represents a runner queue pressure condition.
24#[repr(C)]
25#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
26#[cfg_attr(
27    feature = "python",
28    pyo3::pyclass(
29        frozen,
30        eq,
31        eq_int,
32        module = "nautilus_trader.common",
33        from_py_object,
34        rename_all = "SCREAMING_SNAKE_CASE",
35    )
36)]
37#[cfg_attr(
38    feature = "python",
39    pyo3_stub_gen::derive::gen_stub_pyclass_enum(module = "nautilus_trader.common")
40)]
41pub enum QueueCondition {
42    /// The mean dispatch time crossed its configured threshold.
43    Slow,
44    /// The queue depth crossed its configured threshold.
45    Backlogged,
46}
47
48/// Represents the state of a runner queue pressure condition.
49#[repr(C)]
50#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
51#[cfg_attr(
52    feature = "python",
53    pyo3::pyclass(
54        frozen,
55        eq,
56        eq_int,
57        module = "nautilus_trader.common",
58        from_py_object,
59        rename_all = "SCREAMING_SNAKE_CASE",
60    )
61)]
62#[cfg_attr(
63    feature = "python",
64    pyo3_stub_gen::derive::gen_stub_pyclass_enum(module = "nautilus_trader.common")
65)]
66pub enum QueueState {
67    /// The condition crossed its trigger threshold.
68    Triggered,
69    /// The condition crossed its clear threshold.
70    Cleared,
71}
72
73/// Represents an event where a runner queue pressure condition has changed.
74#[repr(C)]
75#[derive(Debug, Clone, PartialEq, Eq)]
76#[cfg_attr(
77    feature = "python",
78    pyo3::pyclass(module = "nautilus_trader.common", from_py_object)
79)]
80#[cfg_attr(
81    feature = "python",
82    pyo3_stub_gen::derive::gen_stub_pyclass(module = "nautilus_trader.common")
83)]
84pub struct QueueStateChanged {
85    /// The trader ID associated with the event.
86    pub trader_id: TraderId,
87    /// The runner channel associated with the condition.
88    pub channel: SystemChannel,
89    /// The queue pressure condition.
90    pub condition: QueueCondition,
91    /// The condition state.
92    pub state: QueueState,
93    /// The queue depth at the state transition.
94    pub queue_depth: usize,
95    /// The mean dispatch time per message at the state transition, in nanoseconds.
96    pub mean_dispatch_ns: u64,
97    /// The event ID.
98    pub event_id: UUID4,
99    /// UNIX timestamp (nanoseconds) when the event occurred.
100    pub ts_event: UnixNanos,
101    /// UNIX timestamp (nanoseconds) when the instance was initialized.
102    pub ts_init: UnixNanos,
103}
104
105impl QueueStateChanged {
106    /// Creates a new [`QueueStateChanged`] instance.
107    #[expect(clippy::too_many_arguments)]
108    #[must_use]
109    pub const fn new(
110        trader_id: TraderId,
111        channel: SystemChannel,
112        condition: QueueCondition,
113        state: QueueState,
114        queue_depth: usize,
115        mean_dispatch_ns: u64,
116        event_id: UUID4,
117        ts_event: UnixNanos,
118        ts_init: UnixNanos,
119    ) -> Self {
120        Self {
121            trader_id,
122            channel,
123            condition,
124            state,
125            queue_depth,
126            mean_dispatch_ns,
127            event_id,
128            ts_event,
129            ts_init,
130        }
131    }
132
133    pub fn as_any(&self) -> &dyn Any {
134        self
135    }
136}
137
138impl Display for QueueStateChanged {
139    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
140        write!(
141            f,
142            "{}(trader_id={}, channel={:?}, condition={:?}, state={:?}, queue_depth={}, mean_dispatch_ns={}, event_id={})",
143            stringify!(QueueStateChanged),
144            self.trader_id,
145            self.channel,
146            self.condition,
147            self.state,
148            self.queue_depth,
149            self.mean_dispatch_ns,
150            self.event_id,
151        )
152    }
153}
154
155#[cfg(test)]
156#[allow(
157    clippy::too_many_arguments,
158    reason = "constructor cases vary all fields except the fixed trader ID"
159)]
160mod tests {
161    use rstest::rstest;
162
163    use super::*;
164
165    #[rstest]
166    #[case(
167        SystemChannel::DataEvents,
168        QueueCondition::Slow,
169        QueueState::Triggered,
170        17,
171        23,
172        "00000000-0000-4000-8000-000000000001",
173        29,
174        31
175    )]
176    #[case(
177        SystemChannel::DataCommands,
178        QueueCondition::Backlogged,
179        QueueState::Cleared,
180        37,
181        41,
182        "00000000-0000-4000-8000-000000000002",
183        43,
184        47
185    )]
186    fn test_queue_state_changed_new_assigns_all_fields(
187        #[case] channel: SystemChannel,
188        #[case] condition: QueueCondition,
189        #[case] state: QueueState,
190        #[case] queue_depth: usize,
191        #[case] mean_dispatch_ns: u64,
192        #[case] event_id: &str,
193        #[case] ts_event: u64,
194        #[case] ts_init: u64,
195    ) {
196        let trader_id = TraderId::from("TRADER-001");
197        let event_id = UUID4::from(event_id);
198        let ts_event = UnixNanos::from(ts_event);
199        let ts_init = UnixNanos::from(ts_init);
200
201        let event = QueueStateChanged::new(
202            trader_id,
203            channel,
204            condition,
205            state,
206            queue_depth,
207            mean_dispatch_ns,
208            event_id,
209            ts_event,
210            ts_init,
211        );
212
213        assert_eq!(event.trader_id, trader_id);
214        assert_eq!(event.channel, channel);
215        assert_eq!(event.condition, condition);
216        assert_eq!(event.state, state);
217        assert_eq!(event.queue_depth, queue_depth);
218        assert_eq!(event.mean_dispatch_ns, mean_dispatch_ns);
219        assert_eq!(event.event_id, event_id);
220        assert_eq!(event.ts_event, ts_event);
221        assert_eq!(event.ts_init, ts_init);
222    }
223}
nautilus_common/messages/system/queue.rs

nautilus_common/messages/system/
queue.rs
