sipx_testkit/
rtp_echo.rs

1//! A bounded RTP/PCMU echo peer for downstream media tests.
2//!
3//! This is a diagnostic fixture over the public packet and codec APIs, not a SIP user agent or a
4//! production media service. One [`RtpEcho`] owns one UDP socket and [`RtpEcho::run`] spawns no
5//! task, so completion, error and cancellation all have the same cleanup path: drop the future and
6//! the socket is released.
7
8use std::net::{IpAddr, Ipv4Addr, SocketAddr};
9use std::num::NonZeroUsize;
10use std::time::Duration;
11
12use bytes::Bytes;
13use sipx_audio::g711;
14use sipx_rtp::{Packet, RtpError};
15use thiserror::Error;
16use tokio::net::UdpSocket;
17
18/// Largest RTP datagram this telephony fixture admits.
19pub const MAX_DATAGRAM_BYTES: usize = 2048;
20
21const ECHO_SSRC: u32 = 0x5350_5854;
22const PCMU_PAYLOAD_TYPE: u8 = 0;
23
24/// A bounded RTP echo operation that could not be performed.
25#[derive(Debug, Error)]
26#[non_exhaustive]
27pub enum EchoError {
28    /// A configuration value cannot describe a finite, reachable fixture.
29    #[error("invalid {field}: {reason}")]
30    InvalidConfig {
31        /// The rejected field.
32        field: &'static str,
33        /// Why it was rejected.
34        reason: &'static str,
35    },
36    /// Binding, receiving or sending on the sole UDP socket failed.
37    #[error(transparent)]
38    Io(#[from] std::io::Error),
39    /// A datagram from the configured peer was not a valid RTP packet.
40    #[error(transparent)]
41    Rtp(#[from] RtpError),
42    /// A source other than the configured test peer sent a datagram.
43    #[error("received RTP from {actual}, expected {expected}")]
44    UnexpectedPeer {
45        /// The sole admitted peer.
46        expected: SocketAddr,
47        /// The source that actually sent the datagram.
48        actual: SocketAddr,
49    },
50    /// The fixture carries PCMU only.
51    #[error("RTP payload type {0} is unsupported; the echo fixture accepts PCMU payload type 0")]
52    UnsupportedPayloadType(u8),
53    /// A datagram could not be admitted without truncating it.
54    #[error("RTP datagram exceeds the {limit}-byte fixture limit")]
55    DatagramTooLarge {
56        /// Maximum admitted datagram length.
57        limit: usize,
58    },
59    /// UDP reported a partial datagram send.
60    #[error("UDP sent {sent} of {expected} echo bytes")]
61    PartialSend {
62        /// Bytes reported sent.
63        sent: usize,
64        /// Complete encoded packet length.
65        expected: usize,
66    },
67    /// The whole-run deadline elapsed before every configured packet was echoed.
68    #[error("echoed {received} of {expected} packets before the {within:?} run bound elapsed")]
69    TimedOut {
70        /// Packets successfully echoed.
71        received: usize,
72        /// Finite configured packet count.
73        expected: usize,
74        /// Whole-run failure bound.
75        within: Duration,
76    },
77}
78
79/// Explicit, finite configuration for one RTP echo run.
80#[derive(Debug, Clone, Copy)]
81pub struct EchoConfig {
82    bind: SocketAddr,
83    peer: SocketAddr,
84    packets: NonZeroUsize,
85    within: Duration,
86}
87
88impl EchoConfig {
89    /// Validate one finite RTP/PCMU echo run.
90    ///
91    /// Port zero is accepted for `bind`, which is useful in a test that reads [`RtpEcho::local_addr`]
92    /// before it starts its peer. The peer must be a concrete unicast destination with a non-zero
93    /// port, and both addresses must use the same IP family.
94    pub fn new(
95        bind: SocketAddr,
96        peer: SocketAddr,
97        packets: NonZeroUsize,
98        within: Duration,
99    ) -> Result<Self, EchoError> {
100        let broadcast = peer.ip() == IpAddr::V4(Ipv4Addr::BROADCAST);
101        if peer.ip().is_unspecified() || peer.ip().is_multicast() || broadcast || peer.port() == 0 {
102            return Err(EchoError::InvalidConfig {
103                field: "peer",
104                reason: "must be a concrete unicast address with a non-zero port",
105            });
106        }
107        if bind.is_ipv4() != peer.is_ipv4() {
108            return Err(EchoError::InvalidConfig {
109                field: "peer",
110                reason: "must use the bind address family",
111            });
112        }
113        if within.is_zero() {
114            return Err(EchoError::InvalidConfig {
115                field: "within",
116                reason: "must be greater than zero",
117            });
118        }
119        Ok(Self {
120            bind,
121            peer,
122            packets,
123            within,
124        })
125    }
126
127    /// Local address requested for the sole UDP socket.
128    #[must_use]
129    pub const fn bind_addr(self) -> SocketAddr {
130        self.bind
131    }
132
133    /// Sole admitted RTP source and echo destination.
134    #[must_use]
135    pub const fn peer(self) -> SocketAddr {
136        self.peer
137    }
138
139    /// Exact number of valid packets the run echoes.
140    #[must_use]
141    pub const fn packets(self) -> NonZeroUsize {
142        self.packets
143    }
144
145    /// Whole-run failure bound.
146    #[must_use]
147    pub const fn within(self) -> Duration {
148        self.within
149    }
150}
151
152/// A bound, not-yet-running RTP echo fixture.
153#[derive(Debug)]
154pub struct RtpEcho {
155    socket: UdpSocket,
156    local_addr: SocketAddr,
157    config: EchoConfig,
158}
159
160/// What one finite echo run completed.
161#[derive(Debug, Clone, Copy, PartialEq, Eq)]
162pub struct EchoReport {
163    /// RTP packets decoded and echoed.
164    pub packets: usize,
165    /// Decoded PCMU samples carried by those packets.
166    pub samples: usize,
167}
168
169impl RtpEcho {
170    /// Bind the fixture's sole UDP socket.
171    pub async fn bind(config: EchoConfig) -> Result<Self, EchoError> {
172        let socket = UdpSocket::bind(config.bind).await?;
173        let local_addr = socket.local_addr()?;
174        Ok(Self {
175            socket,
176            local_addr,
177            config,
178        })
179    }
180
181    /// The actual bound address, including the OS-selected port when configuration used port zero.
182    #[must_use]
183    pub const fn local_addr(&self) -> SocketAddr {
184        self.local_addr
185    }
186
187    /// Echo exactly the configured number of valid PCMU packets within the whole-run deadline.
188    ///
189    /// This consumes the fixture and spawns no task. Cancelling the future therefore drops the
190    /// sole socket immediately; completion and error take the same ownership path.
191    pub async fn run(self) -> Result<EchoReport, EchoError> {
192        let expected = self.config.packets.get();
193        let within = self.config.within;
194        let deadline = tokio::time::Instant::now() + within;
195        let mut buffer = [0_u8; MAX_DATAGRAM_BYTES + 1];
196        let mut packets = 0_usize;
197        let mut samples = 0_usize;
198        let mut sequence = 0_u16;
199        let mut timestamp = 0_u32;
200
201        while packets < expected {
202            let received = tokio::time::timeout_at(
203                deadline, // failure bound: maximum lifetime of one complete echo run
204                self.socket.recv_from(&mut buffer),
205            )
206            .await;
207            let (length, source) = match received {
208                Ok(result) => result?,
209                Err(_) => {
210                    return Err(EchoError::TimedOut {
211                        received: packets,
212                        expected,
213                        within,
214                    });
215                }
216            };
217            if source != self.config.peer {
218                return Err(EchoError::UnexpectedPeer {
219                    expected: self.config.peer,
220                    actual: source,
221                });
222            }
223            if length > MAX_DATAGRAM_BYTES {
224                return Err(EchoError::DatagramTooLarge {
225                    limit: MAX_DATAGRAM_BYTES,
226                });
227            }
228            let input =
229                Packet::decode(&Bytes::copy_from_slice(buffer.get(..length).unwrap_or(&[])))?;
230            if input.payload_type != PCMU_PAYLOAD_TYPE {
231                return Err(EchoError::UnsupportedPayloadType(input.payload_type));
232            }
233
234            let decoded = g711::ulaw_decode_all(&input.payload);
235            let sample_count = decoded.len();
236            let output = Packet::new(
237                PCMU_PAYLOAD_TYPE,
238                sequence,
239                timestamp,
240                ECHO_SSRC,
241                Bytes::from(g711::ulaw_encode_all(&decoded)),
242            )
243            .encode();
244            let sent = match tokio::time::timeout_at(
245                deadline, // failure bound: sending is inside the same whole-run deadline
246                self.socket.send_to(&output, self.config.peer),
247            )
248            .await
249            {
250                Ok(result) => result?,
251                Err(_) => {
252                    return Err(EchoError::TimedOut {
253                        received: packets,
254                        expected,
255                        within,
256                    });
257                }
258            };
259            if sent != output.len() {
260                return Err(EchoError::PartialSend {
261                    sent,
262                    expected: output.len(),
263                });
264            }
265
266            packets = packets.saturating_add(1);
267            samples = samples.saturating_add(sample_count);
268            sequence = sequence.wrapping_add(1);
269            timestamp = timestamp.wrapping_add(u32::try_from(sample_count).unwrap_or(u32::MAX));
270        }
271        Ok(EchoReport { packets, samples })
272    }
273}
sipx_testkit/rtp_echo.rs

sipx_testkit/
rtp_echo.rs
