brokapi_sansio/
event.rs

1//! Normalized events emitted by sans-io cores.
2
3use brokapi_core::{Event, SubscriptionId};
4
5/// An event produced by a sans-io core, before it is wrapped in an [`Envelope`](brokapi_core::Envelope).
6///
7/// The connector's emitter (`brokapi_runtime::Emitter`) adds the source label, schema version and
8/// timestamps when it wraps the event.
9#[derive(Debug, Clone, PartialEq)]
10pub struct CoreEvent {
11    /// The subscription the event belongs to, or `None` for an event that belongs to no
12    /// subscription (it is not buffered by a subscription policy downstream).
13    pub sub: Option<SubscriptionId>,
14    /// The event itself.
15    pub payload: Event,
16}
17
18/// A list the core pushes events into while it handles a frame or a timeout.
19///
20/// A core receives `&mut EventSink` in [`BrokerCore::on_frame`](crate::BrokerCore::on_frame) and
21/// [`BrokerCore::on_timeout`](crate::BrokerCore::on_timeout). The caller reads and empties it
22/// afterwards. Events keep the order in which they were pushed.
23///
24/// # Examples
25///
26/// ```
27/// use brokapi_core::{Event, SubscriptionId};
28/// use brokapi_sansio::EventSink;
29///
30/// let mut sink = EventSink::new();
31/// assert!(sink.is_empty());
32///
33/// sink.push_payload(Some(SubscriptionId::new(1)), Event::Balances(vec![]));
34/// assert_eq!(sink.len(), 1);
35/// assert_eq!(sink.events()[0].sub, Some(SubscriptionId::new(1)));
36///
37/// // Draining hands the events over and leaves the sink empty.
38/// let events = sink.drain();
39/// assert_eq!(events.len(), 1);
40/// assert!(sink.is_empty());
41/// ```
42#[derive(Debug, Clone, Default)]
43pub struct EventSink {
44    events: Vec<CoreEvent>,
45}
46
47impl EventSink {
48    /// Creates an empty sink.
49    pub fn new() -> Self {
50        Self::default()
51    }
52
53    /// Appends an event.
54    pub fn push(&mut self, event: CoreEvent) {
55        self.events.push(event);
56    }
57
58    /// Appends an event built from its parts.
59    pub fn push_payload(&mut self, sub: Option<SubscriptionId>, payload: Event) {
60        self.push(CoreEvent { sub, payload });
61    }
62
63    /// Whether no event is waiting.
64    pub fn is_empty(&self) -> bool {
65        self.events.is_empty()
66    }
67
68    /// How many events are waiting.
69    pub fn len(&self) -> usize {
70        self.events.len()
71    }
72
73    /// The waiting events, oldest first. They stay in the sink.
74    pub fn events(&self) -> &[CoreEvent] {
75        &self.events
76    }
77
78    /// Takes all waiting events out, oldest first, and leaves the sink empty.
79    pub fn drain(&mut self) -> Vec<CoreEvent> {
80        std::mem::take(&mut self.events)
81    }
82}
brokapi_sansio/event.rs

brokapi_sansio/
event.rs
