forked/reqwest-impersonate
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Update to hyper 0.13

Browse Source
This commit is contained in:
Gleb Pomykalov
2019-12-11 03:24:05 +03:00
committed by Sean McArthur
parent db2de90e42
commit 0f32c4a01a
23 changed files with 469 additions and 219 deletions
Download Patch File Download Diff File Expand all files Collapse all files

18
src/async_impl/body.rs
View File

@@ -6,7 +6,7 @@ use std::task::{Context, Poll};
use bytes::Bytes;
use futures_core::Stream;
use http_body::Body as HttpBody;
use tokio::timer::Delay;
use tokio::time::Delay;
/// An asynchronous request body.
pub struct Body {
@@ -21,7 +21,7 @@ enum Inner {
Streaming {
body: Pin<
Box<
dyn HttpBody<Data = hyper::Chunk, Error = Box<dyn std::error::Error + Send + Sync>>
dyn HttpBody<Data = Bytes, Error = Box<dyn std::error::Error + Send + Sync>>
+ Send
+ Sync,
>,
@@ -73,7 +73,7 @@ impl Body {
where
S: futures_core::stream::TryStream + Send + Sync + 'static,
S::Error: Into<Box<dyn std::error::Error + Send + Sync>>,
hyper::Chunk: From<S::Ok>,
Bytes: From<S::Ok>,
{
Body::stream(stream)
}
@@ -82,12 +82,12 @@ impl Body {
where
S: futures_core::stream::TryStream + Send + Sync + 'static,
S::Error: Into<Box<dyn std::error::Error + Send + Sync>>,
hyper::Chunk: From<S::Ok>,
Bytes: From<S::Ok>,
{
use futures_util::TryStreamExt;
let body = Box::pin(WrapStream(
stream.map_ok(hyper::Chunk::from).map_err(Into::into),
stream.map_ok(Bytes::from).map_err(Into::into),
));
Body {
inner: Inner::Streaming {
@@ -198,7 +198,7 @@ impl fmt::Debug for Body {
// ===== impl ImplStream =====
impl HttpBody for ImplStream {
type Data = hyper::Chunk;
type Data = Bytes;
type Error = crate::Error;
fn poll_data(
@@ -291,10 +291,10 @@ impl Stream for ImplStream {
impl<S, D, E> HttpBody for WrapStream<S>
where
S: Stream<Item = Result<D, E>>,
D: Into<hyper::Chunk>,
D: Into<Bytes>,
E: Into<Box<dyn std::error::Error + Send + Sync>>,
{
type Data = hyper::Chunk;
type Data = Bytes;
type Error = E;
fn poll_data(
@@ -321,7 +321,7 @@ where
// ===== impl WrapHyper =====
impl HttpBody for WrapHyper {
type Data = hyper::Chunk;
type Data = Bytes;
type Error = Box<dyn std::error::Error + Send + Sync>;
fn poll_data(

9
src/async_impl/client.rs
View File

@@ -11,13 +11,14 @@ use http::header::{
CONTENT_TYPE, LOCATION, PROXY_AUTHORIZATION, RANGE, REFERER, TRANSFER_ENCODING, USER_AGENT,
};
use http::Uri;
use http::uri::Scheme;
use hyper::client::ResponseFuture;
#[cfg(feature = "default-tls")]
use native_tls::TlsConnector;
use std::future::Future;
use std::pin::Pin;
use std::task::{Context, Poll};
use tokio::{clock, timer::Delay};
use tokio::time::Delay;
use log::debug;
@@ -726,7 +727,7 @@ impl Client {
// insert default headers in the request headers
// without overwriting already appended headers.
for (key, value) in &self.inner.headers {
if let Ok(Entry::Vacant(entry)) = headers.entry(key) {
if let Entry::Vacant(entry) = headers.entry(key) {
entry.insert(value.clone());
}
}
@@ -772,7 +773,7 @@ impl Client {
let timeout = self
.inner
.request_timeout
.map(|dur| tokio::timer::delay(clock::now() + dur));
.map(|dur| tokio::time::delay_for(dur));
Pending {
inner: PendingInner::Request(PendingRequest {
@@ -799,7 +800,7 @@ impl Client {
// Only set the header here if the destination scheme is 'http',
// since otherwise, the header will be included in the CONNECT tunnel
// request instead.
if dst.scheme_part() != Some(&::http::uri::Scheme::HTTP) {
if dst.scheme() != Some(&Scheme::HTTP) {
return;
}

2
src/async_impl/decoder.rs
View File

@@ -192,7 +192,7 @@ mod imp {
fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
use futures_util::StreamExt;
match futures_core::ready!(Pin::new(&mut self.0).peek(cx)) {
match futures_core::ready!(Pin::new(&mut self.0).poll_peek(cx)) {
Some(Ok(_)) => {
// fallthrough
}

18
src/async_impl/multipart.rs
View File

@@ -3,7 +3,7 @@ use std::borrow::Cow;
use std::fmt;
use std::pin::Pin;
use bytes::Bytes;
use bytes::{Bytes};
use http::HeaderMap;
use mime_guess::Mime;
use percent_encoding::{self, AsciiSet, NON_ALPHANUMERIC};
@@ -536,18 +536,18 @@ mod tests {
use super::*;
use futures_util::TryStreamExt;
use futures_util::{future, stream};
use tokio;
use tokio::{self, runtime};
#[test]
fn form_empty() {
let form = Form::new();
let mut rt = tokio::runtime::current_thread::Runtime::new().expect("new rt");
let mut rt = runtime::Builder::new().basic_scheduler().enable_all().build().expect("new rt");
let body = form.stream().into_stream();
let s = body.map(|try_c| try_c.map(Bytes::from)).try_concat();
let s = body.map_ok(|try_c| try_c.to_vec()).try_concat();
let out = rt.block_on(s);
assert_eq!(out.unwrap(), Vec::new());
assert!(out.unwrap().is_empty());
}
#[test]
@@ -590,9 +590,9 @@ mod tests {
--boundary\r\n\
Content-Disposition: form-data; name=\"key3\"; filename=\"filename\"\r\n\r\n\
value3\r\n--boundary--\r\n";
let mut rt = tokio::runtime::current_thread::Runtime::new().expect("new rt");
let mut rt = runtime::Builder::new().basic_scheduler().enable_all().build().expect("new rt");
let body = form.stream().into_stream();
let s = body.map(|try_c| try_c.map(Bytes::from)).try_concat();
let s = body.map(|try_c| try_c.map(|r| r.to_vec())).try_concat();
let out = rt.block_on(s).unwrap();
// These prints are for debug purposes in case the test fails
@@ -617,9 +617,9 @@ mod tests {
\r\n\
value2\r\n\
--boundary--\r\n";
let mut rt = tokio::runtime::current_thread::Runtime::new().expect("new rt");
let mut rt = runtime::Builder::new().basic_scheduler().enable_all().build().expect("new rt");
let body = form.stream().into_stream();
let s = body.map(|try_c| try_c.map(Bytes::from)).try_concat();
let s = body.map(|try_c| try_c.map(|r| r.to_vec())).try_concat();
let out = rt.block_on(s).unwrap();
// These prints are for debug purposes in case the test fails

15
src/async_impl/request.rs
View File

@@ -1,10 +1,10 @@
use std::convert::TryFrom;
use std::fmt;
use std::future::Future;
use std::io::Write;
use base64;
use base64::write::EncoderWriter as Base64Encoder;
use bytes::Bytes;
use serde::Serialize;
#[cfg(feature = "json")]
use serde_json;
@@ -16,7 +16,6 @@ use super::multipart;
use super::response::Response;
use crate::header::{HeaderMap, HeaderName, HeaderValue, CONTENT_LENGTH, CONTENT_TYPE};
use crate::{Method, Url};
use http::HttpTryFrom;
/// A request which can be executed with `Client::execute()`.
pub struct Request {
@@ -119,13 +118,15 @@ impl RequestBuilder {
/// Add a `Header` to this Request.
pub fn header<K, V>(mut self, key: K, value: V) -> RequestBuilder
where
HeaderName: HttpTryFrom<K>,
HeaderValue: HttpTryFrom<V>,
HeaderName: TryFrom<K>,
<HeaderName as TryFrom<K>>::Error: Into<http::Error>,
HeaderValue: TryFrom<V>,
<HeaderValue as TryFrom<V>>::Error: Into<http::Error>,
{
let mut error = None;
if let Ok(ref mut req) = self.request {
match <HeaderName as HttpTryFrom<K>>::try_from(key) {
Ok(key) => match <HeaderValue as HttpTryFrom<V>>::try_from(value) {
match <HeaderName as TryFrom<K>>::try_from(key) {
Ok(key) => match <HeaderValue as TryFrom<V>>::try_from(value) {
Ok(value) => {
req.headers_mut().append(key, value);
}
@@ -166,7 +167,7 @@ impl RequestBuilder {
}
}
self.header(crate::header::AUTHORIZATION, Bytes::from(header_value))
self.header(crate::header::AUTHORIZATION, header_value)
}
/// Enable HTTP bearer authentication.

19
src/async_impl/response.rs
View File

@@ -2,9 +2,9 @@ use std::borrow::Cow;
use std::fmt;
use std::net::SocketAddr;
use bytes::Bytes;
use bytes::{Bytes, BytesMut};
use encoding_rs::{Encoding, UTF_8};
use futures_util::{StreamExt, TryStreamExt};
use futures_util::stream::StreamExt;
use http;
use hyper::client::connect::HttpInfo;
use hyper::header::CONTENT_LENGTH;
@@ -15,7 +15,7 @@ use mime::Mime;
use serde::de::DeserializeOwned;
#[cfg(feature = "json")]
use serde_json;
use tokio::timer::Delay;
use tokio::time::Delay;
use url::Url;
use super::body::Body;
@@ -260,8 +260,12 @@ impl Response {
/// # Ok(())
/// # }
/// ```
pub async fn bytes(self) -> crate::Result<Bytes> {
self.body.try_concat().await
pub async fn bytes(mut self) -> crate::Result<Bytes> {
let mut buf = BytesMut::new();
while let Some(chunk) = self.body.next().await {
buf.extend(chunk?);
}
Ok(buf.freeze())
}
/// Stream a chunk of the response body.
@@ -408,11 +412,12 @@ struct ResponseUrl(Url);
pub trait ResponseBuilderExt {
/// A builder method for the `http::response::Builder` type that allows the user to add a `Url`
/// to the `http::Response`
fn url(&mut self, url: Url) -> &mut Self;
fn url(self, url: Url) -> Self;
}
impl ResponseBuilderExt for http::response::Builder {
fn url(&mut self, url: Url) -> &mut Self {
fn url(self, url: Url) -> Self {
self.extension(ResponseUrl(url))
}
}

14
src/blocking/body.rs
View File

@@ -2,6 +2,8 @@ use std::fmt;
use std::fs::File;
use std::future::Future;
use std::io::{self, Cursor, Read};
use std::mem::{self, MaybeUninit};
use std::ptr;
use bytes::Bytes;
@@ -246,9 +248,17 @@ async fn send_future(sender: Sender) -> Result<(), crate::Error> {
if buf.is_empty() {
if buf.remaining_mut() == 0 {
buf.reserve(8192);
// zero out the reserved memory
unsafe {
let uninit = mem::transmute::<&mut [MaybeUninit<u8>], &mut [u8]>(buf.bytes_mut());
ptr::write_bytes(uninit.as_mut_ptr(), 0, uninit.len());
}
}
match body.read(unsafe { buf.bytes_mut() }) {
let bytes = unsafe {
mem::transmute::<&mut [MaybeUninit<u8>], &mut [u8]>(buf.bytes_mut())
};
match body.read(bytes) {
Ok(0) => {
// The buffer was empty and nothing's left to
// read. Return.
@@ -270,7 +280,7 @@ async fn send_future(sender: Sender) -> Result<(), crate::Error> {
let buf_len = buf.len() as u64;
tx.as_mut()
.expect("tx only taken on error")
.send_data(buf.take().freeze().into())
.send_data(buf.split().freeze())
.await
.map_err(crate::error::body)?;

8
src/blocking/client.rs
View File

@@ -593,9 +593,8 @@ impl ClientHandle {
let handle = thread::Builder::new()
.name("reqwest-internal-sync-runtime".into())
.spawn(move || {
use tokio::runtime::current_thread::Runtime;
let mut rt = match Runtime::new().map_err(crate::error::builder) {
use tokio::runtime;
let mut rt = match runtime::Builder::new().basic_scheduler().enable_all().build().map_err(crate::error::builder) {
Err(e) => {
if let Err(e) = spawn_tx.send(Err(e)) {
error!("Failed to communicate runtime creation failure: {:?}", e);
@@ -685,7 +684,6 @@ impl ClientHandle {
KeepCoreThreadAlive(Some(self.inner.clone())),
)),
Err(wait::Waited::TimedOut(e)) => Err(crate::error::request(e).with_url(url)),
Err(wait::Waited::Executor(err)) => Err(crate::error::request(err).with_url(url)),
Err(wait::Waited::Inner(err)) => Err(err.with_url(url)),
}
}
@@ -705,7 +703,7 @@ where
Poll::Ready(val) => Poll::Ready(Some(val)),
Poll::Pending => {
// check if the callback is canceled
futures_core::ready!(tx.poll_cancel(cx));
futures_core::ready!(tx.poll_canceled(cx));
Poll::Ready(None)
}
}

12
src/blocking/request.rs
View File

@@ -1,7 +1,7 @@
use std::fmt;
use std::convert::TryFrom;
use base64::encode;
use http::HttpTryFrom;
use serde::Serialize;
#[cfg(feature = "json")]
use serde_json;
@@ -140,13 +140,15 @@ impl RequestBuilder {
/// ```
pub fn header<K, V>(mut self, key: K, value: V) -> RequestBuilder
where
HeaderName: HttpTryFrom<K>,
HeaderValue: HttpTryFrom<V>,
HeaderName: TryFrom<K>,
HeaderValue: TryFrom<V>,
<HeaderName as TryFrom<K>>::Error: Into<http::Error>,
<HeaderValue as TryFrom<V>>::Error: Into<http::Error>,
{
let mut error = None;
if let Ok(ref mut req) = self.request {
match <HeaderName as HttpTryFrom<K>>::try_from(key) {
Ok(key) => match <HeaderValue as HttpTryFrom<V>>::try_from(value) {
match <HeaderName as TryFrom<K>>::try_from(key) {
Ok(key) => match <HeaderValue as TryFrom<V>>::try_from(value) {
Ok(value) => {
req.headers_mut().append(key, value);
}

3
src/blocking/response.rs
View File

@@ -220,7 +220,6 @@ impl Response {
pub fn json<T: DeserializeOwned>(self) -> crate::Result<T> {
wait::timeout(self.inner.json(), self.timeout).map_err(|e| match e {
wait::Waited::TimedOut(e) => crate::error::decode(e),
wait::Waited::Executor(e) => crate::error::decode(e),
wait::Waited::Inner(e) => e,
})
}
@@ -269,7 +268,6 @@ impl Response {
wait::timeout(self.inner.text_with_charset(default_encoding), self.timeout).map_err(|e| {
match e {
wait::Waited::TimedOut(e) => crate::error::decode(e),
wait::Waited::Executor(e) => crate::error::decode(e),
wait::Waited::Inner(e) => e,
}
})
@@ -375,7 +373,6 @@ impl Read for Response {
let timeout = self.timeout;
wait::timeout(self.body_mut().read(buf), timeout).map_err(|e| match e {
wait::Waited::TimedOut(e) => crate::error::decode(e).into_io(),
wait::Waited::Executor(e) => crate::error::decode(e).into_io(),
wait::Waited::Inner(e) => e,
})
}

43
src/blocking/wait.rs
View File

@@ -1,29 +1,27 @@
use std::future::Future;
use std::sync::Arc;
use std::task::{Context, Poll};
use std::thread::{self, Thread};
use std::time::Duration;
use tokio::clock;
use tokio_executor::{
enter,
park::{Park, ParkThread, Unpark, UnparkThread},
};
use tokio::time::Instant;
pub(crate) fn timeout<F, I, E>(fut: F, timeout: Option<Duration>) -> Result<I, Waited<E>>
where
F: Future<Output = Result<I, E>>,
{
let _entered =
enter().map_err(|_| Waited::Executor(crate::error::BlockingClientInAsyncContext))?;
enter();
let deadline = timeout.map(|d| {
log::trace!("wait at most {:?}", d);
clock::now() + d
Instant::now() + d
});
let mut park = ParkThread::new();
// Arc shouldn't be necessary, since UnparkThread is reference counted internally,
let thread = ThreadWaker(thread::current());
// Arc shouldn't be necessary, since `Thread` is reference counted internally,
// but let's just stay safe for now.
let waker = futures_util::task::waker(Arc::new(UnparkWaker(park.unpark())));
let waker = futures_util::task::waker(Arc::new(thread));
let mut cx = Context::from_waker(&waker);
futures_util::pin_mut!(fut);
@@ -36,17 +34,16 @@ where
}
if let Some(deadline) = deadline {
let now = clock::now();
let now = Instant::now();
if now >= deadline {
log::trace!("wait timeout exceeded");
return Err(Waited::TimedOut(crate::error::TimedOut));
}
log::trace!("park timeout {:?}", deadline - now);
park.park_timeout(deadline - now)
.expect("ParkThread doesn't error");
thread::park_timeout(deadline - now);
} else {
park.park().expect("ParkThread doesn't error");
thread::park();
}
}
}
@@ -54,14 +51,24 @@ where
#[derive(Debug)]
pub(crate) enum Waited<E> {
TimedOut(crate::error::TimedOut),
Executor(crate::error::BlockingClientInAsyncContext),
Inner(E),
}
struct UnparkWaker(UnparkThread);
struct ThreadWaker(Thread);
impl futures_util::task::ArcWake for UnparkWaker {
impl futures_util::task::ArcWake for ThreadWaker {
fn wake_by_ref(arc_self: &Arc<Self>) {
arc_self.0.unpark();
}
}
fn enter() {
// Check we aren't already in a runtime
#[cfg(debug_assertions)]
{
tokio::runtime::Builder::new()
.build()
.expect("build shell runtime")
.enter(|| {});
}
}

333
src/connect.rs
View File

@@ -1,24 +1,31 @@
use futures_util::FutureExt;
use hyper::service::Service;
use http::uri::{Scheme, Authority};
use hyper::client::connect::{Connect, Connected, Destination};
use http::Uri;
use hyper::client::connect::{Connected, Connection};
use tokio::io::{AsyncRead, AsyncWrite};
#[cfg(feature = "default-tls")]
use native_tls::{TlsConnector, TlsConnectorBuilder};
#[cfg(feature = "tls")]
use http::header::HeaderValue;
use bytes::{Buf, BufMut};
use std::future::Future;
use std::io;
use std::net::IpAddr;
use std::pin::Pin;
use std::sync::Arc;
use std::task::{Context, Poll};
use std::time::Duration;
use std::mem::MaybeUninit;
use pin_project_lite::pin_project;
//#[cfg(feature = "trust-dns")]
//use crate::dns::TrustDnsResolver;
use crate::proxy::{Proxy, ProxyScheme};
use tokio::future::FutureExt as _;
use crate::error::BoxError;
#[cfg(feature = "default-tls")]
use self::native_tls_conn::NativeTlsConn;
//#[cfg(feature = "trust-dns")]
//type HttpConnector = hyper::client::HttpConnector<TrustDnsResolver>;
@@ -198,24 +205,27 @@ impl Connector {
async fn connect_with_maybe_proxy(
self,
dst: Destination,
dst: Uri,
is_proxy: bool,
) -> Result<(Conn, Connected), io::Error> {
) -> Result<Conn, BoxError> {
match self.inner {
#[cfg(not(feature = "tls"))]
Inner::Http(http) => {
let (io, connected) = http.connect(dst).await?;
Ok((Box::new(io) as Conn, connected.proxy(is_proxy)))
Inner::Http(mut http) => {
let io = http.call(dst).await?;
Ok(Conn {
inner: Box::new(io),
is_proxy,
})
}
#[cfg(feature = "default-tls")]
Inner::DefaultTls(http, tls) => {
let mut http = http.clone();
http.set_nodelay(self.nodelay || (dst.scheme() == "https"));
http.set_nodelay(self.nodelay || (dst.scheme() == Some(&Scheme::HTTPS)));
let tls_connector = tokio_tls::TlsConnector::from(tls.clone());
let http = hyper_tls::HttpsConnector::from((http, tls_connector));
let (io, connected) = http.connect(dst).await?;
let mut http = hyper_tls::HttpsConnector::from((http, tls_connector));
let io = http.call(dst).await?;
//TODO: where's this at now?
//if let hyper_tls::MaybeHttpsStream::Https(_stream) = &io {
// if !no_delay {
@@ -223,7 +233,10 @@ impl Connector {
// }
//}
Ok((Box::new(io) as Conn, connected.proxy(is_proxy)))
Ok(Conn {
inner: Box::new(io),
is_proxy,
})
}
#[cfg(feature = "rustls-tls")]
Inner::RustlsTls { http, tls, .. } => {
@@ -232,10 +245,10 @@ impl Connector {
// Disable Nagle's algorithm for TLS handshake
//
// https://www.openssl.org/docs/man1.1.1/man3/SSL_connect.html#NOTES
http.set_nodelay(no_delay || (dst.scheme() == "https"));
http.set_nodelay(no_delay || (dst.scheme() == Some(&Scheme::HTTPS)));
let http = hyper_rustls::HttpsConnector::from((http, tls.clone()));
let (io, connected) = http.connect(dst).await?;
let io = http.connect(dst).await?;
if let hyper_rustls::MaybeHttpsStream::Https(stream) = &io {
if !no_delay {
let (io, _) = stream.get_ref();
@@ -243,21 +256,24 @@ impl Connector {
}
}
Ok((Box::new(io) as Conn, connected.proxy(is_proxy)))
Ok(Conn {
inner: Box::new(io),
is_proxy,
})
}
}
}
async fn connect_via_proxy(
self,
dst: Destination,
dst: Uri,
proxy_scheme: ProxyScheme,
) -> Result<(Conn, Connected), io::Error> {
) -> Result<Conn, BoxError> {
log::trace!("proxy({:?}) intercepts {:?}", proxy_scheme, dst);
let (proxy_dst, _auth) = match proxy_scheme {
ProxyScheme::Http { host, auth } => (into_dst(Scheme::HTTP, host), auth),
ProxyScheme::Https { host, auth } => (into_dst(Scheme::HTTPS, host), auth),
ProxyScheme::Http { host, auth } => (into_uri(Scheme::HTTP, host), auth),
ProxyScheme::Https { host, auth } => (into_uri(Scheme::HTTPS, host), auth),
#[cfg(feature = "socks")]
ProxyScheme::Socks5 { .. } => return this.connect_socks(dst, proxy_scheme),
};
@@ -269,22 +285,33 @@ impl Connector {
match &self.inner {
#[cfg(feature = "default-tls")]
Inner::DefaultTls(http, tls) => {
if dst.scheme() == "https" {
if dst.scheme() == Some(&Scheme::HTTPS) {
let host = dst.host().to_owned();
let port = dst.port().unwrap_or(443);
let port = dst.port().map(|p| p.as_u16()).unwrap_or(443);
let mut http = http.clone();
http.set_nodelay(self.nodelay);
let tls_connector = tokio_tls::TlsConnector::from(tls.clone());
let http = hyper_tls::HttpsConnector::from((http, tls_connector));
let (conn, connected) = http.connect(proxy_dst).await?;
let mut http = hyper_tls::HttpsConnector::from((http, tls_connector));
let conn = http.call(proxy_dst).await?;
log::trace!("tunneling HTTPS over proxy");
let tunneled = tunnel(conn, host.clone(), port, self.user_agent.clone(), auth).await?;
let tunneled = tunnel(
conn,
host
.ok_or(io::Error::new(io::ErrorKind::Other, "no host in url"))?
.to_string(),
port,
self.user_agent.clone(),
auth
).await?;
let tls_connector = tokio_tls::TlsConnector::from(tls.clone());
let io = tls_connector
.connect(&host, tunneled)
.connect(&host.ok_or(io::Error::new(io::ErrorKind::Other, "no host in url"))?, tunneled)
.await
.map_err(|e| io::Error::new(io::ErrorKind::Other, e))?;
return Ok((Box::new(io) as Conn, connected.proxy(true)));
return Ok(Conn {
inner: Box::new(NativeTlsConn { inner: io }),
is_proxy: false,
});
}
}
#[cfg(feature = "rustls-tls")]
@@ -293,7 +320,7 @@ impl Connector {
tls,
tls_proxy,
} => {
if dst.scheme() == "https" {
if dst.scheme() == Some(&Scheme::HTTPS) {
use rustls::Session;
use tokio_rustls::webpki::DNSNameRef;
use tokio_rustls::TlsConnector as RustlsConnector;
@@ -320,7 +347,10 @@ impl Connector {
} else {
connected
};
return Ok((Box::new(io) as Conn, connected.proxy(true)));
return Ok(Conn {
inner: Box::new(io),
connected: Connected::new(),
});
}
}
#[cfg(not(feature = "tls"))]
@@ -331,9 +361,7 @@ impl Connector {
}
}
fn into_dst(scheme: Scheme, host: Authority) -> Destination {
use std::convert::TryInto;
fn into_uri(scheme: Scheme, host: Authority) -> Uri {
// TODO: Should the `http` crate get `From<(Scheme, Authority)> for Uri`?
http::Uri::builder()
.scheme(scheme)
@@ -341,8 +369,6 @@ fn into_dst(scheme: Scheme, host: Authority) -> Destination {
.path_and_query(http::uri::PathAndQuery::from_static("/"))
.build()
.expect("scheme and authority is valid Uri")
.try_into()
.expect("scheme and authority is valid Destination")
}
//#[cfg(feature = "trust-dns")]
@@ -358,26 +384,32 @@ fn http_connector() -> crate::Result<HttpConnector> {
}
async fn with_timeout<T, F>(f: F, timeout: Option<Duration>) -> Result<T, io::Error>
async fn with_timeout<T, F>(f: F, timeout: Option<Duration>) -> Result<T, BoxError>
where
F: Future<Output = Result<T, io::Error>>,
F: Future<Output = Result<T, BoxError>>,
{
if let Some(to) = timeout {
match f.timeout(to).await {
Err(_elapsed) => Err(io::Error::new(io::ErrorKind::TimedOut, "connect timed out")),
Ok(try_res) => try_res,
match tokio::time::timeout(to, f).await {
Err(_elapsed) => Err(Box::new(io::Error::new(io::ErrorKind::TimedOut, "connect timed out")) as BoxError),
Ok(Ok(try_res)) => Ok(try_res),
Ok(Err(e)) => Err(e),
}
} else {
f.await
}
}
impl Connect for Connector {
type Transport = Conn;
type Error = io::Error;
impl Service<Uri> for Connector
{
type Response = Conn;
type Error = BoxError;
type Future = Connecting;
fn connect(&self, dst: Destination) -> Self::Future {
fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
Poll::Ready(Ok(()))
}
fn call(&mut self, dst: Uri) -> Self::Future {
let timeout = self.timeout;
for prox in self.proxies.iter() {
if let Some(proxy_scheme) = prox.intercept(&dst) {
@@ -397,12 +429,115 @@ impl Connect for Connector {
}
}
pub(crate) trait AsyncConn: AsyncRead + AsyncWrite {}
impl<T: AsyncRead + AsyncWrite> AsyncConn for T {}
pub(crate) type Conn = Box<dyn AsyncConn + Send + Sync + Unpin + 'static>;
//impl Connect for Connector {
// type Transport = Conn;
// type Error = BoxError;
// type Future = Connecting;
//
// fn connect(&self, dst: Uri) -> Self::Future {
// let timeout = self.timeout;
// for prox in self.proxies.iter() {
// if let Some(proxy_scheme) = prox.intercept(&dst) {
// return with_timeout(
// self.clone().connect_via_proxy(dst, proxy_scheme),
// timeout,
// )
// .boxed();
// }
// }
//
// with_timeout(
// self.clone().connect_with_maybe_proxy(dst, false),
// timeout,
// )
// .boxed()
// }
//}
pub(crate) trait AsyncConn: AsyncRead + AsyncWrite + Connection {}
impl<T: AsyncRead + AsyncWrite + Connection> AsyncConn for T {}
pin_project! {
pub(crate) struct Conn {
#[pin]
inner: Box<dyn AsyncConn + Send + Sync + Unpin + 'static>,
is_proxy: bool,
}
}
impl Connection for Conn {
fn connected(&self) -> Connected {
self.inner.connected().proxy(self.is_proxy)
}
}
impl AsyncRead for Conn {
fn poll_read(
self: Pin<&mut Self>,
cx: &mut Context,
buf: &mut [u8]
) -> Poll<tokio::io::Result<usize>> {
let this = self.project();
AsyncRead::poll_read(this.inner, cx, buf)
}
unsafe fn prepare_uninitialized_buffer(
&self,
buf: &mut [MaybeUninit<u8>]
) -> bool {
self.inner.prepare_uninitialized_buffer(buf)
}
fn poll_read_buf<B: BufMut>(
self: Pin<&mut Self>,
cx: &mut Context,
buf: &mut B
) -> Poll<tokio::io::Result<usize>>
where
Self: Sized
{
let this = self.project();
AsyncRead::poll_read_buf(this.inner, cx, buf)
}
}
impl AsyncWrite for Conn {
fn poll_write(
self: Pin<&mut Self>,
cx: &mut Context,
buf: &[u8]
) -> Poll<Result<usize, tokio::io::Error>> {
let this = self.project();
AsyncWrite::poll_write(this.inner, cx, buf)
}
fn poll_flush(self: Pin<&mut Self>, cx: &mut Context) -> Poll<Result<(), tokio::io::Error>> {
let this = self.project();
AsyncWrite::poll_flush(this.inner, cx)
}
fn poll_shutdown(
self: Pin<&mut Self>,
cx: &mut Context
) -> Poll<Result<(), tokio::io::Error>> {
let this = self.project();
AsyncWrite::poll_shutdown(this.inner, cx)
}
fn poll_write_buf<B: Buf>(
self: Pin<&mut Self>,
cx: &mut Context,
buf: &mut B
) -> Poll<Result<usize, tokio::io::Error>> where
Self: Sized {
let this = self.project();
AsyncWrite::poll_write_buf(this.inner, cx, buf)
}
}
pub(crate) type Connecting =
Pin<Box<dyn Future<Output = Result<(Conn, Connected), io::Error>> + Send>>;
Pin<Box<dyn Future<Output = Result<Conn, BoxError>> + Send>>;
#[cfg(feature = "tls")]
async fn tunnel<T>(
@@ -488,6 +623,94 @@ fn tunnel_eof() -> io::Error {
)
}
#[cfg(feature = "default-tls")]
mod native_tls_conn {
use std::mem::MaybeUninit;
use std::{pin::Pin, task::{Context, Poll}};
use bytes::{Buf, BufMut};
use hyper::client::connect::{Connected, Connection};
use pin_project_lite::pin_project;
use tokio::io::{AsyncRead, AsyncWrite};
use tokio_tls::TlsStream;
pin_project! {
pub(super) struct NativeTlsConn<T> {
#[pin] pub(super) inner: TlsStream<T>,
}
}
impl<T: Connection + AsyncRead + AsyncWrite + Unpin> Connection for NativeTlsConn<T> {
fn connected(&self) -> Connected {
self.inner.get_ref().connected()
}
}
impl<T: AsyncRead + AsyncWrite + Unpin> AsyncRead for NativeTlsConn<T> {
fn poll_read(
self: Pin<&mut Self>,
cx: &mut Context,
buf: &mut [u8]
) -> Poll<tokio::io::Result<usize>> {
let this = self.project();
AsyncRead::poll_read(this.inner, cx, buf)
}
unsafe fn prepare_uninitialized_buffer(
&self,
buf: &mut [MaybeUninit<u8>]
) -> bool {
self.inner.prepare_uninitialized_buffer(buf)
}
fn poll_read_buf<B: BufMut>(
self: Pin<&mut Self>,
cx: &mut Context,
buf: &mut B
) -> Poll<tokio::io::Result<usize>>
where
Self: Sized
{
let this = self.project();
AsyncRead::poll_read_buf(this.inner, cx, buf)
}
}
impl<T: AsyncRead + AsyncWrite + Unpin> AsyncWrite for NativeTlsConn<T> {
fn poll_write(
self: Pin<&mut Self>,
cx: &mut Context,
buf: &[u8]
) -> Poll<Result<usize, tokio::io::Error>> {
let this = self.project();
AsyncWrite::poll_write(this.inner, cx, buf)
}
fn poll_flush(self: Pin<&mut Self>, cx: &mut Context) -> Poll<Result<(), tokio::io::Error>> {
let this = self.project();
AsyncWrite::poll_flush(this.inner, cx)
}
fn poll_shutdown(
self: Pin<&mut Self>,
cx: &mut Context
) -> Poll<Result<(), tokio::io::Error>> {
let this = self.project();
AsyncWrite::poll_shutdown(this.inner, cx)
}
fn poll_write_buf<B: Buf>(
self: Pin<&mut Self>,
cx: &mut Context,
buf: &mut B
) -> Poll<Result<usize, tokio::io::Error>> where
Self: Sized {
let this = self.project();
AsyncWrite::poll_write_buf(this.inner, cx, buf)
}
}
}
#[cfg(feature = "socks")]
mod socks {
use std::io;
@@ -510,8 +733,8 @@ mod socks {
proxy: ProxyScheme,
dst: Destination,
dns: DnsResolve,
) -> Result<(super::Conn, Connected), io::Error> {
let https = dst.scheme() == "https";
) -> Result<super::Conn, BoxError> {
let https = dst.scheme() == Some(&Scheme::HTTPS);
let original_host = dst.host().to_owned();
let mut host = original_host.clone();
let port = dst.port().unwrap_or_else(|| if https { 443 } else { 80 });
@@ -555,8 +778,8 @@ mod tests {
use std::io::{Read, Write};
use std::net::TcpListener;
use std::thread;
use tokio::net::tcp::TcpStream;
use tokio::runtime::current_thread::Runtime;
use tokio::net::TcpStream;
use tokio::runtime;
static TUNNEL_UA: &'static str = "tunnel-test/x.y";
static TUNNEL_OK: &[u8] = b"\
@@ -609,7 +832,7 @@ mod tests {
fn test_tunnel() {
let addr = mock_tunnel!();
let mut rt = Runtime::new().unwrap();
let mut rt = runtime::Builder::new().basic_scheduler().enable_all().build().expect("new rt");
let f = async move {
let tcp = TcpStream::connect(&addr).await?;
let host = addr.ip().to_string();
@@ -624,7 +847,7 @@ mod tests {
fn test_tunnel_eof() {
let addr = mock_tunnel!(b"HTTP/1.1 200 OK");
let mut rt = Runtime::new().unwrap();
let mut rt = runtime::Builder::new().basic_scheduler().enable_all().build().expect("new rt");
let f = async move {
let tcp = TcpStream::connect(&addr).await?;
let host = addr.ip().to_string();
@@ -639,7 +862,7 @@ mod tests {
fn test_tunnel_non_http_response() {
let addr = mock_tunnel!(b"foo bar baz hallo");
let mut rt = Runtime::new().unwrap();
let mut rt = runtime::Builder::new().basic_scheduler().enable_all().build().expect("new rt");
let f = async move {
let tcp = TcpStream::connect(&addr).await?;
let host = addr.ip().to_string();
@@ -660,7 +883,7 @@ mod tests {
"
);
let mut rt = Runtime::new().unwrap();
let mut rt = runtime::Builder::new().basic_scheduler().enable_all().build().expect("new rt");
let f = async move {
let tcp = TcpStream::connect(&addr).await?;
let host = addr.ip().to_string();
@@ -679,7 +902,7 @@ mod tests {
"Proxy-Authorization: Basic QWxhZGRpbjpvcGVuIHNlc2FtZQ==\r\n"
);
let mut rt = Runtime::new().unwrap();
let mut rt = runtime::Builder::new().basic_scheduler().enable_all().build().expect("new rt");
let f = async move {
let tcp = TcpStream::connect(&addr).await?;
let host = addr.ip().to_string();

11
src/error.rs
View File

@@ -258,9 +258,6 @@ pub(crate) fn decode_io(e: io::Error) -> Error {
#[derive(Debug)]
pub(crate) struct TimedOut;
#[derive(Debug)]
pub(crate) struct BlockingClientInAsyncContext;
impl fmt::Display for TimedOut {
fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
f.write_str("operation timed out")
@@ -269,14 +266,6 @@ impl fmt::Display for TimedOut {
impl StdError for TimedOut {}
impl fmt::Display for BlockingClientInAsyncContext {
fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
f.write_str("blocking Client used inside a Future context")
}
}
impl StdError for BlockingClientInAsyncContext {}
#[cfg(test)]
mod tests {
use super::*;

20
src/proxy.rs
View File

@@ -5,7 +5,6 @@ use std::sync::Arc;
use crate::{IntoUrl, Url};
use http::{header::HeaderValue, Uri};
use hyper::client::connect::Destination;
use percent_encoding::percent_decode;
use std::collections::HashMap;
use std::env;
@@ -508,25 +507,10 @@ pub(crate) trait Dst {
fn port(&self) -> Option<u16>;
}
#[doc(hidden)]
impl Dst for Destination {
fn scheme(&self) -> &str {
Destination::scheme(self)
}
fn host(&self) -> &str {
Destination::host(self)
}
fn port(&self) -> Option<u16> {
Destination::port(self)
}
}
#[doc(hidden)]
impl Dst for Uri {
fn scheme(&self) -> &str {
self.scheme_part()
self.scheme()
.expect("Uri should have a scheme")
.as_str()
}
@@ -536,7 +520,7 @@ impl Dst for Uri {
}
fn port(&self) -> Option<u16> {
self.port_part().map(|p| p.as_u16())
self.port().map(|p| p.as_u16())
}
}

6
src/wasm/client.rs
View File

@@ -139,8 +139,8 @@ async fn fetch(req: Request) -> crate::Result<Response> {
.map_err(crate::error::request)?;
// Convert from the js Response
let mut resp = http::Response::builder();
resp.status(js_resp.status());
let mut resp = http::Response::builder()
.status(js_resp.status());
let js_headers = js_resp.headers();
let js_iter = js_sys::try_iter(&js_headers)
@@ -150,7 +150,7 @@ async fn fetch(req: Request) -> crate::Result<Response> {
for item in js_iter {
let item = item.expect_throw("headers iterator doesn't throw");
let v: Vec<String> = item.into_serde().expect_throw("headers into_serde");
resp.header(
resp = resp.header(
v.get(0).expect_throw("headers name"),
v.get(1).expect_throw("headers value"),
);

12
src/wasm/request.rs
View File

@@ -1,4 +1,4 @@
use http::HttpTryFrom;
use std::convert::TryFrom;
use std::fmt;
use http::Method;
@@ -96,13 +96,15 @@ impl RequestBuilder {
/// Add a `Header` to this Request.
pub fn header<K, V>(mut self, key: K, value: V) -> RequestBuilder
where
HeaderName: HttpTryFrom<K>,
HeaderValue: HttpTryFrom<V>,
HeaderName: TryFrom<K>,
<HeaderName as TryFrom<K>>::Error: Into<http::Error>,
HeaderValue: TryFrom<V>,
<HeaderValue as TryFrom<V>>::Error: Into<http::Error>,
{
let mut error = None;
if let Ok(ref mut req) = self.request {
match <HeaderName as HttpTryFrom<K>>::try_from(key) {
Ok(key) => match <HeaderValue as HttpTryFrom<V>>::try_from(value) {
match <HeaderName as TryFrom<K>>::try_from(key) {
Ok(key) => match <HeaderValue as TryFrom<V>>::try_from(value) {
Ok(value) => {
req.headers_mut().append(key, value);
}