brokapi_runtime/
ack.rs

1//! Waiting for the broker to confirm a new subscription.
2//!
3//! A `subscribe_*` call that returns before the stream is open gives the caller a promise the connector
4//! cannot keep: an order placed right after it can be filled before the stream is live, and the bot never
5//! gets the event. So every connector holds the call until the broker has confirmed the subscription, or the
6//! limit has passed ([`wait_ack`]), and a subscription that was not confirmed is taken back, not left half open.
7
8use std::sync::Arc;
9use std::time::Duration;
10
11use brokapi_core::{Error, TransportError};
12use tokio::sync::{Mutex, Notify};
13
14/// What a connector knows about one new subscription, for [`wait_ack`].
15#[derive(Debug, Clone, PartialEq, Eq)]
16pub enum AckState {
17    /// The request is out, the broker has not answered.
18    Pending,
19    /// The broker confirmed the subscription (or the data of it has started to come, which proves it).
20    Confirmed,
21    /// The broker refused it. The text is the broker's reason.
22    Refused(String),
23}
24
25/// The code of the [`Error::Rejected`] that a refused subscription gives. The same at every broker.
26pub const SUBSCRIPTION_REFUSED: &str = "subscription_refused";
27
28/// Waits until the broker has confirmed a subscription.
29///
30/// `state` reads the state of the subscription from `core` and is read again after every wake of `notify`
31/// (the core calls `notify_waiters` whenever a subscription changes state). The wait ends with:
32///
33/// - `Ok(())` when the state is [`AckState::Confirmed`];
34/// - [`Error::Rejected`] with the code [`SUBSCRIPTION_REFUSED`] when it is [`AckState::Refused`]: the broker
35///   said no, repeating the call unchanged will not help;
36/// - [`Error::Transport`] when `limit` passes with the state still [`AckState::Pending`]: the stream did not
37///   open in time (the network, or a slow broker). The call may be repeated.
38///
39/// The limit is counted on the tokio clock, so a test moves it with `tokio::time::advance`.
40///
41/// The caller owns the cleanup: on `Err` it takes the subscription back (nothing stays half open).
42///
43/// # Errors
44///
45/// As listed above.
46///
47/// # Examples
48///
49/// ```
50/// use std::sync::Arc;
51/// use std::time::Duration;
52/// use brokapi_runtime::{AckState, wait_ack};
53/// use tokio::sync::{Mutex, Notify};
54///
55/// # #[tokio::main(flavor = "current_thread")]
56/// # async fn main() {
57/// let core = Arc::new(Mutex::new(false)); // "the broker confirmed"
58/// let notify = Arc::new(Notify::new());
59///
60/// // The broker confirms a moment later; the core flips the flag and wakes the waiters.
61/// let (c, n) = (Arc::clone(&core), Arc::clone(&notify));
62/// tokio::spawn(async move {
63///     *c.lock().await = true;
64///     n.notify_waiters();
65/// });
66///
67/// let state = |confirmed: &bool| if *confirmed { AckState::Confirmed } else { AckState::Pending };
68/// wait_ack(&core, notify, Duration::from_secs(3), state).await.unwrap();
69/// # }
70/// ```
71pub async fn wait_ack<C>(
72    core: &Mutex<C>,
73    notify: Arc<Notify>,
74    limit: Duration,
75    state: impl Fn(&C) -> AckState,
76) -> Result<(), Error> {
77    let wait = async {
78        loop {
79            let notified = notify.notified();
80            tokio::pin!(notified);
81            // Registered before the state is read, so a change in between is not missed.
82            notified.as_mut().enable();
83            match state(&*core.lock().await) {
84                AckState::Confirmed => return Ok(()),
85                AckState::Refused(reason) => {
86                    return Err(Error::Rejected {
87                        code: SUBSCRIPTION_REFUSED.into(),
88                        message: if reason.is_empty() {
89                            "the broker refused the subscription".into()
90                        } else {
91                            reason
92                        },
93                    });
94                }
95                AckState::Pending => {}
96            }
97            notified.await;
98        }
99    };
100    match tokio::time::timeout(limit, wait).await {
101        Ok(result) => result,
102        Err(_) => Err(not_confirmed(limit)),
103    }
104}
105
106/// The error of a subscription that was not confirmed within `limit`: [`Error::Transport`].
107#[must_use]
108pub fn not_confirmed(limit: Duration) -> Error {
109    Error::Transport(TransportError::Message(format!(
110        "the subscription was not confirmed by the broker within {limit:?}"
111    )))
112}
brokapi_runtime/ack.rs

brokapi_runtime/
ack.rs
