here.rs

#![feature(conservative_impl_trait)]
extern crate futures;
extern crate tokio_core;

use std::env;
use std::io;
use std::net::SocketAddr;
use futures::{Future, Poll, Async};
use futures::stream::Stream;
use tokio_core::io::{copy, Io, read_exact, WriteHalf};
use tokio_core::net::{TcpListener, TcpStream};
use tokio_core::reactor::{Core, Handle};
use std::io::Write;

struct TestStream {
}

impl Stream for TestStream {
    type Item = u32;
    type Error = io::Error;

    fn poll(&mut self) -> Poll<Option<Self::Item>, Self::Error> {
        Ok(Async::Ready(Some(1)))
    }
}

fn eat_tx(tx: &mut WriteHalf<TcpStream>) -> io::Result<()> {
    tx.write_all(&[1u8; 16]);
    Ok(())
}

fn reply(mut tx: WriteHalf<TcpStream>, hdl: Handle) -> Box<Future<Item = (), Error = io::Error>> {
    let s = TestStream {}
        .for_each(move |x| eat_tx(&mut tx))
        .then(move |r| Ok(()));
    hdl.spawn(s);
    futures::done(Ok(())).boxed()
}

fn main() {
    let addr = env::args().nth(1).unwrap_or("127.0.0.1:8080".to_string());
    let addr = addr.parse::<SocketAddr>().unwrap();
    let mut l = Core::new().unwrap();
    let handle = l.handle();
    let socket = TcpListener::bind(&addr, &handle).unwrap();
    println!("Listening on: {}", addr);

    let done = socket.incoming().for_each(move |(socket, addr)| {
        let handle1 = handle.clone();
        let pair = futures::lazy(|| futures::finished(socket.split()));
        let amt = pair.and_then(move |(reader, mut writer)| {
            let t = futures::done(Ok(())).and_then(move |x| reply(writer, handle1));
            Ok(())
        });
        handle.spawn(amt);
        Ok(())
    });
    l.run(done).unwrap();
}