brokapi_sansio/
core.rs

1//! Sans-io broker core trait.
2
3use std::time::Instant;
4
5use crate::event::EventSink;
6use crate::outbound::Outbound;
7use crate::subscription::SubscriptionSet;
8
9/// Protocol logic of one broker stream, written without sockets, timers or an async runtime.
10///
11/// This is the "sans-io" pattern. A core is a plain state machine. The code that owns the socket
12/// feeds it inbound frames and the current time, and collects what it wants to say: events for the
13/// consumer ([`EventSink`]), messages to put on the wire ([`poll_transmit`](Self::poll_transmit))
14/// and the next time it wants to be woken ([`poll_timeout`](Self::poll_timeout)). Because time is
15/// an argument and there is no I/O, a core can be tested with recorded frames and virtual time, see
16/// [`Harness`](crate::Harness).
17///
18/// How [`Harness`](crate::Harness) drives a core, and what a real driver should do the same way:
19///
20/// - call [`on_frame`](Self::on_frame) for each inbound frame, with the time it arrived;
21/// - after every call into the core, drain [`poll_transmit`](Self::poll_transmit) until it returns
22///   `None` and send the messages;
23/// - sleep until [`poll_timeout`](Self::poll_timeout), then call [`on_timeout`](Self::on_timeout);
24/// - call [`apply`](Self::apply) when the set of wanted subscriptions changes.
25///
26/// A core must never block, sleep or read the clock itself. Use the `now` it is given.
27///
28/// # Examples
29///
30/// A core that answers every frame and reports a resync after 10 seconds of silence, driven with
31/// virtual time:
32///
33/// ```
34/// use std::time::{Duration, Instant};
35/// use brokapi_core::{Event, InstrumentId, InstrumentKey, NativeId, ResyncReason, SubscriptionId};
36/// use brokapi_sansio::{BrokerCore, EventSink, Harness, Outbound, SubscriptionSet};
37///
38/// /// Answers every frame with a pong. If the line is silent for 10 seconds it reports a resync.
39/// struct WatchdogCore {
40///     silent_since: Instant,
41///     armed: bool,
42///     pongs: Vec<String>,
43/// }
44///
45/// impl BrokerCore for WatchdogCore {
46///     type Frame = String;
47///
48///     fn on_frame(&mut self, frame: String, now: Instant, _out: &mut EventSink) {
49///         self.silent_since = now;
50///         self.armed = true;
51///         self.pongs.push(format!("pong {frame}"));
52///     }
53///
54///     fn poll_transmit(&mut self) -> Option<Outbound> {
55///         if self.pongs.is_empty() { None } else { Some(Outbound::WsText(self.pongs.remove(0))) }
56///     }
57///
58///     fn poll_timeout(&self) -> Option<Instant> {
59///         self.armed.then(|| self.silent_since + Duration::from_secs(10))
60///     }
61///
62///     fn on_timeout(&mut self, _now: Instant, out: &mut EventSink) {
63///         // Disarm first: a core that keeps the same deadline would be called again for ever.
64///         self.armed = false;
65///         let instrument = InstrumentId::new(
66///             InstrumentKey::new("SBER", "TQBR"),
67///             NativeId::TickerClass { ticker: "SBER".into(), class_code: "TQBR".into() },
68///         );
69///         out.push_payload(
70///             Some(SubscriptionId::new(1)),
71///             Event::Resync { sub: SubscriptionId::new(1), instrument, reason: ResyncReason::SubscriptionReset },
72///         );
73///     }
74///
75///     fn apply(&mut self, _desired: &SubscriptionSet) {}
76/// }
77///
78/// let t0 = Instant::now();
79/// let core = WatchdogCore { silent_since: t0, armed: false, pongs: Vec::new() };
80/// let mut harness = Harness::new(core, t0);
81///
82/// harness.feed_frame("ping".to_string());
83/// assert_eq!(harness.transmitted(), [Outbound::WsText("pong ping".to_string())]);
84///
85/// harness.advance_by(Duration::from_secs(9));
86/// assert!(harness.events().is_empty());
87/// harness.advance_by(Duration::from_secs(2));
88/// assert_eq!(harness.events().len(), 1);
89/// ```
90pub trait BrokerCore {
91    /// The inbound frame type: for example `String` for WebSocket text, or a decoded message.
92    type Frame;
93
94    /// Handles one inbound transport frame received at `now`.
95    ///
96    /// Events for the consumer are pushed into `out`. Messages for the broker are queued inside the
97    /// core and picked up by [`poll_transmit`](Self::poll_transmit). The method does not return an
98    /// error: a frame the core cannot use is its own business (skip it, or emit an event).
99    fn on_frame(&mut self, frame: Self::Frame, now: Instant, out: &mut EventSink);
100
101    /// Takes the next outbound wire message, if any. Call it until it returns `None`.
102    fn poll_transmit(&mut self) -> Option<Outbound>;
103
104    /// When the core wants [`on_timeout`](Self::on_timeout) to be called next, or `None` if it has
105    /// no timer pending.
106    ///
107    /// After `on_timeout` the core must move this deadline forward or clear it. A core that
108    /// returns the same past deadline again makes a driver loop for ever.
109    fn poll_timeout(&self) -> Option<Instant>;
110
111    /// Handles a timer that fired. `now` is the time the timer was due.
112    fn on_timeout(&mut self, now: Instant, out: &mut EventSink);
113
114    /// Tells the core the full set of subscriptions the consumer wants now.
115    ///
116    /// The argument is the whole target state, not a change. The core works out what to subscribe
117    /// and unsubscribe and queues the messages for [`poll_transmit`](Self::poll_transmit).
118    fn apply(&mut self, desired: &SubscriptionSet);
119}
brokapi_sansio/core.rs

brokapi_sansio/
core.rs
