Struct Listener 

pub struct Listener<S = RandomState> { /* private fields */ }

A sans-IO SRT listener: every connection on one UDP port, with its handshakes, timers and events. S keys the SYN cookies and socket ids; Listener::new makes a random one.

Implementations

impl Listener<RandomState>

pub fn new(config: Config, now: Instant) -> Self

A listener whose cookies and socket ids are keyed by a random key made for this process.

Examples found in repository

examples/listen.rs (line 16)

fn main() -> std::io::Result<()> {
    let socket = UdpSocket::bind("0.0.0.0:9000")?;
    let mut listener = Listener::new(Config::default(), Instant::now());
    // An SRT datagram is at most 1,500 bytes on an ordinary network.
    let mut buf = [0u8; 2048];

    loop {
        // Wait for a datagram or until the listener next needs the time.
        let now = Instant::now();
        let wait = listener
            .poll_timeout()
            .map_or(Duration::from_millis(100), |at| {
                at.saturating_duration_since(now)
            })
            .max(Duration::from_millis(1));
        socket.set_read_timeout(Some(wait))?;
        match socket.recv_from(&mut buf) {
            Ok((n, peer)) => {
                if let Some(datagram) = buf.get(..n) {
                    listener.handle_datagram(Instant::now(), peer, datagram);
                }
            }
            Err(e) if matches!(e.kind(), ErrorKind::WouldBlock | ErrorKind::TimedOut) => {}
            Err(e) => return Err(e),
        }
        listener.handle_timeout(Instant::now());

        while let Some(event) = listener.poll_event() {
            match event {
                Event::Connect {
                    id,
                    peer,
                    stream_id,
                } => {
                    // The stream id is untrusted input: check it here, and
                    // call `listener.reject(id, code)` to refuse a caller.
                    println!("{peer} wants {:?}", String::from_utf8_lossy(&stream_id));
                    listener.accept(id, Instant::now());
                }
                Event::Data { id, payload } => println!("{id:?}: {} bytes", payload.len()),
                Event::Closed { id, reason, .. } => println!("{id:?} closed: {reason:?}"),
            }
        }
        while let Some((to, datagram)) = listener.poll_transmit() {
            socket.send_to(&datagram, to)?;
        }
    }
}

examples/receive.rs (line 37)

fn main() -> Result<(), Box<dyn Error>> {
    let args: Vec<_> = std::env::args().skip(1).collect();
    let (Some(bind), Some(path), Some(expected)) = (args.first(), args.get(1), args.get(2)) else {
        return Err("receive <bind> <output> <expected bytes> [latency ms]".into());
    };
    let bind: SocketAddr = bind.parse()?;
    let expected: usize = expected.parse()?;
    let mut config = Config::default();
    if let Some(ms) = args.get(3) {
        config.latency = Duration::from_millis(ms.parse()?);
    }
    if let Ok(secret) = std::env::var("SAPFLOW_SRT_PASSPHRASE") {
        let size = match std::env::var("SAPFLOW_SRT_KEY_BITS").as_deref() {
            Ok("256") => KeySize::Aes256,
            Ok("128") | Err(_) => KeySize::Aes128,
            _ => return Err("key bits must be 128 or 256".into()),
        };
        config.crypto = Some(CryptoConfig::new(secret.into_bytes(), size)?);
        config.require_encryption = true;
    }
    let socket = UdpSocket::bind(bind)?;
    socket.set_read_timeout(Some(Duration::from_millis(5)))?;
    let mut output = File::create(path)?;
    let start = Instant::now();
    let mut listener = Listener::new(config, start);
    let mut packet = [0; 2048];
    let mut bytes = 0;
    let mut connection = None;
    println!("listening {}", socket.local_addr()?);
    std::io::stdout().flush()?;
    while start.elapsed() < Duration::from_secs(20) {
        let now = Instant::now();
        listener.handle_timeout(now);
        while let Some(event) = listener.poll_event() {
            match event {
                Event::Connect { id, stream_id, .. } => {
                    println!("stream_id={}", String::from_utf8_lossy(&stream_id));
                    if listener.accept(id, now) {
                        connection = Some(id);
                    }
                }
                Event::Data { payload, .. } => {
                    output.write_all(&payload)?;
                    bytes += payload.len();
                }
                Event::Closed { .. } => {}
            }
        }
        if bytes >= expected {
            let stats = connection.and_then(|id| listener.connection_stats(id));
            println!("received_bytes={bytes} stats={stats:?}");
            if let Some(id) = connection {
                listener.close(id, now);
            }
            while let Some((peer, b)) = listener.poll_transmit() {
                socket.send_to(&b, peer)?;
            }
            return Ok(());
        }
        while let Some((peer, b)) = listener.poll_transmit() {
            socket.send_to(&b, peer)?;
        }
        match socket.recv_from(&mut packet) {
            Ok((n, peer)) => {
                listener.handle_datagram(Instant::now(), peer, packet.get(..n).unwrap_or_default())
            }
            Err(e)
                if matches!(
                    e.kind(),
                    ErrorKind::TimedOut | ErrorKind::WouldBlock | ErrorKind::Interrupted
                ) => {}
            Err(e) => return Err(e.into()),
        }
    }
    Err(format!(
        "receiver timeout bytes={bytes} rejected={}",
        listener.stats().rejected
    )
    .into())
}

impl<S: BuildHasher> Listener<S>

