1use crate::channel::ChannelId;
2use crate::ieee80211::FrameLayout;
3use crate::realtek::{parse_rx_aggregate, AggregateError, RealtekRxPacket, RxPacketType};
4use crate::routes::{
5 PayloadRouteError, PayloadRouteEvent, PayloadRouteId, PayloadRouteManager, PayloadRuntimeKey,
6};
7use crate::rtp::{
8 DepacketizedFrame, RtpDepacketizer, RtpDepacketizerStatus, RtpHeader, RtpReorderBuffer,
9 RtpReorderStatus,
10};
11use crate::wfb::{FecCounters, WfbKeypair};
12
13#[derive(Debug, Clone)]
20pub struct ReceiverRuntime {
21 routes: PayloadRouteManager,
22 video_runtime: PayloadRuntimeKey,
23 video_route_id: PayloadRouteId,
24 rtp: RtpDepacketizer,
25 rtp_reorder: Option<RtpReorderBuffer>,
26}
27
28#[derive(Debug, Clone, Default)]
30pub struct ReceiverBatchOptions {
31 pub accept_corrupted: bool,
33 pub raw_payload_routes: Vec<PayloadRouteId>,
35 pub rtp_payload_taps: Vec<RtpPayloadTap>,
37}
38
39#[derive(Debug, Clone, Copy, PartialEq, Eq)]
46pub struct RtpPayloadTap {
47 pub route_id: PayloadRouteId,
49 pub payload_type: u8,
51}
52
53#[derive(Debug, Clone, PartialEq, Eq)]
55pub struct RoutePayload {
56 pub route_id: PayloadRouteId,
58 pub channel_id: ChannelId,
60 pub packet_seq: u64,
62 pub data: Vec<u8>,
64}
65
66#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)]
68pub struct ReceiverBatchCounters {
69 pub packets: usize,
71 pub accepted_packets: usize,
73 pub dropped_packets: usize,
75 pub crc_dropped: usize,
77 pub icv_dropped: usize,
79 pub report_dropped: usize,
81 pub ignored_frames: usize,
83 pub sessions: usize,
85 pub wfb_payloads: usize,
87 pub rtp_packets: usize,
89 pub video_frames: usize,
91 pub raw_payload_count: usize,
93 pub raw_payload_bytes: usize,
95 pub route_errors: usize,
97}
98
99#[derive(Debug, Clone, PartialEq, Eq)]
101pub struct ReceiverBatch {
102 pub frames: Vec<DepacketizedFrame>,
104 pub raw_payloads: Vec<RoutePayload>,
106 pub counters: ReceiverBatchCounters,
108 pub fec_counters: FecCounters,
110 pub rtp_status: RtpDepacketizerStatus,
112 pub rtp_reorder_status: RtpReorderStatus,
114}
115
116impl ReceiverRuntime {
117 pub fn from_routes(
122 routes: PayloadRouteManager,
123 video_runtime: PayloadRuntimeKey,
124 video_route_id: PayloadRouteId,
125 ) -> Self {
126 Self {
127 routes,
128 video_runtime,
129 video_route_id,
130 rtp: RtpDepacketizer::new(),
131 rtp_reorder: None,
132 }
133 }
134
135 pub fn with_plain_video_route(
139 frame_layout: FrameLayout,
140 video_route_id: PayloadRouteId,
141 channel_id: ChannelId,
142 key_slot: u64,
143 fec_k: usize,
144 fec_n: usize,
145 ) -> Result<Self, PayloadRouteError> {
146 let mut routes = PayloadRouteManager::new(frame_layout);
147 let video_runtime =
148 routes.add_plain_route(video_route_id, channel_id, key_slot, fec_k, fec_n)?;
149 Ok(Self::from_routes(routes, video_runtime, video_route_id))
150 }
151
152 pub fn with_keyed_video_route(
154 frame_layout: FrameLayout,
155 video_route_id: PayloadRouteId,
156 channel_id: ChannelId,
157 key_slot: u64,
158 keypair: WfbKeypair,
159 minimum_epoch: u64,
160 ) -> Result<Self, PayloadRouteError> {
161 let mut routes = PayloadRouteManager::new(frame_layout);
162 let video_runtime =
163 routes.add_keyed_route(video_route_id, channel_id, key_slot, keypair, minimum_epoch)?;
164 Ok(Self::from_routes(routes, video_runtime, video_route_id))
165 }
166
167 pub fn with_mock_video_route(
173 frame_layout: FrameLayout,
174 video_route_id: PayloadRouteId,
175 channel_id: ChannelId,
176 key_slot: u64,
177 ) -> Self {
178 let mut routes = PayloadRouteManager::new(frame_layout);
179 let video_runtime = routes.add_mock_route(video_route_id, channel_id, key_slot);
180 Self::from_routes(routes, video_runtime, video_route_id)
181 }
182
183 pub const fn video_runtime(&self) -> PayloadRuntimeKey {
185 self.video_runtime
186 }
187
188 pub const fn video_route_id(&self) -> PayloadRouteId {
190 self.video_route_id
191 }
192
193 pub fn routes(&self) -> &PayloadRouteManager {
195 &self.routes
196 }
197
198 pub fn routes_mut(&mut self) -> &mut PayloadRouteManager {
200 &mut self.routes
201 }
202
203 pub fn rtp_mut(&mut self) -> &mut RtpDepacketizer {
205 &mut self.rtp
206 }
207
208 pub fn rtp_status(&self) -> RtpDepacketizerStatus {
210 self.rtp.status()
211 }
212
213 pub fn rtp_reorder_status(&self) -> RtpReorderStatus {
215 self.rtp_reorder
216 .as_ref()
217 .map(RtpReorderBuffer::status)
218 .unwrap_or_default()
219 }
220
221 pub fn set_rtp_reorder_enabled(&mut self, enabled: bool) {
227 if enabled {
228 self.rtp_reorder
229 .get_or_insert_with(RtpReorderBuffer::default);
230 } else {
231 self.rtp_reorder = None;
232 }
233 }
234
235 pub const fn rtp_reorder_enabled(&self) -> bool {
237 self.rtp_reorder.is_some()
238 }
239
240 pub fn push_rtp_packet(
242 &mut self,
243 packet: &[u8],
244 ) -> Result<Vec<DepacketizedFrame>, crate::rtp::RtpError> {
245 self.push_video_payload(packet)
246 }
247
248 fn push_video_payload(
249 &mut self,
250 payload: &[u8],
251 ) -> Result<Vec<DepacketizedFrame>, crate::rtp::RtpError> {
252 let mut frames = Vec::new();
253 if let Some(reorder) = self.rtp_reorder.as_mut() {
254 for ordered in reorder.push(payload)? {
255 if let Some(frame) = self.rtp.push(&ordered)? {
256 frames.push(frame);
257 }
258 }
259 } else if let Some(frame) = self.rtp.push(payload)? {
260 frames.push(frame);
261 }
262 Ok(frames)
263 }
264
265 pub fn add_plain_route(
267 &mut self,
268 route_id: PayloadRouteId,
269 channel_id: ChannelId,
270 key_slot: u64,
271 fec_k: usize,
272 fec_n: usize,
273 ) -> Result<PayloadRuntimeKey, PayloadRouteError> {
274 self.routes
275 .add_plain_route(route_id, channel_id, key_slot, fec_k, fec_n)
276 }
277
278 pub fn add_keyed_route(
280 &mut self,
281 route_id: PayloadRouteId,
282 channel_id: ChannelId,
283 key_slot: u64,
284 keypair: WfbKeypair,
285 minimum_epoch: u64,
286 ) -> Result<PayloadRuntimeKey, PayloadRouteError> {
287 self.routes
288 .add_keyed_route(route_id, channel_id, key_slot, keypair, minimum_epoch)
289 }
290
291 pub fn add_mock_route(
293 &mut self,
294 route_id: PayloadRouteId,
295 channel_id: ChannelId,
296 key_slot: u64,
297 ) -> PayloadRuntimeKey {
298 self.routes.add_mock_route(route_id, channel_id, key_slot)
299 }
300
301 pub fn video_fec_counters(&self) -> FecCounters {
303 self.routes
304 .fec_counters(self.video_runtime)
305 .unwrap_or_default()
306 }
307
308 pub fn accepts_video_frame(&self, frame: &[u8]) -> bool {
310 self.routes.accepts_80211_frame(self.video_runtime, frame)
311 }
312
313 pub fn push_rx_transfer(
315 &mut self,
316 transfer: &[u8],
317 options: &ReceiverBatchOptions,
318 ) -> Result<ReceiverBatch, AggregateError> {
319 let packets = parse_rx_aggregate(transfer)?;
320 Ok(self.push_rx_packets(packets, options))
321 }
322
323 pub fn push_rx_packets<'a>(
325 &mut self,
326 packets: impl IntoIterator<Item = RealtekRxPacket<'a>>,
327 options: &ReceiverBatchOptions,
328 ) -> ReceiverBatch {
329 let mut batch = self.empty_batch();
330
331 for packet in packets {
332 batch.counters.packets += 1;
333 if packet.attrib.crc_err && !options.accept_corrupted {
334 batch.counters.crc_dropped += 1;
335 continue;
336 }
337 if packet.attrib.icv_err && !options.accept_corrupted {
338 batch.counters.icv_dropped += 1;
339 continue;
340 }
341 if packet.attrib.pkt_rpt_type != RxPacketType::NormalRx {
342 batch.counters.report_dropped += 1;
343 continue;
344 }
345
346 batch.counters.accepted_packets += 1;
347 match self.routes.push_80211_frame(packet.data) {
348 Ok(events) => self.apply_route_events(events, options, &mut batch),
349 Err(_) => {
350 batch.counters.ignored_frames += 1;
351 batch.counters.route_errors += 1;
352 }
353 }
354 }
355
356 batch.counters.dropped_packets =
357 batch.counters.crc_dropped + batch.counters.icv_dropped + batch.counters.report_dropped;
358 batch.fec_counters = self.video_fec_counters();
359 batch
360 }
361
362 pub fn push_80211_frame(
364 &mut self,
365 frame: &[u8],
366 options: &ReceiverBatchOptions,
367 ) -> Result<ReceiverBatch, PayloadRouteError> {
368 let mut batch = self.empty_batch();
369 let events = self.routes.push_80211_frame(frame)?;
370 self.apply_route_events(events, options, &mut batch);
371 batch.fec_counters = self.video_fec_counters();
372 Ok(batch)
373 }
374
375 pub fn push_decrypted_80211_frame(
377 &mut self,
378 runtime: PayloadRuntimeKey,
379 frame: &[u8],
380 decrypted_fragment: &[u8],
381 options: &ReceiverBatchOptions,
382 ) -> Result<ReceiverBatch, PayloadRouteError> {
383 let mut batch = self.empty_batch();
384 let events = self
385 .routes
386 .push_decrypted_80211_frame(runtime, frame, decrypted_fragment)?;
387 self.apply_route_events(events, options, &mut batch);
388 batch.fec_counters = self.video_fec_counters();
389 Ok(batch)
390 }
391
392 pub fn push_decrypted_fragment(
394 &mut self,
395 runtime: PayloadRuntimeKey,
396 data_nonce: u64,
397 decrypted_fragment: &[u8],
398 options: &ReceiverBatchOptions,
399 ) -> Result<ReceiverBatch, PayloadRouteError> {
400 let mut batch = self.empty_batch();
401 let events =
402 self.routes
403 .push_decrypted_fragment(runtime, data_nonce, decrypted_fragment)?;
404 self.apply_route_events(events, options, &mut batch);
405 batch.fec_counters = self.video_fec_counters();
406 Ok(batch)
407 }
408
409 pub fn push_mock_payload(
411 &mut self,
412 runtime: PayloadRuntimeKey,
413 packet_seq: u64,
414 payload: &[u8],
415 options: &ReceiverBatchOptions,
416 ) -> Result<ReceiverBatch, PayloadRouteError> {
417 let mut batch = self.empty_batch();
418 let events = self
419 .routes
420 .push_mock_payload(runtime, packet_seq, payload)?;
421 self.apply_route_events(events, options, &mut batch);
422 batch.fec_counters = self.video_fec_counters();
423 Ok(batch)
424 }
425
426 fn empty_batch(&self) -> ReceiverBatch {
427 ReceiverBatch {
428 frames: Vec::new(),
429 raw_payloads: Vec::new(),
430 counters: ReceiverBatchCounters::default(),
431 fec_counters: self.video_fec_counters(),
432 rtp_status: self.rtp_status(),
433 rtp_reorder_status: self.rtp_reorder_status(),
434 }
435 }
436
437 fn apply_route_events(
438 &mut self,
439 events: Vec<PayloadRouteEvent>,
440 options: &ReceiverBatchOptions,
441 batch: &mut ReceiverBatch,
442 ) {
443 for event in events {
444 match event {
445 PayloadRouteEvent::IgnoredFrame => batch.counters.ignored_frames += 1,
446 PayloadRouteEvent::SessionEstablished { .. } => batch.counters.sessions += 1,
447 PayloadRouteEvent::Payload {
448 route_ids, payload, ..
449 } => {
450 if route_ids.contains(&self.video_route_id) {
451 batch.counters.wfb_payloads += 1;
452 batch.counters.rtp_packets += 1;
453 if let Ok(frames) = self.push_rtp_packet(&payload.data) {
454 batch.counters.video_frames += frames.len();
455 batch.frames.extend(frames);
456 }
457 }
458
459 for &route_id in &route_ids {
460 if !options.raw_payload_routes.contains(&route_id) {
461 continue;
462 }
463 copy_raw_payload(route_id, &payload, batch);
464 }
465
466 if options.rtp_payload_taps.is_empty() {
467 continue;
468 }
469 let Ok(header) = RtpHeader::parse(&payload.data) else {
470 continue;
471 };
472 for tap in &options.rtp_payload_taps {
473 if header.payload_type != tap.payload_type {
474 continue;
475 }
476 if !route_ids.contains(&tap.route_id) {
477 continue;
478 }
479 copy_raw_payload(tap.route_id, &payload, batch);
480 }
481 }
482 }
483 }
484 batch.rtp_status = self.rtp_status();
485 batch.rtp_reorder_status = self.rtp_reorder_status();
486 }
487}
488
489fn copy_raw_payload(
490 route_id: PayloadRouteId,
491 payload: &crate::pipeline::RecoveredPayload,
492 batch: &mut ReceiverBatch,
493) {
494 batch.counters.raw_payload_count += 1;
495 batch.counters.raw_payload_bytes += payload.data.len();
496 batch.raw_payloads.push(RoutePayload {
497 route_id,
498 channel_id: payload.channel_id,
499 packet_seq: payload.packet_seq,
500 data: payload.data.clone(),
501 });
502}
503
504#[cfg(test)]
505mod tests {
506 use super::*;
507 use crate::RadioPort;
508
509 fn plain(payload: &[u8]) -> Vec<u8> {
510 let mut out = Vec::new();
511 out.push(0);
512 out.extend_from_slice(&(payload.len() as u16).to_be_bytes());
513 out.extend_from_slice(payload);
514 out
515 }
516
517 fn rtp(payload_type: u8, payload: &[u8]) -> Vec<u8> {
518 let mut packet = vec![0x80, payload_type & 0x7f];
519 packet.extend_from_slice(&7u16.to_be_bytes());
520 packet.extend_from_slice(&48_000u32.to_be_bytes());
521 packet.extend_from_slice(&0x1122_3344u32.to_be_bytes());
522 packet.extend_from_slice(payload);
523 packet
524 }
525
526 fn h264_stap_a_rtp() -> Vec<u8> {
527 let sps = [0x67, 0x42, 0x00, 0x1e, 0xab];
528 let pps = [0x68, 0xce, 0x06, 0xe2];
529 let idr = [0x65, 0x88, 0x84, 0x21];
530 let mut payload = vec![24];
531 for nalu in [&sps[..], &pps[..], &idr[..]] {
532 payload.extend_from_slice(&(nalu.len() as u16).to_be_bytes());
533 payload.extend_from_slice(nalu);
534 }
535 rtp(crate::rtp::RTP_PAYLOAD_TYPE_H264, &payload)
536 }
537
538 #[test]
539 fn decrypted_fragment_can_fan_out_to_raw_route() {
540 let route = PayloadRouteId::new(7);
541 let mut runtime = ReceiverRuntime::with_plain_video_route(
542 FrameLayout::WithFcs,
543 route,
544 ChannelId::default_video(),
545 0,
546 1,
547 1,
548 )
549 .unwrap();
550 let batch = runtime
551 .push_decrypted_fragment(
552 runtime.video_runtime(),
553 0,
554 &plain(b"payload bytes"),
555 &ReceiverBatchOptions {
556 raw_payload_routes: vec![route],
557 ..ReceiverBatchOptions::default()
558 },
559 )
560 .unwrap();
561
562 assert_eq!(batch.counters.wfb_payloads, 1);
563 assert_eq!(batch.counters.raw_payload_count, 1);
564 assert_eq!(batch.raw_payloads[0].data, b"payload bytes");
565 }
566
567 #[test]
568 fn rtp_payload_tap_copies_only_matching_payload_type() {
569 let video_route = PayloadRouteId::new(1);
570 let audio_route = PayloadRouteId::new(3);
571 let mut runtime = ReceiverRuntime::with_plain_video_route(
572 FrameLayout::WithFcs,
573 video_route,
574 ChannelId::default_video(),
575 0,
576 1,
577 1,
578 )
579 .unwrap();
580 runtime
581 .add_plain_route(audio_route, ChannelId::default_video(), 0, 1, 1)
582 .unwrap();
583
584 let ignored = runtime
585 .push_decrypted_fragment(
586 runtime.video_runtime(),
587 0,
588 &plain(&rtp(crate::rtp::RTP_PAYLOAD_TYPE_H264, b"video")),
589 &ReceiverBatchOptions {
590 rtp_payload_taps: vec![RtpPayloadTap {
591 route_id: audio_route,
592 payload_type: crate::rtp::RTP_PAYLOAD_TYPE_OPUS,
593 }],
594 ..ReceiverBatchOptions::default()
595 },
596 )
597 .unwrap();
598 assert_eq!(ignored.counters.raw_payload_count, 0);
599
600 let packet = rtp(crate::rtp::RTP_PAYLOAD_TYPE_OPUS, b"opus");
601 let batch = runtime
602 .push_decrypted_fragment(
603 runtime.video_runtime(),
604 1 << 8,
605 &plain(&packet),
606 &ReceiverBatchOptions {
607 rtp_payload_taps: vec![RtpPayloadTap {
608 route_id: audio_route,
609 payload_type: crate::rtp::RTP_PAYLOAD_TYPE_OPUS,
610 }],
611 ..ReceiverBatchOptions::default()
612 },
613 )
614 .unwrap();
615
616 assert_eq!(batch.counters.raw_payload_count, 1);
617 assert_eq!(batch.raw_payloads[0].route_id, audio_route);
618 assert_eq!(batch.raw_payloads[0].data, packet);
619 }
620
621 #[test]
622 fn rtp_reorder_is_opt_in() {
623 let mut runtime = ReceiverRuntime::with_plain_video_route(
624 FrameLayout::WithFcs,
625 PayloadRouteId::new(1),
626 ChannelId::default_video(),
627 0,
628 1,
629 1,
630 )
631 .unwrap();
632
633 assert!(!runtime.rtp_reorder_enabled());
634 assert_eq!(runtime.rtp_reorder_status(), RtpReorderStatus::default());
635
636 runtime.set_rtp_reorder_enabled(true);
637 assert!(runtime.rtp_reorder_enabled());
638
639 runtime.set_rtp_reorder_enabled(false);
640 assert!(!runtime.rtp_reorder_enabled());
641 assert_eq!(runtime.rtp_reorder_status(), RtpReorderStatus::default());
642 }
643
644 #[test]
645 fn auxiliary_route_does_not_count_as_video_payload() {
646 let video_route = PayloadRouteId::new(1);
647 let data_route = PayloadRouteId::new(2);
648 let mut runtime = ReceiverRuntime::with_plain_video_route(
649 FrameLayout::WithFcs,
650 video_route,
651 ChannelId::default_video(),
652 0,
653 1,
654 1,
655 )
656 .unwrap();
657 let data_runtime = runtime
658 .add_plain_route(
659 data_route,
660 ChannelId::from_link_port(crate::channel::DEFAULT_LINK_ID, RadioPort::TunnelRx),
661 0,
662 1,
663 1,
664 )
665 .unwrap();
666 let batch = runtime
667 .push_decrypted_fragment(
668 data_runtime,
669 0,
670 &plain(b"data bytes"),
671 &ReceiverBatchOptions {
672 raw_payload_routes: vec![data_route],
673 ..ReceiverBatchOptions::default()
674 },
675 )
676 .unwrap();
677
678 assert_eq!(batch.counters.wfb_payloads, 0);
679 assert_eq!(batch.counters.rtp_packets, 0);
680 assert_eq!(batch.counters.raw_payload_count, 1);
681 assert_eq!(batch.raw_payloads[0].data, b"data bytes");
682 }
683
684 #[test]
685 fn mock_payload_runtime_uses_same_video_route_and_rtp_depacketizer() {
686 let video_route = PayloadRouteId::new(1);
687 let mut runtime = ReceiverRuntime::with_mock_video_route(
688 FrameLayout::WithFcs,
689 video_route,
690 ChannelId::default_video(),
691 0,
692 );
693
694 let packet = h264_stap_a_rtp();
695 let batch = runtime
696 .push_mock_payload(
697 runtime.video_runtime(),
698 123,
699 &packet,
700 &ReceiverBatchOptions {
701 raw_payload_routes: vec![video_route],
702 ..ReceiverBatchOptions::default()
703 },
704 )
705 .unwrap();
706
707 assert_eq!(batch.counters.wfb_payloads, 1);
708 assert_eq!(batch.counters.rtp_packets, 1);
709 assert_eq!(batch.counters.video_frames, 1);
710 assert_eq!(batch.frames.len(), 1);
711 assert_eq!(batch.frames[0].codec, crate::rtp::Codec::H264);
712 assert!(batch.frames[0].is_keyframe);
713 assert_eq!(batch.raw_payloads[0].data, packet);
714 assert_eq!(batch.fec_counters, FecCounters::default());
715 }
716}