reactor_webrtc/encoded.rs
1//! Encoded-frame transform — bypass the codec to forward/receive **encoded**
2//! media (Insertable Streams / Encoded Transform).
3//!
4//! A [`FrameTransform`] attached to a transceiver's sender or receiver
5//! ([`Transceiver::set_sender_transform`](crate::Transceiver::set_sender_transform)
6//! / [`set_receiver_transform`](crate::Transceiver::set_receiver_transform))
7//! runs a closure per **encoded** frame:
8//!
9//! - **Sender** (encoder → packetizer): observe the encoded payload before it's
10//! packetized, and optionally [`replace it`](EncodedFrame::replace_data) with
11//! your own encoded bytes (forwarding). Returning [`FrameAction::Drop`] sends
12//! nothing.
13//! - **Receiver** (depacketizer → decoder): observe the encoded payload before
14//! it's decoded. Returning [`FrameAction::Drop`] **bypasses the decoder**
15//! entirely (right for a forwarding server that never renders locally);
16//! [`FrameAction::Forward`] lets the local decoder run as usual.
17//!
18//! The closure runs on a WebRTC thread and must not block it.
19//!
20//! libwebrtc allows one transformer per sender and one per receiver, so the crate
21//! owns those slots and composes: your callback runs alongside the per-frame
22//! metadata step ([`crate::metadata`]) rather than displacing it. Your callback goes
23//! first in both directions, which means it always sees exactly the bytes that
24//! traverse the network — before a trailer is appended on send, before one is
25//! stripped on receive. Returning [`FrameAction::Drop`] skips the metadata step too.
26
27use std::collections::{HashMap, VecDeque};
28use std::ffi::c_void;
29use std::ffi::CStr;
30use std::os::raw::c_int;
31use std::sync::{Arc, Mutex};
32
33use crate::media::MediaKind;
34
35/// Which side of the pipeline an [`EncodedFrame`] came from.
36#[derive(Debug, Clone, Copy, PartialEq, Eq)]
37pub enum FrameDirection {
38 /// Egress: after the encoder, before packetization.
39 Send,
40 /// Ingress: after depacketization, before the decoder.
41 Receive,
42}
43
44/// What to do with a frame after the callback returns.
45#[derive(Debug, Clone, Copy, PartialEq, Eq)]
46pub enum FrameAction {
47 /// Emit the frame downstream (send it / hand it to the decoder), including
48 /// any [`EncodedFrame::replace_data`] applied.
49 Forward,
50 /// Drop the frame: on receive this bypasses the decoder; on send nothing is
51 /// transmitted.
52 Drop,
53}
54
55/// A borrowed encoded media frame. `data` and `mime_type` are valid only for
56/// the duration of the callback — copy them to retain.
57pub struct EncodedFrame<'a> {
58 pub direction: FrameDirection,
59 pub kind: MediaKind,
60 /// Video only: whether this is a key frame (always `false` for audio).
61 pub is_key_frame: bool,
62 pub payload_type: u8,
63 pub ssrc: u32,
64 /// The frame's RTP timestamp.
65 pub timestamp: u32,
66 /// Capture timestamp in milliseconds (monotonic, same epoch as
67 /// `webrtc::TimeMicros`). Zero when unavailable.
68 pub capture_time_ms: i64,
69 /// e.g. `"video/VP8"`, `"audio/opus"`.
70 pub mime_type: &'a str,
71 /// The encoded payload.
72 pub data: &'a [u8],
73 // Opaque native frame handle, for replace_data.
74 frame: *mut c_void,
75}
76
77impl EncodedFrame<'_> {
78 /// Replace this frame's encoded payload (copied). Combine with
79 /// [`FrameAction::Forward`] to send/forward your own encoded bytes in place
80 /// of the original. No effect if `FrameAction::Drop` is returned.
81 pub fn replace_data(&self, data: &[u8]) {
82 unsafe {
83 reactor_webrtc_sys::reactor_webrtc_encoded_frame_set_data(
84 self.frame,
85 data.as_ptr(),
86 data.len(),
87 );
88 }
89 }
90
91 /// The payload as it stands now: `data` until a
92 /// [`replace_data`](Self::replace_data), then the replacement. A step that
93 /// runs after a caller's callback reads this, not `data`, or it would undo
94 /// whatever the callback rewrote.
95 pub(crate) fn current_data(&self) -> &[u8] {
96 let mut len = 0usize;
97 let ptr =
98 unsafe { reactor_webrtc_sys::reactor_webrtc_encoded_frame_data(self.frame, &mut len) };
99 if ptr.is_null() || len == 0 {
100 &[]
101 } else {
102 unsafe { std::slice::from_raw_parts(ptr, len) }
103 }
104 }
105}
106
107pub(crate) type EncodedCb = Box<dyn for<'a> FnMut(&EncodedFrame<'a>) -> FrameAction + Send>;
108
109// Heap-pinned callback state; owned by the *native* transformer (freed via
110// `free_state_tramp` when its last ref drops), so it outlives every callback
111// even if the `FrameTransform` handle is dropped while still attached.
112struct TransformState {
113 cb: Mutex<EncodedCb>,
114}
115
116extern "C" fn encoded_tramp(
117 ud: *mut c_void,
118 frame: *const reactor_webrtc_sys::ReactorEncodedFrame,
119) -> c_int {
120 // Default to Forward (0) on any error so we never silently break media.
121 let f = match unsafe { frame.as_ref() } {
122 Some(f) => f,
123 None => return 0,
124 };
125 let st = unsafe { &*(ud as *const TransformState) };
126 let data = if f.data.is_null() || f.data_len == 0 {
127 &[][..]
128 } else {
129 unsafe { std::slice::from_raw_parts(f.data, f.data_len) }
130 };
131 let mime = if f.mime_type.is_null() {
132 ""
133 } else {
134 unsafe { CStr::from_ptr(f.mime_type) }
135 .to_str()
136 .unwrap_or("")
137 };
138 let ef = EncodedFrame {
139 direction: if f.direction == 1 {
140 FrameDirection::Receive
141 } else {
142 FrameDirection::Send
143 },
144 kind: if f.is_audio == 1 {
145 MediaKind::Audio
146 } else {
147 MediaKind::Video
148 },
149 is_key_frame: f.is_key_frame == 1,
150 payload_type: f.payload_type,
151 ssrc: f.ssrc,
152 timestamp: f.timestamp,
153 capture_time_ms: f.capture_time_ms,
154 mime_type: mime,
155 data,
156 frame: f.frame,
157 };
158 let action = match st.cb.lock() {
159 Ok(mut cb) => cb(&ef),
160 Err(_) => FrameAction::Forward,
161 };
162 match action {
163 FrameAction::Forward => 0,
164 FrameAction::Drop => 1,
165 }
166}
167
168extern "C" fn free_state_tramp(ud: *mut c_void) {
169 // Reclaim the Box leaked in `FrameTransform::new`.
170 drop(unsafe { Box::from_raw(ud as *mut TransformState) });
171}
172
173/// A callback over encoded frames, to be attached to a transceiver's
174/// sender/receiver; see the [module docs](self).
175///
176/// This is a *registration*, not a native object. Attaching it does not take
177/// libwebrtc's single `SetFrameTransformer` slot — the crate owns one transformer
178/// per sender and per receiver, and composes this callback with its own
179/// frame-metadata step (see [`crate::metadata`]). That is what lets a caller use
180/// encoded-frame access and per-frame metadata on the same transceiver, which a
181/// single slot cannot express.
182///
183/// Attaching the same `FrameTransform` to more than one sender/receiver shares one
184/// callback, serialised by its mutex.
185#[derive(Clone)]
186pub struct FrameTransform {
187 cb: Arc<Mutex<EncodedCb>>,
188}
189
190impl FrameTransform {
191 /// Create a transform running `cb` per encoded frame. The closure runs on a
192 /// WebRTC thread and must not block it.
193 pub fn new(cb: impl for<'a> FnMut(&EncodedFrame<'a>) -> FrameAction + Send + 'static) -> Self {
194 Self {
195 cb: Arc::new(Mutex::new(Box::new(cb))),
196 }
197 }
198
199 pub(crate) fn callback(&self) -> Arc<Mutex<EncodedCb>> {
200 Arc::clone(&self.cb)
201 }
202}
203
204/// The native transformer: one per sender/receiver, owned by the crate.
205///
206/// The composed root that [`FrameTransform`] callbacks and the frame-metadata step
207/// both run under. Dropping this handle releases the binding's reference; the
208/// native object and its callback state live until every sender/receiver it is
209/// attached to also releases it, which is why the install path can attach and drop.
210pub(crate) struct NativeTransform {
211 raw: *mut reactor_webrtc_sys::FrameTransformer,
212}
213
214// SAFETY: the callback is Mutex-guarded and the native transformer is
215// internally thread-safe; the handle only owns a ref-counted pointer.
216unsafe impl Send for NativeTransform {}
217unsafe impl Sync for NativeTransform {}
218
219impl NativeTransform {
220 pub(crate) fn new(
221 cb: impl for<'a> FnMut(&EncodedFrame<'a>) -> FrameAction + Send + 'static,
222 ) -> Self {
223 // Leak the state; the native transformer owns it and frees it via
224 // free_state_tramp when its last ref drops.
225 let state = Box::into_raw(Box::new(TransformState {
226 cb: Mutex::new(Box::new(cb)),
227 }));
228 let raw = unsafe {
229 reactor_webrtc_sys::reactor_webrtc_frame_transformer_create(
230 encoded_tramp,
231 state as *mut c_void,
232 free_state_tramp,
233 )
234 };
235 if raw.is_null() {
236 // Creation failed: reclaim the leaked state so we don't leak it.
237 drop(unsafe { Box::from_raw(state) });
238 }
239 Self { raw }
240 }
241
242 pub(crate) fn raw(&self) -> *mut reactor_webrtc_sys::FrameTransformer {
243 self.raw
244 }
245}
246
247impl Drop for NativeTransform {
248 fn drop(&mut self) {
249 if !self.raw.is_null() {
250 unsafe { reactor_webrtc_sys::reactor_webrtc_frame_transformer_destroy(self.raw) }
251 }
252 }
253}
254
255// ── Custom video encoder ─────────────────────────────────────────────────────
256
257/// Which video codec was negotiated for the session.
258///
259/// The value mirrors `webrtc::VideoCodecType` so it round-trips through the
260/// FFI as a plain `u32`.
261#[derive(Debug, Clone, Copy, PartialEq, Eq)]
262pub enum VideoCodec {
263 Vp8 = 1,
264 Vp9 = 2,
265 Av1 = 3,
266 H264 = 4,
267 H265 = 5,
268}
269
270impl VideoCodec {
271 fn from_u32(v: u32) -> Option<Self> {
272 match v {
273 1 => Some(Self::Vp8),
274 2 => Some(Self::Vp9),
275 3 => Some(Self::Av1),
276 4 => Some(Self::H264),
277 5 => Some(Self::H265),
278 _ => None,
279 }
280 }
281
282 /// The name libwebrtc's codec capabilities carry for this codec (e.g.
283 /// `RtpCodecCapability::name`) — what
284 /// [`Transceiver::set_codec_preferences`](crate::Transceiver::set_codec_preferences)
285 /// matches against, as a plain string rather than a wire-format integer
286 /// that both sides of the FFI boundary have to keep in sync by hand.
287 pub(crate) fn name(self) -> &'static str {
288 match self {
289 Self::Vp8 => "VP8",
290 Self::Vp9 => "VP9",
291 Self::Av1 => "AV1",
292 Self::H264 => "H264",
293 Self::H265 => "H265",
294 }
295 }
296}
297
298/// Raw I420 video frame delivered to an inline encoder callback
299/// ([`TrackVideoEncoder::Inline`]).
300///
301/// Planes are slices into the native frame buffer and are **only valid for the
302/// duration of the callback**. Copy the data if your encoder is asynchronous.
303pub struct RawVideoFrame<'a> {
304 /// Which codec was negotiated — produce a matching bitstream.
305 pub codec: VideoCodec,
306 /// Luma (Y) plane.
307 pub y: &'a [u8],
308 pub y_stride: u32,
309 /// Chroma (U / Cb) plane.
310 pub u: &'a [u8],
311 pub u_stride: u32,
312 /// Chroma (V / Cr) plane.
313 pub v: &'a [u8],
314 pub v_stride: u32,
315 pub width: u32,
316 pub height: u32,
317 pub rtp_timestamp: u32,
318 /// `true` if the media engine is requesting a key frame (IDR / intra).
319 pub request_key_frame: bool,
320}
321
322/// An encoded video frame, produced by an inline encoder callback or pushed
323/// to an [`EncodedVideoTrack`].
324pub struct EncodedVideoFrame {
325 /// Raw H.264 Annex-B or AVCC bitstream bytes.
326 pub data: Vec<u8>,
327 /// `true` for IDR (key) frames.
328 pub is_key_frame: bool,
329 /// Width in pixels (0 = inherit from the raw frame).
330 pub width: u32,
331 /// Height in pixels (0 = inherit from the raw frame).
332 pub height: u32,
333 /// RTP timestamp (0 = inherit from the raw frame).
334 pub rtp_timestamp: u32,
335}
336
337/// The boxed encoder callback a [`TrackVideoEncoder::Inline`] track hands
338/// the factory — called synchronously with every raw I420 frame on the
339/// encoder thread; return `Some(encoded)` to forward, `None` to drop.
340pub type InlineEncoderCallback =
341 Box<dyn FnMut(&RawVideoFrame<'_>) -> Option<EncodedVideoFrame> + Send + 'static>;
342
343pub(crate) struct CustomEncoderState {
344 cb: Mutex<InlineEncoderCallback>,
345 feedback: Arc<FeedbackListeners>,
346}
347
348/// Borrowed view of a glue raw frame, for the inline-encoder callback.
349/// Planes are valid only for the duration of the native Encode() call.
350fn raw_view(r: &reactor_webrtc_sys::ReactorRawVideoFrame) -> RawVideoFrame<'_> {
351 let y_len = (r.y_stride.max(0) as usize) * r.height as usize;
352 let uv_len = (r.u_stride.max(0) as usize) * (r.height as usize).div_ceil(2);
353
354 RawVideoFrame {
355 codec: VideoCodec::from_u32(r.codec).unwrap_or(VideoCodec::H264),
356 y: if r.y.is_null() {
357 &[]
358 } else {
359 unsafe { std::slice::from_raw_parts(r.y, y_len) }
360 },
361 y_stride: r.y_stride as u32,
362 u: if r.u.is_null() {
363 &[]
364 } else {
365 unsafe { std::slice::from_raw_parts(r.u, uv_len) }
366 },
367 u_stride: r.u_stride as u32,
368 v: if r.v.is_null() {
369 &[]
370 } else {
371 unsafe { std::slice::from_raw_parts(r.v, uv_len) }
372 },
373 v_stride: r.v_stride as u32,
374 width: r.width,
375 height: r.height,
376 rtp_timestamp: r.rtp_timestamp,
377 request_key_frame: r.request_key_frame != 0,
378 }
379}
380
381/// Called by C++ after `EncodedImageBuffer::Create` has copied the bytes.
382extern "C" fn free_encoded_data(data: *const u8, len: usize) {
383 // Reconstruct the Vec we leaked in encode_tramp and drop it.
384 // SAFETY: this pointer+len was produced by a Vec with capacity==len
385 // (we called shrink_to_fit before forgetting it).
386 unsafe { drop(Vec::from_raw_parts(data as *mut u8, len, len)) };
387}
388
389// ── Encoder feedback ─────────────────────────────────────────────────────────
390
391/// Feedback from the BWE / rate-control machinery for a custom-encoded
392/// ([`TrackVideoEncoder`]) track's encoder instance. The builtin encoder
393/// pipeline consumes the same signals internally; custom encoders hear them
394/// here through [`VideoTrack::on_encoder_feedback`] /
395/// [`EncodedVideoTrack::on_encoder_feedback`].
396#[derive(Debug, Clone)]
397#[non_exhaustive]
398pub enum EncoderFeedback {
399 /// The remote peer (or libwebrtc's recovery logic) asked for a key frame —
400 /// PLI/FIR routed by the media engine. **You must answer with an IDR
401 /// promptly** (`EncodedVideoFrame::is_key_frame = true` on the pre-encoded
402 /// path; produce a key frame from your encoder on the inline path) or new
403 /// receivers wait a whole GOP for decodable video.
404 KeyFrameRequest,
405 /// The congestion controller's current allocation for this encoder:
406 /// adapt your own encoder's target to match. Fires on BWE changes and
407 /// `PeerConnection::set_bitrate` calls.
408 RateUpdate {
409 /// Target bitrate in bits per second for this encoder.
410 bitrate_bps: u32,
411 /// Target framerate in frames per second.
412 framerate_fps: f64,
413 },
414}
415
416/// A boxed encoder-feedback listener (single-slot per slot).
417pub(crate) type FeedbackCallback = Box<dyn FnMut(EncoderFeedback) + Send>;
418
419/// Callbacks registered for encoder feedback, boxed and shareable across the
420/// registry → track wrapper lifetime boundary.
421pub(crate) struct FeedbackListeners {
422 cb: Mutex<Option<FeedbackCallback>>,
423}
424
425impl FeedbackListeners {
426 pub(crate) fn new() -> Arc<Self> {
427 Arc::new(Self {
428 cb: Mutex::new(None),
429 })
430 }
431
432 /// Overwrite the listener. Latest call wins (sinks and callbacks are
433 /// single-slot in this API).
434 pub(crate) fn set(&self, cb: FeedbackCallback) {
435 *self.cb.lock().unwrap() = Some(cb);
436 }
437
438 /// Deliver feedback to the registered listener, if any. Runs on
439 /// whatever WebRTC thread the signal arrived on — the callback re-enters
440 /// the app synchronously there, like every other sink.
441 fn fire(&self, feedback: EncoderFeedback) {
442 if let Some(cb) = self.cb.lock().unwrap().as_mut() {
443 cb(feedback);
444 }
445 }
446}
447
448/// Native-track identity → the encoder feedback listeners of the slot that
449/// track occupies. Mirrors the sender_meta registration pattern: created
450/// when a track with a custom encoder slot is created, removed at drop.
451type FeedbackBindings = Mutex<std::collections::HashMap<usize, Arc<FeedbackListeners>>>;
452
453fn feedback_bindings() -> &'static FeedbackBindings {
454 static BINDINGS: std::sync::OnceLock<FeedbackBindings> = std::sync::OnceLock::new();
455 BINDINGS.get_or_init(FeedbackBindings::default)
456}
457
458/// Bind a track's feedback listeners to its native-track identity.
459pub(crate) fn register_feedback_binding(native_id: usize, listeners: &Arc<FeedbackListeners>) {
460 if native_id != 0 {
461 feedback_bindings()
462 .lock()
463 .unwrap()
464 .insert(native_id, listeners.clone());
465 }
466}
467
468/// Remove the binding for `native_id` (the track is gone).
469pub(crate) fn deregister_feedback_binding(native_id: usize) {
470 feedback_bindings().lock().unwrap().remove(&native_id);
471}
472
473/// The feedback listener registration for `native_id`, if the track's slot
474/// has one (only tracks created with a [`TrackVideoEncoder`] have some).
475/// Used by [`VideoTrack::on_encoder_feedback`] and
476/// [`EncodedVideoTrack::on_encoder_feedback`].
477pub(crate) fn feedback_binding(native_id: usize) -> Option<Arc<FeedbackListeners>> {
478 feedback_bindings().lock().unwrap().get(&native_id).cloned()
479}
480
481// ── Multi-track encoder registry ─────────────────────────────────────────────
482
483/// A pending slot for one video transceiver in an [`EncoderRegistry`].
484///
485/// - `Custom` — frames are read from the associated queue (push via
486/// [`EncodedVideoTrack`]); the queue drain itself is the "encoder", and the
487/// feedback listeners include everything the pull-based API cannot say.
488/// - `Inline` — the registry calls the user callback synchronously with every
489/// raw I420 frame ([`TrackVideoEncoder::Inline`] tracks); keyframe requests
490/// travel inside that callback's frame, feedback for the rest is attached to
491/// the slot.
492/// - `Builtin` — the factory delegates to a backend encoder (builtin
493/// VP8/VP9/AV1, or H264 via the slot's backend preference); push raw BGRA
494/// frames via the returned [`Track`](crate::media::Track).
495pub(crate) enum RegistrySlot {
496 Custom(CustomSlotState),
497 Inline(Arc<CustomEncoderState>),
498 Builtin(H264BackendPref),
499}
500
501/// Registry state for a [`TrackVideoEncoder::PreEncoded`] slot: the queue the
502/// app pushes into, plus the listeners the queue-pull API cannot reach.
503pub(crate) struct CustomSlotState {
504 pub(crate) queue: Arc<Mutex<VecDeque<EncodedVideoFrame>>>,
505 pub(crate) feedback: Arc<FeedbackListeners>,
506}
507
508impl Clone for RegistrySlot {
509 fn clone(&self) -> Self {
510 match self {
511 Self::Custom(s) => Self::Custom(CustomSlotState {
512 queue: s.queue.clone(),
513 feedback: s.feedback.clone(),
514 }),
515 Self::Inline(s) => Self::Inline(s.clone()),
516 Self::Builtin(pref) => Self::Builtin(*pref),
517 }
518 }
519}
520
521/// Which H.264 backend a builtin-routed slot prefers, mirroring the public
522/// [`H264Backend`] (None → `Auto`). `Auto` resolves at encoder creation:
523/// VideoToolbox on Apple, registered OpenH264 elsewhere.
524#[derive(Clone, Copy, Debug, PartialEq, Eq)]
525pub(crate) enum H264BackendPref {
526 Auto,
527 VideoToolbox,
528 #[cfg(feature = "openh264")]
529 OpenH264,
530}
531
532impl H264BackendPref {
533 /// The `reactor_video_backend_cb` wire value (0 = auto, 1 = VT, 2 = OH).
534 fn as_c_int(self) -> c_int {
535 match self {
536 Self::Auto => 0,
537 Self::VideoToolbox => 1,
538 #[cfg(feature = "openh264")]
539 Self::OpenH264 => 2,
540 }
541 }
542}
543
544/// Routes encoder instances to per-track slots using the per-encoder-instance ID
545/// stamped by the C++ factory.
546///
547/// Slots are assigned lazily: when a given `encoder_id` appears for the first
548/// time (either in `use_builtin_for` or `encode_for`), the next pending slot is
549/// consumed and bound to that ID. The assignment order matches the order
550/// libwebrtc calls `VideoEncoderFactory::Create()` — one call per video
551/// transceiver, in negotiation order — which in turn matches the order
552/// [`add_encoded_slot`] / [`add_raw_slot`] were called on the builder.
553///
554/// Every reservation carries the owning track's native id so the registry
555/// can retract a dead track's slots ([`EncoderRegistry::retract`]): an
556/// unreserved-pending slot for a dead track gets dropped, and an assigned
557/// slot degrades to Builtin with a loud fallback, never silently routing
558/// another track's frames.
559pub(crate) struct EncoderRegistry {
560 pending: Mutex<VecDeque<RegistryReservation>>,
561 assigned: Mutex<HashMap<u64, AssignedSlot>>,
562}
563
564struct RegistryReservation {
565 native_id: usize,
566 slot: RegistrySlot,
567}
568
569struct AssignedSlot {
570 native_id: usize,
571 slot: RegistrySlot,
572}
573
574impl Clone for AssignedSlot {
575 fn clone(&self) -> Self {
576 AssignedSlot {
577 native_id: self.native_id,
578 slot: self.slot.clone(),
579 }
580 }
581}
582
583impl EncoderRegistry {
584 pub(crate) fn new() -> Arc<Self> {
585 Arc::new(Self {
586 pending: Mutex::new(VecDeque::new()),
587 assigned: Mutex::new(HashMap::new()),
588 })
589 }
590
591 /// Wire this registry into factory-create options: leaks a
592 /// [`RegistryState`] the native side frees via `free_registry_state_tramp`
593 /// when the last encoder goes away (or the caller frees manually after a
594 /// failed create), and points every encode/`use_builtin`/`has_custom`
595 /// trampoline at it.
596 pub(crate) fn install_into(
597 self: &Arc<Self>,
598 opts: &mut reactor_webrtc_sys::ReactorFactoryOptions,
599 ) {
600 let state = Box::into_raw(Box::new(RegistryState {
601 registry: Arc::clone(self),
602 }));
603 opts.encode_cb = Some(registry_encode_tramp);
604 opts.encode_userdata = state as *mut c_void;
605 opts.encode_free_ud = Some(free_registry_state_tramp);
606 opts.encode_use_builtin = Some(registry_use_builtin_tramp);
607 opts.encode_has_custom_slots = Some(registry_has_custom_tramp);
608 opts.encode_video_backend_for = Some(registry_backend_for_tramp);
609 opts.encode_rate_update = Some(registry_rate_update_tramp);
610 }
611
612 /// Reserve a custom (pre-encoded) slot for the track `native_id`.
613 /// Returns the queue the [`EncodedVideoTrack`] will push frames into
614 /// plus the slot's feedback listeners
615 /// ([`EncodedVideoTrack::on_encoder_feedback`]).
616 pub(crate) fn add_encoded_slot(
617 &self,
618 native_id: usize,
619 ) -> (
620 Arc<Mutex<VecDeque<EncodedVideoFrame>>>,
621 Arc<FeedbackListeners>,
622 ) {
623 let slot = CustomSlotState {
624 queue: Arc::new(Mutex::new(VecDeque::new())),
625 feedback: FeedbackListeners::new(),
626 };
627 let out = (slot.queue.clone(), slot.feedback.clone());
628 self.pending.lock().unwrap().push_back(RegistryReservation {
629 native_id,
630 slot: RegistrySlot::Custom(slot),
631 });
632 out
633 }
634
635 /// Reserve a builtin (raw BGRA) slot for the track `native_id`. The C++
636 /// factory delegates to the builtin VP8/VP9/AV1 encoder for this
637 /// transceiver.
638 pub(crate) fn add_raw_slot(&self, native_id: usize) {
639 self.add_raw_slot_with_backend(native_id, H264BackendPref::Auto);
640 }
641
642 /// Reserve a builtin (raw BGRA) slot with an explicit H264 backend
643 /// preference, for tracks created with `h264_backend` set.
644 pub(crate) fn add_raw_slot_with_backend(&self, native_id: usize, backend: H264BackendPref) {
645 self.pending.lock().unwrap().push_back(RegistryReservation {
646 native_id,
647 slot: RegistrySlot::Builtin(backend),
648 });
649 }
650
651 /// Reserve an inline-encoder slot for the track `native_id`. The
652 /// registry calls `cb` synchronously with every raw I420 frame for the
653 /// encoder instance that binds here; the returned feedback listeners
654 /// surface the rest (rate updates) through
655 /// [`VideoTrack::on_encoder_feedback`](crate::media::VideoTrack::on_encoder_feedback).
656 pub(crate) fn add_inline_slot(
657 &self,
658 native_id: usize,
659 cb: InlineEncoderCallback,
660 ) -> Arc<FeedbackListeners> {
661 let feedback = FeedbackListeners::new();
662 let out = feedback.clone();
663 let state = Arc::new(CustomEncoderState {
664 cb: Mutex::new(cb),
665 feedback,
666 });
667 self.pending.lock().unwrap().push_back(RegistryReservation {
668 native_id,
669 slot: RegistrySlot::Inline(state),
670 });
671 out
672 }
673
674 /// Drop everything reserved for a track that no longer exists.
675 /// Pending reservations are simply removed; an already-assigned slot is
676 /// degraded to Builtin with a loud fallback — libwebrtc may legitimately
677 /// recreate an encoder for a dead stream, and that must never consume
678 /// another track's slot (the old grey-silence failure). The last living
679 /// custom lane closing is a teardown, not a hazard: it stays quiet
680 /// instead of app woke-and-wail on every shutdown.
681 pub(crate) fn retract(&self, native_id: usize) {
682 self.pending
683 .lock()
684 .unwrap()
685 .retain(|r| r.native_id != native_id);
686 // Any other custom lane matters? Check pending first, then assigned
687 // (same traversal order as has_custom — no ABBA regression).
688 let pending_has_custom = self
689 .pending
690 .lock()
691 .unwrap()
692 .iter()
693 .any(|r| !matches!(r.slot, RegistrySlot::Builtin(_)));
694 let any_live_custom =
695 pending_has_custom
696 || self.assigned.lock().unwrap().values().any(|a| {
697 a.native_id != native_id && !matches!(a.slot, RegistrySlot::Builtin(_))
698 });
699 let mut assigned = self.assigned.lock().unwrap();
700 for a in assigned.values_mut() {
701 if a.native_id == native_id && !matches!(a.slot, RegistrySlot::Builtin(_)) {
702 if any_live_custom {
703 eprintln!(
704 "[reactor-webrtc] retracting encoder slot for dropped track {native_id:#x}: \
705 degrading to builtin with other custom slots still routed"
706 );
707 }
708 a.slot = RegistrySlot::Builtin(H264BackendPref::Auto);
709 }
710 }
711 }
712
713 /// Whether any custom (pre-encoded or inline) slot exists — pending or
714 /// already assigned. Drives the `has_custom_slots` predicate the native
715 /// factory consults when advertising codecs.
716 pub(crate) fn has_custom(&self) -> bool {
717 let is_custom = |s: &RegistrySlot| !matches!(s, RegistrySlot::Builtin(_));
718 // Drop pending's guard before touching `assigned`: holding both is an
719 // ABBA deadlock against use_builtin_for/encode_for under concurrent
720 // negotiation + encode.
721 let pending_has = self
722 .pending
723 .lock()
724 .unwrap()
725 .iter()
726 .any(|r| is_custom(&r.slot));
727 pending_has
728 || self
729 .assigned
730 .lock()
731 .unwrap()
732 .values()
733 .any(|a| is_custom(&a.slot))
734 }
735
736 /// Called by `registry_use_builtin_tramp`. Assigns the next pending slot to
737 /// `encoder_id` if it has not been seen before, then returns whether the
738 /// C++ factory should delegate to the builtin encoder.
739 pub(crate) fn use_builtin_for(&self, encoder_id: u64) -> bool {
740 let mut assigned = self.assigned.lock().unwrap();
741 if let Some(a) = assigned.get(&encoder_id) {
742 return matches!(a.slot, RegistrySlot::Builtin(_));
743 }
744 let next = {
745 let mut pending = self.pending.lock().unwrap();
746 pending.pop_front()
747 };
748 let Some(reservation) = next else {
749 // Empty queue: the positional fallback. Loud only when custom
750 // plumbing exists to lose — an entirely builtin factory must
751 // never hear this (its fallback is the intentional answer).
752 //
753 // `assigned` is already locked on this thread, so consult it (and
754 // `pending`, in the usual assigned→pending order) directly instead
755 // of going through has_custom(): std::sync::Mutex is non-recursive,
756 // and re-locking `assigned` here self-deadlocks this thread. That
757 // wedged the encoder queue whenever libwebrtc recreated an encoder
758 // with an empty reservation queue (normal on stream teardown with
759 // cadence frames in flight), cascading into
760 // VideoStreamEncoder::Stop and PeerConnection::Close never
761 // returning.
762 let is_custom = |s: &RegistrySlot| !matches!(s, RegistrySlot::Builtin(_));
763 let any_custom_left = assigned.values().any(|a| is_custom(&a.slot))
764 || self
765 .pending
766 .lock()
767 .unwrap()
768 .iter()
769 .any(|r| is_custom(&r.slot));
770 if any_custom_left {
771 eprintln!(
772 "[reactor-webrtc] no encoder slot reservation left for encoder {encoder_id}: delegating to builtin — custom-encoded tracks after this point may not produce video"
773 );
774 }
775 return true;
776 };
777 let is_builtin = matches!(reservation.slot, RegistrySlot::Builtin(_));
778 assigned.insert(
779 encoder_id,
780 AssignedSlot {
781 native_id: reservation.native_id,
782 slot: reservation.slot,
783 },
784 );
785 is_builtin
786 }
787
788 /// Called by `registry_encode_tramp`. Produces the next encoded frame for
789 /// this encoder instance — draining the queue for pre-encoded slots,
790 /// invoking the user callback for inline slots — assigning its slot on
791 /// first call if needed.
792 fn encode_for(&self, encoder_id: u64, raw: &RawVideoFrame) -> Option<EncodedVideoFrame> {
793 let mut assigned = self.assigned.lock().unwrap();
794 let a = assigned.entry(encoder_id).or_insert_with(|| {
795 let reservation = self.pending.lock().unwrap().pop_front();
796 match reservation {
797 Some(r) => AssignedSlot {
798 native_id: r.native_id,
799 slot: r.slot,
800 },
801 None => {
802 eprintln!(
803 "[reactor-webrtc] no encoder slot reservation left for encoder {encoder_id}: delegating to builtin"
804 );
805 AssignedSlot {
806 native_id: 0,
807 slot: RegistrySlot::Builtin(H264BackendPref::Auto),
808 }
809 }
810 }
811 });
812 match &a.slot {
813 RegistrySlot::Custom(slot) => {
814 let (q, feedback) = (slot.queue.clone(), slot.feedback.clone());
815 drop(assigned);
816 if raw.request_key_frame {
817 // Queue-mode has no other channel: clone, drop the guard,
818 // then fire — never run the callback under the registry
819 // lock (mirrors `rate_update`'s shape below).
820 feedback.fire(EncoderFeedback::KeyFrameRequest);
821 }
822 let frame = q.lock().unwrap().pop_front();
823 frame
824 }
825 RegistrySlot::Inline(state) => {
826 let state = state.clone();
827 drop(assigned);
828 let Ok(mut cb) = state.cb.lock() else {
829 return None;
830 };
831 cb(raw)
832 }
833 RegistrySlot::Builtin(_) => None,
834 }
835 }
836
837 /// Called by `registry_rate_update_tramp`. Routes a rate-control update
838 /// to the encoder instance's slot's feedback listeners.
839 fn rate_update(&self, encoder_id: u64, bitrate_bps: u32, framerate_fps: f64) {
840 let assigned = self.assigned.lock().unwrap();
841 let slot = match assigned.get(&encoder_id) {
842 Some(s) => AssignedSlot {
843 native_id: s.native_id,
844 slot: s.slot.clone(),
845 },
846 None => return,
847 };
848 drop(assigned);
849 let feedback = match &slot.slot {
850 RegistrySlot::Custom(slot) => Some(slot.feedback.clone()),
851 RegistrySlot::Inline(state) => Some(state.feedback.clone()),
852 RegistrySlot::Builtin(_) => None,
853 };
854 if let Some(fb) = feedback {
855 fb.fire(EncoderFeedback::RateUpdate {
856 bitrate_bps,
857 framerate_fps,
858 });
859 }
860 }
861
862 /// Called by the native composite when building an H264 encoder for a
863 /// backend-routed instance — the slot's stored preference.
864 pub(crate) fn backend_for(&self, encoder_id: u64) -> H264BackendPref {
865 match self.assigned.lock().unwrap().get(&encoder_id) {
866 Some(a) => match &a.slot {
867 RegistrySlot::Builtin(pref) => *pref,
868 _ => H264BackendPref::Auto,
869 },
870 None => H264BackendPref::Auto,
871 }
872 }
873}
874
875struct RegistryState {
876 registry: Arc<EncoderRegistry>,
877}
878
879/// Fills `out` from an `EncodedVideoFrame`. Returns 0 (forward) on success.
880///
881/// Extracted so both trampolines share the same output-filling logic.
882fn fill_output(
883 encoded: EncodedVideoFrame,
884 out: *mut reactor_webrtc_sys::ReactorEncodedVideoOutput,
885) -> c_int {
886 let mut v = encoded.data;
887 v.shrink_to_fit();
888 let ptr = v.as_ptr();
889 let len = v.len();
890 std::mem::forget(v);
891 unsafe {
892 let o = &mut *out;
893 o.data = ptr;
894 o.len = len;
895 o.is_key_frame = encoded.is_key_frame as c_int;
896 o.width = encoded.width;
897 o.height = encoded.height;
898 o.rtp_timestamp = encoded.rtp_timestamp;
899 o.free_data = Some(free_encoded_data);
900 }
901 0
902}
903
904pub(crate) extern "C" fn registry_encode_tramp(
905 ud: *mut c_void,
906 raw: *const reactor_webrtc_sys::ReactorRawVideoFrame,
907 out: *mut reactor_webrtc_sys::ReactorEncodedVideoOutput,
908) -> c_int {
909 let Some(r) = (unsafe { raw.as_ref() }) else {
910 return 1;
911 };
912 let st = unsafe { &*(ud as *const RegistryState) };
913 let frame = raw_view(r);
914 match st.registry.encode_for(r.encoder_id, &frame) {
915 None => 1,
916 Some(encoded) => fill_output(encoded, out),
917 }
918}
919
920/// Called by C++ before creating each encoder instance. Returns 1 if the
921/// builtin VP8/VP9/AV1 encoder should be used for this slot, 0 for custom.
922pub(crate) extern "C" fn registry_use_builtin_tramp(ud: *mut c_void, encoder_id: u64) -> c_int {
923 let st = unsafe { &*(ud as *const RegistryState) };
924 st.registry.use_builtin_for(encoder_id) as c_int
925}
926
927pub(crate) extern "C" fn free_registry_state_tramp(ud: *mut c_void) {
928 drop(unsafe { Box::from_raw(ud as *mut RegistryState) });
929}
930
931/// Called by the native factory when enumerating advertised video codecs.
932/// Returns nonzero while any custom (pre-encoded or inline) slot exists.
933pub(crate) extern "C" fn registry_has_custom_tramp(ud: *mut c_void) -> c_int {
934 let st = unsafe { &*(ud as *const RegistryState) };
935 st.registry.has_custom() as c_int
936}
937
938/// Called by the native composite when it builds an H264 encoder for a
939/// backend-routed instance — returns the slot's stored preference as the
940/// wire code (0 = auto, 1 = VideoToolbox, 2 = OpenH264).
941pub(crate) extern "C" fn registry_backend_for_tramp(ud: *mut c_void, encoder_id: u64) -> c_int {
942 let st = unsafe { &*(ud as *const RegistryState) };
943 st.registry.backend_for(encoder_id).as_c_int()
944}
945
946/// Called by the native encoder on every rate-control (BWE) update.
947pub(crate) extern "C" fn registry_rate_update_tramp(
948 ud: *mut c_void,
949 encoder_id: u64,
950 bitrate_bps: u32,
951 framerate_fps: f64,
952) {
953 let st = unsafe { &*(ud as *const RegistryState) };
954 st.registry
955 .rate_update(encoder_id, bitrate_bps, framerate_fps);
956}
957
958// ── Per-track encoder selection ─────────────────────────────────────────────
959
960/// Geometry for a pre-encoded video track
961/// ([`TrackVideoEncoder::PreEncoded`]).
962///
963/// `width`/`height` set the resolution libwebrtc's encoder pipeline is
964/// configured for. They must match what your encoder actually produces. Pass
965/// 0 in [`EncodedVideoFrame`] fields to inherit them per frame.
966#[derive(Debug, Clone, Copy)]
967#[non_exhaustive]
968pub struct PreEncodedOptions {
969 /// Encoded width in pixels.
970 pub width: u32,
971 /// Encoded height in pixels.
972 pub height: u32,
973}
974
975impl PreEncodedOptions {
976 /// Pre-encoded track encoded at `width`×`height`.
977 pub fn new(width: u32, height: u32) -> Self {
978 Self { width, height }
979 }
980}
981
982/// How a video track produces its encoded bitstream, replacing libwebrtc's
983/// builtin encoder **for that track only**.
984///
985/// One mechanism, two feeding styles:
986///
987/// - [`PreEncoded`](TrackVideoEncoder::PreEncoded) — **asynchronous**: you
988/// push already-encoded bytes at your own pace, from any thread (a
989/// hardware encoder callback, a relay, a file dumper). A queue decouples
990/// your producer from the encoder thread. The track comes back as
991/// [`LocalVideoTrack::Encoded`] and you feed it with
992/// [`EncodedVideoTrack::push_encoded_frame`].
993/// - [`Inline`](TrackVideoEncoder::Inline) — **synchronous**: libwebrtc calls
994/// your closure on the encoder thread with every raw I420 frame; return
995/// `Some(encoded)` to inject bytes into the RTP stack or `None` to drop the
996/// frame. Right when your pipeline is raw-in/encoded-out and you want
997/// libwebrtc to drive the cadence. The track comes back as
998/// [`LocalVideoTrack::Raw`] and you feed it like any raw track.
999///
1000/// Either way, the bytes **you** produce must match the codec negotiated for
1001/// the track's transceiver — read [`RawVideoFrame::codec`] (inline) or set
1002/// codec preferences to pin it.
1003///
1004/// # Slot assignment order — positional, with fresh upgrades
1005///
1006/// Slots route encoder instances ↔ tracks **positionally**: reservation order
1007/// (track creation) must match the negotiation order of the transceivers
1008/// carrying those tracks. Create encoder-carrying tracks **before** the
1009/// peer connection they're attached to negotiates (before `create_offer`),
1010/// or the registry may not yet see a reservation to bind to; ordering across
1011/// **multiple** peer connections sharing one factory follows whichever
1012/// negotiates first, not track creation order.
1013///
1014/// Dropping an encoder-carrying track retracts its pending slots and degrades
1015/// any surviving assigned slot to Builtin (loud fallback, never another
1016/// track's). libwebrtc may recreate encoder instances on internal errors or
1017/// renegotiation past a dead track; that path degrades the same way rather
1018/// than re-routing another live track's pipeline.
1019#[non_exhaustive]
1020pub enum TrackVideoEncoder {
1021 /// You push already-encoded bytes at your own pace.
1022 PreEncoded(PreEncodedOptions),
1023 /// libwebrtc calls your encoder synchronously with every raw I420 frame.
1024 Inline(InlineEncoderCallback),
1025}
1026
1027/// Which H.264 backend a **raw** video track encodes with.
1028///
1029/// `None` (Auto, the default) picks the platform default: **VideoToolbox**
1030/// on Apple, **OpenH264** — when registered via
1031/// [`PeerConnectionFactoryBuilder::with_openh264`] — elsewhere. Without any
1032/// usable backend, H264 is simply not negotiated (VP8/VP9/AV1 take over).
1033///
1034/// Choose explicitly to force a backend per track — a camera on VideoToolbox
1035/// (hardware, battery-friendly) beside a screen share on OpenH264
1036/// (bitrate/intra control, cross-platform parity). An explicit-but-unusable
1037/// backend (VideoToolbox off Apple, OpenH264 unregistered) fails track
1038/// creation with an error.
1039///
1040/// Meaningless — and rejected — on tracks with
1041/// [`TrackVideoEncoder`] set, since the track's bytes come from your own
1042/// pipeline.
1043#[derive(Debug, Clone, Copy, PartialEq, Eq)]
1044#[non_exhaustive]
1045pub enum H264Backend {
1046 /// Apple's hardware VideoToolbox (macOS/iOS).
1047 VideoToolbox,
1048 /// Cisco's software OpenH264 — requires registering the shared library
1049 /// with [`PeerConnectionFactoryBuilder::with_openh264`].
1050 #[cfg(feature = "openh264")]
1051 OpenH264,
1052 // Future: MediaFoundation (Windows hw), NvEnc (NVIDIA), …
1053}
1054
1055/// Options for [`PeerConnectionFactory::create_video_track_with_options`].
1056///
1057/// All fields optional; `Default::default()` produces exactly what
1058/// [`PeerConnectionFactory::create_video_track`] produces. The struct is
1059/// `#[non_exhaustive]` — construct via `Default` and assign fields:
1060///
1061/// ```rust,ignore
1062/// let mut opts = VideoTrackOptions::default();
1063/// opts.encoder = Some(TrackVideoEncoder::PreEncoded(PreEncodedOptions::new(1280, 720)));
1064/// ```
1065#[derive(Default)]
1066#[non_exhaustive]
1067pub struct VideoTrackOptions {
1068 /// Encoder override for this track. `None` (default) = libwebrtc's
1069 /// builtin/software pipeline encodes.
1070 pub encoder: Option<TrackVideoEncoder>,
1071 /// Which H.264 backend encodes this track when it is raw and H264 wins
1072 /// negotiation. Errors when set together with `encoder`.
1073 pub h264_backend: Option<H264Backend>,
1074 /// Per-track frame-metadata switch. `None` (default) inherits the
1075 /// factory's kill switch
1076 /// ([`PeerConnectionFactoryBuilder::with_metadata`]); `Some(false)`
1077 /// disables trailers for this track only — pushes with `user_data` drop
1078 /// it silently (the frame itself still flows) and the peer sees no
1079 /// trailer whatever the connection negotiated. The SDP stays
1080 /// session-level either way.
1081 pub frame_metadata: Option<bool>,
1082}
1083
1084/// One video track created by
1085/// [`PeerConnectionFactory::create_video_track_with_options`].
1086///
1087/// - [`Raw`](LocalVideoTrack::Raw) — push BGRA frames; either libwebrtc
1088/// encodes them itself (default) or they go through an
1089/// [`Inline`](TrackVideoEncoder::Inline) callback.
1090/// - [`Encoded`](LocalVideoTrack::Encoded) — push pre-encoded bytes;
1091/// libwebrtc only packetises.
1092pub enum LocalVideoTrack {
1093 /// A raw (BGRA-in) video track.
1094 Raw(crate::media::VideoTrack),
1095 /// A pre-encoded (bytes-in) video track.
1096 Encoded(EncodedVideoTrack),
1097}
1098
1099impl LocalVideoTrack {
1100 /// The underlying video track handle — attach this to a transceiver with
1101 /// [`Transceiver::set_track`](crate::Transceiver::set_track).
1102 pub fn track(&self) -> &crate::media::VideoTrack {
1103 match self {
1104 Self::Raw(t) => t,
1105 Self::Encoded(e) => e.track(),
1106 }
1107 }
1108
1109 /// Returns the inner [`EncodedVideoTrack`], or `None` if this is a raw track.
1110 pub fn as_encoded(&self) -> Option<&EncodedVideoTrack> {
1111 match self {
1112 Self::Encoded(e) => Some(e),
1113 Self::Raw(_) => None,
1114 }
1115 }
1116
1117 /// Returns the inner raw [`VideoTrack`](crate::media::VideoTrack), or
1118 /// `None` if this is a pre-encoded track.
1119 pub fn as_raw(&self) -> Option<&crate::media::VideoTrack> {
1120 match self {
1121 Self::Raw(t) => Some(t),
1122 Self::Encoded(_) => None,
1123 }
1124 }
1125}
1126
1127// ── Push-based encoded video track ───────────────────────────────────────────
1128
1129/// A video track that accepts **pre-encoded** frames directly, bypassing the
1130/// libwebrtc software encoder pipeline entirely.
1131///
1132/// Obtain one via
1133/// [`PeerConnectionFactory::create_video_track_with_options`] with
1134/// [`TrackVideoEncoder::PreEncoded`], then:
1135///
1136/// 1. Attach `EncodedVideoTrack::track()` to a send-only transceiver.
1137/// 2. Call `push_encoded_frame` whenever your encoder (VideoToolbox, NVENC,
1138/// GStreamer, libvpx, …) produces a frame — at your rate, on any thread.
1139///
1140/// # Timing
1141///
1142/// The WebRTC encoder thread is triggered internally each time you call
1143/// `push_encoded_frame`, so you do **not** need to call
1144/// `push_video_frame` separately. The dummy raw frame used to trigger it
1145/// is cheap (pre-allocated; the I420 data is discarded by the encoder
1146/// callback before it ever touches your encoded bytes).
1147pub struct EncodedVideoTrack {
1148 pub(crate) track: crate::media::VideoTrack,
1149 pub(crate) queue: Arc<Mutex<VecDeque<EncodedVideoFrame>>>,
1150 // Pre-allocated BGRA buffer used to trigger the WebRTC encoder thread.
1151 // The dimensions are kept in sync with the track's configured resolution
1152 // so the pipeline doesn't reject the frame due to a size mismatch.
1153 dummy: Vec<u8>,
1154 width: u32,
1155 height: u32,
1156 // FIFO metadata queue for push_frame_with_metadata: the sender
1157 // FrameTransform pops one entry per encoded frame in push order. A FIFO rather
1158 // than timestamp correlation because capture_time_ms is unreliable here —
1159 // VideoStreamEncoder clamps future timestamps back to post_time, which can
1160 // collide when two pushes land in the same millisecond.
1161 sender_meta_fifo: Arc<FifoMeta>,
1162 feedback: Arc<FeedbackListeners>,
1163}
1164
1165/// An [`EncodedVideoTrack`]'s outgoing metadata, in push order.
1166///
1167/// Registered in [`crate::sender_meta`] under the inner track's native identity,
1168/// *replacing* the timestamp-keyed source that [`crate::Track`] registered for it:
1169/// for a pre-encoded track, push order is the correlation that holds.
1170#[derive(Default)]
1171pub(crate) struct FifoMeta(Mutex<VecDeque<crate::metadata::FrameMetadata>>);
1172
1173impl crate::sender_meta::SenderMetaSource for FifoMeta {
1174 fn take(&self, _frame: &EncodedFrame) -> Option<crate::metadata::FrameMetadata> {
1175 // Pops whether or not the caller will use the result. The queue is filled
1176 // by push_encoded_frame_with_metadata regardless of what the peer declared,
1177 // so leaving entries in place would grow it for as long as the peer
1178 // declines and then emit stale metadata if it ever accepted.
1179 self.0.lock().ok()?.pop_front()
1180 }
1181}
1182
1183// SAFETY: the queue is Mutex-guarded and the dummy buffer is owned; both are
1184// safe to move across threads.
1185unsafe impl Send for EncodedVideoTrack {}
1186unsafe impl Sync for EncodedVideoTrack {}
1187
1188impl EncodedVideoTrack {
1189 pub(crate) fn new(
1190 track: crate::media::VideoTrack,
1191 queue: Arc<Mutex<VecDeque<EncodedVideoFrame>>>,
1192 feedback: Arc<FeedbackListeners>,
1193 width: u32,
1194 height: u32,
1195 ) -> Self {
1196 let dummy = vec![0u8; (width * height * 4) as usize];
1197 let sender_meta_fifo = Arc::new(FifoMeta::default());
1198 // Overwrite the inner Track's timestamp-keyed registration: both describe
1199 // the same native track, and for pre-encoded frames the FIFO is the one
1200 // that correlates correctly.
1201 let source: Arc<dyn crate::sender_meta::SenderMetaSource> = sender_meta_fifo.clone();
1202 crate::sender_meta::register(track.native_id(), &source);
1203 crate::encoded::register_feedback_binding(track.native_id(), &feedback);
1204 Self {
1205 track,
1206 queue,
1207 dummy,
1208 width,
1209 height,
1210 sender_meta_fifo,
1211 feedback,
1212 }
1213 }
1214
1215 /// Listen for encoder feedback for this track —
1216 /// [`EncoderFeedback::KeyFrameRequest`] (answer with an IDR promptly) and
1217 /// [`EncoderFeedback::RateUpdate`] (adapt your encoder's target). Latest
1218 /// registration wins. The callback fires on WebRTC threads, like every
1219 /// other sink in this API.
1220 pub fn on_encoder_feedback(&self, cb: impl FnMut(EncoderFeedback) + Send + 'static) {
1221 self.feedback.set(Box::new(cb));
1222 }
1223
1224 /// The underlying video track handle — pass `track()` (or the
1225 /// [`VideoTrack`](crate::VideoTrack) itself) to
1226 /// [`Transceiver::set_track`](crate::Transceiver::set_track); it derefs to
1227 /// [`&Track`](crate::Track).
1228 pub fn track(&self) -> &crate::media::VideoTrack {
1229 &self.track
1230 }
1231
1232 /// Inject a pre-encoded frame into the WebRTC RTP stack.
1233 ///
1234 /// The call returns immediately; the frame is queued and forwarded to the
1235 /// RTP packetizer on the WebRTC encoder thread. Thread-safe — call from
1236 /// any thread, including a hardware encoder callback.
1237 ///
1238 /// Set `frame.width` / `frame.height` to 0 to inherit from the track's
1239 /// configured resolution — the
1240 /// [`PreEncodedOptions`] the track was created with.
1241 pub fn push_frame(&self, frame: EncodedVideoFrame) -> crate::Result<()> {
1242 // Queue first so the frame is always present when the encoder thread
1243 // dequeues it (the two operations are not atomic, but the encoder
1244 // thread is asynchronous, so the queue push always wins the race).
1245 self.queue.lock().unwrap().push_back(frame);
1246 // Push a dummy raw frame to wake the WebRTC encoder thread.
1247 // The I420 data is thrown away in the encoder callback — the actual
1248 // encoded bytes come from the queue above.
1249 self.track.push_frame(crate::media::VideoFrame::new(
1250 &self.dummy,
1251 self.width,
1252 self.height,
1253 ))?;
1254 Ok(())
1255 }
1256
1257 /// Inject a pre-encoded frame with per-frame metadata embedded as a
1258 /// protobuf trailer.
1259 ///
1260 /// `frame_id` and `capture_time_us` are filled in internally, the latter from
1261 /// [`time_micros`](crate::time_micros); the caller supplies only `user_data`.
1262 /// A capture time cannot be declared here the way it can on a raw
1263 /// [`Track`](crate::Track): these frames arrive already encoded and the
1264 /// encoder clamps future capture timestamps, which is why this path
1265 /// correlates its metadata by FIFO order instead. Requires
1266 /// [`sender_metadata_transform`](Self::sender_metadata_transform) to be
1267 /// called and the returned [`FrameTransform`](crate::FrameTransform) to
1268 /// be attached to the sender transceiver before the first SDP exchange.
1269 pub fn push_frame_with_metadata(
1270 &self,
1271 frame: EncodedVideoFrame,
1272 user_data: &[u8],
1273 ) -> crate::Result<()> {
1274 let meta = crate::metadata::FrameMetadata {
1275 frame_id: self.track.next_frame_id(),
1276 capture_time_us: crate::time_micros().max(0) as u64,
1277 user_data: user_data.to_vec(),
1278 };
1279 // Enqueue metadata first so the FIFO entry is always present when the
1280 // sender FrameTransform fires on the encoder thread. Off-track
1281 // (frame_metadata off) drops user_data here, mirroring the raw path.
1282 if self.track.metadata_enabled() {
1283 if let Ok(mut fifo) = self.sender_meta_fifo.0.lock() {
1284 fifo.push_back(meta);
1285 }
1286 }
1287 self.queue.lock().unwrap().push_back(frame);
1288 self.track.push_frame(crate::media::VideoFrame::new(
1289 &self.dummy,
1290 self.width,
1291 self.height,
1292 ))?;
1293 Ok(())
1294 }
1295}
1296
1297#[cfg(test)]
1298mod registry_locking_tests {
1299 use super::*;
1300
1301 /// A new encoder id arriving with an empty reservation queue takes the
1302 /// positional fallback. That branch must not consult has_custom() while
1303 /// holding the `assigned` lock: std::sync::Mutex is non-recursive, so the
1304 /// re-lock self-deadlocked the encoder queue whenever libwebrtc
1305 /// recreated an encoder during stream teardown, cascading into
1306 /// VideoStreamEncoder::Stop and PeerConnection::Close never returning
1307 /// (sampled as the CI pytest hang).
1308 #[test]
1309 fn use_builtin_for_empty_pending_does_not_self_deadlock() {
1310 let registry = EncoderRegistry::new();
1311 let _live = registry.add_encoded_slot(0xdead);
1312 assert!(
1313 !registry.use_builtin_for(1),
1314 "the single pending reservation is custom"
1315 );
1316 // Pending is now empty while a custom slot stays assigned — exactly
1317 // the state a teardown-time encoder recreate hits.
1318 let (tx, rx) = std::sync::mpsc::channel();
1319 let reg = Arc::clone(®istry);
1320 std::thread::spawn(move || {
1321 let _ = tx.send(reg.use_builtin_for(2));
1322 });
1323 let fell_back = rx
1324 .recv_timeout(std::time::Duration::from_secs(5))
1325 .expect("use_builtin_for self-deadlocked on an empty pending queue");
1326 assert!(fell_back, "no reservation left → builtin fallback");
1327 }
1328}