pub fn with_key(config: Config, now: Instant, key: S) -> Self

A listener keyed by key: give it a random one (tests and fuzzing use a fixed one, for repeatable runs).

pub fn stats(&self) -> ListenerStats

Counts of datagrams discarded so far, by kind.

Examples found in repository

examples/receive.rs (line 89)

fn main() -> Result<(), Box<dyn Error>> {
    let args: Vec<_> = std::env::args().skip(1).collect();
    let (Some(bind), Some(path), Some(expected)) = (args.first(), args.get(1), args.get(2)) else {
        return Err("receive <bind> <output> <expected bytes> [latency ms]".into());
    };
    let bind: SocketAddr = bind.parse()?;
    let expected: usize = expected.parse()?;
    let mut config = Config::default();
    if let Some(ms) = args.get(3) {
        config.latency = Duration::from_millis(ms.parse()?);
    }
    if let Ok(secret) = std::env::var("SAPFLOW_SRT_PASSPHRASE") {
        let size = match std::env::var("SAPFLOW_SRT_KEY_BITS").as_deref() {
            Ok("256") => KeySize::Aes256,
            Ok("128") | Err(_) => KeySize::Aes128,
            _ => return Err("key bits must be 128 or 256".into()),
        };
        config.crypto = Some(CryptoConfig::new(secret.into_bytes(), size)?);
        config.require_encryption = true;
    }
    let socket = UdpSocket::bind(bind)?;
    socket.set_read_timeout(Some(Duration::from_millis(5)))?;
    let mut output = File::create(path)?;
    let start = Instant::now();
    let mut listener = Listener::new(config, start);
    let mut packet = [0; 2048];
    let mut bytes = 0;
    let mut connection = None;
    println!("listening {}", socket.local_addr()?);
    std::io::stdout().flush()?;
    while start.elapsed() < Duration::from_secs(20) {
        let now = Instant::now();
        listener.handle_timeout(now);
        while let Some(event) = listener.poll_event() {
            match event {
                Event::Connect { id, stream_id, .. } => {
                    println!("stream_id={}", String::from_utf8_lossy(&stream_id));
                    if listener.accept(id, now) {
                        connection = Some(id);
                    }
                }
                Event::Data { payload, .. } => {
                    output.write_all(&payload)?;
                    bytes += payload.len();
                }
                Event::Closed { .. } => {}
            }
        }
        if bytes >= expected {
            let stats = connection.and_then(|id| listener.connection_stats(id));
            println!("received_bytes={bytes} stats={stats:?}");
            if let Some(id) = connection {
                listener.close(id, now);
            }
            while let Some((peer, b)) = listener.poll_transmit() {
                socket.send_to(&b, peer)?;
            }
            return Ok(());
        }
        while let Some((peer, b)) = listener.poll_transmit() {
            socket.send_to(&b, peer)?;
        }
        match socket.recv_from(&mut packet) {
            Ok((n, peer)) => {
                listener.handle_datagram(Instant::now(), peer, packet.get(..n).unwrap_or_default())
            }
            Err(e)
                if matches!(
                    e.kind(),
                    ErrorKind::TimedOut | ErrorKind::WouldBlock | ErrorKind::Interrupted
                ) => {}
            Err(e) => return Err(e.into()),
        }
    }
    Err(format!(
        "receiver timeout bytes={bytes} rejected={}",
        listener.stats().rejected
    )
    .into())
}

pub fn connections(&self) -> usize

Connections open or waiting for an answer.

pub fn connection_stats(&self, id: ConnId) -> Option<Stats>

Counters for an open connection; None for an unknown id or one still waiting for an answer.

Examples found in repository

examples/receive.rs (line 62)

fn main() -> Result<(), Box<dyn Error>> {
    let args: Vec<_> = std::env::args().skip(1).collect();
    let (Some(bind), Some(path), Some(expected)) = (args.first(), args.get(1), args.get(2)) else {
        return Err("receive <bind> <output> <expected bytes> [latency ms]".into());
    };
    let bind: SocketAddr = bind.parse()?;
    let expected: usize = expected.parse()?;
    let mut config = Config::default();
    if let Some(ms) = args.get(3) {
        config.latency = Duration::from_millis(ms.parse()?);
    }
    if let Ok(secret) = std::env::var("SAPFLOW_SRT_PASSPHRASE") {
        let size = match std::env::var("SAPFLOW_SRT_KEY_BITS").as_deref() {
            Ok("256") => KeySize::Aes256,
            Ok("128") | Err(_) => KeySize::Aes128,
            _ => return Err("key bits must be 128 or 256".into()),
        };
        config.crypto = Some(CryptoConfig::new(secret.into_bytes(), size)?);
        config.require_encryption = true;
    }
    let socket = UdpSocket::bind(bind)?;
    socket.set_read_timeout(Some(Duration::from_millis(5)))?;
    let mut output = File::create(path)?;
    let start = Instant::now();
    let mut listener = Listener::new(config, start);
    let mut packet = [0; 2048];
    let mut bytes = 0;
    let mut connection = None;
    println!("listening {}", socket.local_addr()?);
    std::io::stdout().flush()?;
    while start.elapsed() < Duration::from_secs(20) {
        let now = Instant::now();
        listener.handle_timeout(now);
        while let Some(event) = listener.poll_event() {
            match event {
                Event::Connect { id, stream_id, .. } => {
                    println!("stream_id={}", String::from_utf8_lossy(&stream_id));
                    if listener.accept(id, now) {
                        connection = Some(id);
                    }
                }
                Event::Data { payload, .. } => {
                    output.write_all(&payload)?;
                    bytes += payload.len();
                }
                Event::Closed { .. } => {}
            }
        }
        if bytes >= expected {
            let stats = connection.and_then(|id| listener.connection_stats(id));
            println!("received_bytes={bytes} stats={stats:?}");
            if let Some(id) = connection {
                listener.close(id, now);
            }
            while let Some((peer, b)) = listener.poll_transmit() {
                socket.send_to(&b, peer)?;
            }
            return Ok(());
        }
        while let Some((peer, b)) = listener.poll_transmit() {
            socket.send_to(&b, peer)?;
        }
        match socket.recv_from(&mut packet) {
            Ok((n, peer)) => {
                listener.handle_datagram(Instant::now(), peer, packet.get(..n).unwrap_or_default())
            }
            Err(e)
                if matches!(
                    e.kind(),
                    ErrorKind::TimedOut | ErrorKind::WouldBlock | ErrorKind::Interrupted
                ) => {}
            Err(e) => return Err(e.into()),
        }
    }
    Err(format!(
        "receiver timeout bytes={bytes} rejected={}",
        listener.stats().rejected
    )
    .into())
}

