Как я могу читать данные из TCP-соединения tokio без использования контейнера tokio_proto?

Я пытаюсь написать TCP-клиент для печати входящих сообщений. Я придумал следующий код:

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.


rust rust-tokio
person Sergey    schedule 19.10.2017    source источник
comment
Почему вы используете TcpStream::connect, если хотите запустить сервер?   -  person Shepmaster    schedule 21.10.2017
comment
@Shepmaster сейчас я пытаюсь что-нибудь прочитать. Так удобнее - я просто запускаю слушателя с помощью ncat. Я думаю, что сокет чтения должен быть одинаковым для сервера и клиента.   -  person Sergey    schedule 21.10.2017

Ответы (1)


arrow_upward
3
arrow_downward

Rust - это компилируемый язык, а это значит, что вам следует обратить внимание на предупреждения, которые генерирует компилятор:

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 есть целая глава, посвященная низкоуровневому вводу-выводу которую, я полагаю, вы читали, чтобы не утомлять вас деталями, которые вы уже знаете.

Сначала мы берем connection Future и преобразуем его в Stream. Поток может выдавать несколько значений - в этом случае мы возвращаем одно значение для каждого успешного чтения. Мы создаем AsWeGetIt для простейшей реализации этого.

Затем мы распечатываем каждое значение потока, используя Stream::for_each. Удобно, что при этом выполняется соответствующее преобразование обратно в Future, что и требуется для 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");
}
person Shepmaster    schedule 21.10.2017