如何使用 Tokio 的 TcpStream 发送数据流？

Question

我正在尝试在 Rust 中实现一个 TCP 客户端。我能够读取来自服务器的数据，但我无法发送数据。

这是我正在处理的代码：

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

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

#[derive(Default)]
pub struct TcpClient {}

struct AsWeGetIt<R>(R);

impl<R> Stream for AsWeGetIt<R>
where
    R: AsyncRead,
{
    type Item = BytesMut;
    type Error = 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)))
    }
}

impl TcpClient {
    pub fn new() -> Self {
        Self {}
    }

    pub fn connectToTcpServer(&mut self) -> bool {
        let mut core = Core::new().unwrap();
        let handle = core.handle();

        let address = "127.0.0.1:2323".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!("{:?}", buf);
                    Ok(())
                })
            })
            .map_err(|e| eprintln!("Error: {}", e));

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

如何添加异步数据发送功能？

Answer 1

如果你想在socket上有两个完全独立的数据流，你可以在当前版本的Tokio中使用TcpStream上的split()方法：

let connection = TcpStream::connect(&address);
connection.and_then(|socket| {
    let (rx, tx) = socket.split();
    //Independently use tx/rx for sending/receiving
    return Ok(());
});

拆分后，您可以独立使用rx（接收端）和tx（发送端）。这是一个将发送和接收视为完全独立的小示例。发送方只是定期发送相同的消息，而接收方只是打印所有传入的数据：

extern crate futures;
extern crate tokio;

use self::futures::{Future, Poll, Stream};
use self::tokio::net::TcpStream;
use tokio::io::{AsyncRead, AsyncWrite, Error, ReadHalf};
use tokio::prelude::*;
use tokio::timer::Interval;

//Receiver struct that implements the future trait
//this exclusively handles incomming data and prints it to stdout
struct Receiver {
    rx: ReadHalf<TcpStream>, //receiving half of the socket stream
}
impl Future for Receiver {
    type Item = ();
    type Error = Error;

    fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
        let mut buffer = vec![0u8; 1000]; //reserve 1000 bytes in the receive buffer
                                          //get all data that is available to us at the moment...
        while let Async::Ready(num_bytes_read) = self.rx.poll_read(&mut buffer)? {
            if num_bytes_read == 0 {
                return Ok(Async::Ready(()));
            } //socket closed
            print!("{}", String::from_utf8_lossy(&buffer[..num_bytes_read]));
        }
        return Ok(Async::NotReady);
    }
}

fn main() {
    let address = "127.0.0.1:2323".parse().expect("Unable to parse address");
    let connection = TcpStream::connect(&address);
    //wait for the connection to be established
    let client = connection
        .and_then(|socket| {
            //split the successfully connected socket in half (receive / send)
            let (rx, mut tx) = socket.split();

            //set up a simple sender, that periodically (1sec) sends the same message
            let sender = Interval::new_interval(std::time::Duration::from_millis(1000))
                .for_each(move |_| {
                    //this lambda is invoked once per passed second
                    tx.poll_write(&vec![82, 117, 115, 116, 10]).map_err(|_| {
                        //shut down the timer if an error occured (e.g. socket was closed)
                        tokio::timer::Error::shutdown()
                    })?;
                    return Ok(());
                }).map_err(|e| println!("{}", e));
            //start the sender
            tokio::spawn(sender);

            //start the receiver
            let receiver = Receiver { rx };
            tokio::spawn(receiver.map_err(|e| println!("{}", e)));

            return Ok(());
        }).map_err(|e| println!("{}", e));

    tokio::run(client);
}

对于某些应用程序来说，这就足够了。但是，通常您将在连接上定义协议/格式。例如，HTTP 连接总是由请求和响应组成，每个请求和响应都由标头和正文组成。 Tokio 不是直接在字节级别上工作，而是提供了适合您的特征 Encoder 和 Decoder 到一个套接字上，它解码您的协议并直接为您提供您想要使用的实体。例如，您可以查看非常基本的 HTTP implementation 或 line-based 编解码器。

当传入消息触发传出消息时，它会变得更加复杂。对于最简单的情况（每条传入消息都会导致一条传出消息），您可以查看this官方请求/响应示例。