pub fn latency(&self, id: ConnId) -> Option<Duration>

The negotiated latency of an open connection.

pub fn poll_transmit(&mut self) -> Option<(SocketAddr, Vec<u8>)>

The next datagram to send and the address to send it to, if any.

Examples found in repository

examples/listen.rs (line 57)

fn main() -> std::io::Result<()> {
    let socket = UdpSocket::bind("0.0.0.0:9000")?;
    let mut listener = Listener::new(Config::default(), Instant::now());
    // An SRT datagram is at most 1,500 bytes on an ordinary network.
    let mut buf = [0u8; 2048];

    loop {
        // Wait for a datagram or until the listener next needs the time.
        let now = Instant::now();
        let wait = listener
            .poll_timeout()
            .map_or(Duration::from_millis(100), |at| {
                at.saturating_duration_since(now)
            })
            .max(Duration::from_millis(1));
        socket.set_read_timeout(Some(wait))?;
        match socket.recv_from(&mut buf) {
            Ok((n, peer)) => {
                if let Some(datagram) = buf.get(..n) {
                    listener.handle_datagram(Instant::now(), peer, datagram);
                }
            }
            Err(e) if matches!(e.kind(), ErrorKind::WouldBlock | ErrorKind::TimedOut) => {}
            Err(e) => return Err(e),
        }
        listener.handle_timeout(Instant::now());

        while let Some(event) = listener.poll_event() {
            match event {
                Event::Connect {
                    id,
                    peer,
                    stream_id,
                } => {
                    // The stream id is untrusted input: check it here, and
                    // call `listener.reject(id, code)` to refuse a caller.
                    println!("{peer} wants {:?}", String::from_utf8_lossy(&stream_id));
                    listener.accept(id, Instant::now());
                }
                Event::Data { id, payload } => println!("{id:?}: {} bytes", payload.len()),
                Event::Closed { id, reason, .. } => println!("{id:?} closed: {reason:?}"),
            }
        }
        while let Some((to, datagram)) = listener.poll_transmit() {
            socket.send_to(&datagram, to)?;
        }
    }
}

examples/receive.rs (line 67)

fn main() -> Result<(), Box<dyn Error>> {
    let args: Vec<_> = std::env::args().skip(1).collect();
    let (Some(bind), Some(path), Some(expected)) = (args.first(), args.get(1), args.get(2)) else {
        return Err("receive <bind> <output> <expected bytes> [latency ms]".into());
    };
    let bind: SocketAddr = bind.parse()?;
    let expected: usize = expected.parse()?;
    let mut config = Config::default();
    if let Some(ms) = args.get(3) {
        config.latency = Duration::from_millis(ms.parse()?);
    }
    if let Ok(secret) = std::env::var("SAPFLOW_SRT_PASSPHRASE") {
        let size = match std::env::var("SAPFLOW_SRT_KEY_BITS").as_deref() {
            Ok("256") => KeySize::Aes256,
            Ok("128") | Err(_) => KeySize::Aes128,
            _ => return Err("key bits must be 128 or 256".into()),
        };
        config.crypto = Some(CryptoConfig::new(secret.into_bytes(), size)?);
        config.require_encryption = true;
    }
    let socket = UdpSocket::bind(bind)?;
    socket.set_read_timeout(Some(Duration::from_millis(5)))?;
    let mut output = File::create(path)?;
    let start = Instant::now();
    let mut listener = Listener::new(config, start);
    let mut packet = [0; 2048];
    let mut bytes = 0;
    let mut connection = None;
    println!("listening {}", socket.local_addr()?);
    std::io::stdout().flush()?;
    while start.elapsed() < Duration::from_secs(20) {
        let now = Instant::now();
        listener.handle_timeout(now);
        while let Some(event) = listener.poll_event() {
            match event {
                Event::Connect { id, stream_id, .. } => {
                    println!("stream_id={}", String::from_utf8_lossy(&stream_id));
                    if listener.accept(id, now) {
                        connection = Some(id);
                    }
                }
                Event::Data { payload, .. } => {
                    output.write_all(&payload)?;
                    bytes += payload.len();
                }
                Event::Closed { .. } => {}
            }
        }
        if bytes >= expected {
            let stats = connection.and_then(|id| listener.connection_stats(id));
            println!("received_bytes={bytes} stats={stats:?}");
            if let Some(id) = connection {
                listener.close(id, now);
            }
            while let Some((peer, b)) = listener.poll_transmit() {
                socket.send_to(&b, peer)?;
            }
            return Ok(());
        }
        while let Some((peer, b)) = listener.poll_transmit() {
            socket.send_to(&b, peer)?;
        }
        match socket.recv_from(&mut packet) {
            Ok((n, peer)) => {
                listener.handle_datagram(Instant::now(), peer, packet.get(..n).unwrap_or_default())
            }
            Err(e)
                if matches!(
                    e.kind(),
                    ErrorKind::TimedOut | ErrorKind::WouldBlock | ErrorKind::Interrupted
                ) => {}
            Err(e) => return Err(e.into()),
        }
    }
    Err(format!(
        "receiver timeout bytes={bytes} rejected={}",
        listener.stats().rejected
    )
    .into())
}

