egui_charts/data/inbox/
mod.rs

1//! Where feed answers and live updates wait for the chart.
2//!
3//! Pushes coalesce on arrival: ticks fold into the bar of their slot and a bar replaces the
4//! pending bar of its slot, so the inbox holds the bars that really changed, however fast a
5//! feed pushes. Trades wait in a store as bounded as the chart's, and depth updates apply to the
6//! subscription's book.
7
8mod sink;
9
10pub(crate) use sink::TradeBars;
11
12pub use sink::{Sink, aggregate_trades, aggregate_trades_with_tick_size};
13
14use super::depth::Book;
15use super::depth_history::{DepthColumn, DepthHistory, DepthHistoryConfig, Sampler};
16use super::merge::Origin;
17use super::trades::Trades;
18use super::{FeedError, FeedStatus, OrderQueue, SubscriptionId};
19use crate::model::{Bar, BarClock, Level1Quote, OrderBook};
20use chrono::{DateTime, TimeDelta, Utc};
21use std::any::Any;
22use std::collections::{BTreeMap, HashMap, HashSet, VecDeque};
23use std::marker::PhantomData;
24use std::sync::atomic::{AtomicBool, Ordering};
25use std::sync::{Arc, Mutex, MutexGuard, OnceLock, PoisonError};
26
27/// A `Result<T, FeedError>` answering a request for `T`, or a `T` pushed through an event stream.
28pub(crate) type Answer = Box<dyn Any + Send>;
29
30/// One chart's inbox, shared with every `Reply` and `Sink` it hands out.
31#[derive(Default)]
32pub(crate) struct Inbox {
33    state: Mutex<State>,
34    waker: OnceLock<Box<dyn Fn() + Send + Sync>>,
35}
36
37#[derive(Default)]
38struct State {
39    /// Set by the first push after a drain, so a burst wakes the chart once.
40    woken: bool,
41    answers: Vec<(u64, Answer)>,
42    /// Pushes queued in `answers` per event stream.
43    streamed: HashMap<u64, usize>,
44    subscriptions: HashMap<SubscriptionId, Live>,
45}
46
47struct Live {
48    symbol: String,
49    clock: BarClock,
50    /// Open of the newest slot pushed, kept across drains: trades of older slots are late.
51    forming: Option<DateTime<Utc>>,
52    /// Feed clock minus system clock, as last reported; zero until then.
53    offset: TimeDelta,
54    count_bar: Option<(Bar, u64)>,
55    delta: f64,
56    event_closed: bool,
57    last_trade: Option<DateTime<Utc>>,
58    last_close: Option<f64>,
59    empty: bool,
60    book: Book,
61    orders: Option<OrderQueue>,
62    orders_changed: bool,
63    tick_size: Option<f64>,
64    /// Samples the book for the chart's depth history, when it keeps one.
65    sampler: Option<Sampler>,
66    study_inputs: Vec<crate::data::OrderFlowInput>,
67    taken: TakenIds,
68    updates: Updates,
69}
70
71/// Ids of the trades folded into bars, as many as the trade store holds, the oldest forgotten
72/// first: a feed that replays its last trades after a reconnect must not count them twice.
73struct TakenIds {
74    order: VecDeque<u64>,
75    ids: HashSet<u64>,
76    capacity: usize,
77}
78
79/// What arrived since the last drain.
80pub(crate) struct Drained {
81    /// In arrival order, each with the key of its request or event stream.
82    pub(crate) answers: Vec<(u64, Answer)>,
83    pub(crate) updates: Vec<(SubscriptionId, Updates)>,
84}
85
86/// What one subscription pushed since the last drain.
87pub(crate) struct Updates {
88    /// A replay seek discards data from the previous playback position.
89    pub(crate) reset: bool,
90    /// One per slot, keyed by its open; the last one is the forming bar.
91    pub(crate) bars: BTreeMap<DateTime<Utc>, (Bar, Origin)>,
92    /// The trades folded into `bars`, in arrival order.
93    pub(crate) trades: Trades,
94    /// The order book, if it changed.
95    pub(crate) depth: Option<OrderBook>,
96    pub(crate) orders: Option<OrderQueue>,
97    pub(crate) orders_changed: bool,
98    /// Columns sampled for the depth history, oldest first, as many as its newest resolution
99    /// holds: the oldest make room for new ones.
100    pub(crate) depth_columns: VecDeque<DepthColumn>,
101    /// A seek rebuilds every resolution before the next drain, instead of dropping old columns.
102    pub(crate) replay_depth: Option<DepthHistory>,
103    /// Latest top-of-book quote pushed in this frame.
104    pub(crate) quote: Option<Level1Quote>,
105    /// Latest accepted heartbeat, applied to studies after the queued trades are merged.
106    pub(crate) time: Option<DateTime<Utc>>,
107    /// The book lost its snapshot: call `Datafeed::request_depth_snapshot`.
108    pub(crate) resnapshot: bool,
109    pub(crate) symbols_changed: bool,
110    pub(crate) status: Option<FeedStatus>,
111    pub(crate) error: Option<FeedError>,
112    /// Trades older than the last accepted execution or forming bar, dropped.
113    pub(crate) late_ticks: u64,
114    /// Trades whose id is among those the subscription remembers, dropped.
115    pub(crate) repeated_ticks: u64,
116    /// Bars and ticks that failed validation or were stamped too far ahead of the feed's clock,
117    /// dropped.
118    pub(crate) rejected: u64,
119    /// Bar slots evicted before a drain, reported as data loss even after another stream error.
120    dropped_bars: u64,
121    /// Depth snapshots and deltas that failed validation, dropped.
122    pub(crate) rejected_depth: u64,
123}
124
125impl Updates {
126    fn new(trades: Trades) -> Self {
127        Self {
128            reset: false,
129            bars: BTreeMap::new(),
130            trades,
131            depth: None,
132            orders: None,
133            orders_changed: false,
134            depth_columns: VecDeque::new(),
135            replay_depth: None,
136            quote: None,
137            time: None,
138            resnapshot: false,
139            symbols_changed: false,
140            status: None,
141            error: None,
142            late_ticks: 0,
143            repeated_ticks: 0,
144            rejected: 0,
145            dropped_bars: 0,
146            rejected_depth: 0,
147        }
148    }
149}
150
151impl Inbox {
152    /// Current time in a subscription's server clock, including between pushes.
153    #[cfg(feature = "widget")]
154    pub(crate) fn server_time(&self, id: SubscriptionId) -> Option<DateTime<Utc>> {
155        Utc::now().checked_add_signed(self.state().subscriptions.get(&id)?.offset)
156    }
157
158    /// Called from the pushing thread by the first push after each drain.
159    pub(crate) fn set_waker(&self, waker: impl Fn() + Send + Sync + 'static) {
160        let _ = self.waker.set(Box::new(waker));
161    }
162
163    /// A reply that delivers its answer here under `request`.
164    pub(crate) fn reply<T: Send + 'static>(self: &Arc<Self>, request: u64) -> Reply<T> {
165        Reply {
166            inbox: Arc::clone(self),
167            request,
168            sent: false,
169            answer: PhantomData,
170        }
171    }
172
173    /// Samples the book of `id` into depth history columns from its next update, or stops. The
174    /// columns sampled before and not yet drained are dropped.
175    pub(crate) fn sample_depth(&self, id: SubscriptionId, config: Option<DepthHistoryConfig>) {
176        if let Some(live) = self.state().subscriptions.get_mut(&id) {
177            live.sampler = config.map(Sampler::new);
178            live.updates.depth_columns.clear();
179            live.updates.replay_depth = None;
180        }
181    }
182
183    /// Connects active stream studies before their inputs are coalesced for the frame.
184    #[cfg(any(feature = "widget", feature = "alert-engine"))]
185    pub(crate) fn study_inputs(
186        &self,
187        id: SubscriptionId,
188        inputs: Vec<crate::data::OrderFlowInput>,
189    ) {
190        if let Some(live) = self.state().subscriptions.get_mut(&id) {
191            live.connect_study_inputs(inputs);
192        }
193    }
194
195    /// Discards what `id` pushed and ignores whatever it pushes from now on.
196    pub(crate) fn unsubscribe(&self, id: SubscriptionId) {
197        self.state().subscriptions.remove(&id);
198    }
199
200    /// Takes everything that arrived since the last drain.
201    pub(crate) fn drain(&self) -> Drained {
202        let mut state = self.state();
203        state.woken = false;
204        let answers = std::mem::take(&mut state.answers);
205        state.streamed.clear();
206        let updates = state
207            .subscriptions
208            .iter_mut()
209            .map(|(id, live)| {
210                let next = Updates::new(live.updates.trades.empty());
211                let mut updates = std::mem::replace(&mut live.updates, next);
212                if updates.dropped_bars > 0 {
213                    updates.rejected += updates.dropped_bars;
214                    updates.error = Some(FeedError::InvalidData(format!(
215                        "pending bar limit exceeded: {} bars dropped; reload history",
216                        updates.dropped_bars,
217                    )));
218                }
219                updates.depth = live.book.take_changed();
220                updates.orders_changed = std::mem::take(&mut live.orders_changed);
221                if updates.orders_changed {
222                    updates.orders = live.orders.clone();
223                }
224                (*id, updates)
225            })
226            .collect();
227        Drained { answers, updates }
228    }
229
230    fn state(&self) -> MutexGuard<'_, State> {
231        // A panic on one feed thread must not take every other producer and the chart with it.
232        self.state.lock().unwrap_or_else(PoisonError::into_inner)
233    }
234
235    /// Applies `push` under the lock, then wakes the chart if it is the first push since a drain.
236    fn push(&self, push: impl FnOnce(&mut State) -> bool) {
237        let wake = {
238            let mut state = self.state();
239            push(&mut state) && !std::mem::replace(&mut state.woken, true)
240        };
241        if wake && let Some(waker) = self.waker.get() {
242            waker();
243        }
244    }
245
246    fn answer<T: Send + 'static>(&self, request: u64, result: Result<T, FeedError>) {
247        self.push(|state| {
248            state.answers.push((request, Box::new(result)));
249            true
250        });
251    }
252
253    /// Queues `event` under `key` while `active` holds, up to 10,000 per drain. A push over the
254    /// bound is dropped and clears `active`: the stream lagged.
255    pub(crate) fn push_answer(&self, key: u64, event: Answer, active: &AtomicBool) {
256        self.push(|state| {
257            if !active.load(Ordering::Relaxed) {
258                return false;
259            }
260            let queued = state.streamed.entry(key).or_default();
261            if *queued == STREAM_BOUND {
262                active.store(false, Ordering::Relaxed);
263                return false;
264            }
265            *queued += 1;
266            state.answers.push((key, event));
267            true
268        });
269    }
270}
271
272/// Pushes one event stream may queue between two drains.
273const STREAM_BOUND: usize = 10_000;
274
275/// The answer to one request, usable from any thread. Dropping it unanswered reports
276/// [`FeedError::Unanswered`], so a lost request never leaves the chart loading forever.
277#[must_use = "answer the request with Reply::send"]
278pub struct Reply<T: Send + 'static> {
279    inbox: Arc<Inbox>,
280    request: u64,
281    sent: bool,
282    answer: PhantomData<fn(T)>,
283}
284
285impl<T: Send + 'static> Reply<T> {
286    /// Delivers `result` to the chart.
287    pub fn send(mut self, result: Result<T, FeedError>) {
288        self.sent = true;
289        self.inbox.answer(self.request, result);
290    }
291}
292
293impl<T: Send + 'static> Drop for Reply<T> {
294    fn drop(&mut self) {
295        if !self.sent {
296            self.inbox
297                .answer(self.request, Err::<T, _>(FeedError::Unanswered));
298        }
299    }
300}
301
302#[cfg(test)]
303mod tests;
egui_charts/data/inbox/mod.rs

egui_charts/data/inbox/
mod.rs
