acta_maker_sdk/ws/managed/
inbound.rs

1use std::sync::Arc;
2
3use tokio::sync::{broadcast, mpsc};
4
5use crate::ws::types::{ReferenceChangedData, ServerMessage};
6
7#[derive(Debug, Clone)]
8pub struct ManagedInbound {
9    pub connection_epoch: u64,
10    pub sequence: u64,
11    pub received_at: std::time::Instant,
12    pub(super) message: Arc<ServerMessage>,
13}
14
15impl ManagedInbound {
16    #[must_use]
17    pub fn message(&self) -> &ServerMessage {
18        &self.message
19    }
20
21    #[must_use]
22    pub fn message_arc(&self) -> Arc<ServerMessage> {
23        Arc::clone(&self.message)
24    }
25}
26
27impl AsRef<ServerMessage> for ManagedInbound {
28    fn as_ref(&self) -> &ServerMessage {
29        &self.message
30    }
31}
32
33#[derive(Debug, thiserror::Error)]
34#[non_exhaustive]
35pub enum ManagedReceiveError {
36    #[error("managed ws message stream closed")]
37    Closed,
38    #[error("managed ws subscriber lost {skipped} messages")]
39    Gap { skipped: u64 },
40}
41
42pub struct ManagedMessageReceiver {
43    pub(super) inner: broadcast::Receiver<ManagedInbound>,
44    /// Set under `GapPolicy::Reconnect`: a lag asks the session to reconnect.
45    pub(super) gap_tx: Option<mpsc::UnboundedSender<()>>,
46}
47
48impl ManagedMessageReceiver {
49    pub async fn recv(&mut self) -> Result<ManagedInbound, ManagedReceiveError> {
50        match self.inner.recv().await {
51            Ok(message) => Ok(message),
52            Err(broadcast::error::RecvError::Closed) => Err(ManagedReceiveError::Closed),
53            Err(broadcast::error::RecvError::Lagged(skipped)) => {
54                if let Some(gap_tx) = &self.gap_tx {
55                    let _ = gap_tx.send(());
56                }
57                Err(ManagedReceiveError::Gap { skipped })
58            }
59        }
60    }
61}
62
63pub struct ReferenceChangeReceiver {
64    inner: ManagedMessageReceiver,
65}
66
67impl ReferenceChangeReceiver {
68    pub(crate) const fn new(inner: ManagedMessageReceiver) -> Self {
69        Self { inner }
70    }
71
72    pub async fn recv(&mut self) -> Result<ReferenceChangedData, ManagedReceiveError> {
73        loop {
74            let inbound = self.inner.recv().await?;
75            if let ServerMessage::ReferenceChanged(data) = inbound.message() {
76                return Ok(data.clone());
77            }
78        }
79    }
80}
acta_maker_sdk/ws/managed/inbound.rs

acta_maker_sdk/ws/managed/
inbound.rs
