brokapi_bybit/
subs.rs

1//! The subscription records a connector keeps, and the generators of subscription ids and request 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 is for.
11#[derive(Debug, Clone, Copy, PartialEq, Eq)]
12pub enum MdSubKind {
13    /// An order book stream.
14    OrderBook,
15    /// Public trades.
16    MarketTrades,
17    /// Best bid and ask.
18    Quotes,
19    /// Candles.
20    Candles,
21}
22
23/// A market data subscription as the connector keeps it.
24#[derive(Debug, Clone)]
25pub struct MdSubscription {
26    /// The id given to the caller.
27    pub id: SubscriptionId,
28    /// The instrument.
29    pub instrument: InstrumentId,
30    /// The purpose.
31    pub kind: MdSubKind,
32    /// The options the caller gave.
33    pub opts: SubscribeOpts,
34    /// The topic of the stream at the exchange.
35    pub topic: String,
36}
37
38/// What a user data subscription is for.
39#[derive(Debug, Clone, Copy, PartialEq, Eq)]
40pub enum UdSubKind {
41    /// Own orders.
42    OrderState,
43    /// Positions and balances.
44    Portfolio,
45    /// Own fills.
46    Executions,
47}
48
49/// A user data subscription as the connector keeps it.
50#[derive(Debug, Clone)]
51pub struct UdSubscription {
52    /// The id given to the caller.
53    pub id: SubscriptionId,
54    /// The purpose.
55    pub kind: UdSubKind,
56    /// The options the caller gave.
57    pub opts: SubscribeOpts,
58    /// The topics of the streams at the exchange.
59    pub topics: Vec<String>,
60}
61
62/// Hands out subscription ids, counting up from `start + 1`.
63pub struct SubIdGen(AtomicU64);
64
65impl SubIdGen {
66    /// A generator whose first id is `start + 1`.
67    pub fn new(start: u64) -> Self {
68        Self(AtomicU64::new(start))
69    }
70
71    /// The next id.
72    pub fn next(&self) -> SubscriptionId {
73        SubscriptionId::new(self.0.fetch_add(1, Ordering::Relaxed) + 1)
74    }
75}
76
77/// Hands out the `req_id` that ties an answer of the exchange to its request.
78#[derive(Default)]
79pub struct ReqIdGen(AtomicU64);
80
81impl ReqIdGen {
82    /// The next request id, a short text.
83    pub fn next(&self) -> String {
84        format!("r{}", self.0.fetch_add(1, Ordering::Relaxed) + 1)
85    }
86}
87
88/// Waits until the exchange has answered a subscription: `Ok` when it is open, [`brokapi_core::Error::Rejected`]
89/// (`subscription_refused`) when the exchange refused it, [`brokapi_core::Error::Transport`] when nothing came in
90/// `limit`.
91pub async fn wait_ack<C>(
92    core: &tokio::sync::Mutex<C>,
93    notify: std::sync::Arc<tokio::sync::Notify>,
94    limit: std::time::Duration,
95    state: impl Fn(&C) -> (crate::core::SubState, Option<String>),
96) -> Result<(), brokapi_core::Error> {
97    use crate::core::SubState;
98    use brokapi_runtime::AckState;
99    brokapi_runtime::wait_ack(core, notify, limit, |c: &C| match state(c) {
100        (SubState::Active, _) => AckState::Confirmed,
101        (SubState::Failed, reason) => AckState::Refused(reason.unwrap_or_default()),
102        (SubState::Pending, _) => AckState::Pending,
103    })
104    .await
105}
106
107#[cfg(test)]
108mod tests {
109    use super::*;
110
111    #[test]
112    fn ids_count_up_from_the_start() {
113        let g = SubIdGen::new(1000);
114        assert_eq!(g.next().value(), 1001);
115        assert_eq!(g.next().value(), 1002);
116        let r = ReqIdGen::default();
117        assert_eq!((r.next().as_str(), r.next().as_str()), ("r1", "r2"));
118    }
119}
brokapi_bybit/subs.rs

brokapi_bybit/
subs.rs
