brokapi_alor/
subs.rs

1//! The subscription records a connector keeps, and the generator of subscription ids.
2//!
3//! These types are the bookkeeping of the connectors. A bot does not build them: it gets a
4//! [`SubscriptionId`] from a `subscribe_*` call and passes it back to `unsubscribe`.
5
6use std::sync::atomic::{AtomicU64, Ordering};
7
8use brokapi_core::{InstrumentId, SubscribeOpts, SubscriptionId};
9
10/// What a market data subscription delivers.
11#[derive(Debug, Clone, Copy, PartialEq, Eq)]
12pub enum MdSubKind {
13    /// Order book snapshots of one instrument.
14    OrderBook,
15    /// Trades of the market (the tape) of one instrument.
16    MarketTrades,
17    /// Best bid and ask of one instrument.
18    Quotes,
19    /// The latest candle of one instrument and timeframe.
20    Candles,
21}
22
23/// One live market data subscription, as the connector remembers it.
24#[derive(Debug, Clone)]
25pub struct MdSubscription {
26    /// The id the bot got from `subscribe_*`.
27    pub id: SubscriptionId,
28    /// The instrument the data is about.
29    pub instrument: InstrumentId,
30    /// What the subscription delivers.
31    pub kind: MdSubKind,
32    /// The options given at subscribe time (backpressure policy, depth).
33    pub opts: SubscribeOpts,
34    /// The `guid` of the WebSocket subscription: it names the data frames and the `unsubscribe`.
35    pub guid: String,
36}
37
38/// What a user data subscription delivers.
39#[derive(Debug, Clone, Copy, PartialEq, Eq)]
40pub enum UdSubKind {
41    /// Status changes of orders.
42    OrderState,
43    /// Portfolio updates.
44    Portfolio,
45    /// Fills of orders.
46    Executions,
47}
48
49/// One live user data subscription, as the connector remembers it.
50#[derive(Debug, Clone)]
51pub struct UdSubscription {
52    /// The id the bot got from `subscribe_*`.
53    pub id: SubscriptionId,
54    /// What the subscription delivers.
55    pub kind: UdSubKind,
56    /// The options given at subscribe time. Order state and executions use a lossless buffer.
57    pub opts: SubscribeOpts,
58    /// The `guid` of the WebSocket subscription.
59    pub guid: String,
60}
61
62/// Hands out subscription ids of one connector, in increasing order, from a thread-safe counter.
63///
64/// The market data connector counts from 1, the user data connector from 1001, so the ids of the two
65/// connectors of one bot do not collide.
66pub struct SubIdGen(AtomicU64);
67
68impl SubIdGen {
69    /// Creates a generator whose first id is `start + 1`.
70    pub fn new(start: u64) -> Self {
71        Self(AtomicU64::new(start))
72    }
73
74    /// Returns the next id. Safe to call from several tasks.
75    pub fn next(&self) -> SubscriptionId {
76        SubscriptionId::new(self.0.fetch_add(1, Ordering::Relaxed) + 1)
77    }
78}
79
80/// A new `guid` for a WebSocket subscription.
81pub fn new_guid() -> String {
82    uuid::Uuid::new_v4().to_string()
83}
84
85/// Waits for the broker to confirm a new subscription.
86///
87/// `state` reads the state of the subscription from the core. The wait ends with `Ok` when the subscription is
88/// active (the broker confirmed it, or data came), with [`brokapi_core::Error::Rejected`] (`subscription_refused`)
89/// when the broker refused it, and with [`brokapi_core::Error::Transport`] when `limit` passes with no answer.
90/// The caller then takes the subscription back: it is not left half open. See [`brokapi_runtime::wait_ack`].
91///
92/// # Errors
93///
94/// [`brokapi_core::Error::Rejected`] with the text of the broker when the subscription was refused,
95/// [`brokapi_core::Error::Transport`] when it was not confirmed in time.
96pub async fn wait_ack<C>(
97    core: &tokio::sync::Mutex<C>,
98    notify: std::sync::Arc<tokio::sync::Notify>,
99    limit: std::time::Duration,
100    state: impl Fn(&C) -> (crate::core::SubState, Option<String>),
101) -> Result<(), brokapi_core::Error> {
102    use crate::core::SubState;
103    use brokapi_runtime::AckState;
104    brokapi_runtime::wait_ack(core, notify, limit, |c: &C| match state(c) {
105        (SubState::Active, _) => AckState::Confirmed,
106        (SubState::Failed, reason) => AckState::Refused(reason.unwrap_or_default()),
107        (SubState::Pending, _) => AckState::Pending,
108    })
109    .await
110}
brokapi_alor/subs.rs

brokapi_alor/
subs.rs
