brokapi_tbank/
raw.rs

1//! Direct access to any T-Bank API method ("back door"): same token, same endpoint, same
2//! [`BrokerLimiter`] as the connector it came from.
3//!
4//! The connectors wrap what the shared traits can express. Everything else the API offers (bond
5//! coupons, forecasts, favorites, reports, signals, sandbox, ...) is reachable here, so nothing the
6//! broker supports is out of reach.
7//!
8//! # Examples
9//!
10//! The example runs against the in-process mock. With a real connector the call is the same, only the
11//! config differs. For a method of your own, pass its proto name and call the generated client.
12//!
13//! ```
14//! use brokapi_runtime::BrokerLimiter;
15//! use brokapi_tbank::limits::default_limits;
16//! use brokapi_tbank::{MarketDataConfig, MarketDataConnector};
17//! use brokapi_tbank_protocol::v1 as pb;
18//! use tokio::sync::mpsc;
19//! # use brokapi_core::ShutdownMode;
20//! # use brokapi_mocks::tbank::{MockServer, TOKEN};
21//! # use brokapi_tbank_protocol::TbankConfig;
22//!
23//! # #[tokio::main(flavor = "current_thread")]
24//! # async fn main() -> Result<(), Box<dyn std::error::Error>> {
25//! # let server = MockServer::start(Default::default()).await;
26//! # let cfg = TbankConfig::for_mock(TOKEN, &server.endpoint);
27//! let (events_tx, _events) = mpsc::channel(64);
28//! let limiter = BrokerLimiter::new(&default_limits());
29//! let md = MarketDataConnector::start(MarketDataConfig::new(cfg, "tbank-md"), limiter, events_tx).await?;
30//! let raw = md.raw();
31//!
32//! // The quota is taken by the proto method name, then the call runs with the connector's token.
33//! let accounts = raw
34//!     .call("GetAccounts", |c| async move {
35//!         c.users().get_accounts(pb::GetAccountsRequest { status: None }).await
36//!     })
37//!     .await?;
38//! assert!(!accounts.accounts.is_empty());
39//! # md.shutdown(ShutdownMode::Immediate).await;
40//! # Ok(())
41//! # }
42//! ```
43//!
44//! Rules of a call:
45//! 1. The quota is taken first, by the proto method name (`"GetBondCoupons"`), from the route in
46//!    `limits.toml`. A name without a route spends the `"*"` default route, so a call
47//!    is never free. A refusal is `Error::RateLimited` and the request is not sent.
48//! 2. One closure is one API method. Do not call several methods inside one closure: only one quota
49//!    unit is taken for it.
50//! 3. The answer is cut by the request timeout of the connector.
51//! 4. Status codes become [`brokapi_core::Error`] like everywhere else in the connector.
52//!
53//! [`call`](TbankRaw::call) is for reads: timeout and transport trouble are retryable errors.
54//! [`call_command`](TbankRaw::call_command) is for anything that changes the account: a timeout or a
55//! break is [`Error::Unknown`] (the broker may have acted), and the library never repeats it.
56//!
57//! The raw layer does not know about the connector's order registry: orders placed through it are
58//! not tracked (no events, no `reconcile`). Use the shared traits for orders.
59
60use std::future::Future;
61use std::sync::Arc;
62use std::time::Duration;
63
64use brokapi_core::Error;
65use brokapi_runtime::BrokerLimiter;
66use brokapi_tbank_protocol::TbankClient;
67use tonic::{Response, Status};
68
69use crate::util::{CommandOutcome, command, query};
70
71/// Direct access to any T-Bank API method, made by [`MarketDataConnector::raw`] or
72/// [`UserDataConnector::raw`]. See the [module docs](self). Cheap to clone.
73///
74/// Reads go through [`call`](Self::call). Anything that changes the account goes through
75/// [`call_command`](Self::call_command), which never lets a lost answer look like a refusal.
76///
77/// [`MarketDataConnector::raw`]: crate::MarketDataConnector::raw
78/// [`UserDataConnector::raw`]: crate::UserDataConnector::raw
79///
80/// # Examples
81///
82/// A command that the broker refuses is [`Error::Rejected`]. Here the order does not exist.
83///
84/// ```
85/// use brokapi_core::Error;
86/// use brokapi_runtime::BrokerLimiter;
87/// use brokapi_tbank::limits::default_limits;
88/// use brokapi_tbank::{UserDataConfig, UserDataConnector};
89/// use brokapi_tbank_protocol::v1 as pb;
90/// use tokio::sync::mpsc;
91/// # use brokapi_core::ShutdownMode;
92/// # use brokapi_mocks::tbank::{MockServer, ACCOUNT_A, TOKEN};
93/// # use brokapi_tbank_protocol::TbankConfig;
94///
95/// # #[tokio::main(flavor = "current_thread")]
96/// # async fn main() -> Result<(), Box<dyn std::error::Error>> {
97/// # let server = MockServer::start(Default::default()).await;
98/// # let cfg = TbankConfig::for_mock(TOKEN, &server.endpoint).with_account(ACCOUNT_A);
99/// let (events_tx, _events) = mpsc::channel(64);
100/// let limiter = BrokerLimiter::new(&default_limits());
101/// let ud = UserDataConnector::start(UserDataConfig::new(cfg, "tbank-ud"), limiter, events_tx).await?;
102/// let raw = ud.raw();
103/// let account = raw.account_id().expect("a user data connector has an account").to_string();
104///
105/// let result = raw
106///     .call_command("CancelOrder", |c| async move {
107///         c.orders()
108///             .cancel_order(pb::CancelOrderRequest {
109///                 account_id: account,
110///                 order_id: "no-such-order".into(),
111///                 order_id_type: Some(pb::OrderIdType::Request as i32),
112///             })
113///             .await
114///     })
115///     .await;
116/// assert!(matches!(result, Err(Error::Rejected { .. })));
117/// # ud.shutdown(ShutdownMode::Immediate).await;
118/// # Ok(())
119/// # }
120/// ```
121#[derive(Clone)]
122pub struct TbankRaw {
123    client: TbankClient,
124    limiter: Arc<BrokerLimiter>,
125    timeout: Duration,
126    account: Option<String>,
127}
128
129impl TbankRaw {
130    pub(crate) fn new(
131        client: TbankClient,
132        limiter: Arc<BrokerLimiter>,
133        timeout: Duration,
134        account: Option<String>,
135    ) -> Self {
136        Self {
137            client,
138            limiter,
139            timeout,
140            account,
141        }
142    }
143
144    /// Account of the connector this came from (user data connector only).
145    ///
146    /// `None` for a market data connector, which has no account. The raw layer does not add the
147    /// account to your requests: put it into the request yourself.
148    ///
149    /// # Examples
150    ///
151    /// ```
152    /// use brokapi_runtime::BrokerLimiter;
153    /// use brokapi_tbank::limits::default_limits;
154    /// use brokapi_tbank::{MarketDataConfig, MarketDataConnector};
155    /// use tokio::sync::mpsc;
156    /// # use brokapi_core::ShutdownMode;
157    /// # use brokapi_mocks::tbank::{MockServer, TOKEN};
158    /// # use brokapi_tbank_protocol::TbankConfig;
159    ///
160    /// # #[tokio::main(flavor = "current_thread")]
161    /// # async fn main() -> Result<(), Box<dyn std::error::Error>> {
162    /// # let server = MockServer::start(Default::default()).await;
163    /// # let cfg = TbankConfig::for_mock(TOKEN, &server.endpoint);
164    /// let (events_tx, _events) = mpsc::channel(64);
165    /// let md = MarketDataConnector::start(
166    ///     MarketDataConfig::new(cfg, "tbank-md"),
167    ///     BrokerLimiter::new(&default_limits()),
168    ///     events_tx,
169    /// )
170    /// .await?;
171    /// assert_eq!(md.raw().account_id(), None);
172    /// # md.shutdown(ShutdownMode::Immediate).await;
173    /// # Ok(())
174    /// # }
175    /// ```
176    pub fn account_id(&self) -> Option<&str> {
177        self.account.as_deref()
178    }
179
180    /// The deadline applied to every call: the `request_timeout` of the connector's config (30 s by
181    /// default).
182    ///
183    /// # Examples
184    ///
185    /// ```
186    /// use std::time::Duration;
187    /// use brokapi_runtime::BrokerLimiter;
188    /// use brokapi_tbank::limits::default_limits;
189    /// use brokapi_tbank::{MarketDataConfig, MarketDataConnector};
190    /// use tokio::sync::mpsc;
191    /// # use brokapi_core::ShutdownMode;
192    /// # use brokapi_mocks::tbank::{MockServer, TOKEN};
193    /// # use brokapi_tbank_protocol::TbankConfig;
194    ///
195    /// # #[tokio::main(flavor = "current_thread")]
196    /// # async fn main() -> Result<(), Box<dyn std::error::Error>> {
197    /// # let server = MockServer::start(Default::default()).await;
198    /// # let cfg = TbankConfig::for_mock(TOKEN, &server.endpoint).with_request_timeout(Duration::from_secs(5));
199    /// let (events_tx, _events) = mpsc::channel(64);
200    /// let md = MarketDataConnector::start(
201    ///     MarketDataConfig::new(cfg, "tbank-md"),
202    ///     BrokerLimiter::new(&default_limits()),
203    ///     events_tx,
204    /// )
205    /// .await?;
206    /// assert_eq!(md.raw().timeout(), Duration::from_secs(5));
207    /// # md.shutdown(ShutdownMode::Immediate).await;
208    /// # Ok(())
209    /// # }
210    /// ```
211    pub fn timeout(&self) -> Duration {
212        self.timeout
213    }
214
215    /// Reads: takes quota for `method`, runs the call, maps errors.
216    ///
217    /// For a server-streaming method the result type is `tonic::Streaming<_>`: the quota covers
218    /// opening the stream, the stream is then yours to read.
219    ///
220    /// # Errors
221    ///
222    /// - [`Error::RateLimited`] when the limiter has no quota for `method`. The request is not sent.
223    ///   Wait `retry_after` and call again.
224    /// - [`Error::Transport`] on a timeout or a network status. A read may be repeated.
225    /// - [`Error::Rejected`] when the broker refuses the request (`InvalidArgument`,
226    ///   `NotFound`, which is the code `not_found`, `PermissionDenied`, ...).
227    /// - [`Error::Config`] when the token is not accepted (`Unauthenticated`).
228    ///
229    /// # Examples
230    ///
231    /// ```
232    /// use brokapi_core::Error;
233    /// use brokapi_runtime::BrokerLimiter;
234    /// use brokapi_tbank::limits::limits_from_toml_str;
235    /// use brokapi_tbank::{MarketDataConfig, MarketDataConnector};
236    /// use brokapi_tbank_protocol::v1 as pb;
237    /// use tokio::sync::mpsc;
238    /// # use brokapi_core::ShutdownMode;
239    /// # use brokapi_mocks::tbank::{MockServer, TOKEN};
240    /// # use brokapi_tbank_protocol::TbankConfig;
241    ///
242    /// # #[tokio::main(flavor = "current_thread")]
243    /// # async fn main() -> Result<(), Box<dyn std::error::Error>> {
244    /// # let server = MockServer::start(Default::default()).await;
245    /// # let cfg = TbankConfig::for_mock(TOKEN, &server.endpoint);
246    /// // A tiny quota for the demo: one call per minute for every method.
247    /// let limits = limits_from_toml_str(
248    ///     r#"
249    /// [tbank.buckets.one]
250    /// rate = 1
251    /// per = "60s"
252    /// burst = 1
253    /// [tbank.route]
254    /// "*" = ["one"]
255    /// [tbank.conns]
256    /// max_conns = 1
257    /// max_subs_per_conn = 1
258    /// supports_multiplex = false
259    /// "#,
260    /// )?;
261    /// let (events_tx, _events) = mpsc::channel(64);
262    /// let md = MarketDataConnector::start(MarketDataConfig::new(cfg, "tbank-md"), BrokerLimiter::new(&limits), events_tx).await?;
263    /// let raw = md.raw();
264    ///
265    /// let list = |c: brokapi_tbank_protocol::TbankClient| async move {
266    ///     c.users().get_accounts(pb::GetAccountsRequest { status: None }).await
267    /// };
268    /// assert!(raw.call("GetAccounts", list).await.is_ok());
269    /// // The second call finds the bucket empty: it is refused locally and never sent.
270    /// assert!(matches!(raw.call("GetAccounts", list).await, Err(Error::RateLimited { .. })));
271    /// # md.shutdown(ShutdownMode::Immediate).await;
272    /// # Ok(())
273    /// # }
274    /// ```
275    pub async fn call<T, F, Fut>(&self, method: &str, f: F) -> Result<T, Error>
276    where
277        F: FnOnce(TbankClient) -> Fut,
278        Fut: Future<Output = Result<Response<T>, Status>>,
279    {
280        self.limiter.acquire(method, brokapi_runtime::now())?;
281        query(self.timeout, f(self.client.clone())).await
282    }
283
284    /// Commands: like [`call`](Self::call), but a timeout, a break or a server fault is
285    /// `Error::Unknown`, because the broker may have executed the command. Never retried.
286    ///
287    /// # Errors
288    ///
289    /// - [`Error::RateLimited`] when the limiter has no quota. Nothing was sent.
290    /// - [`Error::Rejected`] when the broker refuses (`InvalidArgument`, `NotFound`,
291    ///   `PermissionDenied`, `FailedPrecondition`, ...). The broker did not act.
292    /// - [`Error::Config`] when the token is not accepted. The broker did not act.
293    /// - [`Error::Unknown`] on a timeout, a broken connection or a server fault (`Unavailable`,
294    ///   `Internal`, ...). The command may or may not have been done. Do not repeat it. Ask the
295    ///   broker for the state of the order or account first.
296    ///
297    /// # Examples
298    ///
299    /// ```
300    /// use brokapi_core::Error;
301    /// use brokapi_runtime::BrokerLimiter;
302    /// use brokapi_tbank::limits::default_limits;
303    /// use brokapi_tbank::{UserDataConfig, UserDataConnector};
304    /// use brokapi_tbank_protocol::v1 as pb;
305    /// use tokio::sync::mpsc;
306    /// # use brokapi_core::ShutdownMode;
307    /// # use brokapi_mocks::tbank::{MockServer, ACCOUNT_A, TOKEN};
308    /// # use brokapi_tbank_protocol::TbankConfig;
309    ///
310    /// # #[tokio::main(flavor = "current_thread")]
311    /// # async fn main() -> Result<(), Box<dyn std::error::Error>> {
312    /// # let server = MockServer::start(Default::default()).await;
313    /// # let cfg = TbankConfig::for_mock(TOKEN, &server.endpoint).with_account(ACCOUNT_A);
314    /// let (events_tx, _events) = mpsc::channel(64);
315    /// let limiter = BrokerLimiter::new(&default_limits());
316    /// let ud = UserDataConnector::start(UserDataConfig::new(cfg, "tbank-ud"), limiter, events_tx).await?;
317    ///
318    /// // Cancelling an order the broker does not know is a definite refusal.
319    /// let outcome = ud
320    ///     .raw()
321    ///     .call_command("CancelOrder", |c| async move {
322    ///         c.orders()
323    ///             .cancel_order(pb::CancelOrderRequest {
324    ///                 account_id: ACCOUNT_A.into(),
325    ///                 order_id: "no-such-order".into(),
326    ///                 order_id_type: Some(pb::OrderIdType::Request as i32),
327    ///             })
328    ///             .await
329    ///     })
330    ///     .await;
331    /// match outcome {
332    ///     Err(Error::Rejected { code, .. }) => assert_eq!(code, "not_found"),
333    ///     other => panic!("expected a refusal, got {other:?}"),
334    /// }
335    /// # ud.shutdown(ShutdownMode::Immediate).await;
336    /// # Ok(())
337    /// # }
338    /// ```
339    pub async fn call_command<T, F, Fut>(&self, method: &str, f: F) -> Result<T, Error>
340    where
341        F: FnOnce(TbankClient) -> Fut,
342        Fut: Future<Output = Result<Response<T>, Status>>,
343    {
344        self.limiter.acquire(method, brokapi_runtime::now())?;
345        match command(self.timeout, f(self.client.clone())).await {
346            CommandOutcome::Done(v) => Ok(v),
347            CommandOutcome::Refused(_, err) => Err(err),
348            CommandOutcome::Unknown => Err(Error::Unknown),
349        }
350    }
351}
brokapi_tbank/raw.rs

brokapi_tbank/
raw.rs
