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 let mut packet = rtp(crate::rtp::RTP_PAYLOAD_TYPE_H264, &payload);
553 packet[1] |= 0x80;
554 packet
555 }
556
557 #[test]
558 fn decrypted_fragment_can_fan_out_to_raw_route() {
559 let route = PayloadRouteId::new(7);
560 let mut runtime = ReceiverRuntime::with_plain_video_route(
561 FrameLayout::WithFcs,
562 route,
563 ChannelId::default_video(),
564 0,
565 1,
566 1,
567 )
568 .unwrap();
569 let batch = runtime
570 .push_decrypted_fragment(
571 runtime.video_runtime(),
572 0,
573 &plain(b"payload bytes"),
574 &ReceiverBatchOptions {
575 raw_payload_routes: vec![route],
576 ..ReceiverBatchOptions::default()
577 },
578 )
579 .unwrap();
580
581 assert_eq!(batch.counters.wfb_payloads, 1);
582 assert_eq!(batch.counters.raw_payload_count, 1);
583 assert_eq!(batch.raw_payloads[0].data, b"payload bytes");
584 }
585
586 #[test]
587 fn rtp_payload_tap_copies_only_matching_payload_type() {
588 let video_route = PayloadRouteId::new(1);
589 let audio_route = PayloadRouteId::new(3);
590 let mut runtime = ReceiverRuntime::with_plain_video_route(
591 FrameLayout::WithFcs,
592 video_route,
593 ChannelId::default_video(),
594 0,
595 1,
596 1,
597 )
598 .unwrap();
599 runtime
600 .add_plain_route(audio_route, ChannelId::default_video(), 0, 1, 1)
601 .unwrap();
602
603 let ignored = runtime
604 .push_decrypted_fragment(
605 runtime.video_runtime(),
606 0,
607 &plain(&rtp(crate::rtp::RTP_PAYLOAD_TYPE_H264, b"video")),
608 &ReceiverBatchOptions {
609 rtp_payload_taps: vec![RtpPayloadTap {
610 route_id: audio_route,
611 payload_type: crate::rtp::RTP_PAYLOAD_TYPE_OPUS,
612 }],
613 ..ReceiverBatchOptions::default()
614 },
615 )
616 .unwrap();
617 assert_eq!(ignored.counters.raw_payload_count, 0);
618
619 let packet = rtp(crate::rtp::RTP_PAYLOAD_TYPE_OPUS, b"opus");
620 let batch = runtime
621 .push_decrypted_fragment(
622 runtime.video_runtime(),
623 1 << 8,
624 &plain(&packet),
625 &ReceiverBatchOptions {
626 rtp_payload_taps: vec![RtpPayloadTap {
627 route_id: audio_route,
628 payload_type: crate::rtp::RTP_PAYLOAD_TYPE_OPUS,
629 }],
630 ..ReceiverBatchOptions::default()
631 },
632 )
633 .unwrap();
634
635 assert_eq!(batch.counters.raw_payload_count, 1);
636 assert_eq!(batch.raw_payloads[0].route_id, audio_route);
637 assert_eq!(batch.raw_payloads[0].data, packet);
638 }
639
640 #[test]
641 fn rtp_reorder_is_opt_in() {
642 let mut runtime = ReceiverRuntime::with_plain_video_route(
643 FrameLayout::WithFcs,
644 PayloadRouteId::new(1),
645 ChannelId::default_video(),
646 0,
647 1,
648 1,
649 )
650 .unwrap();
651
652 assert!(!runtime.rtp_reorder_enabled());
653 assert_eq!(runtime.rtp_reorder_status(), RtpReorderStatus::default());
654
655 runtime.set_rtp_reorder_enabled(true);
656 assert!(runtime.rtp_reorder_enabled());
657
658 runtime.set_rtp_reorder_enabled(false);
659 assert!(!runtime.rtp_reorder_enabled());
660 assert_eq!(runtime.rtp_reorder_status(), RtpReorderStatus::default());
661 }
662
663 #[test]
664 fn auxiliary_route_does_not_count_as_video_payload() {
665 let video_route = PayloadRouteId::new(1);
666 let data_route = PayloadRouteId::new(2);
667 let mut runtime = ReceiverRuntime::with_plain_video_route(
668 FrameLayout::WithFcs,
669 video_route,
670 ChannelId::default_video(),
671 0,
672 1,
673 1,
674 )
675 .unwrap();
676 let data_runtime = runtime
677 .add_plain_route(
678 data_route,
679 ChannelId::from_link_port(crate::channel::DEFAULT_LINK_ID, RadioPort::TunnelRx),
680 0,
681 1,
682 1,
683 )
684 .unwrap();
685 let batch = runtime
686 .push_decrypted_fragment(
687 data_runtime,
688 0,
689 &plain(b"data bytes"),
690 &ReceiverBatchOptions {
691 raw_payload_routes: vec![data_route],
692 ..ReceiverBatchOptions::default()
693 },
694 )
695 .unwrap();
696
697 assert_eq!(batch.counters.wfb_payloads, 0);
698 assert_eq!(batch.counters.rtp_packets, 0);
699 assert_eq!(batch.counters.raw_payload_count, 1);
700 assert_eq!(batch.raw_payloads[0].data, b"data bytes");
701 }
702
703 #[test]
704 fn mock_payload_runtime_uses_same_video_route_and_rtp_depacketizer() {
705 let video_route = PayloadRouteId::new(1);
706 let mut runtime = ReceiverRuntime::with_mock_video_route(
707 FrameLayout::WithFcs,
708 video_route,
709 ChannelId::default_video(),
710 0,
711 );
712
713 let packet = h264_stap_a_rtp();
714 let batch = runtime
715 .push_mock_payload(
716 runtime.video_runtime(),
717 123,
718 &packet,
719 &ReceiverBatchOptions {
720 raw_payload_routes: vec![video_route],
721 ..ReceiverBatchOptions::default()
722 },
723 )
724 .unwrap();
725
726 assert_eq!(batch.counters.wfb_payloads, 1);
727 assert_eq!(batch.counters.rtp_packets, 1);
728 assert_eq!(batch.counters.video_frames, 1);
729 assert_eq!(batch.frames.len(), 1);
730 assert_eq!(batch.frames[0].codec, crate::rtp::Codec::H264);
731 assert!(batch.frames[0].is_keyframe);
732 assert_eq!(batch.raw_payloads[0].data, packet);
733 assert_eq!(batch.fec_counters, FecCounters::default());
734 }
735
736 #[test]
737 fn rx_transfer_accepts_explicit_jaguar3_descriptor_layout() {
738 let mut runtime = ReceiverRuntime::with_plain_video_route(
739 FrameLayout::WithFcs,
740 PayloadRouteId::new(1),
741 ChannelId::default_video(),
742 0,
743 1,
744 1,
745 )
746 .unwrap();
747 let mut transfer = vec![0u8; 32];
748 transfer[..4].copy_from_slice(&8u32.to_le_bytes());
749 transfer[24..32].copy_from_slice(&[0x08, 0, 0, 0, 0, 0, 0, 0]);
750
751 let batch = runtime
752 .push_rx_transfer_with_kind(
753 &transfer,
754 RxDescriptorKind::Jaguar3,
755 &ReceiverBatchOptions::default(),
756 )
757 .unwrap();
758
759 assert_eq!(batch.counters.packets, 1);
760 assert_eq!(batch.counters.accepted_packets, 1);
761 assert_eq!(batch.counters.ignored_frames, 1);
762 }
763}