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/// What the session broadcasts: a decoded frame, or the place of one it could not decode.
34#[derive(Debug, Clone)]
35pub(crate) enum InboundItem {
36    Message(ManagedInbound),
37    Unreadable,
38}
39
40#[derive(Debug, thiserror::Error)]
41#[non_exhaustive]
42pub enum ManagedReceiveError {
43    #[error("managed ws message stream closed")]
44    Closed,
45    #[error("managed ws subscriber lost {skipped} messages")]
46    Gap { skipped: u64 },
47}
48
49pub struct ManagedMessageReceiver {
50    pub(super) inner: broadcast::Receiver<InboundItem>,
51    /// Set under `GapPolicy::Reconnect`: a lag asks the session to reconnect.
52    pub(super) gap_tx: Option<mpsc::UnboundedSender<()>>,
53}
54
55impl ManagedMessageReceiver {
56    pub async fn recv(&mut self) -> Result<ManagedInbound, ManagedReceiveError> {
57        match self.inner.recv().await {
58            Ok(InboundItem::Message(message)) => Ok(message),
59            // Rebuild state as after any gap; a reconnect would only meet the same frame.
60            Ok(InboundItem::Unreadable) => Err(ManagedReceiveError::Gap { skipped: 1 }),
61            Err(broadcast::error::RecvError::Closed) => Err(ManagedReceiveError::Closed),
62            Err(broadcast::error::RecvError::Lagged(skipped)) => {
63                if let Some(gap_tx) = &self.gap_tx {
64                    let _ = gap_tx.send(());
65                }
66                Err(ManagedReceiveError::Gap { skipped })
67            }
68        }
69    }
70}
71
72pub struct ReferenceChangeReceiver {
73    inner: ManagedMessageReceiver,
74}
75
76impl ReferenceChangeReceiver {
77    pub(crate) const fn new(inner: ManagedMessageReceiver) -> Self {
78        Self { inner }
79    }
80
81    pub async fn recv(&mut self) -> Result<ReferenceChangedData, ManagedReceiveError> {
82        loop {
83            let inbound = self.inner.recv().await?;
84            if let ServerMessage::ReferenceChanged(data) = inbound.message() {
85                return Ok(data.clone());
86            }
87        }
88    }
89}
acta_maker_sdk/ws/managed/inbound.rs

acta_maker_sdk/ws/managed/
inbound.rs
