phoenix_channel_client/client/
presence.rs

1use std::collections::VecDeque;
2
3use phoenix_channel_runtime::{
4    Presence, PresenceError, PresenceState, PresenceTracker, PresenceUpdate, ProtocolEvent,
5};
6use thiserror::Error;
7
8use super::{Channel, ChannelEvent, ChannelEvents, ClientError};
9
10/// A synchronized Presence change or channel lifecycle event.
11#[derive(Clone, Debug, PartialEq)]
12pub enum PresenceEvent {
13    /// A key or one of its metas joined.
14    Joined {
15        /// Application-defined Presence key.
16        key: String,
17        /// Entry before the join, if the key was already present.
18        current: Option<Presence>,
19        /// Entry or metas added by this update.
20        joined: Presence,
21    },
22    /// A key or one of its metas left.
23    Left {
24        /// Application-defined Presence key.
25        key: String,
26        /// Entry before the leave was applied.
27        current: Presence,
28        /// Entry or metas removed by this update.
29        left: Presence,
30    },
31    /// A full state or diff was applied.
32    Synced,
33    /// The socket disconnected and local Presence state was cleared.
34    Disconnected,
35    /// The channel was explicitly left and local state was cleared.
36    ChannelLeft,
37    /// The server closed the channel and local state was cleared.
38    ChannelClosed,
39    /// The channel errored and local state was cleared.
40    ChannelError,
41}
42
43/// Failure while decoding or consuming a Presence event stream.
44#[derive(Clone, Debug, Error, PartialEq)]
45pub enum PresenceStreamError {
46    /// A Presence state or diff payload was invalid.
47    #[error(transparent)]
48    Decode(#[from] PresenceError),
49    /// The bounded channel event stream dropped events.
50    #[error("presence event stream dropped {dropped} events and must be resynchronized")]
51    Desynchronized {
52        /// Number of channel events dropped before lag was observed.
53        dropped: u64,
54    },
55    /// [`ChannelPresence::resync`] is required before consuming more events.
56    #[error("presence state must be resynchronized before consuming more events")]
57    ResyncRequired,
58}
59
60/// Current Presence state and changes for a joined channel.
61///
62/// The value borrows its [`Channel`] so a resynchronization can leave and
63/// rejoin the same channel with a freshly loaded join payload.
64pub struct ChannelPresence<'a> {
65    channel: &'a Channel,
66    events: ChannelEvents,
67    tracker: PresenceTracker,
68    pending: VecDeque<PresenceEvent>,
69    desynchronized: bool,
70}
71
72impl<'a> ChannelPresence<'a> {
73    pub(super) fn new(channel: &'a Channel) -> Result<Self, ClientError> {
74        Ok(Self {
75            channel,
76            events: channel.events()?,
77            tracker: PresenceTracker::new(),
78            pending: VecDeque::new(),
79            desynchronized: false,
80        })
81    }
82
83    /// Returns the current synchronized state.
84    pub fn state(&self) -> &PresenceState {
85        self.tracker.state()
86    }
87
88    /// Returns whether event lag invalidated local state.
89    pub fn requires_resync(&self) -> bool {
90        self.desynchronized
91    }
92
93    /// Clears local state, leaves, and rejoins to request a fresh full state.
94    pub async fn resync(&mut self) -> Result<(), ClientError> {
95        self.tracker.reset();
96        self.pending.clear();
97        self.channel.leave().await?;
98        self.events = self.channel.events()?;
99        self.channel.join().await?;
100        self.desynchronized = false;
101        Ok(())
102    }
103
104    /// Returns the next Presence change or lifecycle event.
105    pub async fn next(&mut self) -> Option<Result<PresenceEvent, PresenceStreamError>> {
106        if self.desynchronized {
107            return Some(Err(PresenceStreamError::ResyncRequired));
108        }
109        loop {
110            if let Some(event) = self.pending.pop_front() {
111                return Some(Ok(event));
112            }
113
114            match self.events.next().await? {
115                ChannelEvent::Protocol(ProtocolEvent::Message(frame)) => {
116                    let previous = self.tracker.state().clone();
117                    match self.tracker.apply(&frame) {
118                        Ok(PresenceUpdate::Synced(diff)) => {
119                            for (key, joined) in diff.joins.0 {
120                                self.pending.push_back(PresenceEvent::Joined {
121                                    current: previous.get(&key).cloned(),
122                                    key,
123                                    joined,
124                                });
125                            }
126                            for (key, left) in diff.leaves.0 {
127                                if let Some(current) = previous.get(&key).cloned() {
128                                    self.pending.push_back(PresenceEvent::Left {
129                                        key,
130                                        current,
131                                        left,
132                                    });
133                                }
134                            }
135                            self.pending.push_back(PresenceEvent::Synced);
136                        }
137                        Ok(PresenceUpdate::Ignored | PresenceUpdate::Pending) => {}
138                        Err(error) => return Some(Err(error.into())),
139                    }
140                }
141                ChannelEvent::Disconnected => {
142                    self.tracker.reset();
143                    return Some(Ok(PresenceEvent::Disconnected));
144                }
145                ChannelEvent::Lagged { dropped } => {
146                    self.tracker.reset();
147                    self.pending.clear();
148                    self.desynchronized = true;
149                    return Some(Err(PresenceStreamError::Desynchronized { dropped }));
150                }
151                ChannelEvent::Protocol(ProtocolEvent::Left { .. }) => {
152                    self.tracker.reset();
153                    return Some(Ok(PresenceEvent::ChannelLeft));
154                }
155                ChannelEvent::Protocol(ProtocolEvent::ChannelClosed { .. }) => {
156                    self.tracker.reset();
157                    return Some(Ok(PresenceEvent::ChannelClosed));
158                }
159                ChannelEvent::Protocol(ProtocolEvent::ChannelError { .. }) => {
160                    self.tracker.reset();
161                    return Some(Ok(PresenceEvent::ChannelError));
162                }
163                ChannelEvent::Protocol(_) | ChannelEvent::JoinPayloadError(_) => {}
164            }
165        }
166    }
167}
phoenix_channel_client/client/presence.rs

phoenix_channel_client/client/
presence.rs
