brokapi_tbank/
util.rs

1//! Small shared helpers: subscription ids, call wrappers, limiter access.
2
3use std::future::Future;
4use std::sync::atomic::{AtomicU64, Ordering};
5use std::time::Duration;
6
7use brokapi_core::{Error, SubscriptionId};
8use tonic::{Response, Status};
9
10use crate::error::{map_command_status, map_status, transport};
11
12/// Hands out subscription ids, each one larger than the last.
13///
14/// A connector keeps one generator. The user data connector starts at 1000 and the market data
15/// connector at 0, so ids of the two kinds do not collide in a log.
16///
17/// # Examples
18///
19/// ```
20/// use brokapi_tbank::util::SubIdGen;
21///
22/// let ids = SubIdGen::new(0);
23/// let (a, b) = (ids.next(), ids.next());
24/// assert_ne!(a, b);
25/// assert_eq!(b.value(), a.value() + 1);
26/// ```
27pub struct SubIdGen(AtomicU64);
28
29impl SubIdGen {
30    /// A generator whose first id is `start + 1`.
31    pub fn new(start: u64) -> Self {
32        Self(AtomicU64::new(start))
33    }
34
35    /// The next id. Safe to call from several tasks at once.
36    pub fn next(&self) -> SubscriptionId {
37        SubscriptionId::new(self.0.fetch_add(1, Ordering::Relaxed) + 1)
38    }
39}
40
41/// Read-only unary call: transport trouble and timeouts are retryable errors.
42///
43/// Waits for `fut` at most `timeout`, then maps a gRPC failure with [`map_status`].
44///
45/// # Errors
46///
47/// - [`Error::Transport`] when the deadline passes, or for a transport status such as `Unavailable`.
48/// - [`Error::RateLimited`], [`Error::Rejected`] and [`Error::Config`] for the matching statuses,
49///   see [`map_status`].
50///
51/// # Examples
52///
53/// ```
54/// use std::time::Duration;
55/// use brokapi_core::Error;
56/// use brokapi_tbank::util::query;
57/// use tonic::{Response, Status};
58///
59/// # #[tokio::main(flavor = "current_thread")]
60/// # async fn main() {
61/// let ok = query(Duration::from_secs(1), async { Ok::<_, Status>(Response::new(42)) }).await;
62/// assert_eq!(ok.unwrap(), 42);
63///
64/// let slow = query(Duration::from_millis(10), async {
65///     tokio::time::sleep(Duration::from_secs(5)).await;
66///     Ok::<_, Status>(Response::new(0))
67/// })
68/// .await;
69/// assert!(matches!(slow, Err(Error::Transport(_))));
70/// # }
71/// ```
72pub async fn query<T>(
73    timeout: Duration,
74    fut: impl Future<Output = Result<Response<T>, Status>>,
75) -> Result<T, Error> {
76    match tokio::time::timeout(timeout, fut).await {
77        Ok(Ok(r)) => Ok(r.into_inner()),
78        Ok(Err(status)) => Err(map_status(&status)),
79        Err(_) => Err(transport(format!("request timed out after {timeout:?}"))),
80    }
81}
82
83/// Result of a trading command: a definite answer, or an outcome nobody knows.
84///
85/// The split exists so that a caller cannot treat a timeout as a refusal. Only `Refused` proves
86/// that the broker did not act.
87pub enum CommandOutcome<T> {
88    /// The broker answered and did the command. Holds the reply.
89    Done(T),
90    /// The broker refused; `status` is its answer.
91    Refused(Status, Error),
92    /// Timeout, break, server fault: the command may or may not have been executed.
93    Unknown,
94}
95
96/// Runs a trading command (place, cancel, replace) under a deadline.
97///
98/// Waits for `fut` at most `timeout`. Statuses that prove the broker did not act become
99/// [`CommandOutcome::Refused`] (see [`map_command_status`]). A timeout, a broken connection or a
100/// server fault is [`CommandOutcome::Unknown`]: the broker may have acted. The caller must never
101/// repeat the command after `Unknown`, because a repeated place could create a second order.
102///
103/// # Examples
104///
105/// ```
106/// use std::time::Duration;
107/// use brokapi_tbank::util::{command, CommandOutcome};
108/// use tonic::{Response, Status};
109///
110/// # #[tokio::main(flavor = "current_thread")]
111/// # async fn main() {
112/// let done = command(Duration::from_secs(1), async { Ok::<_, Status>(Response::new(1)) }).await;
113/// assert!(matches!(done, CommandOutcome::Done(1)));
114///
115/// let refused = command(Duration::from_secs(1), async {
116///     Err::<Response<()>, _>(Status::invalid_argument("30042"))
117/// })
118/// .await;
119/// assert!(matches!(refused, CommandOutcome::Refused(..)));
120///
121/// // A fault after sending hides what the broker did.
122/// let lost = command(Duration::from_secs(1), async { Err::<Response<()>, _>(Status::unavailable("reset")) }).await;
123/// assert!(matches!(lost, CommandOutcome::Unknown));
124/// # }
125/// ```
126pub async fn command<T>(
127    timeout: Duration,
128    fut: impl Future<Output = Result<Response<T>, Status>>,
129) -> CommandOutcome<T> {
130    match tokio::time::timeout(timeout, fut).await {
131        Ok(Ok(r)) => CommandOutcome::Done(r.into_inner()),
132        Ok(Err(status)) => match map_command_status(&status) {
133            Error::Unknown => CommandOutcome::Unknown,
134            err => CommandOutcome::Refused(status, err),
135        },
136        Err(_) => CommandOutcome::Unknown,
137    }
138}
brokapi_tbank/util.rs

brokapi_tbank/
util.rs
