brokapi_sansio/harness.rs
1//! Deterministic replay harness for recorded frames.
2
3use std::time::{Duration, Instant};
4
5use crate::core::BrokerCore;
6use crate::event::{CoreEvent, EventSink};
7use crate::outbound::Outbound;
8use crate::subscription::SubscriptionSet;
9
10/// One inbound frame of a recorded session, with the instant it arrived.
11///
12/// A session is a list of these. Replay it with [`Harness::run_frames`].
13#[derive(Debug, Clone)]
14pub struct FrameRecord<F> {
15 /// When the frame arrived, on the harness's virtual clock.
16 pub at: Instant,
17 /// The frame, in the core's [`BrokerCore::Frame`] type.
18 pub frame: F,
19}
20
21impl<F> FrameRecord<F> {
22 /// Creates a record of `frame` arriving at `at`.
23 pub fn new(at: Instant, frame: F) -> Self {
24 Self { at, frame }
25 }
26}
27
28/// Drives a [`BrokerCore`] with virtual time and records what comes out.
29///
30/// The harness owns the core and a clock that only moves when you move it. Feed frames, move time,
31/// then read the events the core emitted ([`events`](Self::events)) and the messages it wanted to
32/// send ([`transmitted`](Self::transmitted)). There is no socket and no real waiting, so a test of
33/// a minute of silence takes no time.
34///
35/// After every call into the core the harness drains [`BrokerCore::poll_transmit`] into the
36/// `transmitted` log, and it fires the core's timers whenever time passes their deadline.
37///
38/// # Examples
39///
40/// ```
41/// # use std::time::{Duration, Instant};
42/// # use brokapi_core::{Event, InstrumentId, InstrumentKey, NativeId, ResyncReason, SubscriptionId};
43/// # use brokapi_sansio::{BrokerCore, EventSink, Outbound, SubscriptionSet};
44/// # struct WatchdogCore { silent_since: Instant, armed: bool, pongs: Vec<String> }
45/// # impl BrokerCore for WatchdogCore {
46/// # type Frame = String;
47/// # fn on_frame(&mut self, frame: String, now: Instant, _out: &mut EventSink) {
48/// # self.silent_since = now;
49/// # self.armed = true;
50/// # self.pongs.push(format!("pong {frame}"));
51/// # }
52/// # fn poll_transmit(&mut self) -> Option<Outbound> {
53/// # if self.pongs.is_empty() { None } else { Some(Outbound::WsText(self.pongs.remove(0))) }
54/// # }
55/// # fn poll_timeout(&self) -> Option<Instant> {
56/// # self.armed.then(|| self.silent_since + Duration::from_secs(10))
57/// # }
58/// # fn on_timeout(&mut self, _now: Instant, out: &mut EventSink) {
59/// # self.armed = false;
60/// # let instrument = InstrumentId::new(
61/// # InstrumentKey::new("SBER", "TQBR"),
62/// # NativeId::TickerClass { ticker: "SBER".into(), class_code: "TQBR".into() },
63/// # );
64/// # out.push_payload(Some(SubscriptionId::new(1)), Event::Resync { sub: SubscriptionId::new(1), instrument, reason: ResyncReason::SubscriptionReset });
65/// # }
66/// # fn apply(&mut self, _desired: &SubscriptionSet) {}
67/// # }
68/// # let t0 = Instant::now();
69/// # let core = WatchdogCore { silent_since: t0, armed: false, pongs: Vec::new() };
70/// use brokapi_sansio::Harness;
71///
72/// let mut harness = Harness::new(core, t0);
73///
74/// // A frame arrives: the core answers with a pong.
75/// harness.feed_frame("ping".to_string());
76/// assert_eq!(harness.transmitted(), [Outbound::WsText("pong ping".to_string())]);
77///
78/// // 9 seconds of silence: the watchdog has not fired yet. 2 more: it has.
79/// harness.advance_by(Duration::from_secs(9));
80/// assert!(harness.events().is_empty());
81/// harness.advance_by(Duration::from_secs(2));
82/// let events = harness.take_events();
83/// assert_eq!(events.len(), 1);
84/// assert!(matches!(events[0].payload, Event::Resync { .. }));
85/// assert!(harness.events().is_empty());
86/// ```
87pub struct Harness<C: BrokerCore> {
88 core: C,
89 now: Instant,
90 events: Vec<CoreEvent>,
91 transmitted: Vec<Outbound>,
92}
93
94impl<C: BrokerCore> Harness<C> {
95 /// Wraps `core`. The virtual clock starts at `start`.
96 pub fn new(core: C, start: Instant) -> Self {
97 Self {
98 core,
99 now: start,
100 events: Vec::new(),
101 transmitted: Vec::new(),
102 }
103 }
104
105 /// The current virtual time.
106 pub fn now(&self) -> Instant {
107 self.now
108 }
109
110 /// The core, for reading its state.
111 pub fn core(&self) -> &C {
112 &self.core
113 }
114
115 /// The core, for changing it directly. Messages it queues this way are collected on the next
116 /// call that drains transmits, or by [`drain_transmit`](Self::drain_transmit).
117 pub fn core_mut(&mut self) -> &mut C {
118 &mut self.core
119 }
120
121 /// All events the core emitted since the last [`take_events`](Self::take_events), oldest first.
122 pub fn events(&self) -> &[CoreEvent] {
123 &self.events
124 }
125
126 /// The log of every message the core wanted sent, oldest first. It is never cleared.
127 pub fn transmitted(&self) -> &[Outbound] {
128 &self.transmitted
129 }
130
131 /// Tells the core the wanted subscriptions ([`BrokerCore::apply`]) and collects what it queues
132 /// to send.
133 pub fn apply(&mut self, desired: &SubscriptionSet) {
134 self.core.apply(desired);
135 self.drain_transmit();
136 }
137
138 /// Delivers a frame to the core at the current virtual time, and collects its events and
139 /// messages. Time does not move.
140 pub fn feed_frame(&mut self, frame: C::Frame) {
141 let mut sink = EventSink::new();
142 self.core.on_frame(frame, self.now, &mut sink);
143 self.events.extend(sink.drain());
144 self.drain_transmit();
145 }
146
147 /// Moves time to `at` (firing due timers on the way, see [`advance_to`](Self::advance_to)),
148 /// then delivers the frame.
149 pub fn feed_frame_at(&mut self, at: Instant, frame: C::Frame) {
150 self.advance_to(at);
151 self.feed_frame(frame);
152 }
153
154 /// Moves time forward by `delta`, firing every timer that falls due.
155 pub fn advance_by(&mut self, delta: Duration) {
156 self.advance_to(self.now + delta);
157 }
158
159 /// Moves time to `at`, firing every timer that falls due on the way.
160 ///
161 /// While the core's [`poll_timeout`](BrokerCore::poll_timeout) is at or before `at`, the
162 /// harness calls [`on_timeout`](BrokerCore::on_timeout) with that deadline as the time, then
163 /// asks again. So one call can fire several timers, each at its own time.
164 ///
165 /// A core whose `on_timeout` leaves the same deadline in place makes this loop for ever. Move
166 /// or clear the deadline in `on_timeout`.
167 ///
168 /// `at` should not be earlier than [`now`](Self::now): the clock is set to `at` without a check.
169 pub fn advance_to(&mut self, at: Instant) {
170 while let Some(deadline) = self.core.poll_timeout() {
171 if deadline > at {
172 break;
173 }
174 let mut sink = EventSink::new();
175 self.core.on_timeout(deadline, &mut sink);
176 self.events.extend(sink.drain());
177 self.drain_transmit();
178 self.now = deadline;
179 }
180 self.now = at;
181 }
182
183 /// Replays a recorded session: each frame is delivered at its recorded time with
184 /// [`feed_frame_at`](Self::feed_frame_at). Records must be in time order.
185 ///
186 /// # Examples
187 ///
188 /// ```
189 /// # use std::time::{Duration, Instant};
190 /// # use brokapi_core::{Event, InstrumentId, InstrumentKey, NativeId, ResyncReason, SubscriptionId};
191 /// # use brokapi_sansio::{BrokerCore, EventSink, Outbound, SubscriptionSet};
192 /// # struct WatchdogCore { silent_since: Instant, armed: bool, pongs: Vec<String> }
193 /// # impl BrokerCore for WatchdogCore {
194 /// # type Frame = String;
195 /// # fn on_frame(&mut self, frame: String, now: Instant, _out: &mut EventSink) {
196 /// # self.silent_since = now;
197 /// # self.armed = true;
198 /// # self.pongs.push(format!("pong {frame}"));
199 /// # }
200 /// # fn poll_transmit(&mut self) -> Option<Outbound> {
201 /// # if self.pongs.is_empty() { None } else { Some(Outbound::WsText(self.pongs.remove(0))) }
202 /// # }
203 /// # fn poll_timeout(&self) -> Option<Instant> {
204 /// # self.armed.then(|| self.silent_since + Duration::from_secs(10))
205 /// # }
206 /// # fn on_timeout(&mut self, _now: Instant, out: &mut EventSink) {
207 /// # self.armed = false;
208 /// # let instrument = InstrumentId::new(
209 /// # InstrumentKey::new("SBER", "TQBR"),
210 /// # NativeId::TickerClass { ticker: "SBER".into(), class_code: "TQBR".into() },
211 /// # );
212 /// # out.push_payload(Some(SubscriptionId::new(1)), Event::Resync { sub: SubscriptionId::new(1), instrument, reason: ResyncReason::SubscriptionReset });
213 /// # }
214 /// # fn apply(&mut self, _desired: &SubscriptionSet) {}
215 /// # }
216 /// # let t0 = Instant::now();
217 /// # let core = WatchdogCore { silent_since: t0, armed: false, pongs: Vec::new() };
218 /// use brokapi_sansio::{FrameRecord, Harness};
219 ///
220 /// let mut harness = Harness::new(core, t0);
221 /// let session = [
222 /// FrameRecord::new(t0 + Duration::from_secs(1), "a".to_string()),
223 /// FrameRecord::new(t0 + Duration::from_secs(2), "b".to_string()),
224 /// ];
225 /// harness.run_frames(&session);
226 ///
227 /// assert_eq!(harness.transmitted().len(), 2);
228 /// assert_eq!(harness.now(), t0 + Duration::from_secs(2));
229 /// ```
230 pub fn run_frames(&mut self, frames: &[FrameRecord<C::Frame>])
231 where
232 C::Frame: Clone,
233 {
234 for record in frames {
235 self.feed_frame_at(record.at, record.frame.clone());
236 }
237 }
238
239 /// Polls the core for messages it wants sent and returns the ones found by this call.
240 ///
241 /// Every other harness method already does this and logs the result in
242 /// [`transmitted`](Self::transmitted), so the return value is usually empty. Call it after
243 /// changing the core through [`core_mut`](Self::core_mut). What it returns is also appended to
244 /// the log.
245 pub fn drain_transmit(&mut self) -> Vec<Outbound> {
246 let mut out = Vec::new();
247 while let Some(msg) = self.core.poll_transmit() {
248 out.push(msg);
249 }
250 self.transmitted.extend(out.iter().cloned());
251 out
252 }
253
254 /// Returns all collected events, oldest first, and empties the list.
255 pub fn take_events(&mut self) -> Vec<CoreEvent> {
256 std::mem::take(&mut self.events)
257 }
258}
259
260#[cfg(test)]
261mod tests {
262 use super::*;
263 use crate::event::EventSink;
264 use brokapi_core::Event;
265
266 struct EchoCore {
267 pending: Vec<String>,
268 }
269
270 impl EchoCore {
271 fn new() -> Self {
272 Self {
273 pending: Vec::new(),
274 }
275 }
276 }
277
278 impl BrokerCore for EchoCore {
279 type Frame = String;
280
281 fn on_frame(&mut self, frame: Self::Frame, _now: Instant, out: &mut EventSink) {
282 out.push_payload(
283 None,
284 Event::Resync {
285 sub: brokapi_core::SubscriptionId::new(0),
286 instrument: brokapi_core::InstrumentId::new(
287 brokapi_core::InstrumentKey::new("SBER", "TQBR"),
288 brokapi_core::NativeId::TickerClass {
289 ticker: "SBER".into(),
290 class_code: "TQBR".into(),
291 },
292 ),
293 reason: brokapi_core::ResyncReason::Manual,
294 },
295 );
296 self.pending.push(frame);
297 }
298
299 fn poll_transmit(&mut self) -> Option<Outbound> {
300 self.pending.pop().map(Outbound::WsText)
301 }
302
303 fn poll_timeout(&self) -> Option<Instant> {
304 None
305 }
306
307 fn on_timeout(&mut self, _now: Instant, _out: &mut EventSink) {}
308
309 fn apply(&mut self, _desired: &SubscriptionSet) {}
310 }
311
312 #[test]
313 fn harness_replays_frame_and_transmit() {
314 let start = Instant::now();
315 let mut h = Harness::new(EchoCore::new(), start);
316 h.feed_frame("ping".into());
317 assert_eq!(h.events().len(), 1);
318 assert_eq!(h.transmitted().len(), 1);
319 }
320}