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: EnvelopeBatch = self
76            .desired_subscriptions
77            .iter()
78            .filter_map(|(user_id, states)| {
79                ClientEnvelope::Message(ClientMessage::Subscribe(SubscribePayload {
80                    user_id: user_id.clone(),
81                    states: states.clone(),
82                }))
83                .into_envelope()
84                .ok()
85            })
86            .collect();
87        if let Some(info) = self.desired_info.clone() {
88            let envelope = ClientEnvelope::Message(ClientMessage::Info(info))
89                .into_envelope()
90                .ok()?;
91            replay_batch.push(envelope);
92        }
93        if replay_batch.is_empty() {
94            None
95        } else {
96            Some(replay_batch)
97        }
98    }
99}
o_sfu_protocol/core/sticky_replay.rs

o_sfu_protocol/core/
sticky_replay.rs
