tokio :: Stream上的超时总是发生

Question

我正在尝试接受UDP消息，但是只有当它在5秒钟内发生时，我才能通过手动实现Stream和使用Futures库中的组合器来构建Stream抽象。每种方式，在recv_from将来解决之后，持续时间将到期并且流将返回Err(Elapsed(()))。这不是预期的行为，如果返回值，则不应返回任何错误。

[预期的行为是流将解决任一超时或Vec，但不会解决，而是在5秒钟后解决其他问题。

use futures::{pin_mut, ready, stream::unfold, FutureExt};
use tokio::{
    net::{udp, UdpSocket},
    stream::{Stream, StreamExt},
    time::{self, Duration},
};

use std::{
    io,
    net::SocketAddr,
    pin::Pin,
    task::{Context, Poll},
};

#[derive(Debug)]
pub(crate) struct UdpStream {
    stream: udp::RecvHalf,
}

impl UdpStream {
    fn new(stream: udp::RecvHalf) -> Self {
        Self { stream }
    }

    fn stream(self) -> impl Stream<Item = io::Result<(Vec<u8>, SocketAddr)>> {
        unfold(self.stream, |mut stream| async move {
            let mut buf = [0; 4096];
            match time::timeout(Duration::from_secs(5), stream.recv_from(&mut buf)).await {
                Ok(Ok((len, src))) => {
                    Some((Ok((buf.iter().take(len).cloned().collect(), src)), stream))
                }
                e => {
                    println!("{:?}", e);
                    None
                }
            }
        })
    }
}

impl Stream for UdpStream {
    type Item = io::Result<(Vec<u8>, SocketAddr)>;

    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
        let socket = &mut self.stream;
        pin_mut!(socket);
        let mut buf = [0u8; 4096];
        let (len, src) = ready!(Box::pin(socket.recv_from(&mut buf)).poll_unpin(cx))?;
        Poll::Ready(Some(Ok((buf.iter().take(len).cloned().collect(), src))))
    }
}

async fn listen_udp(addr: SocketAddr) -> io::Result<()> {
    let udp = UdpSocket::bind(addr).await?;
    let (mut udp_recv, mut udp_send) = udp.split();

    let mut msg_stream = Box::pin(UdpStream::new(udp_recv).stream());
    // use the manually implemented stream with this:
    // let mut msg_stream = UdpStream::new(udp_recv).timeout(Duration::from_secs(5));

    while let Some(msg) = msg_stream.next().await {
        match msg {
            Ok((buf, src)) => {
                udp_send.send_to(&buf, &src).await?;
                println!("Message recv: {:?}", buf);
            }
            Err(e) => {
                eprintln!("timed out: {:?}", e);
            }
        }
    }
    Ok(())
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error + Send + Sync + 'static>> {
    listen_udp("127.0.0.1:9953".parse()?).await?;
    Ok(())
}

您可以通过运行此代码并使用echo "foo" | nc 127.0.0.1 9953 -u或dig发出udp请求来尝试

cargo.toml

[package]
name = "udp_test"
version = "0.1.0"
edition = "2018"

# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html

[dependencies]
tokio = { version = "0.2", features = ["full"] }
futures = "0.3"

Answer 1

您的流返回第一个（也是唯一的）元素后，它返回以等待下一个；它直到超时才结束。