brokapi_bcs/
pump.rs

1//! WS event pumps: parse in the receiver task, never block on a slow consumer.
2//!
3//! The frames of the WebSocket sessions come in through a channel. The pump gives each one to its core
4//! and hands the events to the [`Emitter`]: per-subscription buffers in front of the bot's channel
5//!. A bot that reads slowly fills the buffer of a subscription, and the policy of that
6//! subscription applies: `Conflate` keeps the latest, `DropNewest` counts drops, `Lossless` overflows
7//! (`Event::Overflow`, connector `Degraded`).
8
9use std::sync::Arc;
10use std::time::{Duration, Instant};
11
12use brokapi_bcs_protocol::models::messages::IncomingMessage;
13use brokapi_runtime::Emitter;
14use brokapi_sansio::{BrokerCore, EventSink};
15use kanal::AsyncReceiver;
16use tokio::sync::Mutex;
17use tokio_util::sync::CancellationToken;
18
19use crate::core::{MdCore, UdCore};
20
21/// How often the events that wait in the buffers are offered to the bot's channel again.
22const FLUSH_EVERY: Duration = Duration::from_millis(20);
23
24/// Runs the market data event pump until `cancel` fires or the frame channel closes.
25///
26/// Every frame from `rx` goes through `core` and the resulting events to `emitter`. Events that a
27/// slow bot left in the subscription buffers are offered again every 20 ms.
28pub async fn run_md_pump(
29    core: Arc<Mutex<MdCore>>,
30    emitter: Arc<Emitter>,
31    rx: AsyncReceiver<IncomingMessage>,
32    cancel: CancellationToken,
33) {
34    run_pump(core, emitter, rx, cancel, |c, msg, now, sink| {
35        c.on_frame(msg, now, sink)
36    })
37    .await;
38}
39
40/// Runs the user data event pump until `cancel` fires or the frame channel closes.
41///
42/// Same loop as [`run_md_pump`], with the user data core (order status, fills, portfolio).
43pub async fn run_ud_pump(
44    core: Arc<Mutex<UdCore>>,
45    emitter: Arc<Emitter>,
46    rx: AsyncReceiver<IncomingMessage>,
47    cancel: CancellationToken,
48) {
49    run_pump(core, emitter, rx, cancel, |c, msg, now, sink| {
50        c.on_incoming(msg, now, sink)
51    })
52    .await;
53}
54
55async fn run_pump<C>(
56    core: Arc<Mutex<C>>,
57    emitter: Arc<Emitter>,
58    rx: AsyncReceiver<IncomingMessage>,
59    cancel: CancellationToken,
60    step: fn(&mut C, IncomingMessage, Instant, &mut EventSink),
61) {
62    loop {
63        let backlog = emitter.has_backlog();
64        tokio::select! {
65            _ = cancel.cancelled() => break,
66            msg = rx.recv() => {
67                let Ok(msg) = msg else { break };
68                let now = brokapi_runtime::now();
69                let mut sink = EventSink::new();
70                step(&mut *core.lock().await, msg, now, &mut sink);
71                emitter.emit(sink.drain());
72            }
73            _ = tokio::time::sleep(FLUSH_EVERY), if backlog => emitter.flush(),
74        }
75    }
76}
77
78/// Gives the bot's channel time to take what the buffers still hold (shutdown).
79pub async fn drain_backlog(emitter: &Emitter, deadline: Duration) {
80    let end = brokapi_runtime::now() + deadline;
81    while emitter.has_backlog() && brokapi_runtime::now() < end {
82        emitter.flush();
83        tokio::time::sleep(Duration::from_millis(5)).await;
84    }
85}
brokapi_bcs/pump.rs

brokapi_bcs/
pump.rs