pub fn poll_event(&mut self) -> Option<Event>

The next Event, if any.

Examples found in repository

examples/listen.rs (line 41)

fn main() -> std::io::Result<()> {
    let socket = UdpSocket::bind("0.0.0.0:9000")?;
    let mut listener = Listener::new(Config::default(), Instant::now());
    // An SRT datagram is at most 1,500 bytes on an ordinary network.
    let mut buf = [0u8; 2048];

    loop {
        // Wait for a datagram or until the listener next needs the time.
        let now = Instant::now();
        let wait = listener
            .poll_timeout()
            .map_or(Duration::from_millis(100), |at| {
                at.saturating_duration_since(now)
            })
            .max(Duration::from_millis(1));
        socket.set_read_timeout(Some(wait))?;
        match socket.recv_from(&mut buf) {
            Ok((n, peer)) => {
                if let Some(datagram) = buf.get(..n) {
                    listener.handle_datagram(Instant::now(), peer, datagram);
                }
            }
            Err(e) if matches!(e.kind(), ErrorKind::WouldBlock | ErrorKind::TimedOut) => {}
            Err(e) => return Err(e),
        }
        listener.handle_timeout(Instant::now());

        while let Some(event) = listener.poll_event() {
            match event {
                Event::Connect {
                    id,
                    peer,
                    stream_id,
                } => {
                    // The stream id is untrusted input: check it here, and
                    // call `listener.reject(id, code)` to refuse a caller.
                    println!("{peer} wants {:?}", String::from_utf8_lossy(&stream_id));
                    listener.accept(id, Instant::now());
                }
                Event::Data { id, payload } => println!("{id:?}: {} bytes", payload.len()),
                Event::Closed { id, reason, .. } => println!("{id:?} closed: {reason:?}"),
            }
        }
        while let Some((to, datagram)) = listener.poll_transmit() {
            socket.send_to(&datagram, to)?;
        }
    }
}

examples/receive.rs (line 46)

fn main() -> Result<(), Box<dyn Error>> {
    let args: Vec<_> = std::env::args().skip(1).collect();
    let (Some(bind), Some(path), Some(expected)) = (args.first(), args.get(1), args.get(2)) else {
        return Err("receive <bind> <output> <expected bytes> [latency ms]".into());
    };
    let bind: SocketAddr = bind.parse()?;
    let expected: usize = expected.parse()?;
    let mut config = Config::default();
    if let Some(ms) = args.get(3) {
        config.latency = Duration::from_millis(ms.parse()?);
    }
    if let Ok(secret) = std::env::var("SAPFLOW_SRT_PASSPHRASE") {
        let size = match std::env::var("SAPFLOW_SRT_KEY_BITS").as_deref() {
            Ok("256") => KeySize::Aes256,
            Ok("128") | Err(_) => KeySize::Aes128,
            _ => return Err("key bits must be 128 or 256".into()),
        };
        config.crypto = Some(CryptoConfig::new(secret.into_bytes(), size)?);
        config.require_encryption = true;
    }
    let socket = UdpSocket::bind(bind)?;
    socket.set_read_timeout(Some(Duration::from_millis(5)))?;
    let mut output = File::create(path)?;
    let start = Instant::now();
    let mut listener = Listener::new(config, start);
    let mut packet = [0; 2048];
    let mut bytes = 0;
    let mut connection = None;
    println!("listening {}", socket.local_addr()?);
    std::io::stdout().flush()?;
    while start.elapsed() < Duration::from_secs(20) {
        let now = Instant::now();
        listener.handle_timeout(now);
        while let Some(event) = listener.poll_event() {
            match event {
                Event::Connect { id, stream_id, .. } => {
                    println!("stream_id={}", String::from_utf8_lossy(&stream_id));
                    if listener.accept(id, now) {
                        connection = Some(id);
                    }
                }
                Event::Data { payload, .. } => {
                    output.write_all(&payload)?;
                    bytes += payload.len();
                }
                Event::Closed { .. } => {}
            }
        }
        if bytes >= expected {
            let stats = connection.and_then(|id| listener.connection_stats(id));
            println!("received_bytes={bytes} stats={stats:?}");
            if let Some(id) = connection {
                listener.close(id, now);
            }
            while let Some((peer, b)) = listener.poll_transmit() {
                socket.send_to(&b, peer)?;
            }
            return Ok(());
        }
        while let Some((peer, b)) = listener.poll_transmit() {
            socket.send_to(&b, peer)?;
        }
        match socket.recv_from(&mut packet) {
            Ok((n, peer)) => {
                listener.handle_datagram(Instant::now(), peer, packet.get(..n).unwrap_or_default())
            }
            Err(e)
                if matches!(
                    e.kind(),
                    ErrorKind::TimedOut | ErrorKind::WouldBlock | ErrorKind::Interrupted
                ) => {}
            Err(e) => return Err(e.into()),
        }
    }
    Err(format!(
        "receiver timeout bytes={bytes} rejected={}",
        listener.stats().rejected
    )
    .into())
}

