#![deny(warnings)]
extern crate bytes;
extern crate http;
extern crate httparse;
#[macro_use]
extern crate serde_derive;
extern crate serde_json;
extern crate time;
extern crate tokio;
extern crate tokio_io;
use std::net::SocketAddr;
use std::{env, fmt, io};
use tokio::codec::{Decoder, Encoder};
use tokio::net::{TcpListener, TcpStream};
use tokio::prelude::*;
use bytes::BytesMut;
use http::header::HeaderValue;
use http::{Request, Response, StatusCode};
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 listener = TcpListener::bind(&addr)?;
println!("Listening on: {}", addr);
tokio::run({
listener
.incoming()
.map_err(|e| println!("failed to accept socket; error = {:?}", e))
.for_each(|socket| {
process(socket);
Ok(())
})
});
Ok(())
}
fn process(socket: TcpStream) {
let (tx, rx) =
Http.framed(socket)
.split();
let task = tx.send_all(rx.and_then(respond)).then(|res| {
if let Err(e) = res {
println!("failed to process connection; error = {:?}", e);
}
Ok(())
});
tokio::spawn(task);
}
fn respond(req: Request<()>) -> Box<Future<Item = Response<String>, Error = io::Error> + Send> {
let f = future::lazy(move || {
let mut response = Response::builder();
let body = match req.uri().path() {
"/plaintext" => {
response.header("Content-Type", "text/plain");
"Hello, World!".to_string()
}
"/json" => {
response.header("Content-Type", "application/json");
#[derive(Serialize)]
struct Message {
message: &'static str,
}
serde_json::to_string(&Message {
message: "Hello, World!",
})?
}
_ => {
response.status(StatusCode::NOT_FOUND);
String::new()
}
};
let response = response
.body(body)
.map_err(|err| io::Error::new(io::ErrorKind::Other, err))?;
Ok(response)
});
Box::new(f)
}
struct Http;
impl Encoder for Http {
type Item = Response<String>;
type Error = io::Error;
fn encode(&mut self, item: Response<String>, dst: &mut BytesMut) -> io::Result<()> {
use std::fmt::Write;
write!(
BytesWrite(dst),
"\
HTTP/1.1 {}\r\n\
Server: Example\r\n\
Content-Length: {}\r\n\
Date: {}\r\n\
",
item.status(),
item.body().len(),
date::now()
)
.unwrap();
for (k, v) in item.headers() {
dst.extend_from_slice(k.as_str().as_bytes());
dst.extend_from_slice(b": ");
dst.extend_from_slice(v.as_bytes());
dst.extend_from_slice(b"\r\n");
}
dst.extend_from_slice(b"\r\n");
dst.extend_from_slice(item.body().as_bytes());
return Ok(());
struct BytesWrite<'a>(&'a mut BytesMut);
impl<'a> fmt::Write for BytesWrite<'a> {
fn write_str(&mut self, s: &str) -> fmt::Result {
self.0.extend_from_slice(s.as_bytes());
Ok(())
}
fn write_fmt(&mut self, args: fmt::Arguments) -> fmt::Result {
fmt::write(self, args)
}
}
}
}
impl Decoder for Http {
type Item = Request<()>;
type Error = io::Error;
fn decode(&mut self, src: &mut BytesMut) -> io::Result<Option<Request<()>>> {
let mut headers = [None; 16];
let (method, path, version, amt) = {
let mut parsed_headers = [httparse::EMPTY_HEADER; 16];
let mut r = httparse::Request::new(&mut parsed_headers);
let status = r.parse(src).map_err(|e| {
let msg = format!("failed to parse http request: {:?}", e);
io::Error::new(io::ErrorKind::Other, msg)
})?;
let amt = match status {
httparse::Status::Complete(amt) => amt,
httparse::Status::Partial => return Ok(None),
};
let toslice = |a: &[u8]| {
let start = a.as_ptr() as usize - src.as_ptr() as usize;
assert!(start < src.len());
(start, start + a.len())
};
for (i, header) in r.headers.iter().enumerate() {
let k = toslice(header.name.as_bytes());
let v = toslice(header.value);
headers[i] = Some((k, v));
}
(
toslice(r.method.unwrap().as_bytes()),
toslice(r.path.unwrap().as_bytes()),
r.version.unwrap(),
amt,
)
};
if version != 1 {
return Err(io::Error::new(
io::ErrorKind::Other,
"only HTTP/1.1 accepted",
));
}
let data = src.split_to(amt).freeze();
let mut ret = Request::builder();
ret.method(&data[method.0..method.1]);
ret.uri(data.slice(path.0, path.1));
ret.version(http::Version::HTTP_11);
for header in headers.iter() {
let (k, v) = match *header {
Some((ref k, ref v)) => (k, v),
None => break,
};
let value = unsafe { HeaderValue::from_shared_unchecked(data.slice(v.0, v.1)) };
ret.header(&data[k.0..k.1], value);
}
let req = ret
.body(())
.map_err(|e| io::Error::new(io::ErrorKind::Other, e))?;
Ok(Some(req))
}
}
mod date {
use std::cell::RefCell;
use std::fmt::{self, Write};
use std::str;
use time::{self, Duration};
pub struct Now(());
pub fn now() -> Now {
Now(())
}
struct LastRenderedNow {
bytes: [u8; 128],
amt: usize,
next_update: time::Timespec,
}
thread_local!(static LAST: RefCell<LastRenderedNow> = RefCell::new(LastRenderedNow {
bytes: [0; 128],
amt: 0,
next_update: time::Timespec::new(0, 0),
}));
impl fmt::Display for Now {
fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
LAST.with(|cache| {
let mut cache = cache.borrow_mut();
let now = time::get_time();
if now >= cache.next_update {
cache.update(now);
}
f.write_str(cache.buffer())
})
}
}
impl LastRenderedNow {
fn buffer(&self) -> &str {
str::from_utf8(&self.bytes[..self.amt]).unwrap()
}
fn update(&mut self, now: time::Timespec) {
self.amt = 0;
write!(LocalBuffer(self), "{}", time::at(now).rfc822()).unwrap();
self.next_update = now + Duration::seconds(1);
self.next_update.nsec = 0;
}
}
struct LocalBuffer<'a>(&'a mut LastRenderedNow);
impl<'a> fmt::Write for LocalBuffer<'a> {
fn write_str(&mut self, s: &str) -> fmt::Result {
let start = self.0.amt;
let end = start + s.len();
self.0.bytes[start..end].copy_from_slice(s.as_bytes());
self.0.amt += s.len();
Ok(())
}
}
}