receive/
receive.rs

1//! Lab receiver for independent encrypted interoperability, not a server.
2//! receive <bind ip:port> <output file> <expected bytes> [latency ms]
3//! Optional SAPFLOW_SRT_PASSPHRASE and SAPFLOW_SRT_KEY_BITS=128|256.
4use sapflow_srt::crypto::{CryptoConfig, KeySize};
5use sapflow_srt::{Config, Event, Listener};
6use std::{
7    error::Error,
8    fs::File,
9    io::{ErrorKind, Write},
10    net::{SocketAddr, UdpSocket},
11    time::{Duration, Instant},
12};
13fn main() -> Result<(), Box<dyn Error>> {
14    let args: Vec<_> = std::env::args().skip(1).collect();
15    let (Some(bind), Some(path), Some(expected)) = (args.first(), args.get(1), args.get(2)) else {
16        return Err("receive <bind> <output> <expected bytes> [latency ms]".into());
17    };
18    let bind: SocketAddr = bind.parse()?;
19    let expected: usize = expected.parse()?;
20    let mut config = Config::default();
21    if let Some(ms) = args.get(3) {
22        config.latency = Duration::from_millis(ms.parse()?);
23    }
24    if let Ok(secret) = std::env::var("SAPFLOW_SRT_PASSPHRASE") {
25        let size = match std::env::var("SAPFLOW_SRT_KEY_BITS").as_deref() {
26            Ok("256") => KeySize::Aes256,
27            Ok("128") | Err(_) => KeySize::Aes128,
28            _ => return Err("key bits must be 128 or 256".into()),
29        };
30        config.crypto = Some(CryptoConfig::new(secret.into_bytes(), size)?);
31        config.require_encryption = true;
32    }
33    let socket = UdpSocket::bind(bind)?;
34    socket.set_read_timeout(Some(Duration::from_millis(5)))?;
35    let mut output = File::create(path)?;
36    let start = Instant::now();
37    let mut listener = Listener::new(config, start);
38    let mut packet = [0; 2048];
39    let mut bytes = 0;
40    let mut connection = None;
41    println!("listening {}", socket.local_addr()?);
42    std::io::stdout().flush()?;
43    while start.elapsed() < Duration::from_secs(20) {
44        let now = Instant::now();
45        listener.handle_timeout(now);
46        while let Some(event) = listener.poll_event() {
47            match event {
48                Event::Connect { id, stream_id, .. } => {
49                    println!("stream_id={}", String::from_utf8_lossy(&stream_id));
50                    if listener.accept(id, now) {
51                        connection = Some(id);
52                    }
53                }
54                Event::Data { payload, .. } => {
55                    output.write_all(&payload)?;
56                    bytes += payload.len();
57                }
58                Event::Closed { .. } => {}
59            }
60        }
61        if bytes >= expected {
62            let stats = connection.and_then(|id| listener.connection_stats(id));
63            println!("received_bytes={bytes} stats={stats:?}");
64            if let Some(id) = connection {
65                listener.close(id, now);
66            }
67            while let Some((peer, b)) = listener.poll_transmit() {
68                socket.send_to(&b, peer)?;
69            }
70            return Ok(());
71        }
72        while let Some((peer, b)) = listener.poll_transmit() {
73            socket.send_to(&b, peer)?;
74        }
75        match socket.recv_from(&mut packet) {
76            Ok((n, peer)) => {
77                listener.handle_datagram(Instant::now(), peer, packet.get(..n).unwrap_or_default())
78            }
79            Err(e)
80                if matches!(
81                    e.kind(),
82                    ErrorKind::TimedOut | ErrorKind::WouldBlock | ErrorKind::Interrupted
83                ) => {}
84            Err(e) => return Err(e.into()),
85        }
86    }
87    Err(format!(
88        "receiver timeout bytes={bytes} rejected={}",
89        listener.stats().rejected
90    )
91    .into())
92}
receive/receive.rs

receive/
receive.rs
