rust-tokio read-timeout

//! A chat server that broadcasts a message to all connections.
//!
//! This example is explicitly more verbose than it has to be. This is to
//! illustrate more concepts.
//!
//! A chat server for telnet clients. After a telnet client connects, the first
//! line should contain the client's name. After that, all lines send by a
//! client are broadcasted to all other connected clients.
//!
//! Because the client is telnet, lines are delimited by "\r\n".
//!
//! You can test this out by running:
//!
//!     cargo run --example chat
//!
//! And then in another terminal run:
//!
//!     telnet localhost 6142
//!
//! You can run the `telnet` command in any number of additional windows.
//!
//! You can run the second command in multiple windows and then chat between the
//! two, seeing the messages from the other client as they're received. For all
//! connected clients they'll all join the same room and see everyone else's
//! messages.

#![allow(warnings)]

extern crate tokio;

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

use tokio::io;
use tokio::net::{TcpListener, TcpStream};
use tokio::prelude::*;
use futures::sync::mpsc;
use futures::future::*;
use bytes::{BytesMut, Bytes, BufMut};

use std::collections::HashMap;
use std::net::SocketAddr;
use std::sync::{Arc, Mutex};
use std::time::Duration;


pub fn main() {
    let addr = "127.0.0.1:6142".parse().unwrap();
    let mut core = tokio_core::reactor::Core::new().unwrap();
    let handle = core.handle();

    // Bind a TCP listener to the socket address.
    //
    // Note that this is the Tokio TcpListener, which is fully async.
    let listener = TcpListener::bind(&addr).unwrap();

    let server = listener
        .incoming()
        .for_each(|socket| {
            println!("accepted socket; addr={:?}", socket.peer_addr().unwrap());
            let timeout = tokio_core::reactor::Timeout::new(Duration::from_millis(3000), &handle).unwrap();

            let buf = vec![0; 5];

            let connection
            = io::read_exact(socket, buf)
                .select2(timeout)
                .then(|res| match res {
                    Ok(Either::A(((socket, buf), _timeout))) => ok((socket, buf)),
                    Ok(Either::B((timeout_event, _stream))) => {
                        println!("time out waiting for message");
                        err(io::Error::new(
                            io::ErrorKind::TimedOut, "timed out waiting for message"))
                    }
                    Err(Either::A((read_error, _timeout))) => err(read_error),
                    Err(Either::B((timeout_error, _get))) => err(timeout_error),
                })
                .and_then(|(socket, buf)| {
                    io::write_all(socket, buf)
                })
                .then(|_| Ok(())); // Just discard the socket and buffer

            // Spawn a new task that processes the socket:
            tokio::spawn(connection);

            Ok(())
        })
        .map_err(|err| {
            // All tasks must have an `Error` type of `()`. This forces error
            // handling and helps avoid silencing failures.
            //
            // In our example, we are only going to log the error to STDOUT.
            println!("accept error = {:?}", err);
        });


    println!("server running on localhost:6142");

    //tokio::run(server);

    match core.run(server) {
        Ok(status) => println!("OK: {:?}", status),
        Err(e) => println!("Error: {:?}", e)
    }
}