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

feat(error): revamp hyper::Error type

Browse Source 
**The `Error` is now an opaque struct**, which allows for more variants to
be added freely, and the internal representation to change without being
breaking changes.

For inspecting an `Error`, there are several `is_*` methods to check for
certain classes of errors, such as `Error::is_parse()`. The `cause` can
also be inspected, like before. This likely seems like a downgrade, but
more inspection can be added as needed!

The `Error` now knows about more states, which gives much more context
around when a certain error occurs. This is also expressed in the
description and `fmt` messages.

**Most places where a user would provide an error to hyper can now pass
any error type** (`E: Into<Box<std::error::Error>>`). This error is passed
back in relevant places, and can be useful for logging. This should make
it much clearer about what error a user should provide to hyper: any it
feels is relevant!

Closes #1128
Closes #1130
Closes #1431
Closes #1338

BREAKING CHANGE: `Error` is no longer an enum to pattern match over, or
  to construct. Code will need to be updated accordingly.

  For body streams or `Service`s, inference might be unable to determine
  what error type you mean to return. Starting in Rust 1.26, you could
  just label that as `!` if you never return an error.
This commit is contained in:
Sean McArthur
2018-04-10 14:29:34 -07:00
parent 33874f9a75
commit 5d3c472228
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) {
bench_server!(b, ("content-length", "1000000"), || {
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) {
bench_server!(b, ("transfer-encoding", "chunked"), || {
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!("Headers: {:#?}", res.headers());
res.into_parts().1.into_stream().for_each(|chunk| {
io::stdout().write_all(&chunk).map_err(From::from)
res.into_body().into_stream().for_each(|chunk| {
io::stdout().write_all(&chunk)
.map_err(|e| panic!("example expects stdout is open, error={}", e))
})
}).map(|_| {
println!("\n\nDone.");

3
examples/hello.rs
View File

@@ -17,7 +17,8 @@ fn main() {
let addr = ([127, 0, 0, 1], 3000).into();
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 || {

9
examples/send_file.rs
View File

@@ -9,7 +9,6 @@ use futures::future::lazy;
use futures::sync::oneshot;
use hyper::{Body, /*Chunk,*/ Method, Request, Response, StatusCode};
use hyper::error::Error;
use hyper::server::{Http, Service};
use std::fs::File;
@@ -19,7 +18,7 @@ use std::thread;
static NOTFOUND: &[u8] = b"Not Found";
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
// this is limited to serving small files, but it is somewhat
// 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;
@@ -64,7 +63,7 @@ struct ResponseExamples;
impl Service for ResponseExamples {
type Request = Request<Body>;
type Response = Response<Body>;
type Error = hyper::Error;
type Error = io::Error;
type Future = Box<Future<Item = Self::Response, Error = Self::Error> + Send>;
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") => {
// 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>
where
T: AsyncRead + AsyncWrite,
B: Entity<Error=::Error> + 'static,
B: Entity + 'static,
{
inner: proto::dispatch::Dispatcher<
proto::dispatch::Client<B>,
@@ -138,7 +138,7 @@ impl<B> SendRequest<B>
impl<B> SendRequest<B>
where
B: Entity<Error=::Error> + 'static,
B: Entity + 'static,
{
/// Sends a `Request` on the associated connection.
///
@@ -262,7 +262,7 @@ impl<B> fmt::Debug for SendRequest<B> {
impl<T, B> Connection<T, B>
where
T: AsyncRead + AsyncWrite,
B: Entity<Error=::Error> + 'static,
B: Entity + 'static,
{
/// Return the inner IO object, and additional information.
pub fn into_parts(self) -> Parts<T> {
@@ -289,7 +289,7 @@ where
impl<T, B> Future for Connection<T, B>
where
T: AsyncRead + AsyncWrite,
B: Entity<Error=::Error> + 'static,
B: Entity + 'static,
{
type Item = ();
type Error = ::Error;
@@ -302,7 +302,7 @@ where
impl<T, B> fmt::Debug for Connection<T, B>
where
T: AsyncRead + AsyncWrite + fmt::Debug,
B: Entity<Error=::Error> + 'static,
B: Entity + 'static,
{
fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
f.debug_struct("Connection")
@@ -331,7 +331,7 @@ impl Builder {
pub fn handshake<T, B>(&self, io: T) -> Handshake<T, B>
where
T: AsyncRead + AsyncWrite,
B: Entity<Error=::Error> + 'static,
B: Entity + 'static,
{
Handshake {
inner: HandshakeInner {
@@ -345,7 +345,7 @@ impl Builder {
pub(super) fn handshake_no_upgrades<T, B>(&self, io: T) -> HandshakeNoUpgrades<T, B>
where
T: AsyncRead + AsyncWrite,
B: Entity<Error=::Error> + 'static,
B: Entity + 'static,
{
HandshakeNoUpgrades {
inner: HandshakeInner {
@@ -362,7 +362,7 @@ impl Builder {
impl<T, B> Future for Handshake<T, B>
where
T: AsyncRead + AsyncWrite,
B: Entity<Error=::Error> + 'static,
B: Entity + 'static,
{
type Item = (SendRequest<B>, Connection<T, B>);
type Error = ::Error;
@@ -387,7 +387,7 @@ impl<T, B> fmt::Debug for Handshake<T, B> {
impl<T, B> Future for HandshakeNoUpgrades<T, B>
where
T: AsyncRead + AsyncWrite,
B: Entity<Error=::Error> + 'static,
B: Entity + 'static,
{
type Item = (SendRequest<B>, proto::dispatch::Dispatcher<
proto::dispatch::Client<B>,
@@ -406,7 +406,7 @@ where
impl<T, B, R> Future for HandshakeInner<T, B, R>
where
T: AsyncRead + AsyncWrite,
B: Entity<Error=::Error> + 'static,
B: Entity + 'static,
R: proto::Http1Transaction<
Incoming=StatusCode,
Outgoing=proto::RequestLine,
@@ -470,7 +470,7 @@ impl<B: Send> AssertSendSync for SendRequest<B> {}
impl<T: Send, B: Send> AssertSend for Connection<T, B>
where
T: AsyncRead + AsyncWrite,
B: Entity<Error=::Error> + 'static,
B: Entity + 'static,
B::Data: Send + 'static,
{}
@@ -478,7 +478,7 @@ where
impl<T: Send + Sync, B: Send + Sync> AssertSendSync for Connection<T, B>
where
T: AsyncRead + AsyncWrite,
B: Entity<Error=::Error> + 'static,
B: Entity + '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.
type Transport: AsyncRead + AsyncWrite + Send + 'static;
/// 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.
type Future: Future<Item=(Self::Transport, Connected), Error=Self::Error> + Send;
/// 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
// want a message yet?
self.giver.poll_want()
.map_err(|_| ::Error::Closed)
.map_err(|_| ::Error::new_closed())
},
Ok(Async::NotReady) => Ok(Async::NotReady),
Err(_) => Err(::Error::Closed),
Err(_) => Err(::Error::new_closed()),
}
}
@@ -184,9 +184,12 @@ mod tests {
drop(rx);
promise.then(|fulfilled| {
let res = fulfilled.expect("fulfilled");
match res.unwrap_err() {
(::Error::Cancel(_), Some(_)) => (),
let err = fulfilled
.expect("fulfilled")
.expect_err("promise should error");
match (err.0.kind(), err.1) {
(&::error::Kind::Canceled, Some(_)) => (),
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>
where C: Connect<Error=io::Error> + Sync + 'static,
where C: Connect + Sync + 'static,
C::Transport: 'static,
C::Future: 'static,
B: Entity<Error=::Error> + Send + 'static,
B: Entity + Send + 'static,
B::Data: Send,
{
@@ -139,13 +139,14 @@ where C: Connect<Error=io::Error> + Sync + 'static,
Version::HTTP_11 => (),
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 {
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();
@@ -154,7 +155,8 @@ where C: Connect<Error=io::Error> + Sync + 'static,
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::ErrorKind::InvalidInput,
"invalid URI for Client Request"
@@ -203,13 +205,13 @@ where C: Connect<Error=io::Error> + Sync + 'static,
};
future::lazy(move || {
connector.connect(dst)
.from_err()
.map_err(::Error::new_connect)
.and_then(move |(io, connected)| {
conn::Builder::new()
.h1_writev(h1_writev)
.handshake_no_upgrades(io)
.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 {
is_proxied: connected.is_proxied,
tx: tx,
@@ -260,9 +262,7 @@ where C: Connect<Error=io::Error> + Sync + 'static,
} else if !res.body().is_empty() {
let (delayed_tx, delayed_rx) = oneshot::channel();
res.body_mut().delayed_eof(delayed_rx);
// If the executor doesn't have room, oh well. Things will likely
// be blowing up soon, but this specific task isn't required.
let _ = executor.execute(
executor.execute(
future::poll_fn(move || {
pooled.tx.poll_ready()
})
@@ -277,7 +277,6 @@ where C: Connect<Error=io::Error> + Sync + 'static,
Ok(res)
});
fut
});
@@ -290,9 +289,9 @@ where C: Connect<Error=io::Error> + Sync + 'static,
}
impl<C, B> Service for Client<C, B>
where C: Connect<Error=io::Error> + 'static,
where C: Connect + 'static,
C::Future: 'static,
B: Entity<Error=::Error> + Send + 'static,
B: Entity + Send + 'static,
B::Data: Send,
{
type Request = Request<B>;
@@ -353,9 +352,9 @@ struct RetryableSendRequest<C, B> {
impl<C, B> Future for RetryableSendRequest<C, B>
where
C: Connect<Error=io::Error> + 'static,
C: Connect + 'static,
C::Future: 'static,
B: Entity<Error=::Error> + Send + 'static,
B: Entity + Send + 'static,
B::Data: Send,
{
type Item = Response<Body>;
@@ -564,10 +563,10 @@ 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::Future: 'static,
B: Entity<Error=::Error> + Send,
B: Entity + Send,
B::Data: Send,
{
/// Construct the Client with this configuration.
@@ -590,7 +589,7 @@ where C: Connect<Error=io::Error>,
impl<B> Config<UseDefaultConnector, B>
where
B: Entity<Error=::Error> + Send,
B: Entity + Send,
B::Data: Send,
{
/// Construct the Client with this configuration.
@@ -640,7 +639,6 @@ impl<C: Clone, B> Clone for Config<C, B> {
}
}
// ===== impl Exec =====
#[derive(Clone)]
@@ -655,24 +653,19 @@ impl Exec {
Exec::Executor(Arc::new(executor))
}
fn execute<F>(&self, fut: F) -> io::Result<()>
fn execute<F>(&self, fut: F)
where
F: Future<Item=(), Error=()> + Send + 'static,
{
match *self {
Exec::Default => spawn(fut),
Exec::Executor(ref e) => {
e.execute(bg(Box::new(fut)))
let _ = e.execute(bg(Box::new(fut)))
.map_err(|err| {
debug!("executor error: {:?}", err.kind());
io::Error::new(
io::ErrorKind::Other,
"executor error",
)
})?
panic!("executor error: {:?}", err.kind());
});
},
}
Ok(())
}
}

2
src/client/pool.rs
View File

@@ -245,7 +245,7 @@ impl<T: Closed + Send + 'static> Pool<T> {
interval: interval,
pool: Arc::downgrade(&self.inner),
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.write(b"HTTP/1.1 200 OK\r\nContent-Length: 0\r\n\r\n"));
Ok(Async::Ready(()))
});
}).map_err(|e: ::std::io::Error| panic!("srv1 poll_fn error: {}", e));
res1.join(srv1).wait().expect("res1");
}
drop(sock1);
@@ -52,7 +52,7 @@ fn retryable_request() {
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"));
Ok(Async::Ready(()))
});
}).map_err(|e: ::std::io::Error| panic!("srv2 poll_fn error: {}", e));
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.write(b"HTTP/1.1 200 OK\r\nContent-Length: 0\r\n\r\n"));
Ok(Async::Ready(()))
});
}).map_err(|e: ::std::io::Error| panic!("srv1 poll_fn error: {}", e));
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]));
sock1.take();
Ok(Async::Ready(()))
});
}).map_err(|e: ::std::io::Error| panic!("srv2 poll_fn error: {}", e));
let err = res2.join(srv2).wait().expect_err("res2");
match err {
::Error::Incomplete => (),
match err.kind() {
&::error::Kind::Incomplete => (),
other => panic!("expected Incomplete, found {:?}", other)
}
}

437
src/error.rs
View File

@@ -1,188 +1,316 @@
//! Error and Result module.
use std::error::Error as StdError;
use std::fmt;
use std::io::Error as IoError;
use std::str::Utf8Error;
use std::string::FromUtf8Error;
use std::io;
use httparse;
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.
pub type Result<T> = ::std::result::Result<T, Error>;
/// A set of errors that can occur parsing HTTP streams.
#[derive(Debug)]
pub enum Error {
/// An invalid `Method`, such as `GE,T`.
Method,
/// An invalid `HttpVersion`, such as `HTP/1.1`
Version,
/// Uri Errors
Uri,
/// An invalid `Header`.
Header,
/// A message head is too large to be reasonable.
TooLarge,
type Cause = Box<StdError + Send + Sync>;
/// Represents errors that can occur handling HTTP streams.
pub struct Error {
inner: Box<ErrorImpl>,
}
struct ErrorImpl {
kind: Kind,
cause: Option<Cause>,
}
#[derive(Debug, PartialEq)]
pub(crate) enum Kind {
Parse(Parse),
/// A message reached EOF, but is not complete.
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.
Upgrade,
/// A client connection received a response when not waiting for one.
MismatchedResponse,
/// A pending item was dropped before ever being processed.
Cancel(Canceled),
Canceled,
/// Indicates a connection is closed.
Closed,
/// An `io::Error` that occurred while trying to read or write to a network stream.
Io(IoError),
/// Parsing a field as string failed
Utf8(Utf8Error),
Io,
/// Error occurred while connecting.
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)]
__Nonexhaustive(Void)
/// User tried to create a Request with bad version.
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 {
pub(crate) fn new_canceled<E: Into<Box<StdError + Send + Sync>>>(cause: Option<E>) -> Error {
Error::Cancel(Canceled {
cause: cause.map(Into::into),
})
//TODO(error): should there be these kinds of inspection methods?
//
// - 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.
///
/// 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 {
impl fmt::Debug for Error {
fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
f.pad(self.description())
}
}
#[doc(hidden)]
pub struct Void(());
impl fmt::Debug for Void {
fn fmt(&self, _: &mut fmt::Formatter) -> fmt::Result {
unreachable!()
f.debug_struct("Error")
.field("kind", &self.inner.kind)
.field("cause", &self.inner.cause)
.finish()
}
}
impl fmt::Display for Error {
fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
match *self {
Io(ref e) => fmt::Display::fmt(e, f),
Utf8(ref e) => fmt::Display::fmt(e, f),
ref e => f.write_str(e.description()),
if let Some(ref cause) = self.inner.cause {
write!(f, "{}: {}", self.description(), cause)
} else {
f.write_str(self.description())
}
}
}
impl StdError for Error {
fn description(&self) -> &str {
match *self {
Method => "invalid Method specified",
Version => "invalid HTTP version specified",
Uri => "invalid URI",
Header => "invalid Header provided",
TooLarge => "message head is too large",
Status => "invalid Status provided",
Incomplete => "message is incomplete",
Timeout => "timeout",
Upgrade => "unsupported protocol upgrade",
Closed => "connection is closed",
Cancel(ref e) => e.description(),
Io(ref e) => e.description(),
Utf8(ref e) => e.description(),
Error::__Nonexhaustive(..) => unreachable!(),
match self.inner.kind {
Kind::Parse(Parse::Method) => "invalid Method specified",
Kind::Parse(Parse::Version) => "invalid HTTP version specified",
Kind::Parse(Parse::Uri) => "invalid URI",
Kind::Parse(Parse::Header) => "invalid Header provided",
Kind::Parse(Parse::TooLarge) => "message head is too large",
Kind::Parse(Parse::Status) => "invalid Status provided",
Kind::Incomplete => "message is incomplete",
Kind::Upgrade => "unsupported protocol upgrade",
Kind::MismatchedResponse => "response received without matching request",
Kind::Closed => "connection closed",
Kind::Connect => "an error occurred trying to connect",
Kind::Canceled => "an operation was canceled internally before starting",
Kind::Listen => "error creating server listener",
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> {
match *self {
Io(ref error) => Some(error),
Utf8(ref error) => Some(error),
Cancel(ref e) => e.cause.as_ref().map(|e| &**e as &StdError),
Error::__Nonexhaustive(..) => unreachable!(),
_ => None,
}
self
.inner
.cause
.as_ref()
.map(|cause| &**cause as &StdError)
}
}
impl From<IoError> for Error {
fn from(err: IoError) -> Error {
Io(err)
#[doc(hidden)]
impl From<Parse> for Error {
fn from(err: Parse) -> Error {
Error::new(Kind::Parse(err), None)
}
}
impl From<Utf8Error> for Error {
fn from(err: Utf8Error) -> Error {
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 {
impl From<httparse::Error> for Parse {
fn from(err: httparse::Error) -> Parse {
match err {
httparse::Error::HeaderName |
httparse::Error::HeaderValue |
httparse::Error::NewLine |
httparse::Error::Token => Header,
httparse::Error::Status => Status,
httparse::Error::TooManyHeaders => TooLarge,
httparse::Error::Version => Version,
httparse::Error::Token => Parse::Header,
httparse::Error::Status => Parse::Status,
httparse::Error::TooManyHeaders => Parse::TooLarge,
httparse::Error::Version => Parse::Version,
}
}
}
impl From<http::method::InvalidMethod> for Error {
fn from(_: http::method::InvalidMethod) -> Error {
Error::Method
impl From<http::method::InvalidMethod> for Parse {
fn from(_: http::method::InvalidMethod) -> Parse {
Parse::Method
}
}
impl From<http::uri::InvalidUriBytes> for Error {
fn from(_: http::uri::InvalidUriBytes) -> Error {
Error::Uri
impl From<http::status::InvalidStatusCode> for Parse {
fn from(_: http::status::InvalidStatusCode) -> Parse {
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)]
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]>;
/// The error type of this stream.
//TODO: add bounds Into<::error::User> (or whatever it is called)
type Error;
type Error: Into<Box<::std::error::Error + Send + Sync>>;
/// Poll for a `Data` buffer.
///
@@ -141,7 +140,7 @@ enum Kind {
_close_tx: oneshot::Sender<()>,
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>),
Empty,
}
@@ -212,17 +211,22 @@ impl Body {
/// " ",
/// "world",
/// ];
/// let stream = futures::stream::iter_ok(chunks);
///
/// let stream = futures::stream::iter_ok::<_, ::std::io::Error>(chunks);
///
/// let body = Body::wrap_stream(stream);
/// # }
/// ```
pub fn wrap_stream<S>(stream: S) -> Body
where
S: Stream<Error=::Error> + Send + 'static,
S: Stream + Send + 'static,
S::Error: Into<Box<::std::error::Error + Send + Sync>>,
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>`.
@@ -327,7 +331,7 @@ impl Body {
Async::Ready(None) => Ok(Async::Ready(None)),
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::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
/// determine if this connection can be kept alive after the message,
/// or if it is complete.
pub struct Conn<I, B, T> {
pub(crate) struct Conn<I, B, T> {
io: Buffered<I, EncodedBuf<Cursor<B>>>,
state: State,
_marker: PhantomData<T>
@@ -146,7 +146,8 @@ where I: AsyncRead + AsyncWrite,
_ => {
error!("unimplemented HTTP Version = {:?}", version);
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;
@@ -245,7 +246,7 @@ where I: AsyncRead + AsyncWrite,
if self.is_mid_message() {
self.maybe_park_read();
} else {
self.require_empty_read()?;
self.require_empty_read().map_err(::Error::new_io)?;
}
Ok(())
}

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

@@ -1,5 +1,3 @@
use std::io;
use bytes::Bytes;
use futures::{Async, Future, Poll, Stream};
use http::{Request, Response, StatusCode};
@@ -9,7 +7,7 @@ use tokio_service::Service;
use proto::body::Entity;
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>,
dispatch: D,
body_tx: Option<::proto::body::Sender>,
@@ -17,7 +15,7 @@ pub struct Dispatcher<D, Bs, I, B, T> {
is_closing: bool,
}
pub trait Dispatch {
pub(crate) trait Dispatch {
type PollItem;
type PollBody;
type RecvItem;
@@ -47,7 +45,7 @@ where
I: AsyncRead + AsyncWrite,
B: AsRef<[u8]>,
T: Http1Transaction,
Bs: Entity<Data=B, Error=::Error>,
Bs: Entity<Data=B>,
{
pub fn new(dispatch: D, conn: Conn<I, B, T>) -> Self {
Dispatcher {
@@ -98,7 +96,7 @@ where
if self.is_done() {
if should_shutdown {
try_ready!(self.conn.shutdown());
try_ready!(self.conn.shutdown().map_err(::Error::new_shutdown));
}
self.conn.take_error()?;
Ok(Async::Ready(()))
@@ -152,7 +150,7 @@ where
return Ok(Async::NotReady);
}
Err(e) => {
body.send_error(::Error::Io(e));
body.send_error(::Error::new_body(e));
}
}
} else {
@@ -225,14 +223,14 @@ where
} else if !self.conn.can_buffer_body() {
try_ready!(self.poll_flush());
} 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)) => {
self.body_rx = Some(body);
chunk
},
Async::Ready(None) => {
if self.conn.can_write_body() {
self.conn.write_body(None)?;
self.conn.write_body(None).map_err(::Error::new_body_write)?;
}
continue;
},
@@ -243,7 +241,7 @@ where
};
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([])`.
} else if chunk.as_ref().len() == 0 {
// ok
@@ -259,7 +257,7 @@ where
fn poll_flush(&mut self) -> Poll<(), ::Error> {
self.conn.flush().map_err(|err| {
debug!("error writing: {}", err);
err.into()
::Error::new_body_write(err)
})
}
@@ -294,7 +292,7 @@ where
I: AsyncRead + AsyncWrite,
B: AsRef<[u8]>,
T: Http1Transaction,
Bs: Entity<Data=B, Error=::Error>,
Bs: Entity<Data=B>,
{
type Item = ();
type Error = ::Error;
@@ -318,8 +316,9 @@ impl<S> Server<S> where S: Service {
impl<S, Bs> Dispatch for Server<S>
where
S: Service<Request=Request<Body>, Response=Response<Bs>, Error=::Error>,
Bs: Entity<Error=::Error>,
S: Service<Request=Request<Body>, Response=Response<Bs>>,
S::Error: Into<Box<::std::error::Error + Send + Sync>>,
Bs: Entity,
{
type PollItem = MessageHead<StatusCode>;
type PollBody = Bs;
@@ -327,7 +326,7 @@ where
fn poll_msg(&mut self) -> Poll<Option<(Self::PollItem, Option<Self::PollBody>)>, ::Error> {
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::NotReady => {
self.in_flight = Some(fut);
@@ -389,7 +388,7 @@ impl<B> Client<B> {
impl<B> Dispatch for Client<B>
where
B: Entity<Error=::Error>,
B: Entity,
{
type PollItem = RequestHead;
type PollBody = B;
@@ -443,7 +442,7 @@ where
let _ = cb.send(Ok(res));
Ok(())
} else {
Err(::Error::Io(io::Error::new(io::ErrorKind::InvalidData, "response received without matching request")))
Err(::Error::new_mismatched_response())
}
},
Err(err) => {
@@ -507,8 +506,15 @@ mod tests {
.expect("callback poll")
.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),
}
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 {
match try!(S::parse(&mut self.read_buf)) {
Some((head, len)) => {
@@ -118,14 +118,14 @@ where
None => {
if self.read_buf.capacity() >= 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 => {
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::encode::{EncodedBuf, Encoder};
mod conn;
mod date;
mod decode;
pub mod dispatch;
pub(crate) mod dispatch;
mod encode;
mod io;
pub mod role;

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

@@ -40,7 +40,7 @@ where
let mut headers = [httparse::EMPTY_HEADER; MAX_HEADERS];
trace!("Request.parse([Header; {}], [u8; {}])", headers.len(), buf.len());
let mut req = httparse::Request::new(&mut headers);
match try!(req.parse(&buf)) {
match req.parse(&buf)? {
httparse::Status::Complete(len) => {
trace!("Request.parse Complete({})", len);
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.
if head.version == Version::HTTP_10 {
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) {
Ok(Decode::Normal(Decoder::chunked()))
} else {
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) {
Ok(Decode::Normal(Decoder::length(len)))
} else if head.headers.contains_key(CONTENT_LENGTH) {
debug!("illegal Content-Length header");
Err(::Error::Header)
Err(::Error::new_header())
} else {
Ok(Decode::Normal(Decoder::length(0)))
}
@@ -146,7 +146,8 @@ where
head = MessageHead::default();
head.subject = StatusCode::INTERNAL_SERVER_ERROR;
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 {
Ok(Server::set_length(&mut head, body, method.as_ref()))
};
@@ -184,14 +185,15 @@ where
}
fn on_error(err: &::Error) -> Option<MessageHead<Self::Outgoing>> {
let status = match err {
&::Error::Method |
&::Error::Version |
&::Error::Header /*|
&::Error::Uri(_)*/ => {
use ::error::{Kind, Parse};
let status = match *err.kind() {
Kind::Parse(Parse::Method) |
Kind::Parse(Parse::Version) |
Kind::Parse(Parse::Header) |
Kind::Parse(Parse::Uri) => {
StatusCode::BAD_REQUEST
},
&::Error::TooLarge => {
Kind::Parse(Parse::TooLarge) => {
StatusCode::REQUEST_HEADER_FIELDS_TOO_LARGE
}
_ => return None,
@@ -271,7 +273,7 @@ where
match try!(res.parse(bytes)) {
httparse::Status::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 {
Version::HTTP_11
} else {
@@ -343,7 +345,7 @@ where
// mal-formed. A server should respond with 400 Bad Request.
if inc.version == Version::HTTP_10 {
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) {
Ok(Decode::Normal(Decoder::chunked()))
} else {
@@ -354,7 +356,7 @@ where
Ok(Decode::Normal(Decoder::length(len)))
} else if inc.headers.contains_key(CONTENT_LENGTH) {
debug!("illegal Content-Length header");
Err(::Error::Header)
Err(::Error::new_header())
} else {
trace!("neither Transfer-Encoding nor Content-Length");
Ok(Decode::Normal(Decoder::eof()))
@@ -577,12 +579,13 @@ impl OnUpgrade for NoUpgrades {
*head = MessageHead::default();
head.subject = ::StatusCode::INTERNAL_SERVER_ERROR;
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> {
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::chunk::Chunk;
pub use self::h1::{dispatch, Conn};
pub(crate) use self::h1::{dispatch, Conn};
pub mod body;
mod chunk;
@@ -60,14 +60,14 @@ pub fn expecting_continue(version: Version, headers: &HeaderMap) -> bool {
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 ServerUpgradeTransaction = h1::role::Server<h1::role::YesUpgrades>;
pub type ClientTransaction = h1::role::Client<h1::role::NoUpgrades>;
pub type ClientUpgradeTransaction = h1::role::Client<h1::role::YesUpgrades>;
pub(crate) type ClientTransaction = h1::role::Client<h1::role::NoUpgrades>;
pub(crate) type ClientUpgradeTransaction = h1::role::Client<h1::role::YesUpgrades>;
pub trait Http1Transaction {
pub(crate) trait Http1Transaction {
type Incoming;
type Outgoing: Default;
fn parse(bytes: &mut BytesMut) -> ParseResult<Self::Incoming>;
@@ -84,7 +84,7 @@ pub trait Http1Transaction {
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)]
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>
where
S: HyperService,
S::ResponseBody: Entity<Error=::Error>,
S::ResponseBody: Entity,
{
pub(super) conn: proto::dispatch::Dispatcher<
proto::dispatch::Server<S>,
@@ -59,9 +59,11 @@ pub struct Parts<T> {
// ===== impl Connection =====
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,
B: Entity<Error=::Error> + 'static,
B: Entity + 'static,
{
/// Disables keep-alive for this connection.
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>
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,
B: Entity<Error=::Error> + 'static,
B: Entity + 'static,
{
type Item = ();
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>
where
S: HyperService,
S::ResponseBody: Entity<Error=::Error>,
S::ResponseBody: Entity,
{
fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
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.
pub struct Server<S, B>
where
B: Entity<Error=::Error>,
B: Entity,
{
protocol: Http<B::Data>,
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
/// actually run the server.
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,
Bd: Entity<Data=B, Error=::Error>,
where
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 std_listener = StdTcpListener::bind(addr)?;
let listener = try!(TcpListener::from_std(std_listener, &handle));
let std_listener = StdTcpListener::bind(addr).map_err(::Error::new_listen)?;
let listener = TcpListener::from_std(std_listener, &handle).map_err(::Error::new_listen)?;
Ok(Server {
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
/// connection.
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>,
Bd: Entity<Data=B, Error=::Error>,
where
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 std_listener = StdTcpListener::bind(addr)?;
let listener = TcpListener::from_std(std_listener, &handle)?;
let mut incoming = AddrIncoming::new(listener, handle.clone(), self.sleep_on_errors)?;
let std_listener = StdTcpListener::bind(addr).map_err(::Error::new_listen)?;
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).map_err(::Error::new_listen)?;
if self.keep_alive {
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
/// connection.
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>,
Bd: Entity<Data=B, Error=::Error>,
where
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 listener = TcpListener::from_std(std_listener, &handle)?;
let mut incoming = AddrIncoming::new(listener, handle.clone(), self.sleep_on_errors)?;
let std_listener = StdTcpListener::bind(addr).map_err(::Error::new_listen)?;
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).map_err(::Error::new_listen)?;
if self.keep_alive {
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.
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,
S: NewService<Request = Request<Body>, Response = Response<Bd>, Error = ::Error>,
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>,
{
Serve {
incoming: incoming,
@@ -279,8 +288,10 @@ impl<B: AsRef<[u8]> + 'static> Http<B> {
/// # fn main() {}
/// ```
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>,
Bd: Entity<Error=::Error>,
where
S: Service<Request = Request<Body>, Response = Response<Bd>>,
S::Error: Into<Box<::std::error::Error + Send + Sync>>,
Bd: Entity,
I: AsyncRead + AsyncWrite,
{
let mut conn = proto::Conn::new(io);
@@ -341,15 +352,19 @@ impl Future for Run {
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 as Service>::Future: Send,
B: Entity<Error=::Error> + Send + 'static,
B: Entity + Send + 'static,
B::Data: Send,
{
/// Returns the local address that this server is bound to.
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
@@ -393,7 +408,7 @@ impl<S, B> Server<S, B>
let mut incoming = match AddrIncoming::new(listener, handle.clone(), protocol.sleep_on_errors) {
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 {
@@ -439,7 +454,7 @@ impl<S, B> Server<S, B>
let main_execution = shutdown_signal.select(srv).then(move |result| {
match result {
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
@@ -454,7 +469,8 @@ impl<S, B> Server<S, B>
future::Either::B(wait.select(timeout).then(|result| {
match result {
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 {
f.debug_struct("Server")
.field("reactor", &"...")
.field("listener", &self.listener)
.field("new_service", &self.new_service)
.field("protocol", &self.protocol)
@@ -499,15 +514,16 @@ impl<I, S, B> Stream for Serve<I, S>
where
I: Stream<Error=io::Error>,
I::Item: AsyncRead + AsyncWrite,
S: NewService<Request=Request<Body>, Response=Response<B>, Error=::Error>,
B: Entity<Error=::Error>,
S: NewService<Request=Request<Body>, Response=Response<B>>,
S::Error: Into<Box<::std::error::Error + Send + Sync>>,
B: Entity,
{
type Item = Connection<I::Item, S::Instance>;
type Error = ::Error;
fn poll(&mut self) -> Poll<Option<Self::Item>, Self::Error> {
if let Some(io) = try_ready!(self.incoming.poll()) {
let service = self.new_service.new_service()?;
if let Some(io) = try_ready!(self.incoming.poll().map_err(::Error::new_accept)) {
let service = self.new_service.new_service().map_err(::Error::new_user_new_service)?;
Ok(Async::Ready(Some(self.protocol.serve_connection(io, service))))
} else {
Ok(Async::Ready(None))
@@ -579,6 +595,12 @@ impl AddrIncoming {
fn set_keepalive(&mut self, dur: Option<Duration>) {
self.keep_alive_timeout = dur;
}
/*
fn set_sleep_on_errors(&mut self, val: bool) {
self.sleep_on_errors = val;
}
*/
}
impl Stream for AddrIncoming {
@@ -802,9 +824,9 @@ mod hyper_service {
S: Service<
Request=Request<Body>,
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
@@ -812,10 +834,10 @@ mod hyper_service {
S: Service<
Request=Request<Body>,
Response=Response<B>,
Error=::Error,
>,
S::Error: Into<Box<::std::error::Error + Send + Sync>>,
S: Sealed,
B: Entity<Error=::Error>,
B: Entity,
{
type ResponseBody = B;
type Sealed = Opaque;

106
tests/client.rs
View File

@@ -8,7 +8,7 @@ extern crate tokio;
extern crate tokio_io;
extern crate pretty_env_logger;
use std::io::{self, Read, Write};
use std::io::{Read, Write};
use std::net::{SocketAddr, TcpListener};
use std::thread;
use std::time::Duration;
@@ -142,7 +142,7 @@ macro_rules! test {
let _ = pretty_env_logger::try_init();
let runtime = Runtime::new().expect("runtime new");
let err = test! {
let err: ::hyper::Error = test! {
INNER;
name: $name,
runtime: &runtime,
@@ -157,7 +157,11 @@ macro_rules! test {
headers: { $($request_header_name => $request_header_val,)* },
body: $request_body,
}.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)
}
}
@@ -228,7 +232,7 @@ macro_rules! test {
let _ = tx.send(());
}).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()
});
@@ -485,10 +489,7 @@ test! {
url: "http://{addr}/err",
headers: {},
body: None,
error: |err| match err {
&hyper::Error::Incomplete => true,
_ => false,
},
error: |err| err.to_string() == "message is incomplete",
}
test! {
@@ -511,10 +512,8 @@ test! {
url: "http://{addr}/err",
headers: {},
body: None,
error: |err| match err {
&hyper::Error::Version => true,
_ => false,
},
// should get a Parse(Version) error
error: |err| err.is_parse(),
}
@@ -574,10 +573,7 @@ test! {
url: "http://{addr}/upgrade",
headers: {},
body: None,
error: |err| match err {
&hyper::Error::Upgrade => true,
_ => false,
},
error: |err| err.to_string() == "unsupported protocol upgrade",
}
@@ -599,10 +595,7 @@ test! {
url: "http://{addr}/",
headers: {},
body: None,
error: |err| match err {
&hyper::Error::Method => true,
_ => false,
},
error: |err| err.is_user(),
}
@@ -689,9 +682,9 @@ mod dispatch_impl {
let res = client.request(req).and_then(move |res| {
assert_eq!(res.status(), hyper::StatusCode::OK);
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();
closes.into_future().wait().unwrap().0.expect("closes");
@@ -736,11 +729,11 @@ mod dispatch_impl {
res.into_body().into_stream().concat2()
}).and_then(|_| {
Delay::new(Duration::from_secs(1))
.from_err()
.expect("timeout")
})
};
// 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();
closes.into_future().wait().unwrap().0.expect("closes");
@@ -788,7 +781,7 @@ mod dispatch_impl {
assert_eq!(res.status(), hyper::StatusCode::OK);
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();
// not closed yet, just idle
@@ -904,7 +897,7 @@ mod dispatch_impl {
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();
let t = Delay::new(Duration::from_millis(100))
@@ -955,7 +948,7 @@ mod dispatch_impl {
assert_eq!(res.status(), hyper::StatusCode::OK);
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();
let t = Delay::new(Duration::from_millis(100))
@@ -1003,10 +996,9 @@ mod dispatch_impl {
assert_eq!(res.status(), hyper::StatusCode::OK);
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();
let t = Delay::new(Duration::from_millis(100))
.map(|_| panic!("time out"));
let close = closes.into_future()
@@ -1049,13 +1041,12 @@ mod dispatch_impl {
assert_eq!(res.status(), hyper::StatusCode::OK);
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 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();
let t = Delay::new(Duration::from_millis(100))
.map(|_| panic!("time out"));
let close = closes.into_future()
@@ -1129,7 +1120,7 @@ mod dispatch_impl {
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()
.uri(&*format!("http://{}/a", addr))
.body(Body::empty())
@@ -1143,7 +1134,7 @@ mod dispatch_impl {
// state and back into client pool
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()
.uri(&*format!("http://{}/b", addr))
.body(Body::empty())
@@ -1194,7 +1185,7 @@ mod dispatch_impl {
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()
.method("HEAD")
.uri(&*format!("http://{}/a", addr))
@@ -1205,7 +1196,7 @@ mod dispatch_impl {
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()
.uri(&*format!("http://{}/b", addr))
.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()
.uri(&*format!("http://{}/foo/bar", addr))
.body(Body::empty())
@@ -1354,7 +1345,7 @@ mod conn {
use hyper::{self, Request};
use hyper::client::conn;
use super::{s, tcp_connect};
use super::{s, tcp_connect, FutureHyperExt};
#[test]
fn get() {
@@ -1395,10 +1386,10 @@ mod conn {
assert_eq!(res.status(), hyper::StatusCode::OK);
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 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();
}
@@ -1441,10 +1432,10 @@ mod conn {
assert_eq!(res.status(), hyper::StatusCode::OK);
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 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();
}
@@ -1490,17 +1481,14 @@ mod conn {
let res2 = client.send_request(req)
.then(|result| {
let err = result.expect_err("res2");
match err {
hyper::Error::Cancel(..) => (),
other => panic!("expected Cancel, found {:?}", other),
}
assert!(err.is_canceled(), "err not canceled, {:?}", err);
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 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();
}
@@ -1558,10 +1546,10 @@ mod conn {
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 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();
// should not be ready now
@@ -1641,10 +1629,10 @@ mod conn {
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 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();
// should not be ready now
@@ -1697,3 +1685,17 @@ mod conn {
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();
Http::<hyper::Chunk>::new()
.serve_connection(socket, service_fn(|_| {
Ok(Response::builder()
Ok::<_, hyper::Error>(Response::builder()
.status(StatusCode::CONTINUE)
.body(Body::empty())
.unwrap())
@@ -988,7 +988,7 @@ fn upgrades() {
.header("upgrade", "foobar")
.body(hyper::Body::empty())
.unwrap();
Ok(res)
Ok::<_, hyper::Error>(res)
}));
let mut conn_opt = Some(conn);
@@ -1144,10 +1144,10 @@ fn streaming_body() {
.keep_alive(false)
.serve_connection(socket, service_fn(|_| {
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);
let b = hyper::Body::wrap_stream(b);
Ok(Response::new(b))
Ok::<_, hyper::Error>(Response::new(b))
}))
.map(|_| ())
});