pub fn poll_timeout(&self) -> Option<Instant>

When to call handle_timeout; None when there are no connections.

Examples found in repository

examples/listen.rs (line 24)

fn main() -> std::io::Result<()> {
    let socket = UdpSocket::bind("0.0.0.0:9000")?;
    let mut listener = Listener::new(Config::default(), Instant::now());
    // An SRT datagram is at most 1,500 bytes on an ordinary network.
    let mut buf = [0u8; 2048];

    loop {
        // Wait for a datagram or until the listener next needs the time.
        let now = Instant::now();
        let wait = listener
            .poll_timeout()
            .map_or(Duration::from_millis(100), |at| {
                at.saturating_duration_since(now)
            })
            .max(Duration::from_millis(1));
        socket.set_read_timeout(Some(wait))?;
        match socket.recv_from(&mut buf) {
            Ok((n, peer)) => {
                if let Some(datagram) = buf.get(..n) {
                    listener.handle_datagram(Instant::now(), peer, datagram);
                }
            }
            Err(e) if matches!(e.kind(), ErrorKind::WouldBlock | ErrorKind::TimedOut) => {}
            Err(e) => return Err(e),
        }
        listener.handle_timeout(Instant::now());

        while let Some(event) = listener.poll_event() {
            match event {
                Event::Connect {
                    id,
                    peer,
                    stream_id,
                } => {
                    // The stream id is untrusted input: check it here, and
                    // call `listener.reject(id, code)` to refuse a caller.
                    println!("{peer} wants {:?}", String::from_utf8_lossy(&stream_id));
                    listener.accept(id, Instant::now());
                }
                Event::Data { id, payload } => println!("{id:?}: {} bytes", payload.len()),
                Event::Closed { id, reason, .. } => println!("{id:?} closed: {reason:?}"),
            }
        }
        while let Some((to, datagram)) = listener.poll_transmit() {
            socket.send_to(&datagram, to)?;
        }
    }
}

pub fn handle_datagram( &mut self, now: Instant, peer: SocketAddr, datagram: &[u8], )

Takes a datagram that arrived from peer at now; bad ones are counted in stats and dropped.

Examples found in repository

examples/listen.rs (line 33)

fn main() -> std::io::Result<()> {
    let socket = UdpSocket::bind("0.0.0.0:9000")?;
    let mut listener = Listener::new(Config::default(), Instant::now());
    // An SRT datagram is at most 1,500 bytes on an ordinary network.
    let mut buf = [0u8; 2048];

    loop {
        // Wait for a datagram or until the listener next needs the time.
        let now = Instant::now();
        let wait = listener
            .poll_timeout()
            .map_or(Duration::from_millis(100), |at| {
                at.saturating_duration_since(now)
            })
            .max(Duration::from_millis(1));
        socket.set_read_timeout(Some(wait))?;
        match socket.recv_from(&mut buf) {
            Ok((n, peer)) => {
                if let Some(datagram) = buf.get(..n) {
                    listener.handle_datagram(Instant::now(), peer, datagram);
                }
            }
            Err(e) if matches!(e.kind(), ErrorKind::WouldBlock | ErrorKind::TimedOut) => {}
            Err(e) => return Err(e),
        }
        listener.handle_timeout(Instant::now());

        while let Some(event) = listener.poll_event() {
            match event {
                Event::Connect {
                    id,
                    peer,
                    stream_id,
                } => {
                    // The stream id is untrusted input: check it here, and
                    // call `listener.reject(id, code)` to refuse a caller.
                    println!("{peer} wants {:?}", String::from_utf8_lossy(&stream_id));
                    listener.accept(id, Instant::now());
                }
                Event::Data { id, payload } => println!("{id:?}: {} bytes", payload.len()),
                Event::Closed { id, reason, .. } => println!("{id:?} closed: {reason:?}"),
            }
        }
        while let Some((to, datagram)) = listener.poll_transmit() {
            socket.send_to(&datagram, to)?;
        }
    }
}

examples/receive.rs (line 77)

