o_sfu_core/engine/room/media_graph/
source_index.rs

1use std::collections::{BTreeMap, BTreeSet};
2
3use super::{PublishedSource, SourceKey, remove_from_index_set};
4#[cfg(any(test, feature = "testing-transport"))]
5use crate::engine::ConnectionId;
6use crate::engine::{
7    UserId,
8    media_transport::TransportMediaId,
9    source_model::{
10        ActiveSpeakerGroup, ActiveSpeakerSourceRole, PublishedSourceId, SourceEncodingId,
11        UserStreamId,
12    },
13};
14
15#[derive(Debug)]
16pub(super) struct PublishedSources {
17    records: BTreeMap<PublishedSourceId, PublishedSource>,
18    id_by_key: BTreeMap<SourceKey, PublishedSourceId>,
19    ids_by_owner: BTreeMap<UserId, BTreeSet<PublishedSourceId>>,
20    id_by_transport: BTreeMap<TransportMediaId, PublishedSourceId>,
21    next_id: u64,
22    next_encoding_id: u64,
23}
24
25impl Default for PublishedSources {
26    fn default() -> Self {
27        Self {
28            records: BTreeMap::new(),
29            id_by_key: BTreeMap::new(),
30            ids_by_owner: BTreeMap::new(),
31            id_by_transport: BTreeMap::new(),
32            next_id: 1,
33            next_encoding_id: 1,
34        }
35    }
36}
37
38impl PublishedSources {
39    pub(super) fn allocate_id(&mut self) -> PublishedSourceId {
40        PublishedSourceId::allocate(&mut self.next_id)
41    }
42
43    pub(super) fn allocate_encoding_id(&mut self) -> SourceEncodingId {
44        SourceEncodingId::allocate(&mut self.next_encoding_id)
45    }
46
47    pub(super) fn publication_count(&self) -> usize {
48        self.records.len()
49    }
50
51    pub(super) fn iter(&self) -> impl Iterator<Item = &PublishedSource> {
52        self.records.values()
53    }
54
55    pub(super) fn source(&self, id: PublishedSourceId) -> Option<&PublishedSource> {
56        self.records.get(&id)
57    }
58
59    pub(super) fn source_mut(&mut self, id: PublishedSourceId) -> Option<&mut PublishedSource> {
60        self.records.get_mut(&id)
61    }
62
63    pub(super) fn source_for_transport(&self, media: TransportMediaId) -> Option<&PublishedSource> {
64        self.source(*self.id_by_transport.get(&media)?)
65    }
66
67    #[cfg(any(test, feature = "testing-transport"))]
68    pub(super) fn first_transport_media_id(&self) -> Option<TransportMediaId> {
69        self.records
70            .values()
71            .next()
72            .map(|source| source.transport.transport_media_id())
73    }
74
75    #[cfg(any(test, feature = "testing-transport"))]
76    pub(super) fn transport_media_id(
77        &self,
78        user: &UserId,
79        connection: ConnectionId,
80        stream: &UserStreamId,
81    ) -> Option<TransportMediaId> {
82        let source = self.source(*self.id_by_key.get(&SourceKey::new(user, stream))?)?;
83        (source.transport.session_key().connection_id() == connection)
84            .then(|| source.transport.transport_media_id())
85    }
86
87    pub(super) fn id_for_owner_stream(
88        &self,
89        owner: &UserId,
90        stream: &UserStreamId,
91    ) -> Option<PublishedSourceId> {
92        self.id_by_key.get(&SourceKey::new(owner, stream)).copied()
93    }
94
95    pub(super) fn owner_has_promotable_source_in_group(
96        &self,
97        owner: &UserId,
98        group: ActiveSpeakerGroup,
99    ) -> bool {
100        self.ids_for_owner(owner)
101            .filter_map(|id| self.source(id))
102            .filter(|source| source.active)
103            .map(|source| &source.descriptor)
104            .any(|source| {
105                source.policy().active_speaker().is_some_and(|policy| {
106                    policy.group() == group && policy.role() == ActiveSpeakerSourceRole::Promotable
107                })
108            })
109    }
110
111    pub(super) fn insert(&mut self, source: PublishedSource) {
112        let id = source.descriptor.source_id();
113        let owner = source.descriptor.owner().user_id().clone();
114        self.id_by_key
115            .insert(SourceKey::new(&owner, source.descriptor.stream_id()), id);
116        self.ids_by_owner.entry(owner).or_default().insert(id);
117        self.id_by_transport
118            .insert(source.transport.transport_media_id(), id);
119        self.records.insert(id, source);
120    }
121
122    pub(super) fn ids_for_owner(
123        &self,
124        user: &UserId,
125    ) -> impl Iterator<Item = PublishedSourceId> + '_ {
126        self.ids_by_owner
127            .get(user)
128            .into_iter()
129            .flat_map(|ids| ids.iter().copied())
130    }
131
132    pub(super) fn remove(&mut self, id: PublishedSourceId) -> Option<PublishedSource> {
133        let source = self.records.remove(&id)?;
134        let owner = source.descriptor.owner().user_id();
135        self.id_by_key
136            .remove(&SourceKey::new(owner, source.descriptor.stream_id()));
137        remove_from_index_set(&mut self.ids_by_owner, owner, &id);
138        self.id_by_transport
139            .remove(&source.transport.transport_media_id());
140        Some(source)
141    }
142}
o_sfu_core/engine/room/media_graph/source_index.rs

o_sfu_core/engine/room/media_graph/
source_index.rs
