o_sfu_protocol/core/
sticky_replay.rs

1use std::collections::BTreeMap;
2
3use crate::{
4    shared::{DownloadStates, StreamType, UserId, UserInfo},
5    signaling::{ClientEnvelope, ClientMessage, EnvelopeBatch, SubscribePayload},
6};
7
8/// Retains client intent across recoverable WebSocket replacement.
9///
10/// Authentication replays subscriptions and user info. Transport readiness
11/// replays active publications.
12#[derive(Debug, Clone, Default, PartialEq, Eq)]
13pub(super) struct StickyReplayState {
14    audio_publication_active: bool,
15    camera_publication_active: bool,
16    screen_publication_active: bool,
17    desired_subscriptions: BTreeMap<UserId, DownloadStates>,
18    desired_info: Option<UserInfo>,
19}
20
21impl StickyReplayState {
22    pub(super) fn new() -> Self {
23        Self::default()
24    }
25
26    pub(super) fn clear(&mut self) {
27        self.audio_publication_active = false;
28        self.camera_publication_active = false;
29        self.screen_publication_active = false;
30        self.desired_subscriptions.clear();
31        self.desired_info = None;
32    }
33
34    pub(super) fn set_publish_active(&mut self, stream_type: StreamType, active: bool) {
35        match stream_type {
36            StreamType::Audio => self.audio_publication_active = active,
37            StreamType::Camera => self.camera_publication_active = active,
38            StreamType::Screen => self.screen_publication_active = active,
39        }
40    }
41
42    pub(super) fn active_publications(&self) -> impl Iterator<Item = StreamType> + '_ {
43        [
44            (StreamType::Audio, self.audio_publication_active),
45            (StreamType::Camera, self.camera_publication_active),
46            (StreamType::Screen, self.screen_publication_active),
47        ]
48        .into_iter()
49        .filter_map(|(stream_type, active)| active.then_some(stream_type))
50    }
51
52    pub(super) fn remember_subscription_states(
53        &mut self,
54        user_id: &UserId,
55        states: &DownloadStates,
56    ) {
57        let existing_states = self
58            .desired_subscriptions
59            .entry(user_id.clone())
60            .or_default();
61        existing_states.apply_partial_update(states);
62        if *existing_states == DownloadStates::default() {
63            self.desired_subscriptions.remove(user_id);
64        }
65    }
66
67    pub(super) fn remember_info(&mut self, info: &UserInfo) {
68        let existing_info = self.desired_info.get_or_insert_with(UserInfo::default);
69        let is_featured = existing_info.is_featured;
70        existing_info.apply_partial_update(info);
71        existing_info.is_featured = is_featured;
72    }
73
74    pub(super) fn replay_session_batch(&self) -> Option<EnvelopeBatch> {
75        let mut replay_batch = Vec::new();
76
77        for (user_id, states) in &self.desired_subscriptions {
78            let Some(envelope) =
79                ClientEnvelope::Message(ClientMessage::Subscribe(SubscribePayload {
80                    user_id: user_id.clone(),
81                    states: states.clone(),
82                }))
83                .into_envelope()
84                .ok()
85            else {
86                continue;
87            };
88            replay_batch.push(envelope);
89        }
90
91        if let Some(info) = self.desired_info.clone() {
92            let envelope = ClientEnvelope::Message(ClientMessage::Info(info))
93                .into_envelope()
94                .ok()?;
95            replay_batch.push(envelope);
96        }
97
98        if replay_batch.is_empty() {
99            None
100        } else {
101            Some(replay_batch)
102        }
103    }
104}
o_sfu_protocol/core/sticky_replay.rs

o_sfu_protocol/core/
sticky_replay.rs
