@@ -117,11 +117,17 @@ impl<K: Key, T: Transport, H: MessageHandler<T>> ConnInner<K, T, H> {
|
|||||||
}
|
}
|
||||||
|
|
||||||
fn parse(&mut self) -> ::Result<http::MessageHead<<<H as MessageHandler<T>>::Message as Http1Message>::Incoming>> {
|
fn parse(&mut self) -> ::Result<http::MessageHead<<<H as MessageHandler<T>>::Message as Http1Message>::Incoming>> {
|
||||||
let n = try!(self.buf.read_from(&mut self.transport));
|
match self.buf.read_from(&mut self.transport) {
|
||||||
if n == 0 {
|
Ok(0) => {
|
||||||
trace!("parse eof");
|
trace!("parse eof");
|
||||||
return Err(io::Error::new(io::ErrorKind::UnexpectedEof, "parse eof").into());
|
return Err(io::Error::new(io::ErrorKind::UnexpectedEof, "parse eof").into());
|
||||||
}
|
}
|
||||||
|
Ok(_) => {},
|
||||||
|
Err(e) => match e.kind() {
|
||||||
|
io::ErrorKind::WouldBlock => {},
|
||||||
|
_ => return Err(e.into())
|
||||||
|
}
|
||||||
|
}
|
||||||
match try!(http::parse::<<H as MessageHandler<T>>::Message, _>(self.buf.bytes())) {
|
match try!(http::parse::<<H as MessageHandler<T>>::Message, _>(self.buf.bytes())) {
|
||||||
Some((head, len)) => {
|
Some((head, len)) => {
|
||||||
trace!("parsed {} bytes out of {}", len, self.buf.len());
|
trace!("parsed {} bytes out of {}", len, self.buf.len());
|
||||||
@@ -318,7 +324,7 @@ impl<K: Key, T: Transport, H: MessageHandler<T>> ConnInner<K, T, H> {
|
|||||||
}
|
}
|
||||||
};
|
};
|
||||||
let mut head = http::MessageHead::default();
|
let mut head = http::MessageHead::default();
|
||||||
let interest = handler.on_outgoing(&mut head);
|
let mut interest = handler.on_outgoing(&mut head);
|
||||||
if head.version == HttpVersion::Http11 {
|
if head.version == HttpVersion::Http11 {
|
||||||
let mut buf = Vec::new();
|
let mut buf = Vec::new();
|
||||||
let keep_alive = self.keep_alive_enabled && head.should_keep_alive();
|
let keep_alive = self.keep_alive_enabled && head.should_keep_alive();
|
||||||
@@ -333,6 +339,7 @@ impl<K: Key, T: Transport, H: MessageHandler<T>> ConnInner<K, T, H> {
|
|||||||
bytes: buf,
|
bytes: buf,
|
||||||
pos: 0
|
pos: 0
|
||||||
});
|
});
|
||||||
|
interest = handler.on_encode(&mut Encoder::h1(&mut encoder, &mut self.transport));
|
||||||
Writing::Ready(encoder)
|
Writing::Ready(encoder)
|
||||||
},
|
},
|
||||||
_ => Writing::Chunk(Chunk {
|
_ => Writing::Chunk(Chunk {
|
||||||
@@ -364,7 +371,7 @@ impl<K: Key, T: Transport, H: MessageHandler<T>> ConnInner<K, T, H> {
|
|||||||
}
|
}
|
||||||
Writing::Head => {
|
Writing::Head => {
|
||||||
let mut head = http::MessageHead::default();
|
let mut head = http::MessageHead::default();
|
||||||
let interest = handler.on_outgoing(&mut head);
|
let mut interest = handler.on_outgoing(&mut head);
|
||||||
// if the request wants to close, server cannot stop it
|
// if the request wants to close, server cannot stop it
|
||||||
if *keep_alive {
|
if *keep_alive {
|
||||||
// if the request wants to stay alive, then it depends
|
// if the request wants to stay alive, then it depends
|
||||||
@@ -383,6 +390,7 @@ impl<K: Key, T: Transport, H: MessageHandler<T>> ConnInner<K, T, H> {
|
|||||||
bytes: buf,
|
bytes: buf,
|
||||||
pos: 0
|
pos: 0
|
||||||
});
|
});
|
||||||
|
interest = handler.on_encode(&mut Encoder::h1(&mut encoder, &mut self.transport));
|
||||||
Writing::Ready(encoder)
|
Writing::Ready(encoder)
|
||||||
},
|
},
|
||||||
_ => Writing::Chunk(Chunk {
|
_ => Writing::Chunk(Chunk {
|
||||||
@@ -444,9 +452,9 @@ impl<K: Key, T: Transport, H: MessageHandler<T>> ConnInner<K, T, H> {
|
|||||||
state
|
state
|
||||||
}
|
}
|
||||||
|
|
||||||
fn can_read_more(&self) -> bool {
|
fn can_read_more(&self, was_init: bool) -> bool {
|
||||||
match self.state {
|
match self.state {
|
||||||
State::Init { .. } => false,
|
State::Init { .. } => !was_init && !self.buf.is_empty(),
|
||||||
_ => !self.buf.is_empty()
|
_ => !self.buf.is_empty()
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -549,6 +557,11 @@ impl<K: Key, T: Transport, H: MessageHandler<T>> Conn<K, T, H> {
|
|||||||
events
|
events
|
||||||
};
|
};
|
||||||
|
|
||||||
|
let was_init = match self.0.state {
|
||||||
|
State::Init { .. } => true,
|
||||||
|
_ => false
|
||||||
|
};
|
||||||
|
|
||||||
if events.is_readable() {
|
if events.is_readable() {
|
||||||
self.0.on_readable(scope);
|
self.0.on_readable(scope);
|
||||||
}
|
}
|
||||||
@@ -570,7 +583,7 @@ impl<K: Key, T: Transport, H: MessageHandler<T>> Conn<K, T, H> {
|
|||||||
},
|
},
|
||||||
};
|
};
|
||||||
|
|
||||||
if events.is_readable() && self.0.can_read_more() {
|
if events.is_readable() && self.0.can_read_more(was_init) {
|
||||||
return self.ready(events, scope);
|
return self.ready(events, scope);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -1,5 +1,5 @@
|
|||||||
use std::borrow::Cow;
|
use std::borrow::Cow;
|
||||||
use std::io::Write;
|
use std::fmt::{self, Write};
|
||||||
|
|
||||||
use httparse;
|
use httparse;
|
||||||
|
|
||||||
@@ -95,8 +95,12 @@ impl Http1Message for ServerMessage {
|
|||||||
let init_cap = 30 + head.headers.len() * AVERAGE_HEADER_SIZE;
|
let init_cap = 30 + head.headers.len() * AVERAGE_HEADER_SIZE;
|
||||||
dst.reserve(init_cap);
|
dst.reserve(init_cap);
|
||||||
debug!("writing {:#?}", head.headers);
|
debug!("writing {:#?}", head.headers);
|
||||||
let _ = write!(dst, "{} {}\r\n{}\r\n", head.version, head.subject, head.headers);
|
if head.version == ::HttpVersion::Http11 && head.subject == ::StatusCode::Ok {
|
||||||
|
extend(dst, b"HTTP/1.1 200 OK\r\n");
|
||||||
|
let _ = write!(FastWrite(dst), "{}\r\n", head.headers);
|
||||||
|
} else {
|
||||||
|
let _ = write!(FastWrite(dst), "{} {}\r\n{}\r\n", head.version, head.subject, head.headers);
|
||||||
|
}
|
||||||
body
|
body
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -195,12 +199,37 @@ impl Http1Message for ClientMessage {
|
|||||||
let init_cap = 30 + head.headers.len() * AVERAGE_HEADER_SIZE;
|
let init_cap = 30 + head.headers.len() * AVERAGE_HEADER_SIZE;
|
||||||
dst.reserve(init_cap);
|
dst.reserve(init_cap);
|
||||||
debug!("writing {:#?}", head.headers);
|
debug!("writing {:#?}", head.headers);
|
||||||
let _ = write!(dst, "{} {}\r\n{}\r\n", head.subject, head.version, head.headers);
|
let _ = write!(FastWrite(dst), "{} {}\r\n{}\r\n", head.subject, head.version, head.headers);
|
||||||
|
|
||||||
body
|
body
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
struct FastWrite<'a>(&'a mut Vec<u8>);
|
||||||
|
|
||||||
|
impl<'a> fmt::Write for FastWrite<'a> {
|
||||||
|
fn write_str(&mut self, s: &str) -> fmt::Result {
|
||||||
|
extend(self.0, s.as_bytes());
|
||||||
|
Ok(())
|
||||||
|
}
|
||||||
|
|
||||||
|
fn write_fmt(&mut self, args: fmt::Arguments) -> fmt::Result {
|
||||||
|
fmt::write(self, args)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
fn extend(dst: &mut Vec<u8>, data: &[u8]) {
|
||||||
|
use std::ptr;
|
||||||
|
dst.reserve(data.len());
|
||||||
|
let prev = dst.len();
|
||||||
|
unsafe {
|
||||||
|
ptr::copy_nonoverlapping(data.as_ptr(),
|
||||||
|
dst.as_mut_ptr().offset(prev as isize),
|
||||||
|
data.len());
|
||||||
|
dst.set_len(prev + data.len());
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
#[cfg(test)]
|
#[cfg(test)]
|
||||||
mod tests {
|
mod tests {
|
||||||
use http;
|
use http;
|
||||||
|
|||||||
Reference in New Issue
Block a user