Function wait_ack 

pub async fn wait_ack<C>(
    core: &Mutex<C>,
    notify: Arc<Notify>,
    limit: Duration,
    state: impl Fn(&C) -> AckState,
) -> Result<(), Error>

Waits until the broker has confirmed a subscription.

state reads the state of the subscription from core and is read again after every wake of notify (the core calls notify_waiters whenever a subscription changes state). The wait ends with:

• Ok(()) when the state is AckState::Confirmed;
• [Error::Rejected] with the code SUBSCRIPTION_REFUSED when it is AckState::Refused: the broker said no, repeating the call unchanged will not help;
• [Error::Transport] when limit passes with the state still AckState::Pending: the stream did not open in time (the network, or a slow broker). The call may be repeated.

The limit is counted on the tokio clock, so a test moves it with tokio::time::advance.

The caller owns the cleanup: on Err it takes the subscription back (nothing stays half open).

Errors

As listed above.

Examples

use std::sync::Arc;
use std::time::Duration;
use brokapi_runtime::{AckState, wait_ack};
use tokio::sync::{Mutex, Notify};

let core = Arc::new(Mutex::new(false)); // "the broker confirmed"
let notify = Arc::new(Notify::new());

// The broker confirms a moment later; the core flips the flag and wakes the waiters.
let (c, n) = (Arc::clone(&core), Arc::clone(&notify));
tokio::spawn(async move {
    *c.lock().await = true;
    n.notify_waiters();
});

let state = |confirmed: &bool| if *confirmed { AckState::Confirmed } else { AckState::Pending };
wait_ack(&core, notify, Duration::from_secs(3), state).await.unwrap();