echo.rs

extern crate env_logger;
extern crate futures;
extern crate tokio_core;
extern crate net2;
extern crate num_cpus;

use std::env;
use std::io;
use std::net::SocketAddr;
use std::thread;

use futures::Future;
use futures::stream::Stream;
use tokio_core::io::{copy, Io};
use tokio_core::net::TcpListener;
use tokio_core::reactor::{Core, Handle};

fn main() {
    env_logger::init().unwrap();
    let addr = env::args().nth(1).unwrap_or("127.0.0.1:8080".to_string());
    let addr = addr.parse::<SocketAddr>().unwrap();
    println!("Listening on: {}", addr);
    for _ in 0..num_cpus::get() - 1 {
        thread::spawn(move || echo(addr));
    }
    echo(addr);
}

fn echo(addr: SocketAddr) {
    let mut core = Core::new().unwrap();
    let handle = core.handle();
    let listener = listener(&addr, &handle).unwrap();

    let server = listener.incoming().for_each(move |(socket, _)| {
        let (reader, writer) = socket.split();
        let amt = copy(reader, writer);

        let msg = amt.map(move |amt| {
            println!("wrote {} bytes to {}", amt, addr)
        }).map_err(|e| {
            panic!("error: {}", e);
        });
        handle.spawn(msg);

        Ok(())
    });

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

fn listener(addr: &SocketAddr, handle: &Handle) -> io::Result<TcpListener> {
    use net2::unix::*;

    let listener = match *addr {
        SocketAddr::V4(_) => try!(net2::TcpBuilder::new_v4()),
        SocketAddr::V6(_) => try!(net2::TcpBuilder::new_v6()),
    };
    try!(listener.reuse_port(true));
    try!(listener.reuse_address(true));
    try!(listener.bind(addr));
    listener.listen(1024).and_then(|l| {
        TcpListener::from_listener(l, addr, handle)
    })
}