o_sfu_core/engine/room/media_graph/
consumer_setup.rs

1use std::mem;
2
3use o_sfu_router::{
4    MediaKind as RouterMediaKind, rtp::MediaStream as RouterRtpParameters,
5    topology::RoutedProducerId,
6};
7use tracing::warn;
8
9use super::{
10    super::{
11        outbound::{OutboundSender, VersionedRemoteTrackSnapshot},
12        state::RoomState,
13    },
14    ConsumerId, ConsumerRouteTarget, PublishedSource, SubscriptionKey,
15    route_graph::{ConsumerRouteReservation, RelayRouteKey},
16};
17use crate::engine::{
18    MediaWorkerId,
19    media_transport::{
20        MediaTransport, ProducerActivity, SourceActivityUpdate, TransportConsumerRoute,
21        TransportMediaId, TransportRelayRouteEffect, TransportSessionKey,
22        TransportSourceActivityEffect, TransportSourceKey,
23    },
24    source_model::{PublishedSourceId, SourceSyncPolicy, UserStreamId},
25};
26
27#[derive(Debug)]
28pub struct ConsumerSetupTarget {
29    pub session: TransportSessionKey,
30    pub source: TransportSourceKey,
31    pub source_id: PublishedSourceId,
32    pub stream: UserStreamId,
33    pub kind: RouterMediaKind,
34    pub sync: SourceSyncPolicy,
35    pub routed: RoutedProducerId,
36}
37
38/// Consumer-route reservation awaiting transport declaration.
39///
40/// It carries graph and optional relay ownership across work performed without
41/// the room lock. Callers must commit or release it after declaration.
42#[derive(Debug)]
43#[must_use = "pending consumer setups reserve route graph state and must be committed or released"]
44pub struct PendingConsumerSetup {
45    pub(super) target: ConsumerSetupTarget,
46    pub(super) consumer: ConsumerId,
47    pub(super) reservation: ConsumerRouteReservation,
48    pub(super) sender: OutboundSender,
49    pub(super) rtp: RouterRtpParameters,
50    pub(super) relays: Vec<TransportRelayRouteEffect>,
51}
52
53pub struct DeclaredConsumerSetup {
54    pub(super) pending: PendingConsumerSetup,
55    pub(super) route: TransportConsumerRoute,
56    pub(super) mid: Option<String>,
57}
58
59pub struct CommittedConsumerSetup {
60    pub(super) target: ConsumerSetupTarget,
61    pub(super) route: TransportConsumerRoute,
62    pub(super) sender: OutboundSender,
63    pub(super) transport_activity_update: Option<bool>,
64}
65
66#[expect(
67    clippy::large_enum_variant,
68    reason = "consumer setup outcomes are returned and matched immediately so boxing the committed setup would allocate on every successful consumer setup"
69)]
70#[derive(Debug)]
71pub enum ConsumerSetupOutcome {
72    Committed {
73        target: ConsumerSetupTarget,
74        route: TransportConsumerRoute,
75        sender: OutboundSender,
76        track_snapshot: VersionedRemoteTrackSnapshot,
77        remote_source_activity: Option<TransportSourceActivityEffect>,
78        transport_activity_update: Option<bool>,
79        readiness_keyframe: Option<ConsumerRouteTarget>,
80    },
81    Released(TransportConsumerRoute, Vec<TransportRelayRouteEffect>),
82}
83
84#[derive(Debug, Clone, Copy)]
85pub enum ConsumerSetupOrigin {
86    Readiness,
87    Publish,
88    Subscribe,
89}
90
91impl ConsumerSetupOrigin {
92    pub const fn as_diagnostic_str(self) -> &'static str {
93        match self {
94            Self::Readiness => "readiness",
95            Self::Publish => "publish",
96            Self::Subscribe => "subscribe",
97        }
98    }
99}
100
101impl RoomState {
102    /// Commits a transport-declared consumer route.
103    ///
104    /// Returns [`ConsumerSetupOutcome::Released`] when receiver or source
105    /// identity changed during declaration or the router rejects the dependency.
106    pub fn commit_declared_consumer_setup(
107        &mut self,
108        setup: DeclaredConsumerSetup,
109        origin: ConsumerSetupOrigin,
110    ) -> ConsumerSetupOutcome {
111        let target = &setup.pending.target;
112        let session = &target.session;
113        if self
114            .user_for_connection(session.user_id(), session.connection_id())
115            .is_some_and(|user| user.parsed_client_rtp_capabilities.is_some())
116            && let Some((source_active, source_activity_revision)) = self
117                .topology
118                .published_source(target.source_id)
119                .filter(|source| target.matches_identity(source))
120                .map(|source| (source.active, source.activity_revision))
121        {
122            // Transport declaration ran without the room lock. Recompute from
123            // current intent and policy before committing the pending route.
124            let selection = self.setup_selection(target, source_active);
125            let delivery_active = selection.delivery_active();
126            match self.topology.commit_consumer_setup(setup, selection) {
127                Ok(commit) => {
128                    // A consumer on another media worker must be told the source
129                    // activity because that worker does not host the producer.
130                    let remote_source_activity =
131                        (commit.route.source().session_key().media_worker_id()
132                            != commit.route.consumer_session_key().media_worker_id())
133                        .then(|| TransportSourceActivityEffect {
134                            source: commit.route.source().clone(),
135                            target_media_worker_id: commit
136                                .route
137                                .consumer_session_key()
138                                .media_worker_id(),
139                            update: SourceActivityUpdate::new(
140                                ProducerActivity::from_active(source_active),
141                                source_activity_revision,
142                            ),
143                        });
144                    let track_snapshot =
145                        self.remote_track_snapshot_for_user(commit.target.session.user_id(), true);
146                    // A pending activity correction to active already requests a
147                    // keyframe so avoid a duplicate readiness refresh.
148                    let readiness_keyframe = match origin {
149                        ConsumerSetupOrigin::Readiness
150                            if delivery_active
151                                && commit.transport_activity_update != Some(true)
152                                && commit.target.kind == RouterMediaKind::Video =>
153                        {
154                            Some(commit.target.route_target(commit.route.clone()))
155                        }
156                        ConsumerSetupOrigin::Readiness
157                        | ConsumerSetupOrigin::Publish
158                        | ConsumerSetupOrigin::Subscribe => None,
159                    };
160                    ConsumerSetupOutcome::Committed {
161                        target: commit.target,
162                        route: commit.route,
163                        sender: commit.sender,
164                        track_snapshot,
165                        remote_source_activity,
166                        transport_activity_update: commit.transport_activity_update,
167                        readiness_keyframe,
168                    }
169                }
170                Err((route, relays)) => ConsumerSetupOutcome::Released(route, relays),
171            }
172        } else {
173            let DeclaredConsumerSetup { pending, route, .. } = setup;
174            ConsumerSetupOutcome::Released(route, self.topology.release_consumer_setup(pending))
175        }
176    }
177
178    pub fn release_pending_consumer_setup(
179        &mut self,
180        setup: PendingConsumerSetup,
181    ) -> Vec<TransportRelayRouteEffect> {
182        self.topology.release_consumer_setup(setup)
183    }
184}
185
186impl PendingConsumerSetup {
187    pub(in crate::engine::room) fn take_relays(&mut self) -> Vec<TransportRelayRouteEffect> {
188        mem::take(&mut self.relays)
189    }
190
191    pub(in crate::engine::room) async fn declare(
192        self,
193        media_transport: &MediaTransport,
194        origin: ConsumerSetupOrigin,
195    ) -> Result<DeclaredConsumerSetup, Self> {
196        match media_transport
197            .consume_media_with_mid(
198                &self.target.session,
199                self.target.kind,
200                &self.target.source,
201                self.target.sync,
202                &self.rtp,
203                self.reservation.declared_activity(),
204            )
205            .await
206        {
207            Ok((media, mid)) => Ok(DeclaredConsumerSetup {
208                route: self.target.transport_consumer_route(media),
209                pending: self,
210                mid: Some(mid),
211            }),
212            Err(error) => {
213                warn!(
214                    consumer_user_id = ?self.target.session.user_id(),
215                    consumer_connection_id = ?self.target.session.connection_id(),
216                    producer_user_id = ?self.target.source.session_key().user_id(),
217                    producer_connection_id = ?self.target.source.session_key().connection_id(),
218                    source_transport_media_id = ?self.target.source.transport_media_id(),
219                    error = ?error,
220                    consumer_mid = self.rtp.mid(),
221                    ?origin,
222                    "media transport rejected consume media declaration"
223                );
224                Err(self)
225            }
226        }
227    }
228}
229
230impl ConsumerSetupTarget {
231    pub fn new(session: TransportSessionKey, source: &PublishedSource) -> Self {
232        Self {
233            session,
234            source: source.transport.clone(),
235            source_id: source.descriptor.source_id(),
236            stream: source.descriptor.stream_id().clone(),
237            kind: source.descriptor.media_kind(),
238            sync: source.descriptor.policy().sync(),
239            routed: source.routed,
240        }
241    }
242
243    pub(super) fn subscription_key(&self) -> SubscriptionKey {
244        SubscriptionKey::new(
245            self.session.user_id(),
246            self.source.session_key().user_id(),
247            &self.stream,
248        )
249    }
250
251    pub(super) fn transport_consumer_route(
252        &self,
253        consumer_media: TransportMediaId,
254    ) -> TransportConsumerRoute {
255        TransportConsumerRoute::new(self.session.clone(), consumer_media, self.source.clone())
256    }
257
258    fn route_target(&self, route: TransportConsumerRoute) -> ConsumerRouteTarget {
259        ConsumerRouteTarget::new(route, self.stream.clone(), self.kind)
260    }
261
262    pub(super) fn relay_route_key(&self, target_worker: MediaWorkerId) -> RelayRouteKey {
263        RelayRouteKey {
264            source: self.source.clone(),
265            target_worker,
266        }
267    }
268
269    pub(super) fn matches_identity(&self, source: &PublishedSource) -> bool {
270        source.descriptor.source_id() == self.source_id
271            && source.transport == self.source
272            && source.descriptor.stream_id() == &self.stream
273            && source.descriptor.media_kind() == self.kind
274            && source.routed == self.routed
275    }
276}
o_sfu_core/engine/room/media_graph/consumer_setup.rs

o_sfu_core/engine/room/media_graph/
consumer_setup.rs
