openipc_core/
pipeline.rs

1use crate::channel::ChannelId;
2use crate::ieee80211::{FrameLayout, WifiFrame};
3use crate::wfb::{
4    parse_forwarder_packet, FecCounters, PlainAssembler, WfbError, WfbEvent, WfbKeypair, WfbPacket,
5    WfbReceiver,
6};
7
8/// Payload recovered from one OpenIPC/WFB channel.
9#[derive(Debug, Clone, PartialEq, Eq)]
10pub struct RecoveredPayload {
11    /// WFB channel that produced this payload.
12    pub channel_id: ChannelId,
13    /// Recovered WFB packet sequence number.
14    pub packet_seq: u64,
15    /// Raw application payload bytes.
16    pub data: Vec<u8>,
17}
18
19/// Event emitted by the lower-level single-channel payload pipeline.
20#[derive(Debug, Clone, PartialEq, Eq)]
21pub enum PayloadPipelineEvent {
22    /// Frame was valid enough to inspect but did not match this pipeline.
23    IgnoredFrame,
24    /// A WFB session packet established or refreshed decryption/FEC state.
25    SessionEstablished {
26        /// Session epoch accepted from the transmitter.
27        epoch: u64,
28        /// Number of primary fragments in each FEC block.
29        fec_k: usize,
30        /// Total primary plus parity fragments in each FEC block.
31        fec_n: usize,
32    },
33    /// A raw application payload was recovered.
34    Payload(RecoveredPayload),
35}
36
37/// Single-channel OpenIPC/WFB recovery pipeline.
38///
39/// This type stops at recovered payload bytes. Use [`crate::ReceiverRuntime`]
40/// when you also want route fanout and built-in RTP-to-video depacketization.
41#[derive(Debug, Clone)]
42pub struct PayloadPipeline {
43    channel_id: ChannelId,
44    frame_layout: FrameLayout,
45    assembler: PlainAssembler,
46    wfb_receiver: Option<WfbReceiver>,
47}
48
49impl PayloadPipeline {
50    /// Create a pipeline for already-plain WFB fragments.
51    pub fn new(
52        channel_id: ChannelId,
53        frame_layout: FrameLayout,
54        fec_k: usize,
55        fec_n: usize,
56    ) -> Result<Self, WfbError> {
57        Ok(Self {
58            channel_id,
59            frame_layout,
60            assembler: PlainAssembler::new(fec_k, fec_n)?,
61            wfb_receiver: None,
62        })
63    }
64
65    /// Create a pipeline that accepts encrypted WFB session and data packets.
66    pub fn with_keypair(
67        channel_id: ChannelId,
68        frame_layout: FrameLayout,
69        keypair: WfbKeypair,
70        minimum_epoch: u64,
71    ) -> Result<Self, WfbError> {
72        Ok(Self {
73            channel_id,
74            frame_layout,
75            assembler: PlainAssembler::new(1, 1)?,
76            wfb_receiver: Some(WfbReceiver::new(channel_id, keypair, minimum_epoch)),
77        })
78    }
79
80    /// Return this pipeline's channel id.
81    pub const fn channel_id(&self) -> ChannelId {
82        self.channel_id
83    }
84
85    /// Return cumulative FEC counters.
86    pub fn fec_counters(&self) -> FecCounters {
87        self.wfb_receiver
88            .as_ref()
89            .map(WfbReceiver::counters)
90            .unwrap_or_else(|| self.assembler.counters())
91    }
92
93    /// Return true if an 802.11 frame belongs to this pipeline's channel.
94    pub fn accepts_80211_frame(&self, frame: &[u8]) -> bool {
95        WifiFrame::parse(frame, self.frame_layout)
96            .map(|frame| frame.matches_channel_id(self.channel_id))
97            .unwrap_or(false)
98    }
99
100    /// Process a raw WFB 802.11 frame and stop at recovered payload bytes.
101    ///
102    /// This is the protocol boundary shared by video, telemetry, and custom
103    /// channels. It does not parse RTP, MAVLink, MSP, CRSF, IP, or any other
104    /// application payload. Feed `RecoveredPayload::data` into the next-stage
105    /// protocol handler chosen by the application.
106    pub fn push_80211_frame(
107        &mut self,
108        frame: &[u8],
109    ) -> Result<Vec<PayloadPipelineEvent>, WfbError> {
110        let Ok(frame) = WifiFrame::parse(frame, self.frame_layout) else {
111            return Ok(vec![PayloadPipelineEvent::IgnoredFrame]);
112        };
113        if !frame.matches_channel_id(self.channel_id) {
114            return Ok(vec![PayloadPipelineEvent::IgnoredFrame]);
115        }
116
117        if let Some(receiver) = self.wfb_receiver.as_mut() {
118            let events = receiver.push_forwarder_packet(frame.payload())?;
119            return Ok(self.map_wfb_events(events));
120        }
121
122        match parse_forwarder_packet(frame.payload())? {
123            WfbPacket::Data {
124                data_nonce,
125                encrypted_payload,
126                ..
127            } => self.push_decrypted_fragment(data_nonce, encrypted_payload),
128            WfbPacket::SessionKey { .. } => Ok(Vec::new()),
129        }
130    }
131
132    /// Process an 802.11 frame when the WFB data fragment is already decrypted.
133    pub fn push_decrypted_80211_frame(
134        &mut self,
135        frame: &[u8],
136        decrypted_fragment: &[u8],
137    ) -> Result<Vec<PayloadPipelineEvent>, WfbError> {
138        let Ok(frame) = WifiFrame::parse(frame, self.frame_layout) else {
139            return Ok(vec![PayloadPipelineEvent::IgnoredFrame]);
140        };
141        if !frame.matches_channel_id(self.channel_id) {
142            return Ok(vec![PayloadPipelineEvent::IgnoredFrame]);
143        }
144
145        let packet = match parse_forwarder_packet(frame.payload())? {
146            WfbPacket::Data { data_nonce, .. } => data_nonce,
147            WfbPacket::SessionKey { .. } => return Ok(Vec::new()),
148        };
149        self.push_decrypted_fragment(packet, decrypted_fragment)
150    }
151
152    /// Push an already-decrypted WFB fragment into the plain assembler.
153    pub fn push_decrypted_fragment(
154        &mut self,
155        data_nonce: u64,
156        decrypted_fragment: &[u8],
157    ) -> Result<Vec<PayloadPipelineEvent>, WfbError> {
158        let payloads = self
159            .assembler
160            .push_decrypted_fragment(data_nonce, decrypted_fragment)?;
161        Ok(payloads
162            .into_iter()
163            .map(|payload| {
164                PayloadPipelineEvent::Payload(RecoveredPayload {
165                    channel_id: self.channel_id,
166                    packet_seq: payload.packet_seq,
167                    data: payload.payload,
168                })
169            })
170            .collect())
171    }
172
173    fn map_wfb_events(&self, events: Vec<WfbEvent>) -> Vec<PayloadPipelineEvent> {
174        events
175            .into_iter()
176            .map(|event| match event {
177                WfbEvent::Session(session) => PayloadPipelineEvent::SessionEstablished {
178                    epoch: session.epoch,
179                    fec_k: session.fec_k,
180                    fec_n: session.fec_n,
181                },
182                WfbEvent::Payload(payload) => PayloadPipelineEvent::Payload(RecoveredPayload {
183                    channel_id: self.channel_id,
184                    packet_seq: payload.packet_seq,
185                    data: payload.payload,
186                }),
187            })
188            .collect()
189    }
190}
191
192#[cfg(test)]
193mod tests {
194    use super::*;
195
196    fn plain(payload: &[u8]) -> Vec<u8> {
197        let mut out = Vec::new();
198        out.push(0);
199        out.extend_from_slice(&(payload.len() as u16).to_be_bytes());
200        out.extend_from_slice(payload);
201        out
202    }
203
204    #[test]
205    fn payload_pipeline_emits_raw_recovered_payloads() {
206        let mut pipeline =
207            PayloadPipeline::new(ChannelId::default_video(), FrameLayout::WithFcs, 1, 1).unwrap();
208        let events = pipeline
209            .push_decrypted_fragment(0, &plain(b"telemetry bytes"))
210            .unwrap();
211        assert_eq!(
212            events,
213            vec![PayloadPipelineEvent::Payload(RecoveredPayload {
214                channel_id: ChannelId::default_video(),
215                packet_seq: 0,
216                data: b"telemetry bytes".to_vec(),
217            })]
218        );
219    }
220
221    #[test]
222    fn recovered_payloads_carry_the_pipeline_channel_id() {
223        let channel_id = ChannelId::from_link_port(0x112233, crate::RadioPort::TunnelRx);
224        let mut pipeline = PayloadPipeline::new(channel_id, FrameLayout::WithFcs, 1, 1).unwrap();
225        let events = pipeline
226            .push_decrypted_fragment(0, &plain(b"data bytes"))
227            .unwrap();
228        let PayloadPipelineEvent::Payload(payload) = &events[0] else {
229            panic!("expected payload event");
230        };
231        assert_eq!(payload.channel_id, channel_id);
232        assert_eq!(payload.data, b"data bytes");
233    }
234}
openipc_core/pipeline.rs

openipc_core/
pipeline.rs
