7

我有一个未来,它Framed使用LinesCodec.

当我尝试将其包装在测试中时,大约 20% 的时间会阻塞未来,但因为我没有在尝试连接的套接字上监听任何内容,所以我希望总是会收到错误消息:

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)
    }
}

操场

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

4

3 回答 3

5

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

#[test]
fn my_case() {

    // setup future f
    // ...

    tokio::run(f);
}

我不知道 Rust 生态系统中是否已经有统一的模式;请参阅有关对基于未来的代码的测试支持的演变的讨论。

为什么您的代码无法按预期工作

当您调用poll()时,将查询未来以检查值是否可用。

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

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

  1. TcpStream::connect创造新的未来TcpStreamNew
  2. TcpStreamNew::poll仅在步骤 1 中创建未来时立即调用一次
  3. 未来超出范围,因此下次调用时,MyFuture::poll您永远不会解析先前创建的未来。

您已经注册了对未来的兴趣,如果您第一次轮询它时没有解决,您永远不会再次询问(轮询)已解决的值或错误。

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

看看mio::sys::unix::tcp::TcpStreamTokio 使用的:

 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,在最后一种情况下,必须触发轮询以检索错误的值。

于 2018-09-25T09:18:55.157 回答
4

问题不在于测试,而在于实施。

这个基于你的工作测试用例没有自定义的未来实现,只有调用TcpStream::connect(). 它可以按您的预期工作。

extern crate futures;
extern crate tokio;

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

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

操场

您在poll()方法中一遍又一遍地连接到同一个端点。这不是未来的运作方式。该poll()方法将被重复调用,期望在某些时候它会返回Ok(Async::Ready(..))or Err(..)

如果每次poll()调用都发起一个新的 TCP 连接,就不太可能及时完成。

这是一个修改后的示例,可以满足您的期望

#[macro_use(try_ready)]
extern crate futures;
extern crate tokio;
use std::io;
use std::net::SocketAddr;
use tokio::codec::{Framed, LinesCodec};
use tokio::net::{ConnectFuture, TcpStream};
use tokio::prelude::*;

struct MyFuture {
    tcp: ConnectFuture,
}

impl MyFuture {
    fn new(addr: SocketAddr) -> MyFuture {
        MyFuture {
            tcp: TcpStream::connect(&addr),
        }
    }
}

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!(self.tcp.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::new(addr)
            .and_then(|f| {
                println!("connected");
                let cn = f.get_ref();
                cn.shutdown(Shutdown::Both)
            }).map_err(|e| panic!("error: {:?}", e));
        tokio::run(fut)
    }
}

不过,我不确定你打算做什么。如果这是正确的方法,我无法评论。

于 2018-09-22T13:06:49.257 回答
1

在某种程度上,您可以加入 tokio 的测试库以使这更容易;它支持单元测试中的异步/等待。

#[tokio::test]
async fn my_future_test() {
  let addr: SocketAddr = "127.0.0.1:4222".parse().unwrap();
  match MyFuture { addr }.poll().await {
    Ok(f) => assert!("something good")
    Err(e) => assert!("something bad")
  }
}

https://docs.rs/tokio/0.3.3/tokio/attr.test.html

于 2020-11-03T10:16:38.307 回答