rust - 我如何测试绑定(bind)到 tokio TcpStream 的 future ？

Question

我有一个使用 LinesCodec 将 TCP 流包装在 Framed 中的 future 。

当我尝试将其包装在测试中时，大约有 20% 的时间我会遇到 future 阻塞，但是因为我没有在我尝试连接的套接字上监听任何东西，所以我预计总是会出现错误:

thread 'tokio-runtime-worker-0' panicked at 'error: Os { code: 111, kind: ConnectionRefused, message: "Connection refused" }', src/lib.rs:35:24 note: Run with 'RUST_BACKTRACE=1' for a backtrace.

这是我用过的测试代码:

#[macro_use(try_ready)]
extern crate futures; // 0.1.24
extern crate tokio;   // 0.1.8

use std::io;
use std::net::SocketAddr;
use tokio::codec::{Framed, LinesCodec};
use tokio::net::TcpStream;
use tokio::prelude::*;

struct MyFuture {
    addr: SocketAddr,
}

impl Future for MyFuture {
    type Item = Framed<TcpStream, LinesCodec>;
    type Error = io::Error;
    fn poll(&mut self) -> Result<Async<Framed<TcpStream, LinesCodec>>, io::Error> {
        let strm = try_ready!(TcpStream::connect(&self.addr).poll());
        Ok(Async::Ready(Framed::new(strm, LinesCodec::new())))
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::net::Shutdown;

    #[test]
    fn connect() {
        let addr: SocketAddr = "127.0.0.1:4222".parse().unwrap();
        let fut = MyFuture { addr: addr }
            .and_then(|f| {
                println!("connected");
                let cn = f.get_ref();
                cn.shutdown(Shutdown::Both)
            }).map_err(|e| panic!("error: {:?}", e));
        tokio::run(fut)
    }
}

playground

我在其他语言中看到过一些模式，其中测试二进制文件本身提供了一种异步返回结果的机制，但还没有找到在 Rust 中使用类似机制的好方法。

Answer 1

测试异步代码的一种简单方法可能是为每个测试使用专用的运行时:启动它，等待将来完成并在测试结束时关闭运行时。

#[test]
fn my_case() {

    // setup future f
    // ...

    tokio::run(f);
}

我不知道 Rust 生态系统中是否已经有统一的模式；见this discussion关于对基于 future 的代码的测试支持的演变。

为什么你的代码没有按预期工作

当您调用 poll() 时，将查询 future 以检查值是否可用。

如果某个值不可用，则会注册一个兴趣，以便在发生可以解决 future 问题的事情时再次调用 poll()。

当你的 MyFuture::poll() 被调用时:

1. TcpStream::connect 创建一个新的 future TcpStreamNew
2. TcpStreamNew::poll 在第 1 步的 future 创建时仅一次立即调用。
3. future 超出范围，所以下次调用 MyFuture::poll 时，您永远不会解决之前创建的 future 。

您已经注册了对 future 的兴趣，如果您第一次投票时未解决，您将永远不会再次询问(投票)已解决的值或错误。

“非确定性”行为的原因是因为第一个 poll 有时会立即解决并出现 ConnectionRefused 错误，有时它会永远等待 future 的连接事件或失败它永远不会被检索。

看看 Tokio 使用的 mio::sys::unix::tcp::TcpStream:

 impl TcpStream {
     pub fn connect(stream: net::TcpStream, addr: &SocketAddr) -> io::Result<TcpStream> {
         set_nonblock(stream.as_raw_fd())?;

         match stream.connect(addr) {
             Ok(..) => {}
             Err(ref e) if e.raw_os_error() == Some(libc::EINPROGRESS) => {}
             Err(e) => return Err(e),
         }

         Ok(TcpStream {
             inner: stream,
         })
     }

当你connect在非阻塞套接字上，系统调用可能会立即连接/失败或返回 EINPROGRESS，在最后一种情况下，必须触发轮询以检索错误值。