listen/
listen.rs

1//! The smallest SRT listener: accept every caller and print each payload's
2//! size. `cargo run -p sapflow-srt --example listen` then, for example,
3//! `ffmpeg -re -f lavfi -i testsrc2 -c:v libx264 -f mpegts 'srt://127.0.0.1:9000'`
4//! (an ffmpeg built with libsrt), or pipe MPEG-TS into libsrt's
5//! `srt-live-transmit file://con 'srt://127.0.0.1:9000?streamid=demo'`.
6//!
7//! The listener is sans-IO: this loop owns the socket and the clock, feeds
8//! it datagrams and timeouts, and sends what it asks to send.
9use sapflow_srt::{Config, Event, Listener};
10use std::io::ErrorKind;
11use std::net::UdpSocket;
12use std::time::{Duration, Instant};
13
14fn main() -> std::io::Result<()> {
15    let socket = UdpSocket::bind("0.0.0.0:9000")?;
16    let mut listener = Listener::new(Config::default(), Instant::now());
17    // An SRT datagram is at most 1,500 bytes on an ordinary network.
18    let mut buf = [0u8; 2048];
19
20    loop {
21        // Wait for a datagram or until the listener next needs the time.
22        let now = Instant::now();
23        let wait = listener
24            .poll_timeout()
25            .map_or(Duration::from_millis(100), |at| {
26                at.saturating_duration_since(now)
27            })
28            .max(Duration::from_millis(1));
29        socket.set_read_timeout(Some(wait))?;
30        match socket.recv_from(&mut buf) {
31            Ok((n, peer)) => {
32                if let Some(datagram) = buf.get(..n) {
33                    listener.handle_datagram(Instant::now(), peer, datagram);
34                }
35            }
36            Err(e) if matches!(e.kind(), ErrorKind::WouldBlock | ErrorKind::TimedOut) => {}
37            Err(e) => return Err(e),
38        }
39        listener.handle_timeout(Instant::now());
40
41        while let Some(event) = listener.poll_event() {
42            match event {
43                Event::Connect {
44                    id,
45                    peer,
46                    stream_id,
47                } => {
48                    // The stream id is untrusted input: check it here, and
49                    // call `listener.reject(id, code)` to refuse a caller.
50                    println!("{peer} wants {:?}", String::from_utf8_lossy(&stream_id));
51                    listener.accept(id, Instant::now());
52                }
53                Event::Data { id, payload } => println!("{id:?}: {} bytes", payload.len()),
54                Event::Closed { id, reason, .. } => println!("{id:?} closed: {reason:?}"),
55            }
56        }
57        while let Some((to, datagram)) = listener.poll_transmit() {
58            socket.send_to(&datagram, to)?;
59        }
60    }
61}
listen/listen.rs

listen/
listen.rs