fn main() -> Result<(), Box<dyn Error>> {
    let args: Vec<_> = std::env::args().skip(1).collect();
    let (Some(bind), Some(path), Some(expected)) = (args.first(), args.get(1), args.get(2)) else {
        return Err("receive <bind> <output> <expected bytes> [latency ms]".into());
    };
    let bind: SocketAddr = bind.parse()?;
    let expected: usize = expected.parse()?;
    let mut config = Config::default();
    if let Some(ms) = args.get(3) {
        config.latency = Duration::from_millis(ms.parse()?);
    }
    if let Ok(secret) = std::env::var("SAPFLOW_SRT_PASSPHRASE") {
        let size = match std::env::var("SAPFLOW_SRT_KEY_BITS").as_deref() {
            Ok("256") => KeySize::Aes256,
            Ok("128") | Err(_) => KeySize::Aes128,
            _ => return Err("key bits must be 128 or 256".into()),
        };
        config.crypto = Some(CryptoConfig::new(secret.into_bytes(), size)?);
        config.require_encryption = true;
    }
    let socket = UdpSocket::bind(bind)?;
    socket.set_read_timeout(Some(Duration::from_millis(5)))?;
    let mut output = File::create(path)?;
    let start = Instant::now();
    let mut listener = Listener::new(config, start);
    let mut packet = [0; 2048];
    let mut bytes = 0;
    let mut connection = None;
    println!("listening {}", socket.local_addr()?);
    std::io::stdout().flush()?;
    while start.elapsed() < Duration::from_secs(20) {
        let now = Instant::now();
        listener.handle_timeout(now);
        while let Some(event) = listener.poll_event() {
            match event {
                Event::Connect { id, stream_id, .. } => {
                    println!("stream_id={}", String::from_utf8_lossy(&stream_id));
                    if listener.accept(id, now) {
                        connection = Some(id);
                    }
                }
                Event::Data { payload, .. } => {
                    output.write_all(&payload)?;
                    bytes += payload.len();
                }
                Event::Closed { .. } => {}
            }
        }
        if bytes >= expected {
            let stats = connection.and_then(|id| listener.connection_stats(id));
            println!("received_bytes={bytes} stats={stats:?}");
            if let Some(id) = connection {
                listener.close(id, now);
            }
            while let Some((peer, b)) = listener.poll_transmit() {
                socket.send_to(&b, peer)?;
            }
            return Ok(());
        }
        while let Some((peer, b)) = listener.poll_transmit() {
            socket.send_to(&b, peer)?;
        }
        match socket.recv_from(&mut packet) {
            Ok((n, peer)) => {
                listener.handle_datagram(Instant::now(), peer, packet.get(..n).unwrap_or_default())
            }
            Err(e)
                if matches!(
                    e.kind(),
                    ErrorKind::TimedOut | ErrorKind::WouldBlock | ErrorKind::Interrupted
                ) => {}
            Err(e) => return Err(e.into()),
        }
    }
    Err(format!(
        "receiver timeout bytes={bytes} rejected={}",
        listener.stats().rejected
    )
    .into())
}

pub fn handle_timeout(&mut self, now: Instant)

Does what is due at now: closes callers the application has not answered in time and runs each open connection’s timers.

Examples found in repository

examples/listen.rs (line 39)

fn main() -> std::io::Result<()> {
    let socket = UdpSocket::bind("0.0.0.0:9000")?;
    let mut listener = Listener::new(Config::default(), Instant::now());
    // An SRT datagram is at most 1,500 bytes on an ordinary network.
    let mut buf = [0u8; 2048];

    loop {
        // Wait for a datagram or until the listener next needs the time.
        let now = Instant::now();
        let wait = listener
            .poll_timeout()
            .map_or(Duration::from_millis(100), |at| {
                at.saturating_duration_since(now)
            })
            .max(Duration::from_millis(1));
        socket.set_read_timeout(Some(wait))?;
        match socket.recv_from(&mut buf) {
            Ok((n, peer)) => {
                if let Some(datagram) = buf.get(..n) {
                    listener.handle_datagram(Instant::now(), peer, datagram);
                }
            }
            Err(e) if matches!(e.kind(), ErrorKind::WouldBlock | ErrorKind::TimedOut) => {}
            Err(e) => return Err(e),
        }
        listener.handle_timeout(Instant::now());

        while let Some(event) = listener.poll_event() {
            match event {
                Event::Connect {
                    id,
                    peer,
                    stream_id,
                } => {
                    // The stream id is untrusted input: check it here, and
                    // call `listener.reject(id, code)` to refuse a caller.
                    println!("{peer} wants {:?}", String::from_utf8_lossy(&stream_id));
                    listener.accept(id, Instant::now());
                }
                Event::Data { id, payload } => println!("{id:?}: {} bytes", payload.len()),
                Event::Closed { id, reason, .. } => println!("{id:?} closed: {reason:?}"),
            }
        }
        while let Some((to, datagram)) = listener.poll_transmit() {
            socket.send_to(&datagram, to)?;
        }
    }
}

examples/receive.rs (line 45)

