brokapi_bcs/
raw.rs

1//! Direct access to any BCS API method (the "back door"): same refresh token and access token, same
2//! endpoint, same [`BrokerLimiter`] as the connector it came from.
3//!
4//! The connectors wrap what the shared traits can express. Everything else (quotes snapshot,
5//! non-trade operations, limits, instrument discounts, the limits and marginal-indicators
6//! WebSockets, ...) is reachable here. Get a handle with `raw()` on either connector.
7//!
8//! Rules of a call:
9//!
10//! 1. The quota is taken first, by the operation name (`"GetOperationHistory"`), from the route in
11//!    `limits.toml`. A name without a route spends the `"*"` default route, so a call is
12//!    never free. A refusal is [`Error::RateLimited`] and the request is not sent.
13//! 2. The closure gets the connector's [`BcsHttpClient`]: it holds the access token, refreshes it and
14//!    repeats a request once after a 401. One closure is one API method.
15//! 3. [`call`](BcsRaw::call) is for reads. [`call_command`](BcsRaw::call_command) is for anything that
16//!    changes the account: a break, a timeout or a 5xx is [`Error::Unknown`], and it is never retried.
17//! 4. [`connect_ws`](BcsRaw::connect_ws) opens one of the WebSocket channels with the same token.
18//!
19//! The raw layer does not know about the connector's order state: orders placed through it are not
20//! tracked (no events, no `reconcile`). Use the shared traits for orders.
21//!
22//! # Examples
23//!
24//! A read that no shared trait wraps, the quotes snapshot, through the market data connector:
25//!
26//! ```
27//! use brokapi_bcs_protocol::http::models::{QuoteInstrument, QuotesRequest};
28//! use brokapi_bcs::limits::default_limits;
29//! use brokapi_bcs::{MarketDataConfig, MarketDataConnector};
30//! use brokapi_core::ShutdownMode;
31//! use brokapi_runtime::BrokerLimiter;
32//! use tokio::sync::mpsc;
33//! # use brokapi_bcs_protocol::config::BcsConfig;
34//!
35//! # #[tokio::main(flavor = "current_thread")]
36//! # async fn main() -> Result<(), Box<dyn std::error::Error>> {
37//! # let server = brokapi_mocks::bcs::MockServer::start(Default::default()).await;
38//! # let cfg = BcsConfig::for_mock("mock-refresh-token", &server.http_base);
39//! let (events_tx, _events) = mpsc::channel(256);
40//! let limiter = BrokerLimiter::new(&default_limits());
41//! let md = MarketDataConnector::start(MarketDataConfig::new(cfg, "bcs-md"), limiter, events_tx).await?;
42//!
43//! let raw = md.raw();
44//! // "GetQuotes" is the name of the quota route. The closure is one API method.
45//! let quotes = raw
46//!     .call("GetQuotes", |http| async move {
47//!         http.get_quotes(&QuotesRequest { instruments: vec![QuoteInstrument::new("SBER", "TQBR")] }).await
48//!     })
49//!     .await?;
50//! for q in &quotes.records {
51//!     println!("{:?} bid {:?} offer {:?}", q.ticker, q.bid, q.offer);
52//! }
53//! md.shutdown(ShutdownMode::Immediate).await;
54//! # Ok(())
55//! # }
56//! ```
57
58use std::future::Future;
59use std::sync::Arc;
60
61use brokapi_bcs_protocol::error::BcsApiError;
62use brokapi_bcs_protocol::http::client::BcsHttpClient;
63use brokapi_bcs_protocol::models::messages::IncomingMessage;
64use brokapi_bcs_protocol::ws::client::BcsWsChannel;
65use brokapi_bcs_protocol::ws::session::WsSession;
66use brokapi_core::Error;
67use brokapi_runtime::BrokerLimiter;
68use kanal::AsyncSender;
69
70use crate::error::{map_command_err, map_err};
71
72/// A handle for calling any BCS API method with a connector's token and rate limiter.
73///
74/// Get it from `raw()` on [`MarketDataConnector`](crate::MarketDataConnector) or
75/// [`UserDataConnector`](crate::UserDataConnector). It is cheap to clone, and all clones share the
76/// token and the limiter. See the [module docs](self) for the rules of a call.
77#[derive(Clone)]
78pub struct BcsRaw {
79    http: BcsHttpClient,
80    limiter: Arc<BrokerLimiter>,
81}
82
83impl BcsRaw {
84    pub(crate) fn new(http: BcsHttpClient, limiter: Arc<BrokerLimiter>) -> Self {
85        Self { http, limiter }
86    }
87
88    /// Calls a read-only API method: takes the quota for `method`, then runs `f`.
89    ///
90    /// `method` is the name of the quota route, not a URL. Take a name from `limits.toml`
91    /// (for example `GetQuotes`). An unknown name spends the default route `"*"`. `f` receives the
92    /// connector's [`BcsHttpClient`] and returns the future of one API call. Do not use this for
93    /// anything that changes the account, use [`call_command`](Self::call_command).
94    ///
95    /// # Errors
96    ///
97    /// - [`Error::RateLimited`] when the quota is spent. Nothing was sent. Also when the broker
98    ///   answers HTTP 429.
99    /// - [`Error::Rejected`] with code `validation` when the broker answers HTTP 400 with a
100    ///   validation error.
101    /// - [`Error::Transport`] for everything else: no connection, a timeout, an unreadable answer, HTTP
102    ///   404 (`not found`), a failed token refresh, any other HTTP status. A read can be repeated.
103    /// - [`Error::Config`] when the limits file routes the name to a bucket that does not exist.
104    ///
105    /// The full table is in [`crate::error::map_err`].
106    ///
107    /// # Examples
108    ///
109    /// A quota that is spent refuses the call before anything is sent:
110    ///
111    /// ```
112    /// use brokapi_bcs_protocol::http::models::{QuoteInstrument, QuotesRequest};
113    /// use brokapi_bcs::{MarketDataConfig, MarketDataConnector};
114    /// use brokapi_core::{Error, ShutdownMode};
115    /// use brokapi_runtime::{BrokerLimiter, BrokerLimits};
116    /// use tokio::sync::mpsc;
117    /// # use brokapi_bcs_protocol::config::BcsConfig;
118    ///
119    /// # #[tokio::main(flavor = "current_thread")]
120    /// # async fn main() -> Result<(), Box<dyn std::error::Error>> {
121    /// # let server = brokapi_mocks::bcs::MockServer::start(Default::default()).await;
122    /// # let cfg = BcsConfig::for_mock("mock-refresh-token", &server.http_base);
123    /// // One call per minute for GetQuotes.
124    /// let limits = BrokerLimits::from_toml_str(
125    ///     r#"
126    ///     [bcs.buckets.one]
127    ///     rate = 1
128    ///     per = "60s"
129    ///     burst = 1
130    ///     [bcs.route]
131    ///     "*" = ["one"]
132    ///     [bcs.conns]
133    ///     max_conns = 4
134    ///     max_subs_per_conn = 50
135    ///     supports_multiplex = false
136    ///     "#,
137    /// )?;
138    /// # let (events_tx, _events) = mpsc::channel(16);
139    /// let md = MarketDataConnector::start(
140    ///     MarketDataConfig::new(cfg, "bcs-md"),
141    ///     BrokerLimiter::new(&limits),
142    ///     events_tx,
143    /// )
144    /// .await?;
145    /// let raw = md.raw();
146    /// let quotes = || QuotesRequest { instruments: vec![QuoteInstrument::new("SBER", "TQBR")] };
147    ///
148    /// raw.call("GetQuotes", |http| async move { http.get_quotes(&quotes()).await }).await?;
149    /// let second = raw.call("GetQuotes", |http| async move { http.get_quotes(&quotes()).await }).await;
150    /// assert!(matches!(second, Err(Error::RateLimited { .. })));
151    /// md.shutdown(ShutdownMode::Immediate).await;
152    /// # Ok(())
153    /// # }
154    /// ```
155    pub async fn call<T, F, Fut>(&self, method: &str, f: F) -> Result<T, Error>
156    where
157        F: FnOnce(BcsHttpClient) -> Fut,
158        Fut: Future<Output = Result<T, BcsApiError>>,
159    {
160        self.limiter.acquire(method, brokapi_runtime::now())?;
161        f(self.http.clone()).await.map_err(map_err)
162    }
163
164    /// Calls an API method that changes the account: takes the quota for `method`, then runs `f`.
165    ///
166    /// Like [`call`](Self::call), but anything that may have reached the broker is
167    /// [`Error::Unknown`]. The library never repeats the call: a retry of a lost place request could
168    /// create a second order. Find out what happened (read the order, run `reconcile`) before you
169    /// decide to send it again. The order is not tracked by the connector (see the module docs).
170    ///
171    /// # Errors
172    ///
173    /// - [`Error::Unknown`] when there was no answer, a break or a timeout, HTTP 500 or above, or an
174    ///   answer that cannot be read. The command may have been done.
175    /// - [`Error::Rejected`] when the broker answered and said no: code `validation` for a validation
176    ///   error, code `http_<status>` for another status from 400 to 499 except 429. Nothing was done.
177    /// - [`Error::RateLimited`] when the quota is spent (nothing was sent) or the broker answers HTTP 429.
178    /// - [`Error::Transport`] for a failed token refresh and for HTTP 404 (`not found`).
179    /// - [`Error::Config`] when the limits file routes the name to a bucket that does not exist.
180    ///
181    /// The full table is in [`crate::error::map_command_err`].
182    ///
183    /// # Examples
184    ///
185    /// An order sent through the raw layer, refused by the broker (the mock refuses a limit order at
186    /// `REJECT_PRICE`). The refusal is a definite [`Error::Rejected`], not an unknown outcome:
187    ///
188    /// ```
189    /// use brokapi_bcs_protocol::http::models::CreateOrderRequest;
190    /// use brokapi_bcs_protocol::models::enums::OrderSide;
191    /// use brokapi_bcs::limits::default_limits;
192    /// use brokapi_bcs::{UserDataConfig, UserDataConnector};
193    /// use brokapi_conformance::REJECT_PRICE;
194    /// use brokapi_core::{ClientOrderId, Error, ShutdownMode};
195    /// use brokapi_runtime::BrokerLimiter;
196    /// use tokio::sync::mpsc;
197    /// # use brokapi_bcs_protocol::config::BcsConfig;
198    ///
199    /// # #[tokio::main(flavor = "current_thread")]
200    /// # async fn main() -> Result<(), Box<dyn std::error::Error>> {
201    /// # let server = brokapi_mocks::bcs::MockServer::start(Default::default()).await;
202    /// # let cfg = BcsConfig::for_mock("mock-refresh-token", &server.http_base).with_write(true);
203    /// # let (events_tx, _events) = mpsc::channel(256);
204    /// # let limiter = BrokerLimiter::new(&default_limits());
205    /// let ud = UserDataConnector::start(UserDataConfig::new(cfg, "bcs-ud"), limiter, events_tx).await?;
206    ///
207    /// let order = CreateOrderRequest::limit(
208    ///     ClientOrderId::new_v4(), "SBER", "TQBR", OrderSide::Buy, 1, REJECT_PRICE.decimal(),
209    /// );
210    /// let result = ud
211    ///     .raw()
212    ///     .call_command("PlaceOrder", |http| async move { http.create_order_sync(&order).await })
213    ///     .await;
214    /// assert!(matches!(result, Err(Error::Rejected { .. })));
215    /// ud.shutdown(ShutdownMode::Immediate).await;
216    /// # Ok(())
217    /// # }
218    /// ```
219    pub async fn call_command<T, F, Fut>(&self, method: &str, f: F) -> Result<T, Error>
220    where
221        F: FnOnce(BcsHttpClient) -> Fut,
222        Fut: Future<Output = Result<T, BcsApiError>>,
223    {
224        self.limiter.acquire(method, brokapi_runtime::now())?;
225        f(self.http.clone()).await.map_err(map_command_err)
226    }
227
228    /// Opens a WebSocket channel (`Limits`, `MarginalIndicators`, ...) with the connector's token.
229    ///
230    /// The quota name is `Connect<Channel>` (`ConnectLimits`). The session reconnects and
231    /// resubscribes on its own. Messages arrive on `event_tx`, parsed by the protocol layer. The
232    /// returned [`WsSession`] connects in the background: it does not wait for the connection.
233    /// Read `event_tx`'s receiver, or the session drops messages when the channel is full (see
234    /// `WsSession::dropped_events`).
235    ///
236    /// # Errors
237    ///
238    /// - [`Error::RateLimited`] when the quota `Connect<Channel>` is spent. No socket is opened.
239    /// - [`Error::Config`] when the limits file routes the name to a bucket that does not exist.
240    ///
241    /// # Examples
242    ///
243    /// ```
244    /// use std::time::Duration;
245    /// use brokapi_bcs_protocol::ws::client::BcsWsChannel;
246    /// use brokapi_bcs::limits::default_limits;
247    /// use brokapi_bcs::{MarketDataConfig, MarketDataConnector};
248    /// use brokapi_core::ShutdownMode;
249    /// use brokapi_runtime::BrokerLimiter;
250    /// use tokio::sync::mpsc;
251    /// # use brokapi_bcs_protocol::config::BcsConfig;
252    ///
253    /// # #[tokio::main(flavor = "current_thread")]
254    /// # async fn main() -> Result<(), Box<dyn std::error::Error>> {
255    /// # let server = brokapi_mocks::bcs::MockServer::start(Default::default()).await;
256    /// # let cfg = BcsConfig::for_mock("mock-refresh-token", &server.http_base);
257    /// # let (events_tx, _events) = mpsc::channel(256);
258    /// # let limiter = BrokerLimiter::new(&default_limits());
259    /// let md = MarketDataConnector::start(MarketDataConfig::new(cfg, "bcs-md"), limiter, events_tx).await?;
260    ///
261    /// let (tx, _rx) = kanal::bounded_async(64);
262    /// let session = md.raw().connect_ws(BcsWsChannel::Limits, tx)?;
263    /// // The session connects in the background: wait for it, with a limit.
264    /// tokio::time::timeout(Duration::from_secs(5), async {
265    ///     while !session.is_connected() {
266    ///         tokio::time::sleep(Duration::from_millis(10)).await;
267    ///     }
268    /// })
269    /// .await?;
270    /// md.shutdown(ShutdownMode::Immediate).await;
271    /// # Ok(())
272    /// # }
273    /// ```
274    pub fn connect_ws(
275        &self,
276        channel: BcsWsChannel,
277        event_tx: AsyncSender<IncomingMessage>,
278    ) -> Result<WsSession, Error> {
279        self.limiter
280            .acquire(&format!("Connect{channel:?}"), brokapi_runtime::now())?;
281        Ok(WsSession::spawn(channel, self.http.clone(), event_tx))
282    }
283}
284
285#[cfg(test)]
286mod tests {
287    use super::*;
288
289    fn http(status: Option<u16>) -> BcsApiError {
290        BcsApiError::HttpError {
291            status,
292            message: "x".into(),
293            trace_id: None,
294        }
295    }
296
297    #[test]
298    fn lost_commands_are_unknown_refused_ones_are_not() {
299        assert!(matches!(map_command_err(http(None)), Error::Unknown));
300        assert!(matches!(map_command_err(http(Some(503))), Error::Unknown));
301        assert!(matches!(
302            map_command_err(BcsApiError::ParseError("x".into())),
303            Error::Unknown
304        ));
305        assert!(matches!(
306            map_command_err(http(Some(429))),
307            Error::RateLimited { .. }
308        ));
309        assert!(matches!(
310            map_command_err(BcsApiError::ValidationError {
311                trace_id: None,
312                errors: vec![]
313            }),
314            Error::Rejected { .. }
315        ));
316    }
317}
brokapi_bcs/raw.rs

brokapi_bcs/
raw.rs
