Tokio tcpstream not receiving

Hi,
I'm trying to write a tokio client for a server using protobuf and I have two implementations, the first one with tokio sends a message which is received, a response is sent by the server but the stream.poll_read is always Async::NotReady

pub fn connect(addr: String) {
    let socket_addr = addr.parse().unwrap();
    let client = TcpStream::connect(&socket_addr).and_then(|mut stream| {
        let request_init = protos::init::RequestInit::new();

        let mut os = protobuf::CodedOutputStream::new(&mut stream);
        os.write_raw_bytes(&build_header(1, 1));
        os.write_message_no_tag(&request_init).unwrap();
        os.flush().unwrap();

        let mut buffer = [0; 10];

        loop {
            match stream.poll_read(&mut buffer) {
                Ok(Async::Ready(n)) => {
                    println!("{} bytes read, buffer is : {}", n, String::from_utf8_lossy(&buffer));
                    break;
                },
                Ok(Async::NotReady) => {
                    ()
                },
                Err(err) => {
                    println!("error : {:?}", err);
                    break;
                }
            }
        }

        let mut is = protobuf::CodedInputStream::new(&mut stream);
        let answer = is.read_message::<protos::init::AcceptInit>();

        match answer {
            Ok(_) => println!("init accepted"),
            Err(err) => println!("error : {:?}", err)
        };

        Ok(())
    }).map_err(|err| {
        println!("connection error = {:?}", err);
    });

    tokio::run(client);
}

I then tried to implement it with the std TcpStream, and I can read the response and everything is okay

pub fn connect_std(addr: String) {
    let mut stream = std::net::TcpStream::connect(&addr).unwrap();

    let request_init = protos::init::RequestInit::new();

    let mut os = protobuf::CodedOutputStream::new(&mut stream);
    os.write_raw_bytes(&build_header(1, 1));
    os.write_message_no_tag(&request_init).unwrap();
    os.flush().unwrap();

    let mut is = protobuf::CodedInputStream::new(&mut stream);
    let header = is.read_raw_bytes(4);
    let body = match header {
        Ok(bytes) => {
            match bytes[0] {
                2 => {
                    let mut length_bytes = [0; 4];
                    length_bytes[0..3].copy_from_slice(&bytes[1..]);
                    let length = u32::from_le_bytes(length_bytes);
                    println!("length is : {}", length);
                    is.read_message::<protos::init::AcceptInit>()
                }
                _ => {
                    println!("error : type is {}", bytes[0]);
                    Err(protobuf::ProtobufError::MessageNotInitialized { message: "not a valid accept message" })
                }
            }
        },
        Err(e) => Err(e)
    };

    match body {
        Ok(_) => println!("init accepted"),
        Err(err) => {
            println!("error : {:?}", err);
        }
    };
}

Anybody have a idea why ?

This topic was automatically closed 90 days after the last reply. New replies are no longer allowed.