1use crate::channel::ChannelId;
2use crate::ieee80211::FrameLayout;
3use crate::realtek::{
4 parse_rx_aggregate, parse_rx_aggregate_with_kind, AggregateError, RealtekRxPacket,
5 RxDescriptorKind, RxPacketType,
6};
7use crate::routes::{
8 PayloadRouteError, PayloadRouteEvent, PayloadRouteId, PayloadRouteManager, PayloadRuntimeKey,
9};
10use crate::rtp::{
11 DepacketizedFrame, RtpDepacketizer, RtpDepacketizerStatus, RtpHeader, RtpReorderBuffer,
12 RtpReorderStatus,
13};
14use crate::wfb::{FecCounters, WfbKeypair};
15
16#[derive(Debug, Clone)]
23pub struct ReceiverRuntime {
24 routes: PayloadRouteManager,
25 video_runtime: PayloadRuntimeKey,
26 video_route_id: PayloadRouteId,
27 rtp: RtpDepacketizer,
28 rtp_reorder: Option<RtpReorderBuffer>,
29}
30
31#[derive(Debug, Clone, Default)]
33pub struct ReceiverBatchOptions {
34 pub accept_corrupted: bool,
36 pub raw_payload_routes: Vec<PayloadRouteId>,
38 pub rtp_payload_taps: Vec<RtpPayloadTap>,
40}
41
42#[derive(Debug, Clone, Copy, PartialEq, Eq)]
49pub struct RtpPayloadTap {
50 pub route_id: PayloadRouteId,
52 pub payload_type: u8,
54}
55
56#[derive(Debug, Clone, PartialEq, Eq)]
58pub struct RoutePayload {
59 pub route_id: PayloadRouteId,
61 pub channel_id: ChannelId,
63 pub packet_seq: u64,
65 pub data: Vec<u8>,
67}
68
69#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)]
71pub struct ReceiverBatchCounters {
72 pub packets: usize,
74 pub accepted_packets: usize,
76 pub dropped_packets: usize,
78 pub crc_dropped: usize,
80 pub icv_dropped: usize,
82 pub report_dropped: usize,
84 pub ignored_frames: usize,
86 pub sessions: usize,
88 pub wfb_payloads: usize,
90 pub rtp_packets: usize,
92 pub video_frames: usize,
94 pub raw_payload_count: usize,
96 pub raw_payload_bytes: usize,
98 pub route_errors: usize,
100}
101
102#[derive(Debug, Clone, PartialEq, Eq)]
104pub struct ReceiverBatch {
105 pub frames: Vec<DepacketizedFrame>,
107 pub raw_payloads: Vec<RoutePayload>,
109 pub counters: ReceiverBatchCounters,
111 pub fec_counters: FecCounters,
113 pub rtp_status: RtpDepacketizerStatus,
115 pub rtp_reorder_status: RtpReorderStatus,
117}
118
119impl ReceiverRuntime {
120 pub fn from_routes(
125 routes: PayloadRouteManager,
126 video_runtime: PayloadRuntimeKey,
127 video_route_id: PayloadRouteId,
128 ) -> Self {
129 Self {
130 routes,
131 video_runtime,
132 video_route_id,
133 rtp: RtpDepacketizer::new(),
134 rtp_reorder: None,
135 }
136 }
137
138 pub fn with_plain_video_route(
142 frame_layout: FrameLayout,
143 video_route_id: PayloadRouteId,
144 channel_id: ChannelId,
145 key_slot: u64,
146 fec_k: usize,
147 fec_n: usize,
148 ) -> Result<Self, PayloadRouteError> {
149 let mut routes = PayloadRouteManager::new(frame_layout);
150 let video_runtime =
151 routes.add_plain_route(video_route_id, channel_id, key_slot, fec_k, fec_n)?;
152 Ok(Self::from_routes(routes, video_runtime, video_route_id))
153 }
154
155 pub fn with_keyed_video_route(
157 frame_layout: FrameLayout,
158 video_route_id: PayloadRouteId,
159 channel_id: ChannelId,
160 key_slot: u64,
161 keypair: WfbKeypair,
162 minimum_epoch: u64,
163 ) -> Result<Self, PayloadRouteError> {
164 let mut routes = PayloadRouteManager::new(frame_layout);
165 let video_runtime =
166 routes.add_keyed_route(video_route_id, channel_id, key_slot, keypair, minimum_epoch)?;
167 Ok(Self::from_routes(routes, video_runtime, video_route_id))
168 }
169
170 pub fn with_mock_video_route(
176 frame_layout: FrameLayout,
177 video_route_id: PayloadRouteId,
178 channel_id: ChannelId,
179 key_slot: u64,
180 ) -> Self {
181 let mut routes = PayloadRouteManager::new(frame_layout);
182 let video_runtime = routes.add_mock_route(video_route_id, channel_id, key_slot);
183 Self::from_routes(routes, video_runtime, video_route_id)
184 }
185
186 pub const fn video_runtime(&self) -> PayloadRuntimeKey {
188 self.video_runtime
189 }
190
191 pub const fn video_route_id(&self) -> PayloadRouteId {
193 self.video_route_id
194 }
195
196 pub fn routes(&self) -> &PayloadRouteManager {
198 &self.routes
199 }
200
201 pub fn routes_mut(&mut self) -> &mut PayloadRouteManager {
203 &mut self.routes
204 }
205
206 pub fn rtp_mut(&mut self) -> &mut RtpDepacketizer {
208 &mut self.rtp
209 }
210
211 pub fn rtp_status(&self) -> RtpDepacketizerStatus {
213 self.rtp.status()
214 }
215
216 pub fn rtp_reorder_status(&self) -> RtpReorderStatus {
218 self.rtp_reorder
219 .as_ref()
220 .map(RtpReorderBuffer::status)
221 .unwrap_or_default()
222 }
223
224 pub fn set_rtp_reorder_enabled(&mut self, enabled: bool) {
230 if enabled {
231 self.rtp_reorder
232 .get_or_insert_with(RtpReorderBuffer::default);
233 } else {
234 self.rtp_reorder = None;
235 }
236 }
237
238 pub const fn rtp_reorder_enabled(&self) -> bool {
240 self.rtp_reorder.is_some()
241 }
242
243 pub fn push_rtp_packet(
245 &mut self,
246 packet: &[u8],
247 ) -> Result<Vec<DepacketizedFrame>, crate::rtp::RtpError> {
248 self.push_video_payload(packet)
249 }
250
251 fn push_video_payload(
252 &mut self,
253 payload: &[u8],
254 ) -> Result<Vec<DepacketizedFrame>, crate::rtp::RtpError> {
255 let mut frames = Vec::new();
256 if let Some(reorder) = self.rtp_reorder.as_mut() {
257 for ordered in reorder.push(payload)? {
258 if let Some(frame) = self.rtp.push(&ordered)? {
259 frames.push(frame);
260 }
261 }
262 } else if let Some(frame) = self.rtp.push(payload)? {
263 frames.push(frame);
264 }
265 Ok(frames)
266 }
267
268 pub fn add_plain_route(
270 &mut self,
271 route_id: PayloadRouteId,
272 channel_id: ChannelId,
273 key_slot: u64,
274 fec_k: usize,
275 fec_n: usize,
276 ) -> Result<PayloadRuntimeKey, PayloadRouteError> {
277 self.routes
278 .add_plain_route(route_id, channel_id, key_slot, fec_k, fec_n)
279 }
280
281 pub fn add_keyed_route(
283 &mut self,
284 route_id: PayloadRouteId,
285 channel_id: ChannelId,
286 key_slot: u64,
287 keypair: WfbKeypair,
288 minimum_epoch: u64,
289 ) -> Result<PayloadRuntimeKey, PayloadRouteError> {
290 self.routes
291 .add_keyed_route(route_id, channel_id, key_slot, keypair, minimum_epoch)
292 }
293
294 pub fn add_mock_route(
296 &mut self,
297 route_id: PayloadRouteId,
298 channel_id: ChannelId,
299 key_slot: u64,
300 ) -> PayloadRuntimeKey {
301 self.routes.add_mock_route(route_id, channel_id, key_slot)
302 }
303
304 pub fn video_fec_counters(&self) -> FecCounters {
306 self.routes
307 .fec_counters(self.video_runtime)
308 .unwrap_or_default()
309 }
310
311 pub fn accepts_video_frame(&self, frame: &[u8]) -> bool {
313 self.routes.accepts_80211_frame(self.video_runtime, frame)
314 }
315
316 pub fn push_rx_transfer(
318 &mut self,
319 transfer: &[u8],
320 options: &ReceiverBatchOptions,
321 ) -> Result<ReceiverBatch, AggregateError> {
322 let packets = parse_rx_aggregate(transfer)?;
323 Ok(self.push_rx_packets(packets, options))
324 }
325
326 pub fn push_rx_transfer_with_kind(
331 &mut self,
332 transfer: &[u8],
333 descriptor_kind: RxDescriptorKind,
334 options: &ReceiverBatchOptions,
335 ) -> Result<ReceiverBatch, AggregateError> {
336 let packets = parse_rx_aggregate_with_kind(transfer, descriptor_kind)?;
337 Ok(self.push_rx_packets(packets, options))
338 }
339
340 pub fn push_rx_packets<'a>(
342 &mut self,
343 packets: impl IntoIterator<Item = RealtekRxPacket<'a>>,
344 options: &ReceiverBatchOptions,
345 ) -> ReceiverBatch {
346 let mut batch = self.empty_batch();
347
348 for packet in packets {
349 batch.counters.packets += 1;
350 if packet.attrib.crc_err && !options.accept_corrupted {
351 batch.counters.crc_dropped += 1;
352 continue;
353 }
354 if packet.attrib.icv_err && !options.accept_corrupted {
355 batch.counters.icv_dropped += 1;
356 continue;
357 }
358 if packet.attrib.pkt_rpt_type != RxPacketType::NormalRx {
359 batch.counters.report_dropped += 1;
360 continue;
361 }
362
363 batch.counters.accepted_packets += 1;
364 match self.routes.push_80211_frame(packet.data) {
365 Ok(events) => self.apply_route_events(events, options, &mut batch),
366 Err(_) => {
367 batch.counters.ignored_frames += 1;
368 batch.counters.route_errors += 1;
369 }
370 }
371 }
372
373 batch.counters.dropped_packets =
374 batch.counters.crc_dropped + batch.counters.icv_dropped + batch.counters.report_dropped;
375 batch.fec_counters = self.video_fec_counters();
376 batch
377 }
378
379 pub fn push_80211_frame(
381 &mut self,
382 frame: &[u8],
383 options: &ReceiverBatchOptions,
384 ) -> Result<ReceiverBatch, PayloadRouteError> {
385 let mut batch = self.empty_batch();
386 let events = self.routes.push_80211_frame(frame)?;
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_80211_frame(
394 &mut self,
395 runtime: PayloadRuntimeKey,
396 frame: &[u8],
397 decrypted_fragment: &[u8],
398 options: &ReceiverBatchOptions,
399 ) -> Result<ReceiverBatch, PayloadRouteError> {
400 let mut batch = self.empty_batch();
401 let events = self
402 .routes
403 .push_decrypted_80211_frame(runtime, frame, 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_decrypted_fragment(
411 &mut self,
412 runtime: PayloadRuntimeKey,
413 data_nonce: u64,
414 decrypted_fragment: &[u8],
415 options: &ReceiverBatchOptions,
416 ) -> Result<ReceiverBatch, PayloadRouteError> {
417 let mut batch = self.empty_batch();
418 let events =
419 self.routes
420 .push_decrypted_fragment(runtime, data_nonce, decrypted_fragment)?;
421 self.apply_route_events(events, options, &mut batch);
422 batch.fec_counters = self.video_fec_counters();
423 Ok(batch)
424 }
425
426 pub fn push_mock_payload(
428 &mut self,
429 runtime: PayloadRuntimeKey,
430 packet_seq: u64,
431 payload: &[u8],
432 options: &ReceiverBatchOptions,
433 ) -> Result<ReceiverBatch, PayloadRouteError> {
434 let mut batch = self.empty_batch();
435 let events = self
436 .routes
437 .push_mock_payload(runtime, packet_seq, payload)?;
438 self.apply_route_events(events, options, &mut batch);
439 batch.fec_counters = self.video_fec_counters();
440 Ok(batch)
441 }
442
443 fn empty_batch(&self) -> ReceiverBatch {
444 ReceiverBatch {
445 frames: Vec::new(),
446 raw_payloads: Vec::new(),
447 counters: ReceiverBatchCounters::default(),
448 fec_counters: self.video_fec_counters(),
449 rtp_status: self.rtp_status(),
450 rtp_reorder_status: self.rtp_reorder_status(),
451 }
452 }
453
454 fn apply_route_events(
455 &mut self,
456 events: Vec<PayloadRouteEvent>,
457 options: &ReceiverBatchOptions,
458 batch: &mut ReceiverBatch,
459 ) {
460 for event in events {
461 match event {
462 PayloadRouteEvent::IgnoredFrame => batch.counters.ignored_frames += 1,
463 PayloadRouteEvent::SessionEstablished { .. } => batch.counters.sessions += 1,
464 PayloadRouteEvent::Payload {
465 route_ids, payload, ..
466 } => {
467 if route_ids.contains(&self.video_route_id) {
468 batch.counters.wfb_payloads += 1;
469 batch.counters.rtp_packets += 1;
470 if let Ok(frames) = self.push_rtp_packet(&payload.data) {
471 batch.counters.video_frames += frames.len();
472 batch.frames.extend(frames);
473 }
474 }
475
476 for &route_id in &route_ids {
477 if !options.raw_payload_routes.contains(&route_id) {
478 continue;
479 }
480 copy_raw_payload(route_id, &payload, batch);
481 }
482
483 if options.rtp_payload_taps.is_empty() {
484 continue;
485 }
486 let Ok(header) = RtpHeader::parse(&payload.data) else {
487 continue;
488 };
489 for tap in &options.rtp_payload_taps {
490 if header.payload_type != tap.payload_type {
491 continue;
492 }
493 if !route_ids.contains(&tap.route_id) {
494 continue;
495 }
496 copy_raw_payload(tap.route_id, &payload, batch);
497 }
498 }
499 }
500 }
501 batch.rtp_status = self.rtp_status();
502 batch.rtp_reorder_status = self.rtp_reorder_status();
503 }
504}
505
506fn copy_raw_payload(
507 route_id: PayloadRouteId,
508 payload: &crate::pipeline::RecoveredPayload,
509 batch: &mut ReceiverBatch,
510) {
511 batch.counters.raw_payload_count += 1;
512 batch.counters.raw_payload_bytes += payload.data.len();
513 batch.raw_payloads.push(RoutePayload {
514 route_id,
515 channel_id: payload.channel_id,
516 packet_seq: payload.packet_seq,
517 data: payload.data.clone(),
518 });
519}
520
521#[cfg(test)]
522mod tests {
523 use super::*;
524 use crate::RadioPort;
525
526 fn plain(payload: &[u8]) -> Vec<u8> {
527 let mut out = Vec::new();
528 out.push(0);
529 out.extend_from_slice(&(payload.len() as u16).to_be_bytes());
530 out.extend_from_slice(payload);
531 out
532 }
533
534 fn rtp(payload_type: u8, payload: &[u8]) -> Vec<u8> {
535 let mut packet = vec![0x80, payload_type & 0x7f];
536 packet.extend_from_slice(&7u16.to_be_bytes());
537 packet.extend_from_slice(&48_000u32.to_be_bytes());
538 packet.extend_from_slice(&0x1122_3344u32.to_be_bytes());
539 packet.extend_from_slice(payload);
540 packet
541 }
542
543 fn h264_stap_a_rtp() -> Vec<u8> {
544 let sps = [0x67, 0x42, 0x00, 0x1e, 0xab];
545 let pps = [0x68, 0xce, 0x06, 0xe2];
546 let idr = [0x65, 0x88, 0x84, 0x21];
547 let mut payload = vec![24];
548 for nalu in [&sps[..], &pps[..], &idr[..]] {
549 payload.extend_from_slice(&(nalu.len() as u16).to_be_bytes());
550 payload.extend_from_slice(nalu);
551 }
552 rtp(crate::rtp::RTP_PAYLOAD_TYPE_H264, &payload)
553 }
554
555 #[test]
556 fn decrypted_fragment_can_fan_out_to_raw_route() {
557 let route = PayloadRouteId::new(7);
558 let mut runtime = ReceiverRuntime::with_plain_video_route(
559 FrameLayout::WithFcs,
560 route,
561 ChannelId::default_video(),
562 0,
563 1,
564 1,
565 )
566 .unwrap();
567 let batch = runtime
568 .push_decrypted_fragment(
569 runtime.video_runtime(),
570 0,
571 &plain(b"payload bytes"),
572 &ReceiverBatchOptions {
573 raw_payload_routes: vec![route],
574 ..ReceiverBatchOptions::default()
575 },
576 )
577 .unwrap();
578
579 assert_eq!(batch.counters.wfb_payloads, 1);
580 assert_eq!(batch.counters.raw_payload_count, 1);
581 assert_eq!(batch.raw_payloads[0].data, b"payload bytes");
582 }
583
584 #[test]
585 fn rtp_payload_tap_copies_only_matching_payload_type() {
586 let video_route = PayloadRouteId::new(1);
587 let audio_route = PayloadRouteId::new(3);
588 let mut runtime = ReceiverRuntime::with_plain_video_route(
589 FrameLayout::WithFcs,
590 video_route,
591 ChannelId::default_video(),
592 0,
593 1,
594 1,
595 )
596 .unwrap();
597 runtime
598 .add_plain_route(audio_route, ChannelId::default_video(), 0, 1, 1)
599 .unwrap();
600
601 let ignored = runtime
602 .push_decrypted_fragment(
603 runtime.video_runtime(),
604 0,
605 &plain(&rtp(crate::rtp::RTP_PAYLOAD_TYPE_H264, b"video")),
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 assert_eq!(ignored.counters.raw_payload_count, 0);
616
617 let packet = rtp(crate::rtp::RTP_PAYLOAD_TYPE_OPUS, b"opus");
618 let batch = runtime
619 .push_decrypted_fragment(
620 runtime.video_runtime(),
621 1 << 8,
622 &plain(&packet),
623 &ReceiverBatchOptions {
624 rtp_payload_taps: vec![RtpPayloadTap {
625 route_id: audio_route,
626 payload_type: crate::rtp::RTP_PAYLOAD_TYPE_OPUS,
627 }],
628 ..ReceiverBatchOptions::default()
629 },
630 )
631 .unwrap();
632
633 assert_eq!(batch.counters.raw_payload_count, 1);
634 assert_eq!(batch.raw_payloads[0].route_id, audio_route);
635 assert_eq!(batch.raw_payloads[0].data, packet);
636 }
637
638 #[test]
639 fn rtp_reorder_is_opt_in() {
640 let mut runtime = ReceiverRuntime::with_plain_video_route(
641 FrameLayout::WithFcs,
642 PayloadRouteId::new(1),
643 ChannelId::default_video(),
644 0,
645 1,
646 1,
647 )
648 .unwrap();
649
650 assert!(!runtime.rtp_reorder_enabled());
651 assert_eq!(runtime.rtp_reorder_status(), RtpReorderStatus::default());
652
653 runtime.set_rtp_reorder_enabled(true);
654 assert!(runtime.rtp_reorder_enabled());
655
656 runtime.set_rtp_reorder_enabled(false);
657 assert!(!runtime.rtp_reorder_enabled());
658 assert_eq!(runtime.rtp_reorder_status(), RtpReorderStatus::default());
659 }
660
661 #[test]
662 fn auxiliary_route_does_not_count_as_video_payload() {
663 let video_route = PayloadRouteId::new(1);
664 let data_route = PayloadRouteId::new(2);
665 let mut runtime = ReceiverRuntime::with_plain_video_route(
666 FrameLayout::WithFcs,
667 video_route,
668 ChannelId::default_video(),
669 0,
670 1,
671 1,
672 )
673 .unwrap();
674 let data_runtime = runtime
675 .add_plain_route(
676 data_route,
677 ChannelId::from_link_port(crate::channel::DEFAULT_LINK_ID, RadioPort::TunnelRx),
678 0,
679 1,
680 1,
681 )
682 .unwrap();
683 let batch = runtime
684 .push_decrypted_fragment(
685 data_runtime,
686 0,
687 &plain(b"data bytes"),
688 &ReceiverBatchOptions {
689 raw_payload_routes: vec![data_route],
690 ..ReceiverBatchOptions::default()
691 },
692 )
693 .unwrap();
694
695 assert_eq!(batch.counters.wfb_payloads, 0);
696 assert_eq!(batch.counters.rtp_packets, 0);
697 assert_eq!(batch.counters.raw_payload_count, 1);
698 assert_eq!(batch.raw_payloads[0].data, b"data bytes");
699 }
700
701 #[test]
702 fn mock_payload_runtime_uses_same_video_route_and_rtp_depacketizer() {
703 let video_route = PayloadRouteId::new(1);
704 let mut runtime = ReceiverRuntime::with_mock_video_route(
705 FrameLayout::WithFcs,
706 video_route,
707 ChannelId::default_video(),
708 0,
709 );
710
711 let packet = h264_stap_a_rtp();
712 let batch = runtime
713 .push_mock_payload(
714 runtime.video_runtime(),
715 123,
716 &packet,
717 &ReceiverBatchOptions {
718 raw_payload_routes: vec![video_route],
719 ..ReceiverBatchOptions::default()
720 },
721 )
722 .unwrap();
723
724 assert_eq!(batch.counters.wfb_payloads, 1);
725 assert_eq!(batch.counters.rtp_packets, 1);
726 assert_eq!(batch.counters.video_frames, 1);
727 assert_eq!(batch.frames.len(), 1);
728 assert_eq!(batch.frames[0].codec, crate::rtp::Codec::H264);
729 assert!(batch.frames[0].is_keyframe);
730 assert_eq!(batch.raw_payloads[0].data, packet);
731 assert_eq!(batch.fec_counters, FecCounters::default());
732 }
733
734 #[test]
735 fn rx_transfer_accepts_explicit_jaguar3_descriptor_layout() {
736 let mut runtime = ReceiverRuntime::with_plain_video_route(
737 FrameLayout::WithFcs,
738 PayloadRouteId::new(1),
739 ChannelId::default_video(),
740 0,
741 1,
742 1,
743 )
744 .unwrap();
745 let mut transfer = vec![0u8; 32];
746 transfer[..4].copy_from_slice(&8u32.to_le_bytes());
747 transfer[24..32].copy_from_slice(&[0x08, 0, 0, 0, 0, 0, 0, 0]);
748
749 let batch = runtime
750 .push_rx_transfer_with_kind(
751 &transfer,
752 RxDescriptorKind::Jaguar3,
753 &ReceiverBatchOptions::default(),
754 )
755 .unwrap();
756
757 assert_eq!(batch.counters.packets, 1);
758 assert_eq!(batch.counters.accepted_packets, 1);
759 assert_eq!(batch.counters.ignored_frames, 1);
760 }
761}