forked/hyper
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #1480 from hyperium/error2

Browse Source 
feat(error): revamp `hyper::Error` type
This commit is contained in:
Sean McArthur
2018-04-10 15:13:57 -07:00
committed by GitHub
parent 33874f9a75 5d3c472228
commit 313f82d971
22 changed files with 519 additions and 407 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
benches/server.rs
View File

@@ -78,7 +78,7 @@ fn throughput_fixedsize_large_payload(b: &mut test::Bencher) {
fn throughput_fixedsize_many_chunks(b: &mut test::Bencher) { fn throughput_fixedsize_many_chunks(b: &mut test::Bencher) {
bench_server!(b, ("content-length", "1000000"), || { bench_server!(b, ("content-length", "1000000"), || {
static S: &'static [&'static [u8]] = &[&[b'x'; 1_000] as &[u8]; 1_000] as _; static S: &'static [&'static [u8]] = &[&[b'x'; 1_000] as &[u8]; 1_000] as _;
Body::wrap_stream(stream::iter_ok(S.iter()).map(|&s| s)) Body::wrap_stream(stream::iter_ok::<_, String>(S.iter()).map(|&s| s))
}) })
} }
@@ -96,7 +96,7 @@ fn throughput_chunked_large_payload(b: &mut test::Bencher) {
fn throughput_chunked_many_chunks(b: &mut test::Bencher) { fn throughput_chunked_many_chunks(b: &mut test::Bencher) {
bench_server!(b, ("transfer-encoding", "chunked"), || { bench_server!(b, ("transfer-encoding", "chunked"), || {
static S: &'static [&'static [u8]] = &[&[b'x'; 1_000] as &[u8]; 1_000] as _; static S: &'static [&'static [u8]] = &[&[b'x'; 1_000] as &[u8]; 1_000] as _;
Body::wrap_stream(stream::iter_ok(S.iter()).map(|&s| s)) Body::wrap_stream(stream::iter_ok::<_, String>(S.iter()).map(|&s| s))
}) })
} }

5
examples/client.rs
View File

