brokapi_finam/
raw.rs

1//! Direct access to any Finam Trade API method ("back door"): same secret and token manager, same
2//! endpoint, same [`BrokerLimiter`] as the connector it came from.
3//!
4//! The connectors wrap what the shared traits can express. Everything else (reports, corporate
5//! actions, options chains, constituents, `PlaceSLTPOrder`, usage metrics, ...) is reachable here:
6//!
7//! ```
8//! use brokapi_finam::limits::default_limits;
9//! use brokapi_finam::{MarketDataConfig, MarketDataConnector};
10//! use brokapi_runtime::BrokerLimiter;
11//! use brokapi_finam_protocol::pb::grpc::tradeapi::v1::marketdata::QuoteRequest;
12//! use tokio::sync::mpsc;
13//! # use brokapi_finam::FinamApiConfig;
14//! # use brokapi_mocks::finam::{MockServer, SECRET};
15//!
16//! # #[tokio::main(flavor = "current_thread")]
17//! # async fn main() -> Result<(), Box<dyn std::error::Error>> {
18//! # let server = MockServer::start().await;
19//! # let api = FinamApiConfig::new(SECRET).with_endpoint(server.url.clone());
20//! let (events, _rx) = mpsc::channel(16);
21//! let md = MarketDataConnector::start(
22//!     MarketDataConfig::new(api, "finam-md"),
23//!     BrokerLimiter::new(&default_limits()),
24//!     events,
25//! )
26//! .await?;
27//!
28//! // The name is the proto method name: the quota is looked up by it.
29//! let raw = md.raw();
30//! let quote = raw
31//!     .call("LastQuote", |api| async move {
32//!         api.market_data().last_quote(QuoteRequest { symbol: "SBER@MISX".into() }).await
33//!     })
34//!     .await?;
35//! assert_eq!(quote.symbol, "SBER@MISX");
36//! # Ok(())
37//! # }
38//! ```
39//!
40//! Rules of a call:
41//! 1. The quota is taken first, by the proto method name (`"GetFutureSplits"`), from the route in
42//!    `limits.toml`. A name without a route spends the `"*"` default route, so a call is
43//!    never free. A refusal is `Error::RateLimited` and the request is not sent.
44//! 2. The closure gets a [`FinamApi`] that attaches the CURRENT JWT to every request (the connector's
45//!    token manager renews it ahead of expiry). One closure is one API method.
46//! 3. The answer is cut by the connector's request timeout.
47//! 4. [`call`](FinamRaw::call) is for reads: on `UNAUTHENTICATED` the JWT is renewed and the call is
48//!    made once more. [`call_command`](FinamRaw::call_command) is for anything that changes the
49//!    account: no retry, and a timeout or break is [`Error::Unknown`].
50//!
51//! A stream opened through `call` keeps the JWT it was opened with. Reopen it after the token
52//! changes (the connector does that for its own streams).
53//!
54//! The raw layer does not know about the connector's order index: orders placed through it are not
55//! tracked (no events, no `reconcile`). Use the shared traits for orders.
56
57use std::future::Future;
58use std::sync::Arc;
59use std::time::Duration;
60
61use brokapi_core::Error;
62use brokapi_finam_protocol::FinamApi;
63use brokapi_runtime::BrokerLimiter;
64use tonic::{Response, Status};
65
66use crate::error::map_command_err;
67use crate::rpc::Rpc;
68
69/// Handle for calling any Finam method with the connector's token, endpoint and rate limiter.
70///
71/// Get it from `MarketDataConnector::raw` or `UserDataConnector::raw`. Cheap to clone. It stays
72/// usable while the connector it came from is alive. See the [module docs](self) for the rules of a call.
73///
74/// # Examples
75///
76/// A command goes through [`call_command`](Self::call_command), a read through [`call`](Self::call):
77///
78/// ```
79/// use brokapi_core::Error;
80/// use brokapi_finam::limits::default_limits;
81/// use brokapi_finam::{UserDataConfig, UserDataConnector};
82/// use brokapi_runtime::BrokerLimiter;
83/// use brokapi_finam_protocol::pb::grpc::tradeapi::v1::accounts::GetAccountRequest;
84/// use tokio::sync::mpsc;
85/// # use brokapi_finam::FinamApiConfig;
86/// # use brokapi_mocks::finam::{MockServer, SECRET};
87///
88/// # #[tokio::main(flavor = "current_thread")]
89/// # async fn main() -> Result<(), Box<dyn std::error::Error>> {
90/// # let server = MockServer::start().await;
91/// # let api = FinamApiConfig::new(SECRET).with_endpoint(server.url.clone());
92/// let (events, _rx) = mpsc::channel(16);
93/// let ud = UserDataConnector::start(
94///     UserDataConfig::new(api, "ACC1", "finam-ud"),
95///     BrokerLimiter::new(&default_limits()),
96///     events,
97/// )
98/// .await?;
99/// let raw = ud.raw();
100/// assert_eq!(raw.account_id(), Some("ACC1"));
101///
102/// let account = raw
103///     .call("GetAccount", |api| async move {
104///         api.accounts().get_account(GetAccountRequest { account_id: "ACC1".into() }).await
105///     })
106///     .await?;
107/// assert_eq!(account.account_id, "ACC1");
108///
109/// // An account the broker does not know is a refusal.
110/// let err = raw
111///     .call("GetAccount", |api| async move {
112///         api.accounts().get_account(GetAccountRequest { account_id: "NOPE".into() }).await
113///     })
114///     .await
115///     .unwrap_err();
116/// assert!(matches!(err, Error::Rejected { .. }));
117/// # Ok(())
118/// # }
119/// ```
120#[derive(Clone)]
121pub struct FinamRaw {
122    rpc: Rpc,
123    account: Option<String>,
124}
125
126impl FinamRaw {
127    pub(crate) fn new(rpc: Rpc, account: Option<String>) -> Self {
128        Self { rpc, account }
129    }
130
131    /// Account of the connector this came from. `None` for a market data connector.
132    pub fn account_id(&self) -> Option<&str> {
133        self.account.as_deref()
134    }
135
136    /// Deadline applied to every call (`FinamApiConfig::request_timeout` of the connector).
137    pub fn timeout(&self) -> Duration {
138        self.rpc.timeout
139    }
140
141    /// The limiter shared with the connector.
142    pub fn limiter(&self) -> &Arc<BrokerLimiter> {
143        &self.rpc.limiter
144    }
145
146    /// Reads: quota for `method`, deadline, one re-auth retry, error mapping.
147    ///
148    /// `f` may run twice (after a re-auth), hence `Fn`. For a server-streaming method the result is
149    /// `tonic::Streaming<_>`: the quota covers opening the stream. A stream opened here is not
150    /// supervised: when it breaks or the token changes, open it again yourself.
151    ///
152    /// Do not use it for a call that changes the account, because of the repeat. Use
153    /// [`call_command`](Self::call_command).
154    ///
155    /// # Errors
156    ///
157    /// - [`Error::RateLimited`]: the local quota for `method` is spent. Nothing was sent.
158    /// - [`Error::Rejected`]: the broker refused (bad argument, unknown id, no permission, or the
159    ///   token was refused again after the renewal).
160    /// - [`Error::Transport`]: connection failure, or no answer within [`timeout`](Self::timeout).
161    ///
162    /// A failed token renewal is reported with the same mapping.
163    ///
164    /// # Examples
165    ///
166    /// See the [module docs](self).
167    pub async fn call<T, F, Fut>(&self, method: &str, f: F) -> Result<T, Error>
168    where
169        F: Fn(FinamApi) -> Fut,
170        Fut: Future<Output = Result<Response<T>, Status>>,
171    {
172        self.rpc.read(method, f).await
173    }
174
175    /// Commands: quota for `method`, deadline, NO retry. Anything that may have reached the broker
176    /// (timeout, break, server fault) is `Error::Unknown`.
177    ///
178    /// There is no repeat on purpose: a second send could change the account twice. When the answer
179    /// is `Unknown`, read the state (orders, positions) before deciding anything.
180    ///
181    /// The order index of the connector does not see what you place here: no events, no
182    /// `reconcile`. For plain and stop orders use the shared traits.
183    ///
184    /// # Errors
185    ///
186    /// - [`Error::RateLimited`]: the local quota for `method` is spent. Nothing was sent.
187    /// - [`Error::Rejected`]: the broker certainly did not process the request (for example
188    ///   `INVALID_ARGUMENT` or `NOT_FOUND`, which is the code `not_found`).
189    /// - [`Error::Config`]: `UNAUTHENTICATED`, the token is refused. A command is not repeated after a token
190    ///   renewal.
191    /// - [`Error::Unknown`]: timeout, broken connection or server fault. The command may have been applied.
192    ///
193    /// # Examples
194    ///
195    /// ```
196    /// use brokapi_core::Error;
197    /// use brokapi_finam::limits::default_limits;
198    /// use brokapi_finam::{MarketDataConfig, MarketDataConnector};
199    /// use brokapi_runtime::BrokerLimiter;
200    /// use brokapi_finam_protocol::pb::grpc::tradeapi::v1::marketdata::QuoteRequest;
201    /// use tokio::sync::mpsc;
202    /// # use brokapi_finam::FinamApiConfig;
203    /// # use brokapi_mocks::finam::{MockServer, SECRET};
204    ///
205    /// # #[tokio::main(flavor = "current_thread")]
206    /// # async fn main() -> Result<(), Box<dyn std::error::Error>> {
207    /// # let server = MockServer::start().await;
208    /// # let api = FinamApiConfig::new(SECRET).with_endpoint(server.url.clone());
209    /// # let (events, _rx) = mpsc::channel(16);
210    /// # let md = MarketDataConnector::start(
211    /// #     MarketDataConfig::new(api, "finam-md"),
212    /// #     BrokerLimiter::new(&default_limits()),
213    /// #     events,
214    /// # )
215    /// # .await?;
216    /// # // The mock has no state-changing call that md owns, so a read method stands in for a command.
217    /// // Same shape as a read, but never repeated. The result of a real command is handled the same way.
218    /// let sent = md
219    ///     .raw()
220    ///     .call_command("LastQuote", |api| async move {
221    ///         api.market_data().last_quote(QuoteRequest { symbol: "SBER@MISX".into() }).await
222    ///     })
223    ///     .await;
224    /// match sent {
225    ///     Ok(reply) => assert_eq!(reply.symbol, "SBER@MISX"),
226    ///     Err(Error::Unknown) => { /* read the state before acting */ }
227    ///     Err(other) => panic!("refused: {other}"),
228    /// }
229    /// # Ok(())
230    /// # }
231    /// ```
232    pub async fn call_command<T, F, Fut>(&self, method: &str, f: F) -> Result<T, Error>
233    where
234        F: Fn(FinamApi) -> Fut,
235        Fut: Future<Output = Result<Response<T>, Status>>,
236    {
237        self.rpc.acquire(method)?;
238        self.rpc
239            .send_command(f)
240            .await
241            .map_err(|e| map_command_err(&e))
242    }
243}
brokapi_finam/raw.rs

brokapi_finam/
raw.rs
