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

feat(lib): replace types with those from http crate

Browse Source 
BREAKING CHANGE: `Method`, `Request`, `Response`, `StatusCode`,
  `Version`, and `Uri` have been replaced with types from the `http`
  crate. The `hyper::header` module is gone for now.

  Removed `Client::get`, since it needed to construct a `Request<B>`
  with an empty body. Just use `Client::request` instead.

  Removed `compat` cargo feature, and `compat` related API.
This commit is contained in:
Sean McArthur
2018-02-28 16:37:17 -08:00
parent a37e6b59e6
commit 3cd48b45fb
109 changed files with 1004 additions and 14411 deletions
Download Patch File Download Diff File Expand all files Collapse all files

55
src/client/compat.rs
View File

@@ -1,55 +0,0 @@
//! Wrappers to build compatibility with the `http` crate.
use futures::{Future, Poll, Stream};
use http;
use tokio_service::Service;
use client::{Connect, Client, FutureResponse};
use error::Error;
use proto::Body;
/// A Client to make outgoing HTTP requests.
#[derive(Debug)]
pub struct CompatClient<C, B = Body> {
inner: Client<C, B>
}
pub(super) fn client<C, B>(client: Client<C, B>) -> CompatClient<C, B> {
CompatClient { inner: client }
}
impl<C, B> Service for CompatClient<C, B>
where C: Connect,
B: Stream<Error=Error> + 'static,
B::Item: AsRef<[u8]>,
{
type Request = http::Request<B>;
type Response = http::Response<Body>;
type Error = Error;
type Future = CompatFutureResponse;
fn call(&self, req: Self::Request) -> Self::Future {
future(self.inner.call(req.into()))
}
}
/// A `Future` that will resolve to an `http::Response`.
#[must_use = "futures do nothing unless polled"]
#[derive(Debug)]
pub struct CompatFutureResponse {
inner: FutureResponse
}
pub(super) fn future(fut: FutureResponse) -> CompatFutureResponse {
CompatFutureResponse { inner: fut }
}
impl Future for CompatFutureResponse {
type Item = http::Response<Body>;
type Error = Error;
fn poll(&mut self) -> Poll<Self::Item, Error> {
self.inner.poll()
.map(|a| a.map(|r| r.into()))
}
}

38
src/client/conn.rs
View File

