Code shared from the Rust Playground

playground.rs

#!/usr/bin/env run-cargo-script
// cargo-deps: tokio, tokio-codec, tokio-io, futures-cpupool

extern crate tokio;
extern crate tokio_codec;

use tokio_codec::{Framed, LinesCodec};
use tokio::net::TcpListener;
use tokio::prelude::*;

use std::env;
use std::net::SocketAddr;
use std::sync::{Mutex,Arc};

use std::io::{Error as IoError,ErrorKind};

fn main() -> Result<(),Box<std::error::Error>> {
    let addr = env::args().nth(1).unwrap_or("127.0.0.1:8080".to_string());
    let addr = addr.parse::<SocketAddr>()?;
    let socket = TcpListener::bind(&addr)?;
    println!("Listening on: {}", addr);
    
    let listener = socket
        .incoming()
        .map_err(|e| println!("failed to accept socket; error = {:?}", e))
        .for_each(move |socket| {
            let framed = Framed::new(socket, LinesCodec::new());
            let (writer, reader) = framed.split();
            
            let (work_tx, work_rx) = mpsc::channel(1);
            let (close_tx, close_rx) = oneshot:
            
            let queuer = reader
                .for_each(move |buf| work_tx.send(buf))
                .and_then(|_| {
                    println!("socket closed");
                    close_tx.send(())
                })
                .or_else(|err| {
                    println!("socket closed with error: {:?}", err);
                    Err(err)
                });
            
            tokio::spawn(queuer);
            
            let processor = work_rx
                .for_each(move || {
                    println!("received request: {:?}", buf);
                    for i in {0..19} {
                        if i % 5 == 0 {
                            println!("processing request: {:?} {}%", buf, i*5);
                        }
                        ::std::thread::sleep(std::time::Duration::from_millis(100));
                    }
                    println!("processing finished: {:?}", buf);
                    writer.send(buf)
                })
                .select(close_rx);

            tokio::spawn(processor)
        });

    tokio::run(listener);
    Ok(())
}