@@ -40,8 +40,9 @@ fn main() {
println!("Response: {}", res.status()); println!("Response: {}", res.status());
println!("Headers: {:#?}", res.headers()); println!("Headers: {:#?}", res.headers());
res.into_parts().1.into_stream().for_each(|chunk| { res.into_body().into_stream().for_each(|chunk| {
io::stdout().write_all(&chunk).map_err(From::from) io::stdout().write_all(&chunk)
.map_err(|e| panic!("example expects stdout is open, error={}", e))
}) })
}).map(|_| { }).map(|_| {
println!("\n\nDone."); println!("\n\nDone.");

3
examples/hello.rs
View File

@@ -17,7 +17,8 @@ fn main() {
let addr = ([127, 0, 0, 1], 3000).into(); let addr = ([127, 0, 0, 1], 3000).into();
let new_service = const_service(service_fn(|_| { let new_service = const_service(service_fn(|_| {
Ok(Response::new(Body::from(PHRASE))) //TODO: when `!` is stable, replace error type
Ok::<_, hyper::Error>(Response::new(Body::from(PHRASE)))
})); }));
tokio::run(lazy(move || { tokio::run(lazy(move || {

9
examples/send_file.rs
View File

@@ -9,7 +9,6 @@ use futures::future::lazy;
use futures::sync::oneshot; use futures::sync::oneshot;
use hyper::{Body, /*Chunk,*/ Method, Request, Response, StatusCode}; use hyper::{Body, /*Chunk,*/ Method, Request, Response, StatusCode};
use hyper::error::Error;
use hyper::server::{Http, Service}; use hyper::server::{Http, Service};
use std::fs::File; use std::fs::File;
@@ -19,7 +18,7 @@ use std::thread;
static NOTFOUND: &[u8] = b"Not Found"; static NOTFOUND: &[u8] = b"Not Found";
static INDEX: &str = "examples/send_file_index.html"; static INDEX: &str = "examples/send_file_index.html";
fn simple_file_send(f: &str) -> Box<Future<Item = Response<Body>, Error = hyper::Error> + Send> { fn simple_file_send(f: &str) -> Box<Future<Item = Response<Body>, Error = io::Error> + Send> {
// Serve a file by reading it entirely into memory. As a result // Serve a file by reading it entirely into memory. As a result
// this is limited to serving small files, but it is somewhat // this is limited to serving small files, but it is somewhat
// simpler with a little less overhead. // simpler with a little less overhead.
@@ -56,7 +55,7 @@ fn simple_file_send(f: &str) -> Box<Future<Item = Response<Body>, Error = hyper:
}; };
}); });
Box::new(rx.map_err(|e| Error::from(io::Error::new(io::ErrorKind::Other, e)))) Box::new(rx.map_err(|e| io::Error::new(io::ErrorKind::Other, e)))
} }
struct ResponseExamples; struct ResponseExamples;
@@ -64,7 +63,7 @@ struct ResponseExamples;
impl Service for ResponseExamples { impl Service for ResponseExamples {
type Request = Request<Body>; type Request = Request<Body>;
type Response = Response<Body>; type Response = Response<Body>;
type Error = hyper::Error; type Error = io::Error;
type Future = Box<Future<Item = Self::Response, Error = Self::Error> + Send>; type Future = Box<Future<Item = Self::Response, Error = Self::Error> + Send>;
fn call(&self, req: Request<Body>) -> Self::Future { fn call(&self, req: Request<Body>) -> Self::Future {
@@ -119,7 +118,7 @@ impl Service for ResponseExamples {
*/ */
}); });
Box::new(rx.map_err(|e| Error::from(io::Error::new(io::ErrorKind::Other, e)))) Box::new(rx.map_err(|e| io::Error::new(io::ErrorKind::Other, e)))
}, },
(&Method::GET, "/no_file.html") => { (&Method::GET, "/no_file.html") => {
// Test what happens when file cannot be be found // Test what happens when file cannot be be found

24
src/client/conn.rs
View File

@@ -45,7 +45,7 @@ pub struct SendRequest<B> {
pub struct Connection<T, B> pub struct Connection<T, B>
where where
T: AsyncRead + AsyncWrite, T: AsyncRead + AsyncWrite,
B: Entity<Error=::Error> + 'static, B: Entity + 'static,
{ {
inner: proto::dispatch::Dispatcher< inner: proto::dispatch::Dispatcher<
proto::dispatch::Client<B>, proto::dispatch::Client<B>,
@@ -138,7 +138,7 @@ impl<B> SendRequest<B>
impl<B> SendRequest<B> impl<B> SendRequest<B>
where where
B: Entity<Error=::Error> + 'static, B: Entity + 'static,
{ {
/// Sends a `Request` on the associated connection. /// Sends a `Request` on the associated connection.
/// ///
@@ -262,7 +262,7 @@ impl<B> fmt::Debug for SendRequest<B> {
impl<T, B> Connection<T, B> impl<T, B> Connection<T, B>
where where
T: AsyncRead + AsyncWrite, T: AsyncRead + AsyncWrite,
B: Entity<Error=::Error> + 'static, B: Entity + 'static,
{ {
/// Return the inner IO object, and additional information. /// Return the inner IO object, and additional information.
pub fn into_parts(self) -> Parts<T> { pub fn into_parts(self) -> Parts<T> {
@@ -289,7 +289,7 @@ where
impl<T, B> Future for Connection<T, B> impl<T, B> Future for Connection<T, B>
where where
T: AsyncRead + AsyncWrite, T: AsyncRead + AsyncWrite,
B: Entity<Error=::Error> + 'static, B: Entity + 'static,
{ {
type Item = (); type Item = ();
type Error = ::Error; type Error = ::Error;
@@ -302,7 +302,7 @@ where
impl<T, B> fmt::Debug for Connection<T, B> impl<T, B> fmt::Debug for Connection<T, B>
where where
T: AsyncRead + AsyncWrite + fmt::Debug, T: AsyncRead + AsyncWrite + fmt::Debug,
B: Entity<Error=::Error> + 'static, B: Entity + 'static,
{ {
fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result { fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
f.debug_struct("Connection") f.debug_struct("Connection")
@@ -331,7 +331,7 @@ impl Builder {
pub fn handshake<T, B>(&self, io: T) -> Handshake<T, B> pub fn handshake<T, B>(&self, io: T) -> Handshake<T, B>
where where
T: AsyncRead + AsyncWrite, T: AsyncRead + AsyncWrite,
B: Entity<Error=::Error> + 'static, B: Entity + 'static,
{ {
Handshake { Handshake {
inner: HandshakeInner { inner: HandshakeInner {
@@ -345,7 +345,7 @@ impl Builder {
pub(super) fn handshake_no_upgrades<T, B>(&self, io: T) -> HandshakeNoUpgrades<T, B> pub(super) fn handshake_no_upgrades<T, B>(&self, io: T) -> HandshakeNoUpgrades<T, B>
where where
T: AsyncRead + AsyncWrite, T: AsyncRead + AsyncWrite,
B: Entity<Error=::Error> + 'static, B: Entity + 'static,
{ {
HandshakeNoUpgrades { HandshakeNoUpgrades {
inner: HandshakeInner { inner: HandshakeInner {
@@ -362,7 +362,7 @@ impl Builder {
impl<T, B> Future for Handshake<T, B> impl<T, B> Future for Handshake<T, B>
where where
T: AsyncRead + AsyncWrite, T: AsyncRead + AsyncWrite,
B: Entity<Error=::Error> + 'static, B: Entity + 'static,
{ {
type Item = (SendRequest<B>, Connection<T, B>); type Item = (SendRequest<B>, Connection<T, B>);
type Error = ::Error; type Error = ::Error;
@@ -387,7 +387,7 @@ impl<T, B> fmt::Debug for Handshake<T, B> {
impl<T, B> Future for HandshakeNoUpgrades<T, B> impl<T, B> Future for HandshakeNoUpgrades<T, B>
where where
T: AsyncRead + AsyncWrite, T: AsyncRead + AsyncWrite,
B: Entity<Error=::Error> + 'static, B: Entity + 'static,
{ {
type Item = (SendRequest<B>, proto::dispatch::Dispatcher< type Item = (SendRequest<B>, proto::dispatch::Dispatcher<
proto::dispatch::Client<B>, proto::dispatch::Client<B>,
@@ -406,7 +406,7 @@ where
impl<T, B, R> Future for HandshakeInner<T, B, R> impl<T, B, R> Future for HandshakeInner<T, B, R>
where where
T: AsyncRead + AsyncWrite, T: AsyncRead + AsyncWrite,
B: Entity<Error=::Error> + 'static, B: Entity + 'static,
R: proto::Http1Transaction< R: proto::Http1Transaction<
Incoming=StatusCode, Incoming=StatusCode,
Outgoing=proto::RequestLine, Outgoing=proto::RequestLine,
@@ -470,7 +470,7 @@ impl<B: Send> AssertSendSync for SendRequest<B> {}
impl<T: Send, B: Send> AssertSend for Connection<T, B> impl<T: Send, B: Send> AssertSend for Connection<T, B>
where where
T: AsyncRead + AsyncWrite, T: AsyncRead + AsyncWrite,
B: Entity<Error=::Error> + 'static, B: Entity + 'static,
B::Data: Send + 'static, B::Data: Send + 'static,
{} {}
@@ -478,7 +478,7 @@ where
impl<T: Send + Sync, B: Send + Sync> AssertSendSync for Connection<T, B> impl<T: Send + Sync, B: Send + Sync> AssertSendSync for Connection<T, B>
where where
T: AsyncRead + AsyncWrite, T: AsyncRead + AsyncWrite,
B: Entity<Error=::Error> + 'static, B: Entity + 'static,
B::Data: Send + Sync + 'static, B::Data: Send + Sync + 'static,
{} {}

2
src/client/connect.rs
View File

@@ -36,7 +36,7 @@ pub trait Connect: Send + Sync {
/// The connected IO Stream. /// The connected IO Stream.
type Transport: AsyncRead + AsyncWrite + Send + 'static; type Transport: AsyncRead + AsyncWrite + Send + 'static;
/// An error occured when trying to connect. /// An error occured when trying to connect.
type Error; type Error: Into<Box<StdError + Send + Sync>>;
/// A Future that will resolve to the connected Transport. /// A Future that will resolve to the connected Transport.
type Future: Future<Item=(Self::Transport, Connected), Error=Self::Error> + Send; type Future: Future<Item=(Self::Transport, Connected), Error=Self::Error> + Send;
/// Connect to a destination. /// Connect to a destination.

13
src/client/dispatch.rs
View File

@@ -39,10 +39,10 @@ impl<T, U> Sender<T, U> {
// there's room in the queue, but does the Connection // there's room in the queue, but does the Connection
// want a message yet? // want a message yet?
self.giver.poll_want() self.giver.poll_want()
.map_err(|_| ::Error::Closed) .map_err(|_| ::Error::new_closed())
}, },
Ok(Async::NotReady) => Ok(Async::NotReady), Ok(Async::NotReady) => Ok(Async::NotReady),
Err(_) => Err(::Error::Closed), Err(_) => Err(::Error::new_closed()),
} }
} }
@@ -184,9 +184,12 @@ mod tests {
drop(rx); drop(rx);
promise.then(|fulfilled| { promise.then(|fulfilled| {
let res = fulfilled.expect("fulfilled"); let err = fulfilled
match res.unwrap_err() { .expect("fulfilled")
(::Error::Cancel(_), Some(_)) => (), .expect_err("promise should error");
match (err.0.kind(), err.1) {
(&::error::Kind::Canceled, Some(_)) => (),
e => panic!("expected Error::Cancel(_), found {:?}", e), e => panic!("expected Error::Cancel(_), found {:?}", e),
} }

49
src/client/mod.rs
View File

@@ -96,10 +96,10 @@ impl<C, B> Client<C, B> {
} }
impl<C, B> Client<C, B> impl<C, B> Client<C, B>
where C: Connect<Error=io::Error> + Sync + 'static, where C: Connect + Sync + 'static,
C::Transport: 'static, C::Transport: 'static,
C::Future: 'static, C::Future: 'static,
B: Entity<Error=::Error> + Send + 'static, B: Entity + Send + 'static,
B::Data: Send, B::Data: Send,
{ {
@@ -139,13 +139,14 @@ where C: Connect<Error=io::Error> + Sync + 'static,
Version::HTTP_11 => (), Version::HTTP_11 => (),
other => { other => {
error!("Request has unsupported version \"{:?}\"", other); error!("Request has unsupported version \"{:?}\"", other);
return FutureResponse(Box::new(future::err(::Error::Version))); //TODO: replace this with a proper variant
return FutureResponse(Box::new(future::err(::Error::new_user_unsupported_version())));
} }
} }
if req.method() == &Method::CONNECT { if req.method() == &Method::CONNECT {
debug!("Client does not support CONNECT requests"); debug!("Client does not support CONNECT requests");
return FutureResponse(Box::new(future::err(::Error::Method))); return FutureResponse(Box::new(future::err(::Error::new_user_unsupported_request_method())));
} }
let uri = req.uri().clone(); let uri = req.uri().clone();
@@ -154,7 +155,8 @@ where C: Connect<Error=io::Error> + Sync + 'static,
format!("{}://{}", scheme, auth) format!("{}://{}", scheme, auth)
} }
_ => { _ => {
return FutureResponse(Box::new(future::err(::Error::Io( //TODO: replace this with a proper variant
return FutureResponse(Box::new(future::err(::Error::new_io(
io::Error::new( io::Error::new(
io::ErrorKind::InvalidInput, io::ErrorKind::InvalidInput,
"invalid URI for Client Request" "invalid URI for Client Request"
@@ -203,13 +205,13 @@ where C: Connect<Error=io::Error> + Sync + 'static,
}; };
future::lazy(move || { future::lazy(move || {
connector.connect(dst) connector.connect(dst)
.from_err() .map_err(::Error::new_connect)
.and_then(move |(io, connected)| { .and_then(move |(io, connected)| {
conn::Builder::new() conn::Builder::new()
.h1_writev(h1_writev) .h1_writev(h1_writev)
.handshake_no_upgrades(io) .handshake_no_upgrades(io)
.and_then(move |(tx, conn)| { .and_then(move |(tx, conn)| {
executor.execute(conn.map_err(|e| debug!("client connection error: {}", e)))?; executor.execute(conn.map_err(|e| debug!("client connection error: {}", e)));
Ok(pool.pooled(pool_key, PoolClient { Ok(pool.pooled(pool_key, PoolClient {
is_proxied: connected.is_proxied, is_proxied: connected.is_proxied,
tx: tx, tx: tx,
@@ -260,9 +262,7 @@ where C: Connect<Error=io::Error> + Sync + 'static,
} else if !res.body().is_empty() { } else if !res.body().is_empty() {
let (delayed_tx, delayed_rx) = oneshot::channel(); let (delayed_tx, delayed_rx) = oneshot::channel();
res.body_mut().delayed_eof(delayed_rx); res.body_mut().delayed_eof(delayed_rx);
// If the executor doesn't have room, oh well. Things will likely executor.execute(
// be blowing up soon, but this specific task isn't required.
let _ = executor.execute(
future::poll_fn(move || { future::poll_fn(move || {
pooled.tx.poll_ready() pooled.tx.poll_ready()
}) })
@@ -277,7 +277,6 @@ where C: Connect<Error=io::Error> + Sync + 'static,
Ok(res) Ok(res)
}); });
fut fut
}); });
@@ -290,9 +289,9 @@ where C: Connect<Error=io::Error> + Sync + 'static,
} }
impl<C, B> Service for Client<C, B> impl<C, B> Service for Client<C, B>
where C: Connect<Error=io::Error> + 'static, where C: Connect + 'static,
C::Future: 'static, C::Future: 'static,
B: Entity<Error=::Error> + Send + 'static, B: Entity + Send + 'static,
B::Data: Send, B::Data: Send,
{ {
type Request = Request<B>; type Request = Request<B>;
@@ -353,9 +352,9 @@ struct RetryableSendRequest<C, B> {
impl<C, B> Future for RetryableSendRequest<C, B> impl<C, B> Future for RetryableSendRequest<C, B>
where where
C: Connect<Error=io::Error> + 'static, C: Connect + 'static,
C::Future: 'static, C::Future: 'static,
B: Entity<Error=::Error> + Send + 'static, B: Entity + Send + 'static,
B::Data: Send, B::Data: Send,
{ {
type Item = Response<Body>; type Item = Response<Body>;
@@ -564,10 +563,10 @@ impl<C, B> Config<C, B> {
} }
impl<C, B> Config<C, B> impl<C, B> Config<C, B>
where C: Connect<Error=io::Error>, where C: Connect,
C::Transport: 'static, C::Transport: 'static,
C::Future: 'static, C::Future: 'static,
B: Entity<Error=::Error> + Send, B: Entity + Send,
B::Data: Send, B::Data: Send,
{ {
/// Construct the Client with this configuration. /// Construct the Client with this configuration.
@@ -590,7 +589,7 @@ where C: Connect<Error=io::Error>,
impl<B> Config<UseDefaultConnector, B> impl<B> Config<UseDefaultConnector, B>
where where
B: Entity<Error=::Error> + Send, B: Entity + Send,
B::Data: Send, B::Data: Send,
{ {
/// Construct the Client with this configuration. /// Construct the Client with this configuration.
@@ -640,7 +639,6 @@ impl<C: Clone, B> Clone for Config<C, B> {
} }
} }
// ===== impl Exec ===== // ===== impl Exec =====
#[derive(Clone)] #[derive(Clone)]
@@ -655,24 +653,19 @@ impl Exec {
Exec::Executor(Arc::new(executor)) Exec::Executor(Arc::new(executor))
} }
fn execute<F>(&self, fut: F) -> io::Result<()> fn execute<F>(&self, fut: F)
where where
F: Future<Item=(), Error=()> + Send + 'static, F: Future<Item=(), Error=()> + Send + 'static,
{ {
match *self { match *self {
Exec::Default => spawn(fut), Exec::Default => spawn(fut),
Exec::Executor(ref e) => { Exec::Executor(ref e) => {
e.execute(bg(Box::new(fut))) let _ = e.execute(bg(Box::new(fut)))
.map_err(|err| { .map_err(|err| {
debug!("executor error: {:?}", err.kind()); panic!("executor error: {:?}", err.kind());
io::Error::new( });
io::ErrorKind::Other,
"executor error",
)
})?
}, },
} }
Ok(())
} }
} }

2
src/client/pool.rs
View File

@@ -245,7 +245,7 @@ impl<T: Closed + Send + 'static> Pool<T> {
interval: interval, interval: interval,
pool: Arc::downgrade(&self.inner), pool: Arc::downgrade(&self.inner),
pool_drop_notifier: rx, pool_drop_notifier: rx,
}).unwrap(); });
} }
} }

12
src/client/tests.rs
View File

@@ -35,7 +35,7 @@ fn retryable_request() {
try_ready!(sock1.read(&mut [0u8; 512])); try_ready!(sock1.read(&mut [0u8; 512]));
try_ready!(sock1.write(b"HTTP/1.1 200 OK\r\nContent-Length: 0\r\n\r\n")); try_ready!(sock1.write(b"HTTP/1.1 200 OK\r\nContent-Length: 0\r\n\r\n"));
Ok(Async::Ready(())) Ok(Async::Ready(()))
}); }).map_err(|e: ::std::io::Error| panic!("srv1 poll_fn error: {}", e));
res1.join(srv1).wait().expect("res1"); res1.join(srv1).wait().expect("res1");
} }
drop(sock1); drop(sock1);
@@ -52,7 +52,7 @@ fn retryable_request() {
try_ready!(sock2.read(&mut [0u8; 512])); try_ready!(sock2.read(&mut [0u8; 512]));
try_ready!(sock2.write(b"HTTP/1.1 222 OK\r\nContent-Length: 0\r\n\r\n")); try_ready!(sock2.write(b"HTTP/1.1 222 OK\r\nContent-Length: 0\r\n\r\n"));
Ok(Async::Ready(())) Ok(Async::Ready(()))
}); }).map_err(|e: ::std::io::Error| panic!("srv2 poll_fn error: {}", e));
res2.join(srv2).wait().expect("res2"); res2.join(srv2).wait().expect("res2");
} }
@@ -82,7 +82,7 @@ fn conn_reset_after_write() {
try_ready!(sock1.read(&mut [0u8; 512])); try_ready!(sock1.read(&mut [0u8; 512]));
try_ready!(sock1.write(b"HTTP/1.1 200 OK\r\nContent-Length: 0\r\n\r\n")); try_ready!(sock1.write(b"HTTP/1.1 200 OK\r\nContent-Length: 0\r\n\r\n"));
Ok(Async::Ready(())) Ok(Async::Ready(()))
}); }).map_err(|e: ::std::io::Error| panic!("srv1 poll_fn error: {}", e));
res1.join(srv1).wait().expect("res1"); res1.join(srv1).wait().expect("res1");
} }
@@ -105,10 +105,10 @@ fn conn_reset_after_write() {
try_ready!(sock1.as_mut().unwrap().read(&mut [0u8; 512])); try_ready!(sock1.as_mut().unwrap().read(&mut [0u8; 512]));
sock1.take(); sock1.take();
Ok(Async::Ready(())) Ok(Async::Ready(()))
}); }).map_err(|e: ::std::io::Error| panic!("srv2 poll_fn error: {}", e));
let err = res2.join(srv2).wait().expect_err("res2"); let err = res2.join(srv2).wait().expect_err("res2");
match err { match err.kind() {
::Error::Incomplete => (), &::error::Kind::Incomplete => (),
other => panic!("expected Incomplete, found {:?}", other) other => panic!("expected Incomplete, found {:?}", other)
} }
} }

437
src/error.rs
View File

@@ -1,188 +1,316 @@
//! Error and Result module. //! Error and Result module.
use std::error::Error as StdError; use std::error::Error as StdError;
use std::fmt; use std::fmt;
use std::io::Error as IoError; use std::io;
use std::str::Utf8Error;
use std::string::FromUtf8Error;
use httparse; use httparse;
use http; use http;
use self::Error::{
Method,
Version,
Uri,
Header,
Status,
Timeout,
Upgrade,
Closed,
Cancel,
Io,
TooLarge,
Incomplete,
Utf8
};
/// Result type often returned from methods that can have hyper `Error`s. /// Result type often returned from methods that can have hyper `Error`s.
pub type Result<T> = ::std::result::Result<T, Error>; pub type Result<T> = ::std::result::Result<T, Error>;
/// A set of errors that can occur parsing HTTP streams. type Cause = Box<StdError + Send + Sync>;
#[derive(Debug)]
pub enum Error { /// Represents errors that can occur handling HTTP streams.
/// An invalid `Method`, such as `GE,T`. pub struct Error {
Method, inner: Box<ErrorImpl>,
/// An invalid `HttpVersion`, such as `HTP/1.1` }
Version,
/// Uri Errors struct ErrorImpl {
Uri, kind: Kind,
/// An invalid `Header`. cause: Option<Cause>,
Header, }
/// A message head is too large to be reasonable.
TooLarge, #[derive(Debug, PartialEq)]
pub(crate) enum Kind {
Parse(Parse),
/// A message reached EOF, but is not complete. /// A message reached EOF, but is not complete.
Incomplete, Incomplete,
/// An invalid `Status`, such as `1337 ELITE`.
Status,
/// A timeout occurred waiting for an IO event.
Timeout,
/// A protocol upgrade was encountered, but not yet supported in hyper. /// A protocol upgrade was encountered, but not yet supported in hyper.
Upgrade, Upgrade,
/// A client connection received a response when not waiting for one.
MismatchedResponse,
/// A pending item was dropped before ever being processed. /// A pending item was dropped before ever being processed.
Cancel(Canceled), Canceled,
/// Indicates a connection is closed. /// Indicates a connection is closed.
Closed, Closed,
/// An `io::Error` that occurred while trying to read or write to a network stream. /// An `io::Error` that occurred while trying to read or write to a network stream.
Io(IoError), Io,
/// Parsing a field as string failed /// Error occurred while connecting.
Utf8(Utf8Error), Connect,
/// Error creating a TcpListener.
Listen,
/// Error accepting on an Incoming stream.
Accept,
/// Error calling user's NewService::new_service().
NewService,
/// Error from future of user's Service::call().
Service,
/// Error while reading a body from connection.
Body,
/// Error while writing a body to connection.
BodyWrite,
/// Error calling user's Entity::poll_data().
BodyUser,
/// Error calling AsyncWrite::shutdown()
Shutdown,
#[doc(hidden)] /// User tried to create a Request with bad version.
__Nonexhaustive(Void) UnsupportedVersion,
/// User tried to create a CONNECT Request with the Client.
UnsupportedRequestMethod,
} }
#[derive(Debug, PartialEq)]
pub(crate) enum Parse {
Method,
Version,
Uri,
Header,
TooLarge,
Status,
}
/*
#[derive(Debug)]
pub(crate) enum User {
VersionNotSupported,
MethodNotSupported,
InvalidRequestUri,
}
*/
impl Error { impl Error {
pub(crate) fn new_canceled<E: Into<Box<StdError + Send + Sync>>>(cause: Option<E>) -> Error { //TODO(error): should there be these kinds of inspection methods?
Error::Cancel(Canceled { //
cause: cause.map(Into::into), // - is_io()
}) // - is_connect()
// - is_closed()
// - etc?
/// Returns true if this was an HTTP parse error.
pub fn is_parse(&self) -> bool {
match self.inner.kind {
Kind::Parse(_) => true,
_ => false,
}
}
/// Returns true if this error was caused by user code.
pub fn is_user(&self) -> bool {
match self.inner.kind {
Kind::BodyUser |
Kind::NewService |
Kind::Service |
Kind::Closed |
Kind::UnsupportedVersion |
Kind::UnsupportedRequestMethod => true,
_ => false,
}
}
/// Returns true if this was about a `Request` that was canceled.
pub fn is_canceled(&self) -> bool {
self.inner.kind == Kind::Canceled
}
/// Returns true if a sender's channel is closed.
pub fn is_closed(&self) -> bool {
self.inner.kind == Kind::Closed
}
pub(crate) fn new(kind: Kind, cause: Option<Cause>) -> Error {
Error {
inner: Box::new(ErrorImpl {
kind,
cause,
}),
}
}
pub(crate) fn kind(&self) -> &Kind {
&self.inner.kind
}
pub(crate) fn new_canceled<E: Into<Cause>>(cause: Option<E>) -> Error {
Error::new(Kind::Canceled, cause.map(Into::into))
}
pub(crate) fn new_upgrade() -> Error {
Error::new(Kind::Upgrade, None)
}
pub(crate) fn new_incomplete() -> Error {
Error::new(Kind::Incomplete, None)
}
pub(crate) fn new_too_large() -> Error {
Error::new(Kind::Parse(Parse::TooLarge), None)
}
pub(crate) fn new_header() -> Error {
Error::new(Kind::Parse(Parse::Header), None)
}
pub(crate) fn new_status() -> Error {
Error::new(Kind::Parse(Parse::Status), None)
}
pub(crate) fn new_version() -> Error {
Error::new(Kind::Parse(Parse::Version), None)
}
pub(crate) fn new_mismatched_response() -> Error {
Error::new(Kind::MismatchedResponse, None)
}
pub(crate) fn new_io(cause: io::Error) -> Error {
Error::new(Kind::Io, Some(cause.into()))
}
pub(crate) fn new_listen(err: io::Error) -> Error {
Error::new(Kind::Listen, Some(err.into()))
}
pub(crate) fn new_accept(err: io::Error) -> Error {
Error::new(Kind::Accept, Some(Box::new(err)))
}
pub(crate) fn new_connect<E: Into<Cause>>(cause: E) -> Error {
Error::new(Kind::Connect, Some(cause.into()))
}
pub(crate) fn new_closed() -> Error {
Error::new(Kind::Closed, None)
}
pub(crate) fn new_body<E: Into<Cause>>(cause: E) -> Error {
Error::new(Kind::Body, Some(cause.into()))
}
pub(crate) fn new_body_write(cause: io::Error) -> Error {
Error::new(Kind::BodyWrite, Some(Box::new(cause)))
}
pub(crate) fn new_user_unsupported_version() -> Error {
Error::new(Kind::UnsupportedVersion, None)
}
pub(crate) fn new_user_unsupported_request_method() -> Error {
Error::new(Kind::UnsupportedRequestMethod, None)
}
pub(crate) fn new_user_new_service(err: io::Error) -> Error {
Error::new(Kind::NewService, Some(Box::new(err)))
}
pub(crate) fn new_user_service<E: Into<Cause>>(cause: E) -> Error {
Error::new(Kind::Service, Some(cause.into()))
}
pub(crate) fn new_user_body<E: Into<Cause>>(cause: E) -> Error {
Error::new(Kind::BodyUser, Some(cause.into()))
}
pub(crate) fn new_shutdown(cause: io::Error) -> Error {
Error::new(Kind::Shutdown, Some(Box::new(cause)))
} }
} }
/// A pending item was dropped before ever being processed. impl fmt::Debug for Error {
///
/// For example, a `Request` could be queued in the `Client`, *just*
/// as the related connection gets closed by the remote. In that case,
/// when the connection drops, the pending response future will be
/// fulfilled with this error, signaling the `Request` was never started.
#[derive(Debug)]
pub struct Canceled {
cause: Option<Box<StdError + Send + Sync>>,
}
impl Canceled {
fn description(&self) -> &str {
"an operation was canceled internally before starting"
}
}
impl fmt::Display for Canceled {
fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result { fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
f.pad(self.description()) f.debug_struct("Error")
} .field("kind", &self.inner.kind)
} .field("cause", &self.inner.cause)
.finish()
#[doc(hidden)]
pub struct Void(());
impl fmt::Debug for Void {
fn fmt(&self, _: &mut fmt::Formatter) -> fmt::Result {
unreachable!()
} }
} }
impl fmt::Display for Error { impl fmt::Display for Error {
fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result { fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
match *self { if let Some(ref cause) = self.inner.cause {
Io(ref e) => fmt::Display::fmt(e, f), write!(f, "{}: {}", self.description(), cause)
Utf8(ref e) => fmt::Display::fmt(e, f), } else {
ref e => f.write_str(e.description()), f.write_str(self.description())
} }
} }
} }
impl StdError for Error { impl StdError for Error {
fn description(&self) -> &str { fn description(&self) -> &str {
match *self { match self.inner.kind {
Method => "invalid Method specified", Kind::Parse(Parse::Method) => "invalid Method specified",
Version => "invalid HTTP version specified", Kind::Parse(Parse::Version) => "invalid HTTP version specified",
Uri => "invalid URI", Kind::Parse(Parse::Uri) => "invalid URI",
Header => "invalid Header provided", Kind::Parse(Parse::Header) => "invalid Header provided",
TooLarge => "message head is too large", Kind::Parse(Parse::TooLarge) => "message head is too large",
Status => "invalid Status provided", Kind::Parse(Parse::Status) => "invalid Status provided",
Incomplete => "message is incomplete", Kind::Incomplete => "message is incomplete",
Timeout => "timeout", Kind::Upgrade => "unsupported protocol upgrade",
Upgrade => "unsupported protocol upgrade", Kind::MismatchedResponse => "response received without matching request",
Closed => "connection is closed", Kind::Closed => "connection closed",
Cancel(ref e) => e.description(), Kind::Connect => "an error occurred trying to connect",
Io(ref e) => e.description(), Kind::Canceled => "an operation was canceled internally before starting",
Utf8(ref e) => e.description(), Kind::Listen => "error creating server listener",
Error::__Nonexhaustive(..) => unreachable!(), Kind::Accept => "error accepting connection",
Kind::NewService => "calling user's new_service failed",
Kind::Service => "error from user's server service",
Kind::Body => "error reading a body from connection",
Kind::BodyWrite => "error write a body to connection",
Kind::BodyUser => "error from user's Entity stream",
Kind::Shutdown => "error shutting down connection",
Kind::UnsupportedVersion => "request has unsupported HTTP version",
Kind::UnsupportedRequestMethod => "request has unsupported HTTP method",
Kind::Io => "an IO error occurred",
} }
} }
fn cause(&self) -> Option<&StdError> { fn cause(&self) -> Option<&StdError> {
match *self { self
Io(ref error) => Some(error), .inner
Utf8(ref error) => Some(error), .cause
Cancel(ref e) => e.cause.as_ref().map(|e| &**e as &StdError), .as_ref()
Error::__Nonexhaustive(..) => unreachable!(), .map(|cause| &**cause as &StdError)
_ => None,
}
} }
} }
impl From<IoError> for Error { #[doc(hidden)]
fn from(err: IoError) -> Error { impl From<Parse> for Error {
Io(err) fn from(err: Parse) -> Error {
Error::new(Kind::Parse(err), None)
} }
} }
impl From<Utf8Error> for Error { impl From<httparse::Error> for Parse {
fn from(err: Utf8Error) -> Error { fn from(err: httparse::Error) -> Parse {
Utf8(err)
}
}
impl From<FromUtf8Error> for Error {
fn from(err: FromUtf8Error) -> Error {
Utf8(err.utf8_error())
}
}
impl From<httparse::Error> for Error {
fn from(err: httparse::Error) -> Error {
match err { match err {
httparse::Error::HeaderName | httparse::Error::HeaderName |
httparse::Error::HeaderValue | httparse::Error::HeaderValue |
httparse::Error::NewLine | httparse::Error::NewLine |
httparse::Error::Token => Header, httparse::Error::Token => Parse::Header,
httparse::Error::Status => Status, httparse::Error::Status => Parse::Status,
httparse::Error::TooManyHeaders => TooLarge, httparse::Error::TooManyHeaders => Parse::TooLarge,
httparse::Error::Version => Version, httparse::Error::Version => Parse::Version,
} }
} }
} }
impl From<http::method::InvalidMethod> for Error { impl From<http::method::InvalidMethod> for Parse {
fn from(_: http::method::InvalidMethod) -> Error { fn from(_: http::method::InvalidMethod) -> Parse {
Error::Method Parse::Method
} }
} }
impl From<http::uri::InvalidUriBytes> for Error { impl From<http::status::InvalidStatusCode> for Parse {
fn from(_: http::uri::InvalidUriBytes) -> Error { fn from(_: http::status::InvalidStatusCode) -> Parse {
Error::Uri Parse::Status
}
}
impl From<http::uri::InvalidUriBytes> for Parse {
fn from(_: http::uri::InvalidUriBytes) -> Parse {
Parse::Uri
} }
} }
@@ -191,58 +319,3 @@ trait AssertSendSync: Send + Sync + 'static {}
#[doc(hidden)] #[doc(hidden)]
impl AssertSendSync for Error {} impl AssertSendSync for Error {}
#[cfg(test)]
mod tests {
use std::error::Error as StdError;
use std::io;
use httparse;
use super::Error;
use super::Error::*;
#[test]
fn test_cause() {
let orig = io::Error::new(io::ErrorKind::Other, "other");
let desc = orig.description().to_owned();
let e = Io(orig);
assert_eq!(e.cause().unwrap().description(), desc);
}
macro_rules! from {
($from:expr => $error:pat) => {
match Error::from($from) {
e @ $error => {
assert!(e.description().len() >= 5);
} ,
e => panic!("{:?}", e)
}
}
}
macro_rules! from_and_cause {
($from:expr => $error:pat) => {
match Error::from($from) {
e @ $error => {
let desc = e.cause().unwrap().description();
assert_eq!(desc, $from.description().to_owned());
assert_eq!(desc, e.description());
},
_ => panic!("{:?}", $from)
}
}
}
#[test]
fn test_from() {
from_and_cause!(io::Error::new(io::ErrorKind::Other, "other") => Io(..));
from!(httparse::Error::HeaderName => Header);
from!(httparse::Error::HeaderName => Header);
from!(httparse::Error::HeaderValue => Header);
from!(httparse::Error::NewLine => Header);
from!(httparse::Error::Status => Status);
from!(httparse::Error::Token => Header);
from!(httparse::Error::TooManyHeaders => TooLarge);
from!(httparse::Error::Version => Version);
}
}

18
src/proto/body.rs
View File

@@ -18,8 +18,7 @@ pub trait Entity {
type Data: AsRef<[u8]>; type Data: AsRef<[u8]>;
/// The error type of this stream. /// The error type of this stream.
//TODO: add bounds Into<::error::User> (or whatever it is called) type Error: Into<Box<::std::error::Error + Send + Sync>>;
type Error;
/// Poll for a `Data` buffer. /// Poll for a `Data` buffer.
/// ///
@@ -141,7 +140,7 @@ enum Kind {
_close_tx: oneshot::Sender<()>, _close_tx: oneshot::Sender<()>,
rx: mpsc::Receiver<Result<Chunk, ::Error>>, rx: mpsc::Receiver<Result<Chunk, ::Error>>,
}, },
Wrapped(Box<Stream<Item=Chunk, Error=::Error> + Send>), Wrapped(Box<Stream<Item=Chunk, Error=Box<::std::error::Error + Send + Sync>> + Send>),
Once(Option<Chunk>), Once(Option<Chunk>),
Empty, Empty,
} }
@@ -212,17 +211,22 @@ impl Body {
/// " ", /// " ",
/// "world", /// "world",
/// ]; /// ];
/// let stream = futures::stream::iter_ok(chunks); ///
/// let stream = futures::stream::iter_ok::<_, ::std::io::Error>(chunks);
/// ///
/// let body = Body::wrap_stream(stream); /// let body = Body::wrap_stream(stream);
/// # } /// # }
/// ``` /// ```
pub fn wrap_stream<S>(stream: S) -> Body pub fn wrap_stream<S>(stream: S) -> Body
where where
S: Stream<Error=::Error> + Send + 'static, S: Stream + Send + 'static,
S::Error: Into<Box<::std::error::Error + Send + Sync>>,
Chunk: From<S::Item>, Chunk: From<S::Item>,
{ {
Body::new(Kind::Wrapped(Box::new(stream.map(Chunk::from)))) let mapped = stream
.map(Chunk::from)
.map_err(Into::into);
Body::new(Kind::Wrapped(Box::new(mapped)))
} }
/// Convert this `Body` into a `Stream<Item=Chunk, Error=hyper::Error>`. /// Convert this `Body` into a `Stream<Item=Chunk, Error=hyper::Error>`.
@@ -327,7 +331,7 @@ impl Body {
Async::Ready(None) => Ok(Async::Ready(None)), Async::Ready(None) => Ok(Async::Ready(None)),
Async::NotReady => Ok(Async::NotReady), Async::NotReady => Ok(Async::NotReady),
}, },
Kind::Wrapped(ref mut s) => s.poll(), Kind::Wrapped(ref mut s) => s.poll().map_err(::Error::new_body),
Kind::Once(ref mut val) => Ok(Async::Ready(val.take())), Kind::Once(ref mut val) => Ok(Async::Ready(val.take())),
Kind::Empty => Ok(Async::Ready(None)), Kind::Empty => Ok(Async::Ready(None)),
} }

7
src/proto/h1/conn.rs
View File

@@ -20,7 +20,7 @@ use super::{EncodedBuf, Encoder, Decoder};
/// The connection will determine when a message begins and ends as well as /// The connection will determine when a message begins and ends as well as
/// determine if this connection can be kept alive after the message, /// determine if this connection can be kept alive after the message,
/// or if it is complete. /// or if it is complete.
pub struct Conn<I, B, T> { pub(crate) struct Conn<I, B, T> {
io: Buffered<I, EncodedBuf<Cursor<B>>>, io: Buffered<I, EncodedBuf<Cursor<B>>>,
state: State, state: State,
_marker: PhantomData<T> _marker: PhantomData<T>
@@ -146,7 +146,8 @@ where I: AsyncRead + AsyncWrite,
_ => { _ => {
error!("unimplemented HTTP Version = {:?}", version); error!("unimplemented HTTP Version = {:?}", version);
self.state.close_read(); self.state.close_read();
return Err(::Error::Version); //TODO: replace this with a more descriptive error
return Err(::Error::new_version());
} }
}; };
self.state.version = version; self.state.version = version;
@@ -245,7 +246,7 @@ where I: AsyncRead + AsyncWrite,
if self.is_mid_message() { if self.is_mid_message() {
self.maybe_park_read(); self.maybe_park_read();
} else { } else {
self.require_empty_read()?; self.require_empty_read().map_err(::Error::new_io)?;
} }
Ok(()) Ok(())
} }

44
src/proto/h1/dispatch.rs
View File

@@ -1,5 +1,3 @@
use std::io;
use bytes::Bytes; use bytes::Bytes;
use futures::{Async, Future, Poll, Stream}; use futures::{Async, Future, Poll, Stream};
use http::{Request, Response, StatusCode}; use http::{Request, Response, StatusCode};
@@ -9,7 +7,7 @@ use tokio_service::Service;
use proto::body::Entity; use proto::body::Entity;
use proto::{Body, BodyLength, Conn, Http1Transaction, MessageHead, RequestHead, RequestLine, ResponseHead}; use proto::{Body, BodyLength, Conn, Http1Transaction, MessageHead, RequestHead, RequestLine, ResponseHead};
pub struct Dispatcher<D, Bs, I, B, T> { pub(crate) struct Dispatcher<D, Bs, I, B, T> {
conn: Conn<I, B, T>, conn: Conn<I, B, T>,
dispatch: D, dispatch: D,
body_tx: Option<::proto::body::Sender>, body_tx: Option<::proto::body::Sender>,
@@ -17,7 +15,7 @@ pub struct Dispatcher<D, Bs, I, B, T> {
is_closing: bool, is_closing: bool,
} }
pub trait Dispatch { pub(crate) trait Dispatch {
type PollItem; type PollItem;
type PollBody; type PollBody;
type RecvItem; type RecvItem;
@@ -47,7 +45,7 @@ where
I: AsyncRead + AsyncWrite, I: AsyncRead + AsyncWrite,
B: AsRef<[u8]>, B: AsRef<[u8]>,
T: Http1Transaction, T: Http1Transaction,
Bs: Entity<Data=B, Error=::Error>, Bs: Entity<Data=B>,
{ {
pub fn new(dispatch: D, conn: Conn<I, B, T>) -> Self { pub fn new(dispatch: D, conn: Conn<I, B, T>) -> Self {
Dispatcher { Dispatcher {
@@ -98,7 +96,7 @@ where
if self.is_done() { if self.is_done() {
if should_shutdown { if should_shutdown {
try_ready!(self.conn.shutdown()); try_ready!(self.conn.shutdown().map_err(::Error::new_shutdown));
} }
self.conn.take_error()?; self.conn.take_error()?;
Ok(Async::Ready(())) Ok(Async::Ready(()))
@@ -152,7 +150,7 @@ where
return Ok(Async::NotReady); return Ok(Async::NotReady);
} }
Err(e) => { Err(e) => {
body.send_error(::Error::Io(e)); body.send_error(::Error::new_body(e));
} }
} }
} else { } else {
@@ -225,14 +223,14 @@ where
} else if !self.conn.can_buffer_body() { } else if !self.conn.can_buffer_body() {
try_ready!(self.poll_flush()); try_ready!(self.poll_flush());
} else if let Some(mut body) = self.body_rx.take() { } else if let Some(mut body) = self.body_rx.take() {
let chunk = match body.poll_data()? { let chunk = match body.poll_data().map_err(::Error::new_user_body)? {
Async::Ready(Some(chunk)) => { Async::Ready(Some(chunk)) => {
self.body_rx = Some(body); self.body_rx = Some(body);
chunk chunk
}, },
Async::Ready(None) => { Async::Ready(None) => {
if self.conn.can_write_body() { if self.conn.can_write_body() {
self.conn.write_body(None)?; self.conn.write_body(None).map_err(::Error::new_body_write)?;
} }
continue; continue;
}, },
@@ -243,7 +241,7 @@ where
}; };
if self.conn.can_write_body() { if self.conn.can_write_body() {
assert!(self.conn.write_body(Some(chunk))?.is_ready()); self.conn.write_body(Some(chunk)).map_err(::Error::new_body_write)?;
// This allows when chunk is `None`, or `Some([])`. // This allows when chunk is `None`, or `Some([])`.
} else if chunk.as_ref().len() == 0 { } else if chunk.as_ref().len() == 0 {
// ok // ok
@@ -259,7 +257,7 @@ where
fn poll_flush(&mut self) -> Poll<(), ::Error> { fn poll_flush(&mut self) -> Poll<(), ::Error> {
self.conn.flush().map_err(|err| { self.conn.flush().map_err(|err| {
debug!("error writing: {}", err); debug!("error writing: {}", err);
err.into() ::Error::new_body_write(err)
}) })
} }
@@ -294,7 +292,7 @@ where
I: AsyncRead + AsyncWrite, I: AsyncRead + AsyncWrite,
B: AsRef<[u8]>, B: AsRef<[u8]>,
T: Http1Transaction, T: Http1Transaction,
Bs: Entity<Data=B, Error=::Error>, Bs: Entity<Data=B>,
{ {
type Item = (); type Item = ();
type Error = ::Error; type Error = ::Error;
@@ -318,8 +316,9 @@ impl<S> Server<S> where S: Service {
impl<S, Bs> Dispatch for Server<S> impl<S, Bs> Dispatch for Server<S>
where where
S: Service<Request=Request<Body>, Response=Response<Bs>, Error=::Error>, S: Service<Request=Request<Body>, Response=Response<Bs>>,
Bs: Entity<Error=::Error>, S::Error: Into<Box<::std::error::Error + Send + Sync>>,
Bs: Entity,
{ {
type PollItem = MessageHead<StatusCode>; type PollItem = MessageHead<StatusCode>;
type PollBody = Bs; type PollBody = Bs;
@@ -327,7 +326,7 @@ where
fn poll_msg(&mut self) -> Poll<Option<(Self::PollItem, Option<Self::PollBody>)>, ::Error> { fn poll_msg(&mut self) -> Poll<Option<(Self::PollItem, Option<Self::PollBody>)>, ::Error> {
if let Some(mut fut) = self.in_flight.take() { if let Some(mut fut) = self.in_flight.take() {
let resp = match fut.poll()? { let resp = match fut.poll().map_err(::Error::new_user_service)? {
Async::Ready(res) => res, Async::Ready(res) => res,
Async::NotReady => { Async::NotReady => {
self.in_flight = Some(fut); self.in_flight = Some(fut);
@@ -389,7 +388,7 @@ impl<B> Client<B> {
impl<B> Dispatch for Client<B> impl<B> Dispatch for Client<B>
where where
B: Entity<Error=::Error>, B: Entity,
{ {
type PollItem = RequestHead; type PollItem = RequestHead;
type PollBody = B; type PollBody = B;
@@ -443,7 +442,7 @@ where
let _ = cb.send(Ok(res)); let _ = cb.send(Ok(res));
Ok(()) Ok(())
} else { } else {
Err(::Error::Io(io::Error::new(io::ErrorKind::InvalidData, "response received without matching request"))) Err(::Error::new_mismatched_response())
} }
}, },
Err(err) => { Err(err) => {
@@ -507,8 +506,15 @@ mod tests {
.expect("callback poll") .expect("callback poll")
.expect_err("callback response"); .expect_err("callback response");
match err { /*
(::Error::Cancel(_), Some(_)) => (), let err = match async {
Async::Ready(result) => result.unwrap_err(),
Async::Pending => panic!("callback should be ready"),
};
*/
match (err.0.kind(), err.1) {
(&::error::Kind::Canceled, Some(_)) => (),
other => panic!("expected Canceled, got {:?}", other), other => panic!("expected Canceled, got {:?}", other),
} }
Ok::<(), ()>(()) Ok::<(), ()>(())

8
src/proto/h1/io.rs
View File

@@ -108,7 +108,7 @@ where
} }
} }
pub fn parse<S: Http1Transaction>(&mut self) -> Poll<MessageHead<S::Incoming>, ::Error> { pub(super) fn parse<S: Http1Transaction>(&mut self) -> Poll<MessageHead<S::Incoming>, ::Error> {
loop { loop {
match try!(S::parse(&mut self.read_buf)) { match try!(S::parse(&mut self.read_buf)) {
Some((head, len)) => { Some((head, len)) => {
@@ -118,14 +118,14 @@ where
None => { None => {
if self.read_buf.capacity() >= self.max_buf_size { if self.read_buf.capacity() >= self.max_buf_size {
debug!("max_buf_size ({}) reached, closing", self.max_buf_size); debug!("max_buf_size ({}) reached, closing", self.max_buf_size);
return Err(::Error::TooLarge); return Err(::Error::new_too_large());
} }
}, },
} }
match try_ready!(self.read_from_io()) { match try_ready!(self.read_from_io().map_err(::Error::new_io)) {
0 => { 0 => {
trace!("parse eof"); trace!("parse eof");
return Err(::Error::Incomplete); return Err(::Error::new_incomplete());
} }
_ => {}, _ => {},
} }

4
src/proto/h1/mod.rs
View File

@@ -1,11 +1,11 @@
pub use self::conn::Conn; pub(crate) use self::conn::Conn;
pub use self::decode::Decoder; pub use self::decode::Decoder;
pub use self::encode::{EncodedBuf, Encoder}; pub use self::encode::{EncodedBuf, Encoder};
mod conn; mod conn;
mod date; mod date;
mod decode; mod decode;
pub mod dispatch; pub(crate) mod dispatch;
mod encode; mod encode;
mod io; mod io;
pub mod role; pub mod role;

35
src/proto/h1/role.rs
View File

@@ -40,7 +40,7 @@ where
let mut headers = [httparse::EMPTY_HEADER; MAX_HEADERS]; let mut headers = [httparse::EMPTY_HEADER; MAX_HEADERS];
trace!("Request.parse([Header; {}], [u8; {}])", headers.len(), buf.len()); trace!("Request.parse([Header; {}], [u8; {}])", headers.len(), buf.len());
let mut req = httparse::Request::new(&mut headers); let mut req = httparse::Request::new(&mut headers);
match try!(req.parse(&buf)) { match req.parse(&buf)? {
httparse::Status::Complete(len) => { httparse::Status::Complete(len) => {
trace!("Request.parse Complete({})", len); trace!("Request.parse Complete({})", len);
let method = Method::from_bytes(req.method.unwrap().as_bytes())?; let method = Method::from_bytes(req.method.unwrap().as_bytes())?;
@@ -104,18 +104,18 @@ where
// mal-formed. A server should respond with 400 Bad Request. // mal-formed. A server should respond with 400 Bad Request.
if head.version == Version::HTTP_10 { if head.version == Version::HTTP_10 {
debug!("HTTP/1.0 cannot have Transfer-Encoding header"); debug!("HTTP/1.0 cannot have Transfer-Encoding header");
Err(::Error::Header) Err(::Error::new_header())
} else if headers::transfer_encoding_is_chunked(&head.headers) { } else if headers::transfer_encoding_is_chunked(&head.headers) {
Ok(Decode::Normal(Decoder::chunked())) Ok(Decode::Normal(Decoder::chunked()))
} else { } else {
debug!("request with transfer-encoding header, but not chunked, bad request"); debug!("request with transfer-encoding header, but not chunked, bad request");
Err(::Error::Header) Err(::Error::new_header())
} }
} else if let Some(len) = headers::content_length_parse(&head.headers) { } else if let Some(len) = headers::content_length_parse(&head.headers) {
Ok(Decode::Normal(Decoder::length(len))) Ok(Decode::Normal(Decoder::length(len)))
} else if head.headers.contains_key(CONTENT_LENGTH) { } else if head.headers.contains_key(CONTENT_LENGTH) {
debug!("illegal Content-Length header"); debug!("illegal Content-Length header");
Err(::Error::Header) Err(::Error::new_header())
} else { } else {
Ok(Decode::Normal(Decoder::length(0))) Ok(Decode::Normal(Decoder::length(0)))
} }
@@ -146,7 +146,8 @@ where
head = MessageHead::default(); head = MessageHead::default();
head.subject = StatusCode::INTERNAL_SERVER_ERROR; head.subject = StatusCode::INTERNAL_SERVER_ERROR;
headers::content_length_zero(&mut head.headers); headers::content_length_zero(&mut head.headers);
Err(::Error::Status) //TODO: change this to a more descriptive error than just a parse error
Err(::Error::new_status())
} else { } else {
Ok(Server::set_length(&mut head, body, method.as_ref())) Ok(Server::set_length(&mut head, body, method.as_ref()))
}; };
@@ -184,14 +185,15 @@ where
} }
fn on_error(err: &::Error) -> Option<MessageHead<Self::Outgoing>> { fn on_error(err: &::Error) -> Option<MessageHead<Self::Outgoing>> {
let status = match err { use ::error::{Kind, Parse};
&::Error::Method | let status = match *err.kind() {
&::Error::Version | Kind::Parse(Parse::Method) |
&::Error::Header /*| Kind::Parse(Parse::Version) |
&::Error::Uri(_)*/ => { Kind::Parse(Parse::Header) |
Kind::Parse(Parse::Uri) => {
StatusCode::BAD_REQUEST StatusCode::BAD_REQUEST
}, },
&::Error::TooLarge => { Kind::Parse(Parse::TooLarge) => {
StatusCode::REQUEST_HEADER_FIELDS_TOO_LARGE StatusCode::REQUEST_HEADER_FIELDS_TOO_LARGE
} }
_ => return None, _ => return None,
@@ -271,7 +273,7 @@ where
match try!(res.parse(bytes)) { match try!(res.parse(bytes)) {
httparse::Status::Complete(len) => { httparse::Status::Complete(len) => {
trace!("Response.parse Complete({})", len); trace!("Response.parse Complete({})", len);
let status = try!(StatusCode::from_u16(res.code.unwrap()).map_err(|_| ::Error::Status)); let status = StatusCode::from_u16(res.code.unwrap())?;
let version = if res.version.unwrap() == 1 { let version = if res.version.unwrap() == 1 {
Version::HTTP_11 Version::HTTP_11
} else { } else {
@@ -343,7 +345,7 @@ where
// mal-formed. A server should respond with 400 Bad Request. // mal-formed. A server should respond with 400 Bad Request.
if inc.version == Version::HTTP_10 { if inc.version == Version::HTTP_10 {
debug!("HTTP/1.0 cannot have Transfer-Encoding header"); debug!("HTTP/1.0 cannot have Transfer-Encoding header");
Err(::Error::Header) Err(::Error::new_header())
} else if headers::transfer_encoding_is_chunked(&inc.headers) { } else if headers::transfer_encoding_is_chunked(&inc.headers) {
Ok(Decode::Normal(Decoder::chunked())) Ok(Decode::Normal(Decoder::chunked()))
} else { } else {
@@ -354,7 +356,7 @@ where
Ok(Decode::Normal(Decoder::length(len))) Ok(Decode::Normal(Decoder::length(len)))
} else if inc.headers.contains_key(CONTENT_LENGTH) { } else if inc.headers.contains_key(CONTENT_LENGTH) {
debug!("illegal Content-Length header"); debug!("illegal Content-Length header");
Err(::Error::Header) Err(::Error::new_header())
} else { } else {
trace!("neither Transfer-Encoding nor Content-Length"); trace!("neither Transfer-Encoding nor Content-Length");
Ok(Decode::Normal(Decoder::eof())) Ok(Decode::Normal(Decoder::eof()))
@@ -577,12 +579,13 @@ impl OnUpgrade for NoUpgrades {
*head = MessageHead::default(); *head = MessageHead::default();
head.subject = ::StatusCode::INTERNAL_SERVER_ERROR; head.subject = ::StatusCode::INTERNAL_SERVER_ERROR;
headers::content_length_zero(&mut head.headers); headers::content_length_zero(&mut head.headers);
Err(::Error::Status) //TODO: replace with more descriptive error
return Err(::Error::new_status());
} }
fn on_decode_upgrade() -> ::Result<Decoder> { fn on_decode_upgrade() -> ::Result<Decoder> {
debug!("received 101 upgrade response, not supported"); debug!("received 101 upgrade response, not supported");
return Err(::Error::Upgrade); return Err(::Error::new_upgrade());
} }
} }

12
src/proto/mod.rs
View File

@@ -6,7 +6,7 @@ use headers;
pub use self::body::Body; pub use self::body::Body;
pub use self::chunk::Chunk; pub use self::chunk::Chunk;
pub use self::h1::{dispatch, Conn}; pub(crate) use self::h1::{dispatch, Conn};
pub mod body; pub mod body;
mod chunk; mod chunk;
@@ -60,14 +60,14 @@ pub fn expecting_continue(version: Version, headers: &HeaderMap) -> bool {
version == Version::HTTP_11 && headers::expect_continue(headers) version == Version::HTTP_11 && headers::expect_continue(headers)
} }
pub type ServerTransaction = h1::role::Server<h1::role::YesUpgrades>; pub(crate) type ServerTransaction = h1::role::Server<h1::role::YesUpgrades>;
//pub type ServerTransaction = h1::role::Server<h1::role::NoUpgrades>; //pub type ServerTransaction = h1::role::Server<h1::role::NoUpgrades>;
//pub type ServerUpgradeTransaction = h1::role::Server<h1::role::YesUpgrades>; //pub type ServerUpgradeTransaction = h1::role::Server<h1::role::YesUpgrades>;
pub type ClientTransaction = h1::role::Client<h1::role::NoUpgrades>; pub(crate) type ClientTransaction = h1::role::Client<h1::role::NoUpgrades>;
pub type ClientUpgradeTransaction = h1::role::Client<h1::role::YesUpgrades>; pub(crate) type ClientUpgradeTransaction = h1::role::Client<h1::role::YesUpgrades>;
pub trait Http1Transaction { pub(crate) trait Http1Transaction {
type Incoming; type Incoming;
type Outgoing: Default; type Outgoing: Default;
fn parse(bytes: &mut BytesMut) -> ParseResult<Self::Incoming>; fn parse(bytes: &mut BytesMut) -> ParseResult<Self::Incoming>;
@@ -84,7 +84,7 @@ pub trait Http1Transaction {
fn should_read_first() -> bool; fn should_read_first() -> bool;
} }
pub type ParseResult<T> = ::Result<Option<(MessageHead<T>, usize)>>; pub(crate) type ParseResult<T> = Result<Option<(MessageHead<T>, usize)>, ::error::Parse>;
#[derive(Debug)] #[derive(Debug)]
pub enum BodyLength { pub enum BodyLength {

16
src/server/conn.rs
View File

@@ -25,7 +25,7 @@ use super::{HyperService, Request, Response, Service};
pub struct Connection<I, S> pub struct Connection<I, S>
where where
S: HyperService, S: HyperService,
S::ResponseBody: Entity<Error=::Error>, S::ResponseBody: Entity,
{ {
pub(super) conn: proto::dispatch::Dispatcher< pub(super) conn: proto::dispatch::Dispatcher<
proto::dispatch::Server<S>, proto::dispatch::Server<S>,
@@ -59,9 +59,11 @@ pub struct Parts<T> {
// ===== impl Connection ===== // ===== impl Connection =====
impl<I, B, S> Connection<I, S> impl<I, B, S> Connection<I, S>
where S: Service<Request = Request<Body>, Response = Response<B>, Error = ::Error> + 'static, where
S: Service<Request=Request<Body>, Response=Response<B>> + 'static,
S::Error: Into<Box<::std::error::Error + Send + Sync>>,
I: AsyncRead + AsyncWrite + 'static, I: AsyncRead + AsyncWrite + 'static,
B: Entity<Error=::Error> + 'static, B: Entity + 'static,
{ {
/// Disables keep-alive for this connection. /// Disables keep-alive for this connection.
pub fn disable_keep_alive(&mut self) { pub fn disable_keep_alive(&mut self) {
@@ -96,9 +98,11 @@ where S: Service<Request = Request<Body>, Response = Response<B>, Error = ::Erro
} }
impl<I, B, S> Future for Connection<I, S> impl<I, B, S> Future for Connection<I, S>
where S: Service<Request = Request<Body>, Response = Response<B>, Error = ::Error> + 'static, where
S: Service<Request=Request<Body>, Response=Response<B>> + 'static,
S::Error: Into<Box<::std::error::Error + Send + Sync>>,
I: AsyncRead + AsyncWrite + 'static, I: AsyncRead + AsyncWrite + 'static,
B: Entity<Error=::Error> + 'static, B: Entity + 'static,
{ {
type Item = (); type Item = ();
type Error = ::Error; type Error = ::Error;
@@ -111,7 +115,7 @@ where S: Service<Request = Request<Body>, Response = Response<B>, Error = ::Erro
impl<I, S> fmt::Debug for Connection<I, S> impl<I, S> fmt::Debug for Connection<I, S>
where where
S: HyperService, S: HyperService,
S::ResponseBody: Entity<Error=::Error>, S::ResponseBody: Entity,
{ {
fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result { fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
f.debug_struct("Connection") f.debug_struct("Connection")

94
src/server/mod.rs
View File

@@ -51,7 +51,7 @@ pub struct Http<B = ::Chunk> {
/// address and then serving TCP connections accepted with the service provided. /// address and then serving TCP connections accepted with the service provided.
pub struct Server<S, B> pub struct Server<S, B>
where where
B: Entity<Error=::Error>, B: Entity,
{ {
protocol: Http<B::Data>, protocol: Http<B::Data>,
new_service: S, new_service: S,
@@ -165,12 +165,14 @@ impl<B: AsRef<[u8]> + 'static> Http<B> {
/// The returned `Server` contains one method, `run`, which is used to /// The returned `Server` contains one method, `run`, which is used to
/// actually run the server. /// actually run the server.
pub fn bind<S, Bd>(&self, addr: &SocketAddr, new_service: S) -> ::Result<Server<S, Bd>> pub fn bind<S, Bd>(&self, addr: &SocketAddr, new_service: S) -> ::Result<Server<S, Bd>>
where S: NewService<Request = Request<Body>, Response = Response<Bd>, Error = ::Error> + 'static, where
Bd: Entity<Data=B, Error=::Error>, S: NewService<Request=Request<Body>, Response=Response<Bd>> + 'static,
S::Error: Into<Box<::std::error::Error + Send + Sync>>,
Bd: Entity<Data=B>,
{ {
let handle = Handle::current(); let handle = Handle::current();
let std_listener = StdTcpListener::bind(addr)?; let std_listener = StdTcpListener::bind(addr).map_err(::Error::new_listen)?;
let listener = try!(TcpListener::from_std(std_listener, &handle)); let listener = TcpListener::from_std(std_listener, &handle).map_err(::Error::new_listen)?;
Ok(Server { Ok(Server {
new_service: new_service, new_service: new_service,
@@ -188,13 +190,15 @@ impl<B: AsRef<[u8]> + 'static> Http<B> {
/// `new_service` object provided as well, creating a new service per /// `new_service` object provided as well, creating a new service per
/// connection. /// connection.
pub fn serve_addr<S, Bd>(&self, addr: &SocketAddr, new_service: S) -> ::Result<Serve<AddrIncoming, S>> pub fn serve_addr<S, Bd>(&self, addr: &SocketAddr, new_service: S) -> ::Result<Serve<AddrIncoming, S>>
where S: NewService<Request = Request<Body>, Response = Response<Bd>, Error = ::Error>, where
Bd: Entity<Data=B, Error=::Error>, S: NewService<Request=Request<Body>, Response=Response<Bd>>,
S::Error: Into<Box<::std::error::Error + Send + Sync>>,
Bd: Entity<Data=B>,
{ {
let handle = Handle::current(); let handle = Handle::current();
let std_listener = StdTcpListener::bind(addr)?; let std_listener = StdTcpListener::bind(addr).map_err(::Error::new_listen)?;
let listener = TcpListener::from_std(std_listener, &handle)?; let listener = TcpListener::from_std(std_listener, &handle).map_err(::Error::new_listen)?;
let mut incoming = AddrIncoming::new(listener, handle.clone(), self.sleep_on_errors)?; let mut incoming = AddrIncoming::new(listener, handle.clone(), self.sleep_on_errors).map_err(::Error::new_listen)?;
if self.keep_alive { if self.keep_alive {
incoming.set_keepalive(Some(Duration::from_secs(90))); incoming.set_keepalive(Some(Duration::from_secs(90)));
} }
@@ -210,12 +214,15 @@ impl<B: AsRef<[u8]> + 'static> Http<B> {
/// `new_service` object provided as well, creating a new service per /// `new_service` object provided as well, creating a new service per
/// connection. /// connection.
pub fn serve_addr_handle<S, Bd>(&self, addr: &SocketAddr, handle: &Handle, new_service: S) -> ::Result<Serve<AddrIncoming, S>> pub fn serve_addr_handle<S, Bd>(&self, addr: &SocketAddr, handle: &Handle, new_service: S) -> ::Result<Serve<AddrIncoming, S>>
where S: NewService<Request = Request<Body>, Response = Response<Bd>, Error = ::Error>, where
Bd: Entity<Data=B, Error=::Error>, S: NewService<Request = Request<Body>, Response = Response<Bd>>,
S::Error: Into<Box<::std::error::Error + Send + Sync>>,
Bd: Entity<Data=B>,
{ {
let std_listener = StdTcpListener::bind(addr)?; let std_listener = StdTcpListener::bind(addr).map_err(::Error::new_listen)?;
let listener = TcpListener::from_std(std_listener, &handle)?; let listener = TcpListener::from_std(std_listener, &handle).map_err(::Error::new_listen)?;
let mut incoming = AddrIncoming::new(listener, handle.clone(), self.sleep_on_errors)?; let mut incoming = AddrIncoming::new(listener, handle.clone(), self.sleep_on_errors).map_err(::Error::new_listen)?;
if self.keep_alive { if self.keep_alive {
incoming.set_keepalive(Some(Duration::from_secs(90))); incoming.set_keepalive(Some(Duration::from_secs(90)));
} }
@@ -226,10 +233,12 @@ impl<B: AsRef<[u8]> + 'static> Http<B> {
/// ///
/// This method allows the ability to share a `Core` with multiple servers. /// This method allows the ability to share a `Core` with multiple servers.
pub fn serve_incoming<I, S, Bd>(&self, incoming: I, new_service: S) -> Serve<I, S> pub fn serve_incoming<I, S, Bd>(&self, incoming: I, new_service: S) -> Serve<I, S>
where I: Stream<Error=::std::io::Error>, where
I: Stream<Error=::std::io::Error>,
I::Item: AsyncRead + AsyncWrite, I::Item: AsyncRead + AsyncWrite,
S: NewService<Request = Request<Body>, Response = Response<Bd>, Error = ::Error>, S: NewService<Request = Request<Body>, Response = Response<Bd>>,
Bd: Entity<Data=B, Error=::Error>, S::Error: Into<Box<::std::error::Error + Send + Sync>>,
Bd: Entity<Data=B>,
{ {
Serve { Serve {
incoming: incoming, incoming: incoming,
@@ -279,8 +288,10 @@ impl<B: AsRef<[u8]> + 'static> Http<B> {
/// # fn main() {} /// # fn main() {}
/// ``` /// ```
pub fn serve_connection<S, I, Bd>(&self, io: I, service: S) -> Connection<I, S> pub fn serve_connection<S, I, Bd>(&self, io: I, service: S) -> Connection<I, S>
where S: Service<Request = Request<Body>, Response = Response<Bd>, Error = ::Error>, where
Bd: Entity<Error=::Error>, S: Service<Request = Request<Body>, Response = Response<Bd>>,
S::Error: Into<Box<::std::error::Error + Send + Sync>>,
Bd: Entity,
I: AsyncRead + AsyncWrite, I: AsyncRead + AsyncWrite,
{ {
let mut conn = proto::Conn::new(io); let mut conn = proto::Conn::new(io);
@@ -341,15 +352,19 @@ impl Future for Run {
impl<S, B> Server<S, B> impl<S, B> Server<S, B>
where S: NewService<Request = Request<Body>, Response = Response<B>, Error = ::Error> + Send + 'static, where
S: NewService<Request = Request<Body>, Response = Response<B>> + Send + 'static,
S::Error: Into<Box<::std::error::Error + Send + Sync>>,
<S as NewService>::Instance: Send, <S as NewService>::Instance: Send,
<<S as NewService>::Instance as Service>::Future: Send, <<S as NewService>::Instance as Service>::Future: Send,
B: Entity<Error=::Error> + Send + 'static, B: Entity + Send + 'static,
B::Data: Send, B::Data: Send,
{ {
/// Returns the local address that this server is bound to. /// Returns the local address that this server is bound to.
pub fn local_addr(&self) -> ::Result<SocketAddr> { pub fn local_addr(&self) -> ::Result<SocketAddr> {
Ok(try!(self.listener.local_addr())) //TODO: this shouldn't return an error at all, but should get the
//local_addr at construction
self.listener.local_addr().map_err(::Error::new_io)
} }
/// Configure the amount of time this server will wait for a "graceful /// Configure the amount of time this server will wait for a "graceful
@@ -393,7 +408,7 @@ impl<S, B> Server<S, B>
let mut incoming = match AddrIncoming::new(listener, handle.clone(), protocol.sleep_on_errors) { let mut incoming = match AddrIncoming::new(listener, handle.clone(), protocol.sleep_on_errors) {
Ok(incoming) => incoming, Ok(incoming) => incoming,
Err(err) => return Run(Box::new(future::err(err.into()))), Err(err) => return Run(Box::new(future::err(::Error::new_listen(err)))),
}; };
if protocol.keep_alive { if protocol.keep_alive {
@@ -439,7 +454,7 @@ impl<S, B> Server<S, B>
let main_execution = shutdown_signal.select(srv).then(move |result| { let main_execution = shutdown_signal.select(srv).then(move |result| {
match result { match result {
Ok(((), _incoming)) => {}, Ok(((), _incoming)) => {},
Err((e, _other)) => return future::Either::A(future::err(e.into())) Err((e, _other)) => return future::Either::A(future::err(::Error::new_accept(e))),
} }
// Ok we've stopped accepting new connections at this point, but we want // Ok we've stopped accepting new connections at this point, but we want
@@ -454,7 +469,8 @@ impl<S, B> Server<S, B>
future::Either::B(wait.select(timeout).then(|result| { future::Either::B(wait.select(timeout).then(|result| {
match result { match result {
Ok(_) => Ok(()), Ok(_) => Ok(()),
Err((e, _)) => Err(e.into()) //TODO: error variant should be "timed out waiting for graceful shutdown"
Err((e, _)) => Err(::Error::new_io(e))
} }
})) }))
}); });
@@ -463,11 +479,10 @@ impl<S, B> Server<S, B>
} }
} }
impl<S: fmt::Debug, B: Entity<Error=::Error>> fmt::Debug for Server<S, B> impl<S: fmt::Debug, B: Entity> fmt::Debug for Server<S, B>
{ {
fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result { fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
f.debug_struct("Server") f.debug_struct("Server")
.field("reactor", &"...")
.field("listener", &self.listener) .field("listener", &self.listener)
.field("new_service", &self.new_service) .field("new_service", &self.new_service)
.field("protocol", &self.protocol) .field("protocol", &self.protocol)
@@ -499,15 +514,16 @@ impl<I, S, B> Stream for Serve<I, S>
where where
I: Stream<Error=io::Error>, I: Stream<Error=io::Error>,
I::Item: AsyncRead + AsyncWrite, I::Item: AsyncRead + AsyncWrite,
S: NewService<Request=Request<Body>, Response=Response<B>, Error=::Error>, S: NewService<Request=Request<Body>, Response=Response<B>>,
B: Entity<Error=::Error>, S::Error: Into<Box<::std::error::Error + Send + Sync>>,
B: Entity,
{ {
type Item = Connection<I::Item, S::Instance>; type Item = Connection<I::Item, S::Instance>;
type Error = ::Error; type Error = ::Error;
fn poll(&mut self) -> Poll<Option<Self::Item>, Self::Error> { fn poll(&mut self) -> Poll<Option<Self::Item>, Self::Error> {
if let Some(io) = try_ready!(self.incoming.poll()) { if let Some(io) = try_ready!(self.incoming.poll().map_err(::Error::new_accept)) {
let service = self.new_service.new_service()?; let service = self.new_service.new_service().map_err(::Error::new_user_new_service)?;
Ok(Async::Ready(Some(self.protocol.serve_connection(io, service)))) Ok(Async::Ready(Some(self.protocol.serve_connection(io, service))))
} else { } else {
Ok(Async::Ready(None)) Ok(Async::Ready(None))
@@ -579,6 +595,12 @@ impl AddrIncoming {
fn set_keepalive(&mut self, dur: Option<Duration>) { fn set_keepalive(&mut self, dur: Option<Duration>) {
self.keep_alive_timeout = dur; self.keep_alive_timeout = dur;
} }
/*
fn set_sleep_on_errors(&mut self, val: bool) {
self.sleep_on_errors = val;
}
*/
} }
impl Stream for AddrIncoming { impl Stream for AddrIncoming {
@@ -802,9 +824,9 @@ mod hyper_service {
S: Service< S: Service<
Request=Request<Body>, Request=Request<Body>,
Response=Response<B>, Response=Response<B>,
Error=::Error,
>, >,
B: Entity<Error=::Error>, S::Error: Into<Box<::std::error::Error + Send + Sync>>,
B: Entity,
{} {}
impl<S, B> HyperService for S impl<S, B> HyperService for S
@@ -812,10 +834,10 @@ mod hyper_service {
S: Service< S: Service<
Request=Request<Body>, Request=Request<Body>,
Response=Response<B>, Response=Response<B>,
Error=::Error,
>, >,
S::Error: Into<Box<::std::error::Error + Send + Sync>>,
S: Sealed, S: Sealed,
B: Entity<Error=::Error>, B: Entity,
{ {
type ResponseBody = B; type ResponseBody = B;
type Sealed = Opaque; type Sealed = Opaque;

106
tests/client.rs
View File

@@ -8,7 +8,7 @@ extern crate tokio;
extern crate tokio_io; extern crate tokio_io;
extern crate pretty_env_logger; extern crate pretty_env_logger;
use std::io::{self, Read, Write}; use std::io::{Read, Write};
use std::net::{SocketAddr, TcpListener}; use std::net::{SocketAddr, TcpListener};
use std::thread; use std::thread;
use std::time::Duration; use std::time::Duration;
@@ -142,7 +142,7 @@ macro_rules! test {
let _ = pretty_env_logger::try_init(); let _ = pretty_env_logger::try_init();
let runtime = Runtime::new().expect("runtime new"); let runtime = Runtime::new().expect("runtime new");
let err = test! { let err: ::hyper::Error = test! {
INNER; INNER;
name: $name, name: $name,
runtime: &runtime, runtime: &runtime,
@@ -157,7 +157,11 @@ macro_rules! test {
headers: { $($request_header_name => $request_header_val,)* }, headers: { $($request_header_name => $request_header_val,)* },
body: $request_body, body: $request_body,
}.unwrap_err(); }.unwrap_err();
if !$err(&err) {
fn infer_closure<F: FnOnce(&::hyper::Error) -> bool>(f: F) -> F { f }
let closure = infer_closure($err);
if !closure(&err) {
panic!("expected error, unexpected variant: {:?}", err) panic!("expected error, unexpected variant: {:?}", err)
} }
} }
@@ -228,7 +232,7 @@ macro_rules! test {
let _ = tx.send(()); let _ = tx.send(());
}).expect("thread spawn"); }).expect("thread spawn");
let rx = rx.map_err(|_| hyper::Error::Io(io::Error::new(io::ErrorKind::Other, "thread panicked"))); let rx = rx.expect("thread panicked");
res.join(rx).map(|r| r.0).wait() res.join(rx).map(|r| r.0).wait()
}); });
@@ -485,10 +489,7 @@ test! {
url: "http://{addr}/err", url: "http://{addr}/err",
headers: {}, headers: {},
body: None, body: None,
error: |err| match err { error: |err| err.to_string() == "message is incomplete",
&hyper::Error::Incomplete => true,
_ => false,
},
} }
test! { test! {
@@ -511,10 +512,8 @@ test! {
url: "http://{addr}/err", url: "http://{addr}/err",
headers: {}, headers: {},
body: None, body: None,
error: |err| match err { // should get a Parse(Version) error
&hyper::Error::Version => true, error: |err| err.is_parse(),
_ => false,
},
} }
@@ -574,10 +573,7 @@ test! {
url: "http://{addr}/upgrade", url: "http://{addr}/upgrade",
headers: {}, headers: {},
body: None, body: None,
error: |err| match err { error: |err| err.to_string() == "unsupported protocol upgrade",
&hyper::Error::Upgrade => true,
_ => false,
},
} }
@@ -599,10 +595,7 @@ test! {
url: "http://{addr}/", url: "http://{addr}/",
headers: {}, headers: {},
body: None, body: None,
error: |err| match err { error: |err| err.is_user(),
&hyper::Error::Method => true,
_ => false,
},
} }
@@ -689,9 +682,9 @@ mod dispatch_impl {
let res = client.request(req).and_then(move |res| { let res = client.request(req).and_then(move |res| {
assert_eq!(res.status(), hyper::StatusCode::OK); assert_eq!(res.status(), hyper::StatusCode::OK);
Delay::new(Duration::from_secs(1)) Delay::new(Duration::from_secs(1))
.from_err() .expect("timeout")
}); });
let rx = rx1.map_err(|_| hyper::Error::Io(io::Error::new(io::ErrorKind::Other, "thread panicked"))); let rx = rx1.expect("thread panicked");
res.join(rx).map(|r| r.0).wait().unwrap(); res.join(rx).map(|r| r.0).wait().unwrap();
closes.into_future().wait().unwrap().0.expect("closes"); closes.into_future().wait().unwrap().0.expect("closes");
@@ -736,11 +729,11 @@ mod dispatch_impl {
res.into_body().into_stream().concat2() res.into_body().into_stream().concat2()
}).and_then(|_| { }).and_then(|_| {
Delay::new(Duration::from_secs(1)) Delay::new(Duration::from_secs(1))
.from_err() .expect("timeout")
}) })
}; };
// client is dropped // client is dropped
let rx = rx1.map_err(|_| hyper::Error::Io(io::Error::new(io::ErrorKind::Other, "thread panicked"))); let rx = rx1.expect("thread panicked");
res.join(rx).map(|r| r.0).wait().unwrap(); res.join(rx).map(|r| r.0).wait().unwrap();
closes.into_future().wait().unwrap().0.expect("closes"); closes.into_future().wait().unwrap().0.expect("closes");
@@ -788,7 +781,7 @@ mod dispatch_impl {
assert_eq!(res.status(), hyper::StatusCode::OK); assert_eq!(res.status(), hyper::StatusCode::OK);
res.into_body().into_stream().concat2() res.into_body().into_stream().concat2()
}); });
let rx = rx1.map_err(|_| hyper::Error::Io(io::Error::new(io::ErrorKind::Other, "thread panicked"))); let rx = rx1.expect("thread panicked");
res.join(rx).map(|r| r.0).wait().unwrap(); res.join(rx).map(|r| r.0).wait().unwrap();
// not closed yet, just idle // not closed yet, just idle
@@ -904,7 +897,7 @@ mod dispatch_impl {
client.request(req) client.request(req)
}; };
let rx = rx1.map_err(|_| hyper::Error::Io(io::Error::new(io::ErrorKind::Other, "thread panicked"))); let rx = rx1.expect("thread panicked");
res.join(rx).map(|r| r.0).wait().unwrap(); res.join(rx).map(|r| r.0).wait().unwrap();
let t = Delay::new(Duration::from_millis(100)) let t = Delay::new(Duration::from_millis(100))
@@ -955,7 +948,7 @@ mod dispatch_impl {
assert_eq!(res.status(), hyper::StatusCode::OK); assert_eq!(res.status(), hyper::StatusCode::OK);
res.into_body().into_stream().concat2() res.into_body().into_stream().concat2()
}); });
let rx = rx1.map_err(|_| hyper::Error::Io(io::Error::new(io::ErrorKind::Other, "thread panicked"))); let rx = rx1.expect("thread panicked");
res.join(rx).map(|r| r.0).wait().unwrap(); res.join(rx).map(|r| r.0).wait().unwrap();
let t = Delay::new(Duration::from_millis(100)) let t = Delay::new(Duration::from_millis(100))
@@ -1003,10 +996,9 @@ mod dispatch_impl {
assert_eq!(res.status(), hyper::StatusCode::OK); assert_eq!(res.status(), hyper::StatusCode::OK);
res.into_body().into_stream().concat2() res.into_body().into_stream().concat2()
}); });
let rx = rx1.map_err(|_| hyper::Error::Io(io::Error::new(io::ErrorKind::Other, "thread panicked"))); let rx = rx1.expect("thread panicked");
res.join(rx).map(|r| r.0).wait().unwrap(); res.join(rx).map(|r| r.0).wait().unwrap();
let t = Delay::new(Duration::from_millis(100)) let t = Delay::new(Duration::from_millis(100))
.map(|_| panic!("time out")); .map(|_| panic!("time out"));
let close = closes.into_future() let close = closes.into_future()
@@ -1049,13 +1041,12 @@ mod dispatch_impl {
assert_eq!(res.status(), hyper::StatusCode::OK); assert_eq!(res.status(), hyper::StatusCode::OK);
res.into_body().into_stream().concat2() res.into_body().into_stream().concat2()
}); });
let rx = rx1.map_err(|_| hyper::Error::Io(io::Error::new(io::ErrorKind::Other, "thread panicked"))); let rx = rx1.expect("thread panicked");
let timeout = Delay::new(Duration::from_millis(200)); let timeout = Delay::new(Duration::from_millis(200));
let rx = rx.and_then(move |_| timeout.map_err(|e| e.into())); let rx = rx.and_then(move |_| timeout.expect("timeout"));
res.join(rx).map(|r| r.0).wait().unwrap(); res.join(rx).map(|r| r.0).wait().unwrap();
let t = Delay::new(Duration::from_millis(100)) let t = Delay::new(Duration::from_millis(100))
.map(|_| panic!("time out")); .map(|_| panic!("time out"));
let close = closes.into_future() let close = closes.into_future()
@@ -1129,7 +1120,7 @@ mod dispatch_impl {
assert_eq!(connects.load(Ordering::SeqCst), 0); assert_eq!(connects.load(Ordering::SeqCst), 0);
let rx = rx1.map_err(|_| hyper::Error::Io(io::Error::new(io::ErrorKind::Other, "thread panicked"))); let rx = rx1.expect("thread panicked");
let req = Request::builder() let req = Request::builder()
.uri(&*format!("http://{}/a", addr)) .uri(&*format!("http://{}/a", addr))
.body(Body::empty()) .body(Body::empty())
@@ -1143,7 +1134,7 @@ mod dispatch_impl {
// state and back into client pool // state and back into client pool
thread::sleep(Duration::from_millis(50)); thread::sleep(Duration::from_millis(50));
let rx = rx2.map_err(|_| hyper::Error::Io(io::Error::new(io::ErrorKind::Other, "thread panicked"))); let rx = rx2.expect("thread panicked");
let req = Request::builder() let req = Request::builder()
.uri(&*format!("http://{}/b", addr)) .uri(&*format!("http://{}/b", addr))
.body(Body::empty()) .body(Body::empty())
@@ -1194,7 +1185,7 @@ mod dispatch_impl {
assert_eq!(connects.load(Ordering::Relaxed), 0); assert_eq!(connects.load(Ordering::Relaxed), 0);
let rx = rx1.map_err(|_| hyper::Error::Io(io::Error::new(io::ErrorKind::Other, "thread panicked"))); let rx = rx1.expect("thread panicked");
let req = Request::builder() let req = Request::builder()
.method("HEAD") .method("HEAD")
.uri(&*format!("http://{}/a", addr)) .uri(&*format!("http://{}/a", addr))
@@ -1205,7 +1196,7 @@ mod dispatch_impl {
assert_eq!(connects.load(Ordering::Relaxed), 1); assert_eq!(connects.load(Ordering::Relaxed), 1);
let rx = rx2.map_err(|_| hyper::Error::Io(io::Error::new(io::ErrorKind::Other, "thread panicked"))); let rx = rx2.expect("thread panicked");
let req = Request::builder() let req = Request::builder()
.uri(&*format!("http://{}/b", addr)) .uri(&*format!("http://{}/b", addr))
.body(Body::empty()) .body(Body::empty())
@@ -1246,7 +1237,7 @@ mod dispatch_impl {
}); });
let rx = rx1.map_err(|_| hyper::Error::Io(io::Error::new(io::ErrorKind::Other, "thread panicked"))); let rx = rx1.expect("thread panicked");
let req = Request::builder() let req = Request::builder()
.uri(&*format!("http://{}/foo/bar", addr)) .uri(&*format!("http://{}/foo/bar", addr))
.body(Body::empty()) .body(Body::empty())
@@ -1354,7 +1345,7 @@ mod conn {
use hyper::{self, Request}; use hyper::{self, Request};
use hyper::client::conn; use hyper::client::conn;
use super::{s, tcp_connect}; use super::{s, tcp_connect, FutureHyperExt};
#[test] #[test]
fn get() { fn get() {
@@ -1395,10 +1386,10 @@ mod conn {
assert_eq!(res.status(), hyper::StatusCode::OK); assert_eq!(res.status(), hyper::StatusCode::OK);
res.into_body().into_stream().concat2() res.into_body().into_stream().concat2()
}); });
let rx = rx1.map_err(|_| hyper::Error::Io(io::Error::new(io::ErrorKind::Other, "thread panicked"))); let rx = rx1.expect("thread panicked");
let timeout = Delay::new(Duration::from_millis(200)); let timeout = Delay::new(Duration::from_millis(200));
let rx = rx.and_then(move |_| timeout.map_err(|e| e.into())); let rx = rx.and_then(move |_| timeout.expect("timeout"));
res.join(rx).map(|r| r.0).wait().unwrap(); res.join(rx).map(|r| r.0).wait().unwrap();
} }
@@ -1441,10 +1432,10 @@ mod conn {
assert_eq!(res.status(), hyper::StatusCode::OK); assert_eq!(res.status(), hyper::StatusCode::OK);
res.into_body().into_stream().concat2() res.into_body().into_stream().concat2()
}); });
let rx = rx1.map_err(|_| hyper::Error::Io(io::Error::new(io::ErrorKind::Other, "thread panicked"))); let rx = rx1.expect("thread panicked");
let timeout = Delay::new(Duration::from_millis(200)); let timeout = Delay::new(Duration::from_millis(200));
let rx = rx.and_then(move |_| timeout.map_err(|e| e.into())); let rx = rx.and_then(move |_| timeout.expect("timeout"));
res.join(rx).map(|r| r.0).wait().unwrap(); res.join(rx).map(|r| r.0).wait().unwrap();
} }
@@ -1490,17 +1481,14 @@ mod conn {
let res2 = client.send_request(req) let res2 = client.send_request(req)
.then(|result| { .then(|result| {
let err = result.expect_err("res2"); let err = result.expect_err("res2");
match err { assert!(err.is_canceled(), "err not canceled, {:?}", err);
hyper::Error::Cancel(..) => (),
other => panic!("expected Cancel, found {:?}", other),
}
Ok(()) Ok(())
}); });
let rx = rx1.map_err(|_| hyper::Error::Io(io::Error::new(io::ErrorKind::Other, "thread panicked"))); let rx = rx1.expect("thread panicked");
let timeout = Delay::new(Duration::from_millis(200)); let timeout = Delay::new(Duration::from_millis(200));
let rx = rx.and_then(move |_| timeout.map_err(|e| e.into())); let rx = rx.and_then(move |_| timeout.expect("timeout"));
res1.join(res2).join(rx).map(|r| r.0).wait().unwrap(); res1.join(res2).join(rx).map(|r| r.0).wait().unwrap();
} }
@@ -1558,10 +1546,10 @@ mod conn {
res.into_body().into_stream().concat2() res.into_body().into_stream().concat2()
}); });
let rx = rx1.map_err(|_| hyper::Error::Io(io::Error::new(io::ErrorKind::Other, "thread panicked"))); let rx = rx1.expect("thread panicked");
let timeout = Delay::new(Duration::from_millis(200)); let timeout = Delay::new(Duration::from_millis(200));
let rx = rx.and_then(move |_| timeout.map_err(|e| e.into())); let rx = rx.and_then(move |_| timeout.expect("timeout"));
until_upgrade.join(res).join(rx).map(|r| r.0).wait().unwrap(); until_upgrade.join(res).join(rx).map(|r| r.0).wait().unwrap();
// should not be ready now // should not be ready now
@@ -1641,10 +1629,10 @@ mod conn {
assert_eq!(body.as_ref(), b""); assert_eq!(body.as_ref(), b"");
}); });
let rx = rx1.map_err(|_| hyper::Error::Io(io::Error::new(io::ErrorKind::Other, "thread panicked"))); let rx = rx1.expect("thread panicked");
let timeout = Delay::new(Duration::from_millis(200)); let timeout = Delay::new(Duration::from_millis(200));
let rx = rx.and_then(move |_| timeout.map_err(|e| e.into())); let rx = rx.and_then(move |_| timeout.expect("timeout"));
until_tunneled.join(res).join(rx).map(|r| r.0).wait().unwrap(); until_tunneled.join(res).join(rx).map(|r| r.0).wait().unwrap();
// should not be ready now // should not be ready now
@@ -1697,3 +1685,17 @@ mod conn {
impl AsyncRead for DebugStream {} impl AsyncRead for DebugStream {}
} }
trait FutureHyperExt: Future {
fn expect<E>(self, msg: &'static str) -> Box<Future<Item=Self::Item, Error=E>>;
}
impl<F> FutureHyperExt for F
where
F: Future + 'static,
F::Error: ::std::fmt::Debug,
{
fn expect<E>(self, msg: &'static str) -> Box<Future<Item=Self::Item, Error=E>> {
Box::new(self.map_err(move |e| panic!("expect: {}; error={:?}", msg, e)))
}
}

8
tests/server.rs
View File

@@ -935,7 +935,7 @@ fn returning_1xx_response_is_error() {
let socket = item.unwrap(); let socket = item.unwrap();
Http::<hyper::Chunk>::new() Http::<hyper::Chunk>::new()
.serve_connection(socket, service_fn(|_| { .serve_connection(socket, service_fn(|_| {
Ok(Response::builder() Ok::<_, hyper::Error>(Response::builder()
.status(StatusCode::CONTINUE) .status(StatusCode::CONTINUE)
.body(Body::empty()) .body(Body::empty())
.unwrap()) .unwrap())
@@ -988,7 +988,7 @@ fn upgrades() {
.header("upgrade", "foobar") .header("upgrade", "foobar")
.body(hyper::Body::empty()) .body(hyper::Body::empty())
.unwrap(); .unwrap();
Ok(res) Ok::<_, hyper::Error>(res)
})); }));
let mut conn_opt = Some(conn); let mut conn_opt = Some(conn);
@@ -1144,10 +1144,10 @@ fn streaming_body() {
.keep_alive(false) .keep_alive(false)
.serve_connection(socket, service_fn(|_| { .serve_connection(socket, service_fn(|_| {
static S: &'static [&'static [u8]] = &[&[b'x'; 1_000] as &[u8]; 1_00] as _; static S: &'static [&'static [u8]] = &[&[b'x'; 1_000] as &[u8]; 1_00] as _;
let b = ::futures::stream::iter_ok(S.into_iter()) let b = ::futures::stream::iter_ok::<_, String>(S.into_iter())
.map(|&s| s); .map(|&s| s);
let b = hyper::Body::wrap_stream(b); let b = hyper::Body::wrap_stream(b);
Ok(Response::new(b)) Ok::<_, hyper::Error>(Response::new(b))
})) }))
.map(|_| ()) .map(|_| ())
}); });