@@ -16,7 +16,8 @@ use futures::future::{self, Either};
use tokio_io::{AsyncRead, AsyncWrite};
use proto;
use super::{dispatch, Request, Response};
use super::dispatch;
use {Body, Request, Response, StatusCode};
/// Returns a `Handshake` future over some IO.
///
@@ -31,7 +32,7 @@ where
/// The sender side of an established connection.
pub struct SendRequest<B> {
dispatch: dispatch::Sender<proto::dispatch::ClientMsg<B>, ::Response>,
dispatch: dispatch::Sender<proto::dispatch::ClientMsg<B>, Response<Body>>,
}
@@ -79,7 +80,7 @@ pub struct Handshake<T, B> {
pub struct ResponseFuture {
// for now, a Box is used to hide away the internal `B`
// that can be returned if canceled
inner: Box<Future<Item=Response, Error=::Error> + Send>,
inner: Box<Future<Item=Response<Body>, Error=::Error> + Send>,
}
/// Deconstructed parts of a `Connection`.
@@ -158,17 +159,20 @@ where
/// ```
/// # extern crate futures;
/// # extern crate hyper;
/// # extern crate http;
/// # use http::header::HOST;
/// # use hyper::client::conn::SendRequest;
/// # use hyper::Body;
/// use futures::Future;
/// use hyper::{Method, Request};
/// use hyper::header::Host;
/// use hyper::Request;
///
/// # fn doc(mut tx: SendRequest<Body>) {
/// // build a Request
/// let path = "/foo/bar".parse().expect("valid path");
/// let mut req = Request::new(Method::Get, path);
/// req.headers_mut().set(Host::new("hyper.rs", None));
/// let req = Request::builder()
/// .uri("/foo/bar")
/// .header(HOST, "hyper.rs")
/// .body(Body::empty())
/// .unwrap();
///
/// // send it and get a future back
/// let fut = tx.send_request(req)
@@ -180,7 +184,8 @@ where
/// # }
/// # fn main() {}
/// ```
pub fn send_request(&mut self, mut req: Request<B>) -> ResponseFuture {
pub fn send_request(&mut self, req: Request<B>) -> ResponseFuture {
/* TODO?
// The Connection API does less things automatically than the Client
// API does. For instance, right here, we always assume set_proxy, so
// that if an absolute-form URI is provided, it is serialized as-is.
@@ -191,9 +196,9 @@ where
// It's important that this method isn't called directly from the
// `Client`, so that `set_proxy` there is still respected.
req.set_proxy(true);
*/
let (head, body) = proto::request::split(req);
let inner = match self.dispatch.send((head, body)) {
let inner = match self.dispatch.send(req) {
Ok(rx) => {
Either::A(rx.then(move |res| {
match res {
@@ -210,15 +215,15 @@ where
Either::B(future::err(err))
}
};
ResponseFuture {
inner: Box::new(inner),
}
}
//TODO: replace with `impl Future` when stable
pub(crate) fn send_request_retryable(&mut self, req: Request<B>) -> Box<Future<Item=Response, Error=(::Error, Option<(::proto::RequestHead, Option<B>)>)>> {
let (head, body) = proto::request::split(req);
let inner = match self.dispatch.try_send((head, body)) {
pub(crate) fn send_request_retryable(&mut self, req: Request<B>) -> Box<Future<Item=Response<Body>, Error=(::Error, Option<Request<B>>)>> {
let inner = match self.dispatch.try_send(req) {
Ok(rx) => {
Either::A(rx.then(move |res| {
match res {
@@ -418,7 +423,7 @@ where
B: Stream<Error=::Error> + 'static,
B::Item: AsRef<[u8]>,
R: proto::Http1Transaction<
Incoming=proto::RawStatus,
Incoming=StatusCode,
Outgoing=proto::RequestLine,
>,
{
@@ -451,7 +456,7 @@ where
// ===== impl ResponseFuture
impl Future for ResponseFuture {
type Item = Response;
type Item = Response<Body>;
type Error = ::Error;
#[inline]
@@ -497,3 +502,4 @@ impl AssertSendSync for Builder {}
#[doc(hidden)]
impl AssertSend for ResponseFuture {}

12
src/client/connect.rs
View File

@@ -9,11 +9,12 @@ use futures::{Future, Poll, Async};
use futures::future::{Executor, ExecuteError};
use futures::sync::oneshot;
use futures_cpupool::{Builder as CpuPoolBuilder};
use http::Uri;
use http::uri::Scheme;
use tokio_io::{AsyncRead, AsyncWrite};
use tokio::reactor::Handle;
use tokio::net::{TcpStream, TcpStreamNew};
use tokio_service::Service;
use Uri;
use super::dns;
@@ -118,10 +119,10 @@ impl Service for HttpConnector {
trace!("Http::connect({:?})", uri);
if self.enforce_http {
if uri.scheme() != Some("http") {
if uri.scheme_part() != Some(&Scheme::HTTP) {
return invalid_url(InvalidUrl::NotHttp, &self.handle);
}
} else if uri.scheme().is_none() {
} else if uri.scheme_part().is_none() {
return invalid_url(InvalidUrl::MissingScheme, &self.handle);
}
@@ -131,10 +132,7 @@ impl Service for HttpConnector {
};
let port = match uri.port() {
Some(port) => port,
None => match uri.scheme() {
Some("https") => 443,
_ => 80,
},
None => if uri.scheme_part() == Some(&Scheme::HTTPS) { 443 } else { 80 },
};
HttpConnecting {

138
src/client/mod.rs
View File

@@ -9,21 +9,14 @@ use std::time::Duration;
use futures::{Async, Future, Poll, Stream};
use futures::future::{self, Executor};
#[cfg(feature = "compat")]
use http;
use http::{Method, Request, Response, Uri, Version};
use http::header::{Entry, HeaderValue, HOST};
use tokio::reactor::Handle;
pub use tokio_service::Service;
use header::{Host};
use proto;
use proto::request;
use method::Method;
use proto::{self, Body};
use self::pool::Pool;
use uri::{self, Uri};
use version::HttpVersion;
pub use proto::response::Response;
pub use proto::request::Request;
pub use self::connect::{HttpConnector, Connect};
use self::background::{bg, Background};
@@ -34,8 +27,6 @@ mod connect;
pub(crate) mod dispatch;
mod dns;
mod pool;
#[cfg(feature = "compat")]
pub mod compat;
mod signal;
#[cfg(test)]
mod tests;
@@ -113,14 +104,29 @@ where C: Connect,
B: Stream<Error=::Error> + 'static,
B::Item: AsRef<[u8]>,
{
/// Send a GET Request using this Client.
#[inline]
pub fn get(&self, url: Uri) -> FutureResponse {
self.request(Request::new(Method::Get, url))
/// Send a `GET` request to the supplied `Uri`.
///
/// # Note
///
/// This requires that the `Entity` type have a `Default` implementation.
/// It *should* return an "empty" version of itself, such that
/// `Entity::is_end_stream` is `true`.
pub fn get(&self, uri: Uri) -> FutureResponse
where
B: Default,
{
let body = B::default();
if !body.is_end_stream() {
warn!("default Entity used for get() does not return true for is_end_stream");
}
let mut req = Request::new(body);
*req.uri_mut() = uri;
self.request(req)
}
/// Send a constructed Request using this Client.
#[inline]
pub fn request(&self, mut req: Request<B>) -> FutureResponse {
// TODO(0.12): do this at construction time.
//
@@ -131,23 +137,25 @@ where C: Connect,
self.schedule_pool_timer();
match req.version() {
HttpVersion::Http10 |
HttpVersion::Http11 => (),
Version::HTTP_10 |
Version::HTTP_11 => (),
other => {
error!("Request has unsupported version \"{}\"", other);
error!("Request has unsupported version \"{:?}\"", other);
return FutureResponse(Box::new(future::err(::Error::Version)));
}
}
if req.method() == &Method::Connect {
if req.method() == &Method::CONNECT {
debug!("Client does not support CONNECT requests");
return FutureResponse(Box::new(future::err(::Error::Method)));
}
let domain = match uri::scheme_and_authority(req.uri()) {
Some(uri) => uri,
None => {
debug!("request uri does not include scheme and authority");
let uri = req.uri().clone();
let domain = match (uri.scheme_part(), uri.authority_part()) {
(Some(scheme), Some(auth)) => {
format!("{}://{}", scheme, auth)
}
_ => {
return FutureResponse(Box::new(future::err(::Error::Io(
io::Error::new(
io::ErrorKind::InvalidInput,
@@ -156,45 +164,51 @@ where C: Connect,
))));
}
};
if self.set_host && !req.headers().has::<Host>() {
let host = Host::new(
domain.host().expect("authority implies host").to_owned(),
domain.port(),
);
req.headers_mut().set_pos(0, host);
if self.set_host {
if let Entry::Vacant(entry) = req.headers_mut().entry(HOST).expect("HOST is always valid header name") {
let hostname = uri.host().expect("authority implies host");
let host = if let Some(port) = uri.port() {
let s = format!("{}:{}", hostname, port);
HeaderValue::from_str(&s)
} else {
HeaderValue::from_str(hostname)
}.expect("uri host is valid header value");
entry.insert(host);
}
}
let client = self.clone();
let is_proxy = req.is_proxy();
let uri = req.uri().clone();
//TODO: let is_proxy = req.is_proxy();
//let uri = req.uri().clone();
let fut = RetryableSendRequest {
client: client,
future: self.send_request(req, &domain),
domain: domain,
is_proxy: is_proxy,
uri: uri,
//is_proxy: is_proxy,
//uri: uri,
};
FutureResponse(Box::new(fut))
}
/// Send an `http::Request` using this Client.
#[inline]
#[cfg(feature = "compat")]
pub fn request_compat(&self, req: http::Request<B>) -> compat::CompatFutureResponse {
self::compat::future(self.call(req.into()))
}
/// Convert into a client accepting `http::Request`.
#[cfg(feature = "compat")]
pub fn into_compat(self) -> compat::CompatClient<C, B> {
self::compat::client(self)
}
//TODO: replace with `impl Future` when stable
fn send_request(&self, req: Request<B>, domain: &Uri) -> Box<Future<Item=Response, Error=ClientError<B>>> {
//fn send_request(&self, req: Request<B>, domain: &Uri) -> Box<Future<Item=Response, Error=::Error>> {
fn send_request(&self, mut req: Request<B>, domain: &str) -> Box<Future<Item=Response<Body>, Error=ClientError<B>>> {
let url = req.uri().clone();
let checkout = self.pool.checkout(domain.as_ref());
let path = match url.path_and_query() {
Some(path) => {
let mut parts = ::http::uri::Parts::default();
parts.path_and_query = Some(path.clone());
Uri::from_parts(parts).expect("path is valid uri")
},
None => {
"/".parse().expect("/ is valid path")
}
};
*req.uri_mut() = path;
let checkout = self.pool.checkout(domain);
let connect = {
let executor = self.executor.clone();
let pool = self.pool.clone();
@@ -228,7 +242,6 @@ where C: Connect,
ClientError::Normal(e)
});
let executor = self.executor.clone();
let resp = race.and_then(move |mut pooled| {
let conn_reused = pooled.is_reused();
@@ -284,7 +297,7 @@ where C: Connect,
B::Item: AsRef<[u8]>,
{
type Request = Request<B>;
type Response = Response;
type Response = Response<Body>;
type Error = ::Error;
type Future = FutureResponse;
@@ -315,7 +328,7 @@ impl<C, B> fmt::Debug for Client<C, B> {
/// A `Future` that will resolve to an HTTP Response.
#[must_use = "futures do nothing unless polled"]
pub struct FutureResponse(Box<Future<Item=Response, Error=::Error> + 'static>);
pub struct FutureResponse(Box<Future<Item=Response<Body>, Error=::Error> + 'static>);
impl fmt::Debug for FutureResponse {
fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
@@ -324,7 +337,7 @@ impl fmt::Debug for FutureResponse {
}
impl Future for FutureResponse {
type Item = Response;
type Item = Response<Body>;
type Error = ::Error;
fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
@@ -334,10 +347,10 @@ impl Future for FutureResponse {
struct RetryableSendRequest<C, B> {
client: Client<C, B>,
domain: Uri,
future: Box<Future<Item=Response, Error=ClientError<B>>>,
is_proxy: bool,
uri: Uri,
domain: String,
future: Box<Future<Item=Response<Body>, Error=ClientError<B>>>,
//is_proxy: bool,
//uri: Uri,
}
impl<C, B> Future for RetryableSendRequest<C, B>
@@ -346,7 +359,7 @@ where
B: Stream<Error=::Error> + 'static,
B::Item: AsRef<[u8]>,
{
type Item = Response;
type Item = Response<Body>;
type Error = ::Error;
fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
@@ -367,9 +380,6 @@ where
}
trace!("unstarted request canceled, trying again (reason={:?})", reason);
let mut req = request::join(req);
req.set_proxy(self.is_proxy);
req.set_uri(self.uri.clone());
self.future = self.client.send_request(req, &self.domain);
}
}
@@ -394,7 +404,7 @@ pub(crate) enum ClientError<B> {
Normal(::Error),
Canceled {
connection_reused: bool,
req: (::proto::RequestHead, Option<B>),
req: Request<B>,
reason: ::Error,
}
}

27
src/client/tests.rs
View File

@@ -23,7 +23,12 @@ fn retryable_request() {
{
let res1 = client.get("http://mock.local/a".parse().unwrap());
let req = Request::builder()
.uri("http://mock.local/a")
.body(Default::default())
.unwrap();
let res1 = client.request(req);
let srv1 = poll_fn(|| {
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"));
@@ -33,7 +38,11 @@ fn retryable_request() {
}
drop(sock1);
let res2 = client.get("http://mock.local/b".parse().unwrap())
let req = Request::builder()
.uri("http://mock.local/b")
.body(Default::default())
.unwrap();
let res2 = client.request(req)
.map(|res| {
assert_eq!(res.status().as_u16(), 222);
});
@@ -61,7 +70,13 @@ fn conn_reset_after_write() {
{
let res1 = client.get("http://mock.local/a".parse().unwrap());
let req = Request::builder()
.uri("http://mock.local/a")
//TODO: remove this header when auto lengths are fixed
.header("content-length", "0")
.body(Default::default())
.unwrap();
let res1 = client.request(req);
let srv1 = poll_fn(|| {
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"));
@@ -70,7 +85,11 @@ fn conn_reset_after_write() {
core.run(res1.join(srv1)).expect("res1");
}
let res2 = client.get("http://mock.local/a".parse().unwrap());
let req = Request::builder()
.uri("http://mock.local/a")
.body(Default::default())
.unwrap();
let res2 = client.request(req);
let mut sock1 = Some(sock1);
let srv2 = poll_fn(|| {
// We purposefully keep the socket open until the client