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 let mut frames = Vec::new();
249 self.push_video_payload_into(packet, &mut frames)?;
250 Ok(frames)
251 }
252
253 fn push_video_payload_into(
254 &mut self,
255 payload: &[u8],
256 frames: &mut Vec<DepacketizedFrame>,
257 ) -> Result<usize, crate::rtp::RtpError> {
258 let before = frames.len();
259 if let Some(reorder) = self.rtp_reorder.as_mut() {
260 for ordered in reorder.push(payload)? {
261 if let Some(frame) = self.rtp.push(&ordered)? {
262 frames.push(frame);
263 }
264 }
265 } else if let Some(frame) = self.rtp.push(payload)? {
266 frames.push(frame);
267 }
268 Ok(frames.len() - before)
269 }
270
271 pub fn add_plain_route(
273 &mut self,
274 route_id: PayloadRouteId,
275 channel_id: ChannelId,
276 key_slot: u64,
277 fec_k: usize,
278 fec_n: usize,
279 ) -> Result<PayloadRuntimeKey, PayloadRouteError> {
280 self.routes
281 .add_plain_route(route_id, channel_id, key_slot, fec_k, fec_n)
282 }
283
284 pub fn add_keyed_route(
286 &mut self,
287 route_id: PayloadRouteId,
288 channel_id: ChannelId,
289 key_slot: u64,
290 keypair: WfbKeypair,
291 minimum_epoch: u64,
292 ) -> Result<PayloadRuntimeKey, PayloadRouteError> {
293 self.routes
294 .add_keyed_route(route_id, channel_id, key_slot, keypair, minimum_epoch)
295 }
296
297 pub fn add_mock_route(
299 &mut self,
300 route_id: PayloadRouteId,
301 channel_id: ChannelId,
302 key_slot: u64,
303 ) -> PayloadRuntimeKey {
304 self.routes.add_mock_route(route_id, channel_id, key_slot)
305 }
306
307 pub fn video_fec_counters(&self) -> FecCounters {
309 self.routes
310 .fec_counters(self.video_runtime)
311 .unwrap_or_default()
312 }
313
314 pub fn accepts_video_frame(&self, frame: &[u8]) -> bool {
316 self.routes.accepts_80211_frame(self.video_runtime, frame)
317 }
318
319 pub fn push_rx_transfer(
321 &mut self,
322 transfer: &[u8],
323 options: &ReceiverBatchOptions,
324 ) -> Result<ReceiverBatch, AggregateError> {
325 let packets = parse_rx_aggregate(transfer)?;
326 Ok(self.push_rx_packets(packets, options))
327 }
328
329 pub fn push_rx_transfer_with_kind(
334 &mut self,
335 transfer: &[u8],
336 descriptor_kind: RxDescriptorKind,
337 options: &ReceiverBatchOptions,
338 ) -> Result<ReceiverBatch, AggregateError> {
339 let packets = parse_rx_aggregate_with_kind(transfer, descriptor_kind)?;
340 Ok(self.push_rx_packets(packets, options))
341 }
342
343 pub fn push_rx_packets<'a>(
345 &mut self,
346 packets: impl IntoIterator<Item = RealtekRxPacket<'a>>,
347 options: &ReceiverBatchOptions,
348 ) -> ReceiverBatch {
349 let mut batch = self.empty_batch();
350
351 for packet in packets {
352 batch.counters.packets += 1;
353 if packet.attrib.crc_err && !options.accept_corrupted {
354 batch.counters.crc_dropped += 1;
355 continue;
356 }
357 if packet.attrib.icv_err && !options.accept_corrupted {
358 batch.counters.icv_dropped += 1;
359 continue;
360 }
361 if packet.attrib.pkt_rpt_type != RxPacketType::NormalRx {
362 batch.counters.report_dropped += 1;
363 continue;
364 }
365
366 batch.counters.accepted_packets += 1;
367 match self.routes.push_80211_frame(packet.data) {
368 Ok(events) => self.apply_route_events(events, options, &mut batch),
369 Err(_) => {
370 batch.counters.ignored_frames += 1;
371 batch.counters.route_errors += 1;
372 }
373 }
374 }
375
376 batch.counters.dropped_packets =
377 batch.counters.crc_dropped + batch.counters.icv_dropped + batch.counters.report_dropped;
378 batch.fec_counters = self.video_fec_counters();
379 batch
380 }
381
382 pub fn push_80211_frame(
384 &mut self,
385 frame: &[u8],
386 options: &ReceiverBatchOptions,
387 ) -> Result<ReceiverBatch, PayloadRouteError> {
388 let mut batch = self.empty_batch();
389 let events = self.routes.push_80211_frame(frame)?;
390 self.apply_route_events(events, options, &mut batch);
391 batch.fec_counters = self.video_fec_counters();
392 Ok(batch)
393 }
394
395 pub fn push_decrypted_80211_frame(
397 &mut self,
398 runtime: PayloadRuntimeKey,
399 frame: &[u8],
400 decrypted_fragment: &[u8],
401 options: &ReceiverBatchOptions,
402 ) -> Result<ReceiverBatch, PayloadRouteError> {
403 let mut batch = self.empty_batch();
404 let events = self
405 .routes
406 .push_decrypted_80211_frame(runtime, frame, decrypted_fragment)?;
407 self.apply_route_events(events, options, &mut batch);
408 batch.fec_counters = self.video_fec_counters();
409 Ok(batch)
410 }
411
412 pub fn push_decrypted_fragment(
414 &mut self,
415 runtime: PayloadRuntimeKey,
416 data_nonce: u64,
417 decrypted_fragment: &[u8],
418 options: &ReceiverBatchOptions,
419 ) -> Result<ReceiverBatch, PayloadRouteError> {
420 let mut batch = self.empty_batch();
421 let events =
422 self.routes
423 .push_decrypted_fragment(runtime, data_nonce, decrypted_fragment)?;
424 self.apply_route_events(events, options, &mut batch);
425 batch.fec_counters = self.video_fec_counters();
426 Ok(batch)
427 }
428
429 pub fn push_mock_payload(
431 &mut self,
432 runtime: PayloadRuntimeKey,
433 packet_seq: u64,
434 payload: &[u8],
435 options: &ReceiverBatchOptions,
436 ) -> Result<ReceiverBatch, PayloadRouteError> {
437 let mut batch = self.empty_batch();
438 let events = self
439 .routes
440 .push_mock_payload(runtime, packet_seq, payload)?;
441 self.apply_route_events(events, options, &mut batch);
442 batch.fec_counters = self.video_fec_counters();
443 Ok(batch)
444 }
445
446 fn empty_batch(&self) -> ReceiverBatch {
447 ReceiverBatch {
448 frames: Vec::new(),
449 raw_payloads: Vec::new(),
450 counters: ReceiverBatchCounters::default(),
451 fec_counters: self.video_fec_counters(),
452 rtp_status: self.rtp_status(),
453 rtp_reorder_status: self.rtp_reorder_status(),
454 }
455 }
456
457 fn apply_route_events(
458 &mut self,
459 events: Vec<PayloadRouteEvent>,
460 options: &ReceiverBatchOptions,
461 batch: &mut ReceiverBatch,
462 ) {
463 for event in events {
464 match event {
465 PayloadRouteEvent::IgnoredFrame => batch.counters.ignored_frames += 1,
466 PayloadRouteEvent::SessionEstablished { .. } => batch.counters.sessions += 1,
467 PayloadRouteEvent::Payload {
468 route_ids, payload, ..
469 } => {
470 if route_ids.contains(&self.video_route_id) {
471 batch.counters.wfb_payloads += 1;
472 batch.counters.rtp_packets += 1;
473 if let Ok(frames) =
474 self.push_video_payload_into(&payload.data, &mut batch.frames)
475 {
476 batch.counters.video_frames += frames;
477 }
478 }
479
480 for &route_id in &route_ids {
481 if !options.raw_payload_routes.contains(&route_id) {
482 continue;
483 }
484 copy_raw_payload(route_id, &payload, batch);
485 }
486
487 if options.rtp_payload_taps.is_empty() {
488 continue;
489 }
490 let Ok(header) = RtpHeader::parse(&payload.data) else {
491 continue;
492 };
493 for tap in &options.rtp_payload_taps {
494 if header.payload_type != tap.payload_type {
495 continue;
496 }
497 if !route_ids.contains(&tap.route_id) {
498 continue;
499 }
500 copy_raw_payload(tap.route_id, &payload, batch);
501 }
502 }
503 }
504 }
505 batch.rtp_status = self.rtp_status();
506 batch.rtp_reorder_status = self.rtp_reorder_status();
507 }
508}
509
510fn copy_raw_payload(
511 route_id: PayloadRouteId,
512 payload: &crate::pipeline::RecoveredPayload,
513 batch: &mut ReceiverBatch,
514) {
515 batch.counters.raw_payload_count += 1;
516 batch.counters.raw_payload_bytes += payload.data.len();
517 batch.raw_payloads.push(RoutePayload {
518 route_id,
519 channel_id: payload.channel_id,
520 packet_seq: payload.packet_seq,
521 data: payload.data.clone(),
522 });
523}
524
525#[cfg(test)]
526mod tests {
527 use super::*;
528 use crate::RadioPort;
529
530 fn plain(payload: &[u8]) -> Vec<u8> {
531 let mut out = Vec::new();
532 out.push(0);
533 out.extend_from_slice(&(payload.len() as u16).to_be_bytes());
534 out.extend_from_slice(payload);
535 out
536 }
537
538 fn rtp(payload_type: u8, payload: &[u8]) -> Vec<u8> {
539 let mut packet = vec![0x80, payload_type & 0x7f];
540 packet.extend_from_slice(&7u16.to_be_bytes());
541 packet.extend_from_slice(&48_000u32.to_be_bytes());
542 packet.extend_from_slice(&0x1122_3344u32.to_be_bytes());
543 packet.extend_from_slice(payload);
544 packet
545 }
546
547 fn h264_stap_a_rtp() -> Vec<u8> {
548 let sps = [0x67, 0x42, 0x00, 0x1e, 0xab];
549 let pps = [0x68, 0xce, 0x06, 0xe2];
550 let idr = [0x65, 0x88, 0x84, 0x21];
551 let mut payload = vec![24];
552 for nalu in [&sps[..], &pps[..], &idr[..]] {
553 payload.extend_from_slice(&(nalu.len() as u16).to_be_bytes());
554 payload.extend_from_slice(nalu);
555 }
556 let mut packet = rtp(crate::rtp::RTP_PAYLOAD_TYPE_H264, &payload);
557 packet[1] |= 0x80;
558 packet
559 }
560
561 #[test]
562 fn decrypted_fragment_can_fan_out_to_raw_route() {
563 let route = PayloadRouteId::new(7);
564 let mut runtime = ReceiverRuntime::with_plain_video_route(
565 FrameLayout::WithFcs,
566 route,
567 ChannelId::default_video(),
568 0,
569 1,
570 1,
571 )
572 .unwrap();
573 let batch = runtime
574 .push_decrypted_fragment(
575 runtime.video_runtime(),
576 0,
577 &plain(b"payload bytes"),
578 &ReceiverBatchOptions {
579 raw_payload_routes: vec![route],
580 ..ReceiverBatchOptions::default()
581 },
582 )
583 .unwrap();
584
585 assert_eq!(batch.counters.wfb_payloads, 1);
586 assert_eq!(batch.counters.raw_payload_count, 1);
587 assert_eq!(batch.raw_payloads[0].data, b"payload bytes");
588 }
589
590 #[test]
591 fn rtp_payload_tap_copies_only_matching_payload_type() {
592 let video_route = PayloadRouteId::new(1);
593 let audio_route = PayloadRouteId::new(3);
594 let mut runtime = ReceiverRuntime::with_plain_video_route(
595 FrameLayout::WithFcs,
596 video_route,
597 ChannelId::default_video(),
598 0,
599 1,
600 1,
601 )
602 .unwrap();
603 runtime
604 .add_plain_route(audio_route, ChannelId::default_video(), 0, 1, 1)
605 .unwrap();
606
607 let ignored = runtime
608 .push_decrypted_fragment(
609 runtime.video_runtime(),
610 0,
611 &plain(&rtp(crate::rtp::RTP_PAYLOAD_TYPE_H264, b"video")),
612 &ReceiverBatchOptions {
613 rtp_payload_taps: vec![RtpPayloadTap {
614 route_id: audio_route,
615 payload_type: crate::rtp::RTP_PAYLOAD_TYPE_OPUS,
616 }],
617 ..ReceiverBatchOptions::default()
618 },
619 )
620 .unwrap();
621 assert_eq!(ignored.counters.raw_payload_count, 0);
622
623 let packet = rtp(crate::rtp::RTP_PAYLOAD_TYPE_OPUS, b"opus");
624 let batch = runtime
625 .push_decrypted_fragment(
626 runtime.video_runtime(),
627 1 << 8,
628 &plain(&packet),
629 &ReceiverBatchOptions {
630 rtp_payload_taps: vec![RtpPayloadTap {
631 route_id: audio_route,
632 payload_type: crate::rtp::RTP_PAYLOAD_TYPE_OPUS,
633 }],
634 ..ReceiverBatchOptions::default()
635 },
636 )
637 .unwrap();
638
639 assert_eq!(batch.counters.raw_payload_count, 1);
640 assert_eq!(batch.raw_payloads[0].route_id, audio_route);
641 assert_eq!(batch.raw_payloads[0].data, packet);
642 }
643
644 #[test]
645 fn rtp_reorder_is_opt_in() {
646 let mut runtime = ReceiverRuntime::with_plain_video_route(
647 FrameLayout::WithFcs,
648 PayloadRouteId::new(1),
649 ChannelId::default_video(),
650 0,
651 1,
652 1,
653 )
654 .unwrap();
655
656 assert!(!runtime.rtp_reorder_enabled());
657 assert_eq!(runtime.rtp_reorder_status(), RtpReorderStatus::default());
658
659 runtime.set_rtp_reorder_enabled(true);
660 assert!(runtime.rtp_reorder_enabled());
661
662 runtime.set_rtp_reorder_enabled(false);
663 assert!(!runtime.rtp_reorder_enabled());
664 assert_eq!(runtime.rtp_reorder_status(), RtpReorderStatus::default());
665 }
666
667 #[test]
668 fn auxiliary_route_does_not_count_as_video_payload() {
669 let video_route = PayloadRouteId::new(1);
670 let data_route = PayloadRouteId::new(2);
671 let mut runtime = ReceiverRuntime::with_plain_video_route(
672 FrameLayout::WithFcs,
673 video_route,
674 ChannelId::default_video(),
675 0,
676 1,
677 1,
678 )
679 .unwrap();
680 let data_runtime = runtime
681 .add_plain_route(
682 data_route,
683 ChannelId::from_link_port(crate::channel::DEFAULT_LINK_ID, RadioPort::TunnelRx),
684 0,
685 1,
686 1,
687 )
688 .unwrap();
689 let batch = runtime
690 .push_decrypted_fragment(
691 data_runtime,
692 0,
693 &plain(b"data bytes"),
694 &ReceiverBatchOptions {
695 raw_payload_routes: vec![data_route],
696 ..ReceiverBatchOptions::default()
697 },
698 )
699 .unwrap();
700
701 assert_eq!(batch.counters.wfb_payloads, 0);
702 assert_eq!(batch.counters.rtp_packets, 0);
703 assert_eq!(batch.counters.raw_payload_count, 1);
704 assert_eq!(batch.raw_payloads[0].data, b"data bytes");
705 }
706
707 #[test]
708 fn mock_payload_runtime_uses_same_video_route_and_rtp_depacketizer() {
709 let video_route = PayloadRouteId::new(1);
710 let mut runtime = ReceiverRuntime::with_mock_video_route(
711 FrameLayout::WithFcs,
712 video_route,
713 ChannelId::default_video(),
714 0,
715 );
716
717 let packet = h264_stap_a_rtp();
718 let batch = runtime
719 .push_mock_payload(
720 runtime.video_runtime(),
721 123,
722 &packet,
723 &ReceiverBatchOptions {
724 raw_payload_routes: vec![video_route],
725 ..ReceiverBatchOptions::default()
726 },
727 )
728 .unwrap();
729
730 assert_eq!(batch.counters.wfb_payloads, 1);
731 assert_eq!(batch.counters.rtp_packets, 1);
732 assert_eq!(batch.counters.video_frames, 1);
733 assert_eq!(batch.frames.len(), 1);
734 assert_eq!(batch.frames[0].codec, crate::rtp::Codec::H264);
735 assert!(batch.frames[0].is_keyframe);
736 assert_eq!(batch.raw_payloads[0].data, packet);
737 assert_eq!(batch.fec_counters, FecCounters::default());
738 }
739
740 #[test]
741 fn rx_transfer_accepts_explicit_jaguar3_descriptor_layout() {
742 let mut runtime = ReceiverRuntime::with_plain_video_route(
743 FrameLayout::WithFcs,
744 PayloadRouteId::new(1),
745 ChannelId::default_video(),
746 0,
747 1,
748 1,
749 )
750 .unwrap();
751 let mut transfer = vec![0u8; 32];
752 transfer[..4].copy_from_slice(&8u32.to_le_bytes());
753 transfer[24..32].copy_from_slice(&[0x08, 0, 0, 0, 0, 0, 0, 0]);
754
755 let batch = runtime
756 .push_rx_transfer_with_kind(
757 &transfer,
758 RxDescriptorKind::Jaguar3,
759 &ReceiverBatchOptions::default(),
760 )
761 .unwrap();
762
763 assert_eq!(batch.counters.packets, 1);
764 assert_eq!(batch.counters.accepted_packets, 1);
765 assert_eq!(batch.counters.ignored_frames, 1);
766 }
767}