architect_sdk/
synced.rs

1//! A device for knowing if a subscription is synced.
2
3use anyhow::{bail, Result};
4use std::time::Duration;
5use tokio::sync::watch;
6
7#[derive(Debug, Clone)]
8pub struct Synced<T>(watch::Sender<T>);
9
10impl<T> Synced<T> {
11    pub fn new(value: T) -> Self {
12        let (tx, _) = watch::channel(value);
13        Self(tx)
14    }
15
16    pub fn set(&self, value: T) {
17        self.0.send_replace(value);
18    }
19
20    pub fn handle(&self) -> SyncedHandle<T> {
21        SyncedHandle(self.0.subscribe())
22    }
23}
24
25#[derive(Clone)]
26pub struct SyncedHandle<T>(pub watch::Receiver<T>);
27
28impl<T> SyncedHandle<T> {
29    pub async fn wait_synced_with(
30        &mut self,
31        timeout: Option<Duration>,
32        f: impl FnMut(&T) -> bool,
33    ) -> Result<()> {
34        if let Some(timeout) = timeout {
35            if tokio::time::timeout(timeout, self.0.wait_for(f)).await.is_err() {
36                bail!("timed out waiting for book to sync");
37            }
38        } else {
39            self.0.wait_for(f).await?;
40        }
41        Ok(())
42    }
43
44    pub async fn changed(&mut self) -> Result<()> {
45        Ok(self.0.changed().await?)
46    }
47}
48
49impl SyncedHandle<bool> {
50    pub fn is_synced(&self) -> bool {
51        *self.0.borrow()
52    }
53
54    pub async fn wait_synced(&mut self, timeout: Option<Duration>) -> Result<()> {
55        self.wait_synced_with(timeout, |v| *v).await
56    }
57}
58
59impl SyncedHandle<u64> {
60    pub async fn wait_synced(&mut self, timeout: Option<Duration>) -> Result<()> {
61        self.wait_synced_with(timeout, |v| *v > 0).await
62    }
63}
64
65impl<T> SyncedHandle<Option<T>> {
66    pub async fn wait_synced(&mut self, timeout: Option<Duration>) -> Result<()> {
67        self.wait_synced_with(timeout, |v| v.is_some()).await
68    }
69}
architect_sdk/synced.rs

architect_sdk/
synced.rs