fn main() -> Result<(), Box<dyn Error>> {
    let args: Vec<_> = std::env::args().skip(1).collect();
    let (Some(bind), Some(path), Some(expected)) = (args.first(), args.get(1), args.get(2)) else {
        return Err("receive <bind> <output> <expected bytes> [latency ms]".into());
    };
    let bind: SocketAddr = bind.parse()?;
    let expected: usize = expected.parse()?;
    let mut config = Config::default();
    if let Some(ms) = args.get(3) {
        config.latency = Duration::from_millis(ms.parse()?);
    }
    if let Ok(secret) = std::env::var("SAPFLOW_SRT_PASSPHRASE") {
        let size = match std::env::var("SAPFLOW_SRT_KEY_BITS").as_deref() {
            Ok("256") => KeySize::Aes256,
            Ok("128") | Err(_) => KeySize::Aes128,
            _ => return Err("key bits must be 128 or 256".into()),
        };
        config.crypto = Some(CryptoConfig::new(secret.into_bytes(), size)?);
        config.require_encryption = true;
    }
    let socket = UdpSocket::bind(bind)?;
    socket.set_read_timeout(Some(Duration::from_millis(5)))?;
    let mut output = File::create(path)?;
    let start = Instant::now();
    let mut listener = Listener::new(config, start);
    let mut packet = [0; 2048];
    let mut bytes = 0;
    let mut connection = None;
    println!("listening {}", socket.local_addr()?);
    std::io::stdout().flush()?;
    while start.elapsed() < Duration::from_secs(20) {
        let now = Instant::now();
        listener.handle_timeout(now);
        while let Some(event) = listener.poll_event() {
            match event {
                Event::Connect { id, stream_id, .. } => {
                    println!("stream_id={}", String::from_utf8_lossy(&stream_id));
                    if listener.accept(id, now) {
                        connection = Some(id);
                    }
                }
                Event::Data { payload, .. } => {
                    output.write_all(&payload)?;
                    bytes += payload.len();
                }
                Event::Closed { .. } => {}
            }
        }
        if bytes >= expected {
            let stats = connection.and_then(|id| listener.connection_stats(id));
            println!("received_bytes={bytes} stats={stats:?}");
            if let Some(id) = connection {
                listener.close(id, now);
            }
            while let Some((peer, b)) = listener.poll_transmit() {
                socket.send_to(&b, peer)?;
            }
            return Ok(());
        }
        while let Some((peer, b)) = listener.poll_transmit() {
            socket.send_to(&b, peer)?;
        }
        match socket.recv_from(&mut packet) {
            Ok((n, peer)) => {
                listener.handle_datagram(Instant::now(), peer, packet.get(..n).unwrap_or_default())
            }
            Err(e)
                if matches!(
                    e.kind(),
                    ErrorKind::TimedOut | ErrorKind::WouldBlock | ErrorKind::Interrupted
                ) => {}
            Err(e) => return Err(e.into()),
        }
    }
    Err(format!(
        "receiver timeout bytes={bytes} rejected={}",
        listener.stats().rejected
    )
    .into())
}

pub fn accept(&mut self, id: ConnId, now: Instant) -> bool

Accepts a caller announced by Event::Connect: the handshake response goes out and data is delivered from now on.

Examples found in repository

examples/listen.rs (line 51)

fn main() -> std::io::Result<()> {
    let socket = UdpSocket::bind("0.0.0.0:9000")?;
    let mut listener = Listener::new(Config::default(), Instant::now());
    // An SRT datagram is at most 1,500 bytes on an ordinary network.
    let mut buf = [0u8; 2048];

    loop {
        // Wait for a datagram or until the listener next needs the time.
        let now = Instant::now();
        let wait = listener
            .poll_timeout()
            .map_or(Duration::from_millis(100), |at| {
                at.saturating_duration_since(now)
            })
            .max(Duration::from_millis(1));
        socket.set_read_timeout(Some(wait))?;
        match socket.recv_from(&mut buf) {
            Ok((n, peer)) => {
                if let Some(datagram) = buf.get(..n) {
                    listener.handle_datagram(Instant::now(), peer, datagram);
                }
            }
            Err(e) if matches!(e.kind(), ErrorKind::WouldBlock | ErrorKind::TimedOut) => {}
            Err(e) => return Err(e),
        }
        listener.handle_timeout(Instant::now());

        while let Some(event) = listener.poll_event() {
            match event {
                Event::Connect {
                    id,
                    peer,
                    stream_id,
                } => {
                    // The stream id is untrusted input: check it here, and
                    // call `listener.reject(id, code)` to refuse a caller.
                    println!("{peer} wants {:?}", String::from_utf8_lossy(&stream_id));
                    listener.accept(id, Instant::now());
                }
                Event::Data { id, payload } => println!("{id:?}: {} bytes", payload.len()),
                Event::Closed { id, reason, .. } => println!("{id:?} closed: {reason:?}"),
            }
        }
        while let Some((to, datagram)) = listener.poll_transmit() {
            socket.send_to(&datagram, to)?;
        }
    }
}

examples/receive.rs (line 50)

