acta_maker_sdk/ws/managed/
inbound.rs

1use std::sync::Arc;
2
3use tokio::sync::{broadcast, mpsc};
4
5use crate::ws::types::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}
acta_maker_sdk/ws/managed/inbound.rs

acta_maker_sdk/ws/managed/
inbound.rs
