hyper state machines

1.hello.rs

#![deny(warnings)]
extern crate hyper;
extern crate env_logger;

use std::io::Write;

use hyper::http::{Decoder, Encoder, Next};
use hyper::net::HttpStream;
use hyper::server::{Server, Handler, Request, Response};

static PHRASE: &'static [u8] = b"Hello World!";

struct Hello;

impl Handler for Hello {
    fn on_request(&mut self, _: Request) -> Next {
        Next::write()
    }
    fn on_request_readable(&mut self, _: &mut Decoder) -> Next {
        Next::write()
    }
    fn on_response(&mut self, response: &mut Response) -> Next {
        use hyper::header::ContentLength;
        response.headers_mut().set(ContentLength(PHRASE.len() as u64));
        Next::write()
    }
    fn on_response_writable(&mut self, encoder: &mut Encoder) -> Next {
        let n = encoder.write(PHRASE).unwrap();
        debug_assert_eq!(n, PHRASE.len());
        Next::end()
    }
}

fn main() {
    env_logger::init().unwrap();
    let _listening = Server::http("127.0.0.1:3000").unwrap()
        .handle(|| Hello).unwrap();
    println!("Listening on http://127.0.0.1:3000");
}

2.echo.rs

#![deny(warnings)]
extern crate hyper;
extern crate env_logger;

use std::io::{self, Read, Write};

use hyper::{Get, Post};
use hyper::header::ContentLength;
use hyper::http::{Decoder, Encoder, Next};
use hyper::net::HttpStream;
use hyper::server::{Server, Handler, Request, Response};
use hyper::status::StatusCode;
use hyper::uri::RequestUri::AbsolutePath;

struct Echo {
    buf: Vec,
    read_pos: usize,
    write_pos: usize,
    route: Route,
}

enum Route {
    NotFound,
    Index,
    Echo(Body),
}

#[derive(Clone, Copy)]
enum Body {
    Len(u64),
    Chunked
}

static INDEX: &'static [u8] = b"Try POST /echo";

impl Echo {
    fn new() -> Echo {
        Echo {
            buf: vec![0; 4096],
            read_pos: 0,
            write_pos: 0,
            route: Route::NotFound,
        }
    }
}

impl Handler for Echo {
    fn on_request(&mut self, req: Request) -> Next {
        match *req.uri() {
            AbsolutePath(ref path) => match (req.method(), &path[..]) {
                (&Get, "/") | (&Get, "/echo") => {
                    self.route = Route::Index;
                    Next::write()
                }
                (&Post, "/echo") => {
                    self.route = if let Some(len) = req.headers().get::() {
                        Route::Echo(Body::Len(**len))
                    } else {
                        Route::Echo(Body::Chunked)
                    };
                    Next::read_and_write()
                }
                _ => Next::write(),
            },
            _ => Next::write()
        }
    }
    fn on_request_readable(&mut self, transport: &mut Decoder) -> Next {
        match self.route {
            Route::Echo(..) => {
                if self.read_pos < self.buf.len() {
                    match transport.read(&mut self.buf[self.read_pos..]) {
                        Ok(0) => Next::write(),
                        Ok(n) => {
                            self.read_pos += n;
                            Next::read_and_write()
                        }
                        Err(e) => match e.kind() {
                            io::ErrorKind::WouldBlock => Next::read_and_write(),
                            _ => {
                                println!("read error {:?}", e);
                                Next::end()
                            }
                        }
                    }
                } else {
                    Next::write()
                }
            }
            _ => unreachable!()
        }
    }

    fn on_response(&mut self, res: &mut Response) -> Next {
        match self.route {
            Route::NotFound => {
                res.set_status(StatusCode::NotFound);
                Next::end()
            }
            Route::Index => {
                res.headers_mut().set(ContentLength(INDEX.len() as u64));
                Next::write()
            }
            Route::Echo(body) => {
                if let Body::Len(len) = body {
                    res.headers_mut().set(ContentLength(len));
                }
                Next::read_and_write()
            }
        }
    }

    fn on_response_writable(&mut self, transport: &mut Encoder) -> Next {
        match self.route {
            Route::Index => {
                transport.write(INDEX).unwrap();
                Next::end()
            }
            Route::Echo(..) => {
                if self.write_pos < self.read_pos {
                    match transport.write(&self.buf[self.write_pos..self.read_pos]) {
                        Ok(0) => panic!("write ZERO"),
                        Ok(n) => {
                            self.write_pos += n;
                            Next::write()
                        }
                        Err(e) => match e.kind() {
                            io::ErrorKind::WouldBlock => Next::write(),
                            _ => {
                                println!("write error {:?}", e);
                                Next::end()
                            }
                        }
                    }
                } else {
                    // Next::read().timeout(Duration::from_secs(10))
                    Next::read()
                }
            }
            _ => unreachable!()
        }
    }
}

fn main() {
    env_logger::init().unwrap();
    let server = Server::http("127.0.0.1:1337").unwrap();
    let _guard = server.handle(|| Echo::new());
    println!("Listening on http://127.0.0.1:1337");
}