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::{Io, WriteHalf, write_all};
use tokio_core::net::{TcpListener, TcpStream};
use tokio_core::reactor::{Core, Handle};

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: WriteHalf<TcpStream>) -> impl Future<Item=WriteHalf<TcpStream>, Error=io::Error> {
    let buffer = [1u8; 16];
    write_all(tx, buffer).map(|(tx, _buffer)| tx)
}

fn reply(tx: WriteHalf<TcpStream>, hdl: Handle) -> Box<Future<Item = (), Error = io::Error>> {
    let s = TestStream {}
        .fold(tx, |tx, _| {
            eat_tx(tx)
        })
        .then(|_| 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();
}