fn main() -> Result<(), Box<dyn Error>> {
    let args: Vec<_> = std::env::args().skip(1).collect();
    let (Some(bind), Some(path), Some(expected)) = (args.first(), args.get(1), args.get(2)) else {
        return Err("receive <bind> <output> <expected bytes> [latency ms]".into());
    };
    let bind: SocketAddr = bind.parse()?;
    let expected: usize = expected.parse()?;
    let mut config = Config::default();
    if let Some(ms) = args.get(3) {
        config.latency = Duration::from_millis(ms.parse()?);
    }
    if let Ok(secret) = std::env::var("SAPFLOW_SRT_PASSPHRASE") {
        let size = match std::env::var("SAPFLOW_SRT_KEY_BITS").as_deref() {
            Ok("256") => KeySize::Aes256,
            Ok("128") | Err(_) => KeySize::Aes128,
            _ => return Err("key bits must be 128 or 256".into()),
        };
        config.crypto = Some(CryptoConfig::new(secret.into_bytes(), size)?);
        config.require_encryption = true;
    }
    let socket = UdpSocket::bind(bind)?;
    socket.set_read_timeout(Some(Duration::from_millis(5)))?;
    let mut output = File::create(path)?;
    let start = Instant::now();
    let mut listener = Listener::new(config, start);
    let mut packet = [0; 2048];
    let mut bytes = 0;
    let mut connection = None;
    println!("listening {}", socket.local_addr()?);
    std::io::stdout().flush()?;
    while start.elapsed() < Duration::from_secs(20) {
        let now = Instant::now();
        listener.handle_timeout(now);
        while let Some(event) = listener.poll_event() {
            match event {
                Event::Connect { id, stream_id, .. } => {
                    println!("stream_id={}", String::from_utf8_lossy(&stream_id));
                    if listener.accept(id, now) {
                        connection = Some(id);
                    }
                }
                Event::Data { payload, .. } => {
                    output.write_all(&payload)?;
                    bytes += payload.len();
                }
                Event::Closed { .. } => {}
            }
        }
        if bytes >= expected {
            let stats = connection.and_then(|id| listener.connection_stats(id));
            println!("received_bytes={bytes} stats={stats:?}");
            if let Some(id) = connection {
                listener.close(id, now);
            }
            while let Some((peer, b)) = listener.poll_transmit() {
                socket.send_to(&b, peer)?;
            }
            return Ok(());
        }
        while let Some((peer, b)) = listener.poll_transmit() {
            socket.send_to(&b, peer)?;
        }
        match socket.recv_from(&mut packet) {
            Ok((n, peer)) => {
                listener.handle_datagram(Instant::now(), peer, packet.get(..n).unwrap_or_default())
            }
            Err(e)
                if matches!(
                    e.kind(),
                    ErrorKind::TimedOut | ErrorKind::WouldBlock | ErrorKind::Interrupted
                ) => {}
            Err(e) => return Err(e.into()),
        }
    }
    Err(format!(
        "receiver timeout bytes={bytes} rejected={}",
        listener.stats().rejected
    )
    .into())
}

pub fn accept_with_crypto( &mut self, id: ConnId, now: Instant, crypto: Option<CryptoConfig>, ) -> bool

Select an authenticated admission/session secret after inspecting the bounded stream ID. A configured secret requires matching KM unless allow_plaintext admits a caller that sent none; a missing secret never downgrades an encrypted or encryption-required session. Failed key validation rejects and removes the pending connection.

pub fn reject(&mut self, id: ConnId, code: RejectCode) -> bool

Refuses a caller announced by Event::Connect.

pub fn close(&mut self, id: ConnId, now: Instant)

Ends an open connection from our side (a SHUTDOWN goes to the caller), or forgets one waiting for an answer.

Examples found in repository

examples/receive.rs (line 65)

fn main() -> Result<(), Box<dyn Error>> {
    let args: Vec<_> = std::env::args().skip(1).collect();
    let (Some(bind), Some(path), Some(expected)) = (args.first(), args.get(1), args.get(2)) else {
        return Err("receive <bind> <output> <expected bytes> [latency ms]".into());
    };
    let bind: SocketAddr = bind.parse()?;
    let expected: usize = expected.parse()?;
    let mut config = Config::default();
    if let Some(ms) = args.get(3) {
        config.latency = Duration::from_millis(ms.parse()?);
    }
    if let Ok(secret) = std::env::var("SAPFLOW_SRT_PASSPHRASE") {
        let size = match std::env::var("SAPFLOW_SRT_KEY_BITS").as_deref() {
            Ok("256") => KeySize::Aes256,
            Ok("128") | Err(_) => KeySize::Aes128,
            _ => return Err("key bits must be 128 or 256".into()),
        };
        config.crypto = Some(CryptoConfig::new(secret.into_bytes(), size)?);
        config.require_encryption = true;
    }
    let socket = UdpSocket::bind(bind)?;
    socket.set_read_timeout(Some(Duration::from_millis(5)))?;
    let mut output = File::create(path)?;
    let start = Instant::now();
    let mut listener = Listener::new(config, start);
    let mut packet = [0; 2048];
    let mut bytes = 0;
    let mut connection = None;
    println!("listening {}", socket.local_addr()?);
    std::io::stdout().flush()?;
    while start.elapsed() < Duration::from_secs(20) {
        let now = Instant::now();
        listener.handle_timeout(now);
        while let Some(event) = listener.poll_event() {
            match event {
                Event::Connect { id, stream_id, .. } => {
                    println!("stream_id={}", String::from_utf8_lossy(&stream_id));
                    if listener.accept(id, now) {
                        connection = Some(id);
                    }
                }
                Event::Data { payload, .. } => {
                    output.write_all(&payload)?;
                    bytes += payload.len();
                }
                Event::Closed { .. } => {}
            }
        }
        if bytes >= expected {
            let stats = connection.and_then(|id| listener.connection_stats(id));
            println!("received_bytes={bytes} stats={stats:?}");
            if let Some(id) = connection {
                listener.close(id, now);
            }
            while let Some((peer, b)) = listener.poll_transmit() {
                socket.send_to(&b, peer)?;
            }
            return Ok(());
        }
        while let Some((peer, b)) = listener.poll_transmit() {
            socket.send_to(&b, peer)?;
        }
        match socket.recv_from(&mut packet) {
            Ok((n, peer)) => {
                listener.handle_datagram(Instant::now(), peer, packet.get(..n).unwrap_or_default())
            }
            Err(e)
                if matches!(
                    e.kind(),
                    ErrorKind::TimedOut | ErrorKind::WouldBlock | ErrorKind::Interrupted
                ) => {}
            Err(e) => return Err(e.into()),
        }
    }
    Err(format!(
        "receiver timeout bytes={bytes} rejected={}",
        listener.stats().rejected
    )
    .into())
}