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 = {
63            let mut state = self.room.state.write().await;
64            state.apply_receiver_intent(self.user_id, self.connection_id, target_user_id, intents)
65        };
66        let commit = commit?;
67        RoomEffects::from_receiver_intent(commit)
68            .execute(self.room, RoomEffectContext::runtime(self.media_transport))
69            .await;
70        Some(())
71    }
72}
73
74#[cfg(test)]
75#[path = "TESTS/subscription.rs"]
76mod tests;
o_sfu_core/engine/room/transition/subscription.rs

o_sfu_core/engine/room/transition/
subscription.rs
