send/
send.rs

1//! Lab SRT file sender using our own sans-IO caller.
2//! Run: cargo run -p sapflow-srt --example send -- 127.0.0.1:9000 publish:lab input.ts [bytes/sec] [latency-ms]
3//! Reads bounded chunks; this measures transport interoperability, not live
4//! capture timing or physical-device performance. Stream IDs travel in clear.
5use std::error::Error;
6use std::fs::File;
7use std::io::{ErrorKind, Read};
8use std::net::{SocketAddr, UdpSocket};
9use std::time::{Duration, Instant};
10
11use sapflow_srt::crypto::{CryptoConfig, KeySize};
12use sapflow_srt::{Caller, CallerConfig, CallerEvent};
13
14fn main() -> Result<(), Box<dyn Error>> {
15    let args: Vec<_> = std::env::args().skip(1).collect();
16    let (Some(destination), Some(stream_id), Some(path)) = (args.first(), args.get(1), args.get(2))
17    else {
18        return Err("usage: send <ip:port> <stream-id> <input.ts> [SRT-bytes/sec] [latency-ms]; LAB ONLY; optional SAPFLOW_SRT_PASSPHRASE env".into());
19    };
20    let destination: SocketAddr = destination.parse()?;
21    let mut config = CallerConfig {
22        stream_id: stream_id.clone(),
23        ..CallerConfig::default()
24    };
25    if let Some(bandwidth) = args.get(3) {
26        config.send.bandwidth = bandwidth.parse()?;
27    }
28    if let Some(latency) = args.get(4) {
29        config.latency = Duration::from_millis(latency.parse()?);
30    }
31    if let Ok(secret) = std::env::var("SAPFLOW_SRT_PASSPHRASE") {
32        let size = match std::env::var("SAPFLOW_SRT_KEY_BITS").as_deref() {
33            Ok("256") => KeySize::Aes256,
34            Ok("128") | Err(_) => KeySize::Aes128,
35            _ => return Err("key bits must be 128 or 256".into()),
36        };
37        let mut crypto = CryptoConfig::new(secret.into_bytes(), size)?;
38        if let Ok(refresh) = std::env::var("SAPFLOW_SRT_REFRESH") {
39            crypto.refresh_packets = refresh.parse()?;
40            crypto.preannounce_packets = crypto.refresh_packets / 4;
41        }
42        config.crypto = Some(crypto);
43    }
44    let encrypted = config.crypto.is_some();
45    let mut latency = config.latency;
46    let socket = UdpSocket::bind(if destination.is_ipv4() {
47        "0.0.0.0:0"
48    } else {
49        "[::]:0"
50    })?;
51    let mut caller = Caller::new(config, destination, Instant::now())?;
52    let mut file = File::open(path)?;
53    let mut input = [0u8; 1316];
54    let mut packet = [0u8; 2048];
55    let mut eof = false;
56    let mut drained_at = None;
57    let deadline = Instant::now() + Duration::from_secs(120);
58    eprintln!("LAB ONLY: encrypted={encrypted}; stream ID is visible on the network");
59    loop {
60        let now = Instant::now();
61        if now >= deadline {
62            caller.stop(now);
63            return Err("lab sender exceeded 120-second run budget".into());
64        }
65        caller.handle_timeout(now);
66        while let Some(event) = caller.poll_event() {
67            match event {
68                CallerEvent::Connected {
69                    latency: negotiated_latency,
70                    max_payload,
71                    ..
72                } => {
73                    latency = negotiated_latency;
74                    eprintln!(
75                        "connected latency_ms={} payload_limit={max_payload}",
76                        latency.as_millis()
77                    );
78                }
79                CallerEvent::Closed { reason } => {
80                    return Err(format!("SRT session closed: {reason:?}").into());
81                }
82            }
83        }
84        // Eight payloads at most are read ahead. Retained retransmissions
85        // remain accounted for, so slow receivers cannot drive file growth.
86        while caller.is_connected() && !eof && caller.stats().send.buffered_packets < 8 {
87            let len = file.read(&mut input)?;
88            if len == 0 {
89                eof = true;
90            } else {
91                caller.enqueue(now, input.get(..len).unwrap_or_default().to_vec())?;
92            }
93        }
94        while let Some((peer, bytes)) = caller.poll_transmit(Instant::now()) {
95            socket.send_to(&bytes, peer)?;
96        }
97        if eof && caller.stats().send.buffered_packets == 0 {
98            let drained = *drained_at.get_or_insert(now);
99            if now.saturating_duration_since(drained) >= latency + Duration::from_millis(250) {
100                caller.stop(now);
101                if let Some((peer, bytes)) = caller.poll_transmit(now) {
102                    socket.send_to(&bytes, peer)?;
103                }
104                let stats = caller.stats();
105                println!(
106                    "acknowledged_bytes={} acknowledged_packets={} retransmitted_packets={} rtt_us={} peak_buffered_bytes={}",
107                    stats.send.acknowledged_bytes,
108                    stats.send.acknowledged_packets,
109                    stats.send.retransmitted_packets,
110                    stats.send.rtt_us,
111                    stats.send.peak_buffered_bytes
112                );
113                return Ok(());
114            }
115        }
116        let wait = caller
117            .poll_timeout()
118            .map(|at| at.saturating_duration_since(Instant::now()))
119            .unwrap_or(Duration::from_millis(10))
120            .clamp(Duration::from_micros(100), Duration::from_millis(10));
121        socket.set_read_timeout(Some(wait))?;
122        match socket.recv_from(&mut packet) {
123            Ok((len, peer)) => {
124                caller.handle_datagram(Instant::now(), peer, packet.get(..len).unwrap_or_default())
125            }
126            Err(error)
127                if matches!(
128                    error.kind(),
129                    ErrorKind::TimedOut | ErrorKind::WouldBlock | ErrorKind::Interrupted
130                ) => {}
131            Err(error) => {
132                caller.stop(Instant::now());
133                return Err(error.into());
134            }
135        }
136    }
137}
send/send.rs

send/
send.rs
