o_sfu_core/engine/room/media_graph/
subscription.rs

1use std::collections::{BTreeMap, BTreeSet};
2
3use o_sfu_router::{MediaKind as RouterMediaKind, negotiation::negotiate_consumer_rtp_parameters};
4
5use super::{
6    super::state::{ActiveUser, RoomState},
7    ConsumerId, ConsumerRouteTarget, SubscriptionKey,
8    consumer_setup::{ConsumerSetupTarget, PendingConsumerSetup},
9};
10use crate::engine::{
11    ConnectionId, UserId,
12    media_transport::{TransportMediaId, TransportRelayRouteEffect, TransportTeardown},
13    room::{
14        outbound::{OutboundSender, VersionedRemoteTrackSnapshot},
15        source_policy::VideoAdmissionRank,
16    },
17    source_model::{
18        ConsumerSourceSelection, PolicyPauseReason, PublishedSourceId, SourceRoutePriority,
19        SourceSubscriptionIntent, UserStreamId,
20    },
21};
22
23#[derive(Debug, Clone, PartialEq, Eq)]
24pub struct ReceiverRouteActivity {
25    target: ConsumerRouteTarget,
26    active: bool,
27}
28
29impl ReceiverRouteActivity {
30    pub const fn new(target: ConsumerRouteTarget, active: bool) -> Self {
31        Self { target, active }
32    }
33
34    pub const fn target(&self) -> &ConsumerRouteTarget {
35        &self.target
36    }
37
38    pub const fn active(&self) -> bool {
39        self.active
40    }
41}
42
43/// observable receiver route state for room inspection
44#[cfg(any(test, feature = "testing-transport"))]
45#[derive(Debug, Clone, Copy, PartialEq, Eq)]
46pub enum ConsumerRouteState {
47    Absent,
48    Inactive,
49    Active,
50}
51
52#[derive(Debug, Default)]
53pub struct ReceiverRouteWork {
54    pub(in crate::engine::room) activities: Vec<ReceiverRouteActivity>,
55    pub(in crate::engine::room) setups: Vec<PendingConsumerSetup>,
56    pub(in crate::engine::room) relays: Vec<TransportRelayRouteEffect>,
57    pub(in crate::engine::room) teardown: Vec<TransportTeardown>,
58}
59
60#[derive(Debug)]
61pub struct ReceiverRouteCommit {
62    pub(in crate::engine::room) work: ReceiverRouteWork,
63    pub(in crate::engine::room) track_snapshots:
64        Vec<(OutboundSender, VersionedRemoteTrackSnapshot)>,
65}
66
67#[derive(Clone, Copy)]
68pub(super) enum ReceiverRouteScope<'a> {
69    Source(PublishedSourceId),
70    Receiver(&'a UserId, ConnectionId),
71    SourceUser(&'a UserId, ConnectionId, &'a UserId),
72}
73
74impl RoomState {
75    pub fn apply_receiver_intent(
76        &mut self,
77        user_id: &UserId,
78        connection_id: ConnectionId,
79        target_user_id: &UserId,
80        intents: &BTreeMap<UserStreamId, SourceSubscriptionIntent>,
81    ) -> Option<ReceiverRouteCommit> {
82        self.user_for_connection(user_id, connection_id)?;
83        let work =
84            self.plan_receiver_intent_change(user_id, connection_id, target_user_id, intents);
85        Some(ReceiverRouteCommit {
86            work,
87            track_snapshots: Vec::new(),
88        })
89    }
90
91    #[cfg(test)]
92    pub fn plan_receiver_route_work(
93        &mut self,
94        user_id: &UserId,
95        connection_id: ConnectionId,
96        target_user_id: &UserId,
97        intents: &BTreeMap<UserStreamId, SourceSubscriptionIntent>,
98    ) -> ReceiverRouteWork {
99        if self.user_for_connection(user_id, connection_id).is_none() {
100            return ReceiverRouteWork::default();
101        }
102        self.plan_receiver_intent_change(user_id, connection_id, target_user_id, intents)
103    }
104
105    fn plan_receiver_intent_change(
106        &mut self,
107        user_id: &UserId,
108        connection_id: ConnectionId,
109        target_user_id: &UserId,
110        intents: &BTreeMap<UserStreamId, SourceSubscriptionIntent>,
111    ) -> ReceiverRouteWork {
112        for (stream_id, intent) in intents {
113            self.topology.merge_subscription_intent(
114                SubscriptionKey::new(user_id, target_user_id, stream_id),
115                *intent,
116            );
117        }
118        let (updates, relays) =
119            self.apply_route_updates(user_id, connection_id, target_user_id, intents);
120        let ReceiverRouteWork { setups, .. } = self.plan_missing_receiver_routes(
121            ReceiverRouteScope::SourceUser(user_id, connection_id, target_user_id),
122        );
123        ReceiverRouteWork {
124            activities: updates,
125            setups,
126            relays,
127            ..Default::default()
128        }
129    }
130
131    pub fn refresh_consumer_readiness(
132        &mut self,
133        user_id: &UserId,
134        connection_id: ConnectionId,
135        declined_consumers: &[TransportMediaId],
136    ) -> Option<ReceiverRouteCommit> {
137        let sender = {
138            let user = self.user_for_connection(user_id, connection_id)?;
139            user.parsed_client_rtp_capabilities.as_ref()?;
140            (!declined_consumers.is_empty()).then(|| user.sender.clone())
141        };
142        let session = self
143            .topology
144            .transport_user_key(user_id.clone(), connection_id);
145        // Keep declined routes committed while planning so the answer turn cannot
146        // immediately recreate the same media lines.
147        let mut work =
148            self.plan_missing_receiver_routes(ReceiverRouteScope::Receiver(user_id, connection_id));
149        let (relays, teardown, detached) = self
150            .topology
151            .detach_declined_consumers(&session, declined_consumers);
152        work.relays.extend(relays);
153        work.teardown.extend(teardown);
154        let track_snapshots = sender
155            .filter(|_| detached)
156            .map(|sender| (sender, self.remote_track_snapshot_for_user(user_id, false)))
157            .into_iter()
158            .collect();
159        Some(ReceiverRouteCommit {
160            work,
161            track_snapshots,
162        })
163    }
164
165    #[cfg(test)]
166    pub fn plan_missing_consumers(
167        &mut self,
168        user_id: &UserId,
169        connection_id: ConnectionId,
170    ) -> Option<Vec<PendingConsumerSetup>> {
171        let user = self.user_for_connection(user_id, connection_id)?;
172        if user.parsed_client_rtp_capabilities.is_none() {
173            return Some(Vec::new());
174        }
175        let ReceiverRouteWork { setups, .. } =
176            self.plan_missing_receiver_routes(ReceiverRouteScope::Receiver(user_id, connection_id));
177        Some(setups)
178    }
179
180    pub(super) fn plan_missing_receiver_routes(
181        &mut self,
182        scope: ReceiverRouteScope<'_>,
183    ) -> ReceiverRouteWork {
184        let targets = self.missing_receiver_route_targets(scope);
185        ReceiverRouteWork {
186            setups: self.plan_consumers(targets),
187            ..Default::default()
188        }
189    }
190
191    fn plan_consumers(
192        &mut self,
193        mut targets: Vec<ConsumerSetupTarget>,
194    ) -> Vec<PendingConsumerSetup> {
195        // Initial admission has no active-speaker snapshot. The following
196        // source-policy turn applies the current speaker ranking.
197        let active_speakers = BTreeSet::new();
198        targets.sort_by_key(|target| self.setup_rank(target, &active_speakers));
199        targets
200            .into_iter()
201            .filter_map(|target| self.plan_consumer(target))
202            .collect()
203    }
204
205    fn setup_rank(
206        &self,
207        target: &ConsumerSetupTarget,
208        active_speakers: &BTreeSet<UserId>,
209    ) -> VideoAdmissionRank {
210        if target.kind != RouterMediaKind::Video {
211            return VideoAdmissionRank::new(
212                SourceRoutePriority::PinnedOrFeatured,
213                None,
214                target.source_id,
215            );
216        }
217        let Some(source) = self.topology.source_descriptor(target.source_id) else {
218            return VideoAdmissionRank::new(
219                SourceRoutePriority::HiddenOrOverflow,
220                None,
221                target.source_id,
222            );
223        };
224        VideoAdmissionRank::new(
225            self.receiver_video_layout_role(target.session.user_id(), source, active_speakers)
226                .priority(),
227            None,
228            target.source_id,
229        )
230    }
231
232    fn apply_route_updates(
233        &mut self,
234        user_id: &UserId,
235        connection_id: ConnectionId,
236        target_user_id: &UserId,
237        intents: &BTreeMap<UserStreamId, SourceSubscriptionIntent>,
238    ) -> (Vec<ReceiverRouteActivity>, Vec<TransportRelayRouteEffect>) {
239        let mut updates = Vec::new();
240        let mut relays = Vec::new();
241        let receiver_deafened = self
242            .user_for_connection(user_id, connection_id)
243            .is_some_and(ActiveUser::is_deaf);
244        for (stream_id, intent) in intents {
245            let Some(active) = intent.active() else {
246                continue;
247            };
248            let Some(commit) = self.topology.set_consumer_activity(
249                user_id,
250                connection_id,
251                target_user_id,
252                stream_id,
253                active,
254                active && receiver_deafened,
255            ) else {
256                continue;
257            };
258            relays.extend(commit.relay_effects);
259            if let Some(update) = commit.update {
260                updates.push(update);
261            }
262        }
263        (updates, relays)
264    }
265
266    fn missing_receiver_route_targets(
267        &mut self,
268        scope: ReceiverRouteScope<'_>,
269    ) -> Vec<ConsumerSetupTarget> {
270        match scope {
271            ReceiverRouteScope::Source(source_id) => {
272                let users = &self.users;
273                self.topology.missing_consumer_targets_for_source(
274                    source_id,
275                    users
276                        .iter()
277                        .map(|(user, state)| (user, state.connection_id)),
278                )
279            }
280            ReceiverRouteScope::Receiver(user, connection) => self
281                .topology
282                .missing_consumer_targets(user, connection, |_| true),
283            ReceiverRouteScope::SourceUser(user, connection, source_user_id) => self
284                .topology
285                .missing_consumer_targets(user, connection, |source| {
286                    source.descriptor.owner().user_id() == source_user_id
287                }),
288        }
289    }
290
291    fn plan_consumer(&mut self, target: ConsumerSetupTarget) -> Option<PendingConsumerSetup> {
292        let (sender, client_caps) = {
293            let user = self.users.get(target.session.user_id())?;
294            if user.connection_id != target.session.connection_id() {
295                return None;
296            }
297            (
298                user.sender.clone(),
299                user.parsed_client_rtp_capabilities.as_ref()?,
300            )
301        };
302        let source = self.topology.published_source(target.source_id)?;
303        if !target.matches_identity(source) {
304            return None;
305        }
306        let selection = self.setup_selection(&target, source.active);
307        let rtp = negotiate_consumer_rtp_parameters(&source.rtp, client_caps).ok()?;
308        let consumer = ConsumerId::allocate(&mut self.next_consumer_id);
309        self.topology
310            .reserve_consumer_setup(target, consumer, selection, sender, rtp)
311    }
312
313    pub(super) fn setup_selection(
314        &self,
315        target: &ConsumerSetupTarget,
316        source_active: bool,
317    ) -> ConsumerSourceSelection {
318        let key = target.subscription_key();
319        let selection = self
320            .topology
321            .consumer_source_selection(&key, target.source_id)
322            .unwrap_or_else(|| {
323                ConsumerSourceSelection::open(
324                    self.topology
325                        .subscription_intent(&key)
326                        .active()
327                        .unwrap_or(true),
328                )
329            });
330        let selection = self.apply_initial_video_download_cap(target, source_active, selection);
331        self.apply_initial_receiver_deafened(target, selection)
332    }
333
334    fn apply_initial_receiver_deafened(
335        &self,
336        target: &ConsumerSetupTarget,
337        mut selection: ConsumerSourceSelection,
338    ) -> ConsumerSourceSelection {
339        if target.kind == RouterMediaKind::Audio
340            && selection.delivery_active()
341            && self
342                .user_for_connection(target.session.user_id(), target.session.connection_id())
343                .is_some_and(ActiveUser::is_deaf)
344        {
345            selection.set_policy_pause_reason(Some(PolicyPauseReason::ReceiverDeafened));
346        }
347        selection
348    }
349
350    fn apply_initial_video_download_cap(
351        &self,
352        target: &ConsumerSetupTarget,
353        source_active: bool,
354        mut selection: ConsumerSourceSelection,
355    ) -> ConsumerSourceSelection {
356        if target.kind != RouterMediaKind::Video
357            || !source_active
358            || !selection.delivery_active()
359            || self.active_video_count(target.session.user_id())
360                < self.media_limits.max_video_downloads_per_receiver()
361        {
362            return selection;
363        }
364        selection.set_policy_pause_reason(Some(PolicyPauseReason::VideoDownloadLimit));
365        selection
366    }
367
368    fn active_video_count(&self, consumer_user_id: &UserId) -> usize {
369        let committed = self
370            .topology
371            .committed_consumer_routes_for_user(consumer_user_id)
372            .filter(|route| route.source.descriptor.media_kind() == RouterMediaKind::Video)
373            .filter(|route| route.source.active)
374            .filter(|route| route.selection.delivery_active())
375            .count();
376        // Delivery-active reservations already claim download capacity. Counting
377        // them prevents concurrent setup from oversubscribing the receiver cap.
378        let pending = self
379            .topology
380            .pending_consumer_routes_for_user(consumer_user_id)
381            .filter(|route| route.source.descriptor.media_kind() == RouterMediaKind::Video)
382            .filter(|route| route.source.active)
383            .filter(|route| route.selection.delivery_active())
384            .count();
385        committed + pending
386    }
387
388    #[cfg(any(test, feature = "testing-transport"))]
389    pub fn consumer_route_state(
390        &self,
391        consumer_user_id: &UserId,
392        producer_user_id: &UserId,
393        stream_id: &UserStreamId,
394    ) -> Option<ConsumerRouteState> {
395        self.users.get(consumer_user_id)?;
396        let key = SubscriptionKey::new(consumer_user_id, producer_user_id, stream_id);
397        let Some(route) = self.topology.committed_consumer_route_for_key(&key) else {
398            return Some(ConsumerRouteState::Absent);
399        };
400        let route_active = route.source.active && route.selection.delivery_active();
401        Some(if route_active {
402            ConsumerRouteState::Active
403        } else {
404            ConsumerRouteState::Inactive
405        })
406    }
407}
o_sfu_core/engine/room/media_graph/subscription.rs

o_sfu_core/engine/room/media_graph/
subscription.rs
