Async UdpStream: request for API feedback & code review

code review
1

As part of a larger personal project, I needed (wanted?) a run-time agnostic, ergonomic wrapper around a non-blocking, non-exclusive, async UdpSocket.

I'd love your feedback to both the API and the code itself for UdpStream as defined here ssdp-rs/src/udp.rs · MusicalNinjaDad/splurt

[Edit: of course, within hours of posting I thought ... surely this should be using futures::Stream and futures::Sink with send((tuple, of, values)),
Edit2: now done that and updated the examples below ...]

The main API is

// construct by binding to a socket & specifying max expected
// length for incoming datagrams
let mut stream = UdpStream::<32>::bind(addr).expect("bound to socket");

// next to receive
let (msg: [u8; 32], len: usize, sent_by: SocketAddr) = receiver
            .next()
            .await
            .expect("a message")
            .expect("a valid message");

// sinks don't need a defined buffer size
let mut sink = UdpSink::bind(addr).expect("bound to socket");

// send a tuple of (buf, address)
let msg: &[u8; 17] = b"udp loopback test";
sink.push((msg, rec_addr)).await.expect("send msg");

A more complete example from the tests:

#[futures_net::test]
async fn stream_and_sink() {
    let loopback = Ipv4Addr::new(127, 0, 0, 1);

    // https://doc.rust-lang.org/std/net/struct.TcpListener.html#method.bind
    // Binding with a port number of 0 will request that the OS assigns a port to this listener.
    // The port allocated can be queried via the TcpListener::local_addr method.
    let addr: SocketAddr = SocketAddrV4::new(loopback, 0).into();

    let mut sender = UdpSink::bind(addr).expect("sender");
    let send_addr = sender.local_addr().expect("bound port");
    let msg: &[u8; 17] = b"udp loopback test";
    dbg!(send_addr);

    let mut receiver = UdpStream::<32>::bind(addr).expect("receiver");
    let rec_addr = receiver.local_addr().expect("bound port");
    let mut received: [u8; 17] = [b'\x00'; 17];
    // dummy address (google DNS) should be changed on reception of message from our sender
    let mut outer_sent_by = SocketAddr::from_str("8.8.8.8:80").expect("valid addr");
    dbg!(rec_addr);

    let send = async move {
        println!("sending {}", String::from_utf8_lossy(msg));
        sender.send((msg, &rec_addr)).await.expect("send msg");
    };

    let rec = async {
        println!("initiating receiver");
        let (msg, len, sent_by) = receiver
            .next()
            .await
            .expect("a message")
            .expect("a valid message");
        println!(
            "received: {} from {} ({} bytes)",
            String::from_utf8_lossy(&msg),
            sent_by,
            len
        );
        received = msg[..len].try_into().expect("17 bytes in msg");
        outer_sent_by = sent_by;
    };

    println!("ready to join");
    futures::join!(rec, send);

    assert_eq!(
        String::from_utf8_lossy(&received),
        String::from_utf8_lossy(msg)
    );

    assert_eq!(outer_sent_by, send_addr)
}

(and yes ... a real-world usage would handle the incoming information much more safely)