如何在不使用tokio_proto板条箱的情况下从tokio TCP连接读取? [英] How can I read from a tokio TCP connection without using the tokio_proto crate?

问题描述

我正在尝试编写一个TCP客户端以打印传入消息.我想出了以下代码:

I'm trying to write a TCP client to print incoming messages. I came up with the following code:

extern crate bytes;
extern crate futures;
extern crate tokio_core;
extern crate tokio_io;

use futures::Future;
use tokio_core::net::TcpStream;
use tokio_core::reactor::Core;
use tokio_io::AsyncRead;
use bytes::BytesMut;

fn main() {
    let mut core = Core::new().unwrap();
    let handle = core.handle();

    let connection = TcpStream::connect(&"127.0.0.1:8081".parse().unwrap(), &handle);

    let server = connection.and_then(move |mut stream| {
        let mut buf = BytesMut::with_capacity(1000);
        stream
            .read_buf(&mut buf)
            .map(|buf| print!("Buffer {:?}", buf))
            .map_err(|e| eprintln!("Error: {}", e));
        Ok(())
    });

    core.run(server).unwrap();
}

它可以编译，但是失败，并显示Buffer NotReady错误.

It compiles but it fails with a Buffer NotReady error.

推荐答案

Rust是一种编译语言，这意味着您应注意编译器生成的警告:

Rust is a compiled language, which means that you should pay attention to the warnings that the compiler generates:

warning: unused `std::result::Result` which must be used
  --> src/main.rs:20:9
   |
20 | /         stream
21 | |             .read_buf(&mut buf)
22 | |             .map(|buf| print!("Buffer {:?}", buf))
23 | |             .map_err(|e| eprintln!("Error: {}", e));
   | |____________________________________________________^
   |
   = note: #[warn(unused_must_use)] on by default

此外， tokio整整一章专门介绍低级别IO ，我认为您不会对您已经了解的细节感到厌烦.

Additionally, tokio has an entire chapter dedicated to low-level IO which I'll assume you've read to not bore you with details you already know.

首先，我们将connection Future转换为Stream.一个流可以产生多个值-在这种情况下，每次成功读取我们都会返回一个值.我们为此最简单的实现创建AsWeGetIt.

First we take the connection Future and convert it into a Stream. A stream can yield multiple values — in this case we return one value for every successful read. We create AsWeGetIt for the simplest implementation of this.

然后我们使用Stream::for_each打印出流的每个值.方便地，这将执行相应的转换回Future，这是and_then所需要的.

We then print out each value of the stream using Stream::for_each. Conveniently, this performs the corresponding conversion back to a Future, which is what is needed for and_then.

extern crate bytes;
extern crate futures;
extern crate tokio_core;
extern crate tokio_io;

use futures::{Future, Poll, Stream};
use tokio_core::net::TcpStream;
use tokio_core::reactor::Core;
use tokio_io::AsyncRead;
use bytes::BytesMut;

struct AsWeGetIt<R>(R);

impl<R> Stream for AsWeGetIt<R>
where
    R: AsyncRead,
{
    type Item = BytesMut;
    type Error = std::io::Error;

    fn poll(&mut self) -> Poll<Option<Self::Item>, Self::Error> {
        let mut buf = BytesMut::with_capacity(1000);

        self.0
            .read_buf(&mut buf)
            .map(|async| async.map(|_| Some(buf)))
    }
}

fn main() {
    let mut core = Core::new().unwrap();
    let handle = core.handle();

    let address = "127.0.0.1:8081".parse().expect("Unable to parse address");
    let connection = TcpStream::connect(&address, &handle);

    let client = connection
        .and_then(|tcp_stream| {
            AsWeGetIt(tcp_stream).for_each(|buf| {
                println!("Buffer {:?}", buf);
                Ok(())
            })
        })
        .map_err(|e| eprintln!("Error: {}", e));

    core.run(client).expect("Unable to run the event loop");
}

这篇关于如何在不使用tokio_proto板条箱的情况下从tokio TCP连接读取?的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！