o_sfu_core/engine/room/transition/
subscription.rs

1//! Receiver intent and consumer-route realization.
2//!
3//! Intent survives a missing publication or unready receiver. Readiness reserves
4//! routes under the room lock, declares transport without it and commits only if
5//! the reservation still matches the publication and receiver connection.
6
7use std::collections::BTreeMap;
8
9use o_sfu_router::rtp::MediaCapabilities;
10
11use super::super::{
12    RoomUserOperation,
13    effects::batch::{RoomEffectContext, RoomEffects},
14};
15use crate::engine::{
16    UserId,
17    media_transport::TransportMediaId,
18    source_model::{SourceSubscriptionIntent, UserStreamId},
19};
20
21impl RoomUserOperation<'_> {
22    pub(crate) async fn apply_session_negotiated(
23        self,
24        capabilities: MediaCapabilities,
25        declined_consumers: &[TransportMediaId],
26    ) -> Option<()> {
27        let became_ready = {
28            let mut state = self.room.state.write().await;
29            state.set_user_negotiated(self.user_id, self.connection_id, capabilities)
30        }?;
31        if became_ready {
32            self.apply_receiver_readiness(declined_consumers).await
33        } else {
34            Some(())
35        }
36    }
37
38    pub(crate) async fn apply_session_refreshed(
39        self,
40        declined_consumers: &[TransportMediaId],
41    ) -> Option<()> {
42        self.apply_receiver_readiness(declined_consumers).await
43    }
44
45    async fn apply_receiver_readiness(self, declined_consumers: &[TransportMediaId]) -> Option<()> {
46        let commit = {
47            let mut state = self.room.state.write().await;
48            state.refresh_consumer_readiness(self.user_id, self.connection_id, declined_consumers)
49        };
50        let commit = commit?;
51        RoomEffects::from_consumer_readiness(commit)
52            .execute(self.room, RoomEffectContext::runtime(self.media_transport))
53            .await;
54        Some(())
55    }
56
57    pub(crate) async fn apply_receiver_intent(
58        self,
59        target_user_id: &UserId,
60        intents: &BTreeMap<UserStreamId, SourceSubscriptionIntent>,
61    ) -> Option<()> {
62        let commit = self.room.state.write().await.apply_receiver_intent(
63            self.user_id,
64            self.connection_id,
65            target_user_id,
66            intents,
67        )?;
68        self.room
69            .metrics
70            .record_subscription_intent_evictions(commit.work.evictions);
71        RoomEffects::from_receiver_intent(commit)
72            .execute(self.room, RoomEffectContext::runtime(self.media_transport))
73            .await;
74        Some(())
75    }
76}
77
78#[cfg(test)]
79#[path = "TESTS/subscription.rs"]
80mod tests;
o_sfu_core/engine/room/transition/subscription.rs

o_sfu_core/engine/room/transition/
subscription.rs
