nautilus_persistence/writer/
subscription.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//! Message-bus subscriptions for streaming writer sinks.
17
18use std::{
19    any::Any,
20    cell::RefCell,
21    fmt::Debug,
22    rc::Rc,
23    sync::{
24        Arc,
25        atomic::{AtomicU64, Ordering},
26    },
27};
28
29use nautilus_common::{
30    clock::Clock,
31    msgbus::{
32        MStr, ShareableMessageHandler, TypedHandler, subscribe_account_state, subscribe_any,
33        subscribe_bars, subscribe_book_deltas, subscribe_book_depth as subscribe_book_depths,
34        subscribe_funding_rates, subscribe_greeks, subscribe_index_prices, subscribe_instruments,
35        subscribe_mark_prices, subscribe_option_greeks, subscribe_order_events,
36        subscribe_position_events, subscribe_quotes, subscribe_trades, unsubscribe_account_state,
37        unsubscribe_any, unsubscribe_bars, unsubscribe_book_deltas,
38        unsubscribe_book_depth as unsubscribe_book_depths, unsubscribe_funding_rates,
39        unsubscribe_greeks, unsubscribe_index_prices, unsubscribe_instruments,
40        unsubscribe_mark_prices, unsubscribe_option_greeks, unsubscribe_order_events,
41        unsubscribe_position_events, unsubscribe_quotes, unsubscribe_trades,
42    },
43};
44use nautilus_model::{
45    data::{
46        Bar, CustomData, FundingRateUpdate, IndexPriceUpdate, MarkPriceUpdate, OptionGreeks,
47        OrderBookDeltas, OrderBookDepth, QuoteTick, TradeTick, greeks::GreeksData,
48    },
49    events::{AccountState, OrderEventAny, PositionEvent},
50    instruments::{Instrument, InstrumentAny},
51};
52
53use super::traits::StreamingDataSink;
54
55type ClockBridge = Option<(Rc<RefCell<dyn Clock>>, Arc<AtomicU64>)>;
56
57/// Message-bus subscriptions forwarding supported messages into a streaming sink.
58pub struct StreamingSinkSubscription {
59    sink: Rc<RefCell<StreamingDataSink>>,
60    clock_bridge: ClockBridge,
61    any_handler: ShareableMessageHandler,
62    quotes_handler: TypedHandler<QuoteTick>,
63    trades_handler: TypedHandler<TradeTick>,
64    bars_handler: TypedHandler<Bar>,
65    deltas_handler: TypedHandler<OrderBookDeltas>,
66    depths_handler: TypedHandler<OrderBookDepth>,
67    mark_prices_handler: TypedHandler<MarkPriceUpdate>,
68    index_prices_handler: TypedHandler<IndexPriceUpdate>,
69    funding_rates_handler: TypedHandler<FundingRateUpdate>,
70    option_greeks_handler: TypedHandler<OptionGreeks>,
71    greeks_handler: TypedHandler<GreeksData>,
72    instruments_handler: TypedHandler<InstrumentAny>,
73    account_state_handler: TypedHandler<AccountState>,
74    order_events_handler: TypedHandler<OrderEventAny>,
75    position_events_handler: TypedHandler<PositionEvent>,
76}
77
78impl Debug for StreamingSinkSubscription {
79    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
80        f.debug_struct(stringify!(StreamingSinkSubscription))
81            .finish_non_exhaustive()
82    }
83}
84
85impl StreamingSinkSubscription {
86    /// Subscribes a streaming sink to typed and dynamic message-bus routes.
87    pub fn subscribe(sink: Rc<RefCell<StreamingDataSink>>, clock_bridge: ClockBridge) -> Self {
88        Self::subscribe_named(sink, clock_bridge, "streaming writer".to_string())
89    }
90
91    /// Subscribes a sink and includes its name in write-failure diagnostics.
92    pub fn subscribe_named(
93        sink: Rc<RefCell<StreamingDataSink>>,
94        clock_bridge: ClockBridge,
95        name: String,
96    ) -> Self {
97        let name: Rc<str> = Rc::from(name);
98        let any_handler = any_handler(Rc::clone(&sink), clock_bridge.clone(), name.clone());
99        let quotes_handler =
100            typed_handler::<QuoteTick>(Rc::clone(&sink), clock_bridge.clone(), name.clone());
101        let trades_handler =
102            typed_handler::<TradeTick>(Rc::clone(&sink), clock_bridge.clone(), name.clone());
103        let bars_handler =
104            typed_handler::<Bar>(Rc::clone(&sink), clock_bridge.clone(), name.clone());
105        let deltas_handler =
106            typed_handler::<OrderBookDeltas>(Rc::clone(&sink), clock_bridge.clone(), name.clone());
107        let depths_handler =
108            typed_handler::<OrderBookDepth>(Rc::clone(&sink), clock_bridge.clone(), name.clone());
109        let mark_prices_handler =
110            typed_handler::<MarkPriceUpdate>(Rc::clone(&sink), clock_bridge.clone(), name.clone());
111        let index_prices_handler =
112            typed_handler::<IndexPriceUpdate>(Rc::clone(&sink), clock_bridge.clone(), name.clone());
113        let funding_rates_handler = typed_handler::<FundingRateUpdate>(
114            Rc::clone(&sink),
115            clock_bridge.clone(),
116            name.clone(),
117        );
118        let option_greeks_handler =
119            typed_handler::<OptionGreeks>(Rc::clone(&sink), clock_bridge.clone(), name.clone());
120        let greeks_handler =
121            typed_handler::<GreeksData>(Rc::clone(&sink), clock_bridge.clone(), name.clone());
122        let instruments_handler =
123            typed_handler::<InstrumentAny>(Rc::clone(&sink), clock_bridge.clone(), name.clone());
124        let account_state_handler =
125            typed_handler::<AccountState>(Rc::clone(&sink), clock_bridge.clone(), name.clone());
126        let order_events_handler =
127            typed_handler::<OrderEventAny>(Rc::clone(&sink), clock_bridge.clone(), name.clone());
128        let position_events_handler =
129            typed_handler::<PositionEvent>(Rc::clone(&sink), clock_bridge.clone(), name);
130        let pattern = MStr::pattern("*");
131        // Order events publish once on `events.order.{strategy_id}` and fills re-publish on
132        // `events.order_filled.{instrument_id}`; a `*` pattern captures both and duplicates
133        // every fill row, so the sink subscribes to the strategy-scoped topic only.
134        let order_events_pattern = MStr::pattern("events.order.*");
135
136        subscribe_any(pattern, any_handler.clone(), None);
137        subscribe_quotes(pattern, quotes_handler.clone(), None);
138        subscribe_trades(pattern, trades_handler.clone(), None);
139        subscribe_bars(pattern, bars_handler.clone(), None);
140        subscribe_book_deltas(pattern, deltas_handler.clone(), None);
141        subscribe_book_depths(pattern, depths_handler.clone(), None);
142        subscribe_mark_prices(pattern, mark_prices_handler.clone(), None);
143        subscribe_index_prices(pattern, index_prices_handler.clone(), None);
144        subscribe_funding_rates(pattern, funding_rates_handler.clone(), None);
145        subscribe_option_greeks(pattern, option_greeks_handler.clone(), None);
146        subscribe_greeks(pattern, greeks_handler.clone(), None);
147        subscribe_instruments(pattern, instruments_handler.clone(), None);
148        subscribe_account_state(pattern, account_state_handler.clone(), None);
149        subscribe_order_events(order_events_pattern, order_events_handler.clone(), None);
150        subscribe_position_events(pattern, position_events_handler.clone(), None);
151
152        Self {
153            sink,
154            clock_bridge,
155            any_handler,
156            quotes_handler,
157            trades_handler,
158            bars_handler,
159            deltas_handler,
160            depths_handler,
161            mark_prices_handler,
162            index_prices_handler,
163            funding_rates_handler,
164            option_greeks_handler,
165            greeks_handler,
166            instruments_handler,
167            account_state_handler,
168            order_events_handler,
169            position_events_handler,
170        }
171    }
172
173    /// Flushes the sink without removing its message-bus subscriptions.
174    ///
175    /// # Errors
176    ///
177    /// Returns an error if the underlying sink cannot be flushed.
178    pub fn flush(&self) -> anyhow::Result<()> {
179        refresh_writer_clock(&self.clock_bridge);
180        self.sink.borrow_mut().flush()
181    }
182
183    /// Unsubscribes and closes the sink.
184    /// # Errors
185    ///
186    /// Returns an error if the underlying sink cannot be closed.
187    pub fn close(&self) -> anyhow::Result<()> {
188        self.unsubscribe();
189        refresh_writer_clock(&self.clock_bridge);
190        self.sink.borrow_mut().close()
191    }
192
193    /// Removes all message-bus subscriptions without closing the sink.
194    pub fn unsubscribe(&self) {
195        let pattern = MStr::pattern("*");
196        let order_events_pattern = MStr::pattern("events.order.*");
197
198        unsubscribe_any(pattern, &self.any_handler);
199        unsubscribe_quotes(pattern, &self.quotes_handler);
200        unsubscribe_trades(pattern, &self.trades_handler);
201        unsubscribe_bars(pattern, &self.bars_handler);
202        unsubscribe_book_deltas(pattern, &self.deltas_handler);
203        unsubscribe_book_depths(pattern, &self.depths_handler);
204        unsubscribe_mark_prices(pattern, &self.mark_prices_handler);
205        unsubscribe_index_prices(pattern, &self.index_prices_handler);
206        unsubscribe_funding_rates(pattern, &self.funding_rates_handler);
207        unsubscribe_option_greeks(pattern, &self.option_greeks_handler);
208        unsubscribe_greeks(pattern, &self.greeks_handler);
209        unsubscribe_instruments(pattern, &self.instruments_handler);
210        unsubscribe_account_state(pattern, &self.account_state_handler);
211        unsubscribe_order_events(order_events_pattern, &self.order_events_handler);
212        unsubscribe_position_events(pattern, &self.position_events_handler);
213    }
214}
215
216fn typed_handler<T: 'static>(
217    sink: Rc<RefCell<StreamingDataSink>>,
218    clock_bridge: ClockBridge,
219    name: Rc<str>,
220) -> TypedHandler<T> {
221    TypedHandler::from(move |message: &T| {
222        write_bus_message(&sink, &clock_bridge, &name, message);
223    })
224}
225
226fn any_handler(
227    sink: Rc<RefCell<StreamingDataSink>>,
228    clock_bridge: ClockBridge,
229    name: Rc<str>,
230) -> ShareableMessageHandler {
231    ShareableMessageHandler::from_any(move |message: &dyn Any| {
232        write_bus_message(&sink, &clock_bridge, &name, message);
233    })
234}
235
236fn write_bus_message(
237    sink: &Rc<RefCell<StreamingDataSink>>,
238    clock_bridge: &ClockBridge,
239    name: &str,
240    message: &dyn Any,
241) {
242    refresh_writer_clock(clock_bridge);
243
244    if let Err(e) = sink.borrow_mut().write_any(message) {
245        log::warn!(
246            "Failed to write {name} {} message: {e}",
247            streaming_message_label(message)
248        );
249    }
250}
251
252fn refresh_writer_clock(clock_bridge: &ClockBridge) {
253    if let Some((clock, shared)) = clock_bridge {
254        shared.store(clock.borrow().timestamp_ns().as_u64(), Ordering::Relaxed);
255    }
256}
257
258fn streaming_message_label(message: &dyn Any) -> String {
259    if let Some(custom) = message.downcast_ref::<CustomData>() {
260        return format!(
261            "CustomData({}, identifier={:?})",
262            custom.data.type_name(),
263            custom.data_type.identifier()
264        );
265    }
266
267    if let Some(instrument) = message.downcast_ref::<InstrumentAny>() {
268        return format!("InstrumentAny({})", instrument.id());
269    }
270
271    if let Some(bar) = message.downcast_ref::<Bar>() {
272        return format!("Bar({})", bar.bar_type);
273    }
274
275    if let Some(quotes) = message.downcast_ref::<QuoteTick>() {
276        return format!("QuoteTick({})", quotes.instrument_id);
277    }
278
279    if let Some(trade) = message.downcast_ref::<TradeTick>() {
280        return format!("TradeTick({})", trade.instrument_id);
281    }
282
283    "unknown".to_string()
284}
nautilus_persistence/writer/subscription.rs

nautilus_persistence/writer/
subscription.rs
