diff --git a/Cargo.toml b/Cargo.toml index 0c507df..228863d 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -25,7 +25,7 @@ default-tls-vendored = ["default-tls", "native-tls/vendored"] #rustls-tls = ["hyper-rustls", "tokio-rustls", "webpki-roots", "rustls", "tls"] -blocking = ["futures-channel-preview", "futures-util-preview/io", "tokio/rt-full"] +blocking = ["futures-channel", "futures-util/io", "tokio/rt-threaded", "tokio/rt-core"] cookies = ["cookie_crate", "cookie_store"] @@ -44,25 +44,26 @@ unstable-stream = [] __internal_proxy_sys_no_cache = [] [dependencies] -http = "0.1.15" +http = "0.2" url = "2.1" -bytes = "0.4" +bytes = "0.5" [target.'cfg(not(target_arch = "wasm32"))'.dependencies] base64 = "0.11" encoding_rs = "0.8" -futures-core-preview = { version = "=0.3.0-alpha.19" } -futures-util-preview = { version = "=0.3.0-alpha.19" } -http-body = "=0.2.0-alpha.3" -hyper = { version = "=0.13.0-alpha.4", default-features = false, features = ["tcp"] } +futures-core = { version = "0.3.0" } +futures-util = { version = "0.3.0" } +http-body = "0.3.0" +hyper = { version = "0.13", default-features = false, features = ["tcp"] } lazy_static = "1.4" log = "0.4" mime = "0.3.7" mime_guess = "2.0" percent-encoding = "2.1" -tokio = { version = "=0.2.0-alpha.6", default-features = false, features = ["io", "tcp", "timer"] } -tokio-executor = "=0.2.0-alpha.6" +tokio = { version = "0.2.0", default-features = false, features = ["tcp", "time"] } +#tokio-executor = "0.2.0" time = "0.1.42" +pin-project-lite = "0.1.1" # TODO: candidates for optional features @@ -72,9 +73,9 @@ serde_urlencoded = "0.6.1" # Optional deps... ## default-tls -hyper-tls = { version = "=0.4.0-alpha.4", optional = true } +hyper-tls = { version = "0.4", optional = true } native-tls = { version = "0.2", optional = true } -tokio-tls = { version = "=0.3.0-alpha.6", optional = true } +tokio-tls = { version = "0.3.0", optional = true } ## rustls-tls #hyper-rustls = { version = "=0.18.0-alpha.1", optional = true } @@ -83,14 +84,14 @@ tokio-tls = { version = "=0.3.0-alpha.6", optional = true } #webpki-roots = { version = "0.17", optional = true } ## blocking -futures-channel-preview = { version = "=0.3.0-alpha.19", optional = true } +futures-channel = { version = "0.3.0", optional = true } ## cookies cookie_crate = { version = "0.12", package = "cookie", optional = true } cookie_store = { version = "0.10", optional = true } ## gzip -async-compression = { version = "=0.1.0-alpha.7", default-features = false, features = ["gzip", "stream"], optional = true } +async-compression = { version = "0.2.0", default-features = false, features = ["gzip", "stream"], optional = true } ## json serde_json = { version = "1.0", optional = true } @@ -103,10 +104,11 @@ serde_json = { version = "1.0", optional = true } [target.'cfg(not(target_arch = "wasm32"))'.dev-dependencies] env_logger = "0.6" -hyper = { version = "=0.13.0-alpha.4", features = ["unstable-stream"] } +hyper = { version = "0.13", default-features = false, features = ["tcp", "stream"] } serde = { version = "1.0", features = ["derive"] } libflate = "0.1" doc-comment = "0.3" +tokio = { version = "0.2.0", default-features = false, features = ["macros"] } [target.'cfg(windows)'.dependencies] winreg = "0.6" diff --git a/src/async_impl/body.rs b/src/async_impl/body.rs index 3f26723..8ff2d04 100644 --- a/src/async_impl/body.rs +++ b/src/async_impl/body.rs @@ -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> + dyn HttpBody> + Send + Sync, >, @@ -73,7 +73,7 @@ impl Body { where S: futures_core::stream::TryStream + Send + Sync + 'static, S::Error: Into>, - hyper::Chunk: From, + Bytes: From, { Body::stream(stream) } @@ -82,12 +82,12 @@ impl Body { where S: futures_core::stream::TryStream + Send + Sync + 'static, S::Error: Into>, - hyper::Chunk: From, + Bytes: From, { 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 HttpBody for WrapStream where S: Stream>, - D: Into, + D: Into, E: Into>, { - 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; fn poll_data( diff --git a/src/async_impl/client.rs b/src/async_impl/client.rs index b4d615e..5b03564 100644 --- a/src/async_impl/client.rs +++ b/src/async_impl/client.rs @@ -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; } diff --git a/src/async_impl/decoder.rs b/src/async_impl/decoder.rs index 4efdc95..74c5c9c 100644 --- a/src/async_impl/decoder.rs +++ b/src/async_impl/decoder.rs @@ -192,7 +192,7 @@ mod imp { fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll { 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 } diff --git a/src/async_impl/multipart.rs b/src/async_impl/multipart.rs index 25a64e1..bf2ac7c 100644 --- a/src/async_impl/multipart.rs +++ b/src/async_impl/multipart.rs @@ -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 diff --git a/src/async_impl/request.rs b/src/async_impl/request.rs index 50ed0af..e162c89 100644 --- a/src/async_impl/request.rs +++ b/src/async_impl/request.rs @@ -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(mut self, key: K, value: V) -> RequestBuilder where - HeaderName: HttpTryFrom, - HeaderValue: HttpTryFrom, + HeaderName: TryFrom, + >::Error: Into, + HeaderValue: TryFrom, + >::Error: Into, { let mut error = None; if let Ok(ref mut req) = self.request { - match >::try_from(key) { - Ok(key) => match >::try_from(value) { + match >::try_from(key) { + Ok(key) => match >::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. diff --git a/src/async_impl/response.rs b/src/async_impl/response.rs index 3033a1c..ec9f430 100644 --- a/src/async_impl/response.rs +++ b/src/async_impl/response.rs @@ -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 { - self.body.try_concat().await + pub async fn bytes(mut self) -> crate::Result { + 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)) } } diff --git a/src/blocking/body.rs b/src/blocking/body.rs index 4b48823..623670e 100644 --- a/src/blocking/body.rs +++ b/src/blocking/body.rs @@ -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], &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], &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)?; diff --git a/src/blocking/client.rs b/src/blocking/client.rs index 9fedf53..8c13d6a 100644 --- a/src/blocking/client.rs +++ b/src/blocking/client.rs @@ -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) } } diff --git a/src/blocking/request.rs b/src/blocking/request.rs index b8545d1..d7442a4 100644 --- a/src/blocking/request.rs +++ b/src/blocking/request.rs @@ -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(mut self, key: K, value: V) -> RequestBuilder where - HeaderName: HttpTryFrom, - HeaderValue: HttpTryFrom, + HeaderName: TryFrom, + HeaderValue: TryFrom, + >::Error: Into, + >::Error: Into, { let mut error = None; if let Ok(ref mut req) = self.request { - match >::try_from(key) { - Ok(key) => match >::try_from(value) { + match >::try_from(key) { + Ok(key) => match >::try_from(value) { Ok(value) => { req.headers_mut().append(key, value); } diff --git a/src/blocking/response.rs b/src/blocking/response.rs index 165bdce..57ad638 100644 --- a/src/blocking/response.rs +++ b/src/blocking/response.rs @@ -220,7 +220,6 @@ impl Response { pub fn json(self) -> crate::Result { 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, }) } diff --git a/src/blocking/wait.rs b/src/blocking/wait.rs index 7698509..e417183 100644 --- a/src/blocking/wait.rs +++ b/src/blocking/wait.rs @@ -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(fut: F, timeout: Option) -> Result> where F: Future>, { - 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 { 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) { 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(|| {}); + } +} diff --git a/src/connect.rs b/src/connect.rs index c553fac..b9bdb72 100644 --- a/src/connect.rs +++ b/src/connect.rs @@ -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; @@ -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 { 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 { 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 { } -async fn with_timeout(f: F, timeout: Option) -> Result +async fn with_timeout(f: F, timeout: Option) -> Result where - F: Future>, + F: Future>, { 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 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> { + 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 AsyncConn for T {} -pub(crate) type Conn = Box; +//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 AsyncConn for T {} + +pin_project! { + pub(crate) struct Conn { + #[pin] + inner: Box, + 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> { + let this = self.project(); + AsyncRead::poll_read(this.inner, cx, buf) + } + + unsafe fn prepare_uninitialized_buffer( + &self, + buf: &mut [MaybeUninit] + ) -> bool { + self.inner.prepare_uninitialized_buffer(buf) + } + + fn poll_read_buf( + self: Pin<&mut Self>, + cx: &mut Context, + buf: &mut B + ) -> Poll> + 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> { + let this = self.project(); + AsyncWrite::poll_write(this.inner, cx, buf) + } + + fn poll_flush(self: Pin<&mut Self>, cx: &mut Context) -> Poll> { + let this = self.project(); + AsyncWrite::poll_flush(this.inner, cx) + } + + fn poll_shutdown( + self: Pin<&mut Self>, + cx: &mut Context + ) -> Poll> { + let this = self.project(); + AsyncWrite::poll_shutdown(this.inner, cx) + } + + fn poll_write_buf( + self: Pin<&mut Self>, + cx: &mut Context, + buf: &mut B + ) -> Poll> where + Self: Sized { + let this = self.project(); + AsyncWrite::poll_write_buf(this.inner, cx, buf) + } +} pub(crate) type Connecting = - Pin> + Send>>; + Pin> + Send>>; #[cfg(feature = "tls")] async fn tunnel( @@ -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 { + #[pin] pub(super) inner: TlsStream, + } + } + + impl Connection for NativeTlsConn { + fn connected(&self) -> Connected { + self.inner.get_ref().connected() + } + } + + impl AsyncRead for NativeTlsConn { + fn poll_read( + self: Pin<&mut Self>, + cx: &mut Context, + buf: &mut [u8] + ) -> Poll> { + let this = self.project(); + AsyncRead::poll_read(this.inner, cx, buf) + } + + unsafe fn prepare_uninitialized_buffer( + &self, + buf: &mut [MaybeUninit] + ) -> bool { + self.inner.prepare_uninitialized_buffer(buf) + } + + fn poll_read_buf( + self: Pin<&mut Self>, + cx: &mut Context, + buf: &mut B + ) -> Poll> + where + Self: Sized + { + let this = self.project(); + AsyncRead::poll_read_buf(this.inner, cx, buf) + } + } + + impl AsyncWrite for NativeTlsConn { + fn poll_write( + self: Pin<&mut Self>, + cx: &mut Context, + buf: &[u8] + ) -> Poll> { + let this = self.project(); + AsyncWrite::poll_write(this.inner, cx, buf) + } + + fn poll_flush(self: Pin<&mut Self>, cx: &mut Context) -> Poll> { + let this = self.project(); + AsyncWrite::poll_flush(this.inner, cx) + } + + fn poll_shutdown( + self: Pin<&mut Self>, + cx: &mut Context + ) -> Poll> { + let this = self.project(); + AsyncWrite::poll_shutdown(this.inner, cx) + } + + fn poll_write_buf( + self: Pin<&mut Self>, + cx: &mut Context, + buf: &mut B + ) -> Poll> 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 { + 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(); diff --git a/src/error.rs b/src/error.rs index e668f50..a0ee866 100644 --- a/src/error.rs +++ b/src/error.rs @@ -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::*; diff --git a/src/proxy.rs b/src/proxy.rs index 9cfccdf..dcb512a 100644 --- a/src/proxy.rs +++ b/src/proxy.rs @@ -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; } -#[doc(hidden)] -impl Dst for Destination { - fn scheme(&self) -> &str { - Destination::scheme(self) - } - - fn host(&self) -> &str { - Destination::host(self) - } - - fn port(&self) -> Option { - 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 { - self.port_part().map(|p| p.as_u16()) + self.port().map(|p| p.as_u16()) } } diff --git a/src/wasm/client.rs b/src/wasm/client.rs index 6897703..71bb938 100644 --- a/src/wasm/client.rs +++ b/src/wasm/client.rs @@ -139,8 +139,8 @@ async fn fetch(req: Request) -> crate::Result { .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 { for item in js_iter { let item = item.expect_throw("headers iterator doesn't throw"); let v: Vec = 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"), ); diff --git a/src/wasm/request.rs b/src/wasm/request.rs index 4cd50cc..e9796d6 100644 --- a/src/wasm/request.rs +++ b/src/wasm/request.rs @@ -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(mut self, key: K, value: V) -> RequestBuilder where - HeaderName: HttpTryFrom, - HeaderValue: HttpTryFrom, + HeaderName: TryFrom, + >::Error: Into, + HeaderValue: TryFrom, + >::Error: Into, { let mut error = None; if let Ok(ref mut req) = self.request { - match >::try_from(key) { - Ok(key) => match >::try_from(value) { + match >::try_from(key) { + Ok(key) => match >::try_from(value) { Ok(value) => { req.headers_mut().append(key, value); } diff --git a/tests/blocking.rs b/tests/blocking.rs index 4519094..18ff6ec 100644 --- a/tests/blocking.rs +++ b/tests/blocking.rs @@ -82,12 +82,12 @@ fn test_get() { #[test] fn test_post() { - let server = server::http(move |mut req| { + let server = server::http(move |req| { async move { assert_eq!(req.method(), "POST"); assert_eq!(req.headers()["content-length"], "5"); - let data = req.body_mut().next().await.unwrap().unwrap(); + let data = hyper::body::to_bytes(req.into_body()).await.unwrap(); assert_eq!(&*data, b"Hello"); http::Response::default() @@ -107,7 +107,7 @@ fn test_post() { #[test] fn test_post_form() { - let server = server::http(move |mut req| { + let server = server::http(move |req| { async move { assert_eq!(req.method(), "POST"); assert_eq!(req.headers()["content-length"], "24"); @@ -116,7 +116,7 @@ fn test_post_form() { "application/x-www-form-urlencoded" ); - let data = req.body_mut().next().await.unwrap().unwrap(); + let data = hyper::body::to_bytes(req.into_body()).await.unwrap(); assert_eq!(&*data, b"hello=world&sean=monstar"); http::Response::default() @@ -287,3 +287,17 @@ fn test_appended_headers_not_overwritten() { assert_eq!(res.url().as_str(), &url); assert_eq!(res.status(), reqwest::StatusCode::OK); } + +#[test] +#[should_panic] +fn test_blocking_inside_a_runtime() { + let server = server::http(move |_req| async { http::Response::new("Hello".into()) }); + + let url = format!("http://{}/text", server.addr()); + + let mut rt = tokio::runtime::Builder::new().build().expect("new rt"); + + rt.block_on(async move { + let _should_panic = reqwest::blocking::get(&url); + }); +} diff --git a/tests/client.rs b/tests/client.rs index 1bf26b9..b273a89 100644 --- a/tests/client.rs +++ b/tests/client.rs @@ -1,4 +1,5 @@ mod support; +use futures_util::stream::StreamExt; use support::*; use reqwest::Client; diff --git a/tests/multipart.rs b/tests/multipart.rs index d2f66e4..968a089 100644 --- a/tests/multipart.rs +++ b/tests/multipart.rs @@ -1,4 +1,5 @@ mod support; +use futures_util::stream::StreamExt; use support::*; #[tokio::test] diff --git a/tests/redirect.rs b/tests/redirect.rs index e0c3818..66ba54b 100644 --- a/tests/redirect.rs +++ b/tests/redirect.rs @@ -1,4 +1,5 @@ mod support; +use futures_util::stream::StreamExt; use support::*; #[tokio::test] diff --git a/tests/support/server.rs b/tests/support/server.rs index 84ea36b..c00ca18 100644 --- a/tests/support/server.rs +++ b/tests/support/server.rs @@ -8,6 +8,7 @@ use std::time::Duration; use tokio::sync::oneshot; pub use http::Response; +use tokio::runtime; pub struct Server { addr: net::SocketAddr, @@ -40,41 +41,52 @@ where F: Fn(http::Request) -> Fut + Clone + Send + 'static, Fut: Future> + Send + 'static, { - let srv = hyper::Server::bind(&([127, 0, 0, 1], 0).into()).serve( - hyper::service::make_service_fn(move |_| { - let func = func.clone(); - async move { - Ok::<_, Infallible>(hyper::service::service_fn(move |req| { - let fut = func(req); - async move { Ok::<_, Infallible>(fut.await) } - })) - } - }), - ); + //Spawn new runtime in thread to prevent reactor execution context conflict + thread::spawn(move || { + let mut rt = runtime::Builder::new() + .basic_scheduler() + .enable_all() + .build() + .expect("new rt"); + let srv = rt.block_on(async move { + hyper::Server::bind(&([127, 0, 0, 1], 0).into()).serve(hyper::service::make_service_fn( + move |_| { + let func = func.clone(); + async move { + Ok::<_, Infallible>(hyper::service::service_fn(move |req| { + let fut = func(req); + async move { Ok::<_, Infallible>(fut.await) } + })) + } + }, + )) + }); - let addr = srv.local_addr(); - let (shutdown_tx, shutdown_rx) = oneshot::channel(); - let srv = srv.with_graceful_shutdown(async move { - let _ = shutdown_rx.await; - }); + let addr = srv.local_addr(); + let (shutdown_tx, shutdown_rx) = oneshot::channel(); + let srv = srv.with_graceful_shutdown(async move { + let _ = shutdown_rx.await; + }); - let (panic_tx, panic_rx) = std_mpsc::channel(); - let tname = format!( - "test({})-support-server", - thread::current().name().unwrap_or("") - ); - thread::Builder::new() - .name(tname) - .spawn(move || { - let mut rt = tokio::runtime::current_thread::Runtime::new().expect("rt new"); - rt.block_on(srv).unwrap(); - let _ = panic_tx.send(()); - }) - .expect("thread spawn"); + let (panic_tx, panic_rx) = std_mpsc::channel(); + let tname = format!( + "test({})-support-server", + thread::current().name().unwrap_or("") + ); + thread::Builder::new() + .name(tname) + .spawn(move || { + rt.block_on(srv).unwrap(); + let _ = panic_tx.send(()); + }) + .expect("thread spawn"); - Server { - addr, - panic_rx, - shutdown_tx: Some(shutdown_tx), - } + Server { + addr, + panic_rx, + shutdown_tx: Some(shutdown_tx), + } + }) + .join() + .unwrap() } diff --git a/tests/timeouts.rs b/tests/timeouts.rs index 1628a44..2a68d9b 100644 --- a/tests/timeouts.rs +++ b/tests/timeouts.rs @@ -1,7 +1,7 @@ mod support; use support::*; -use std::time::{Duration, Instant}; +use std::time::Duration; #[tokio::test] async fn request_timeout() { @@ -10,7 +10,7 @@ async fn request_timeout() { let server = server::http(move |_req| { async { // delay returning the response - tokio::timer::delay(Instant::now() + Duration::from_secs(2)).await; + tokio::time::delay_for(Duration::from_secs(2)).await; http::Response::default() } }); @@ -38,7 +38,7 @@ async fn response_timeout() { async { // immediate response, but delayed body let body = hyper::Body::wrap_stream(futures_util::stream::once(async { - tokio::timer::delay(Instant::now() + Duration::from_secs(2)).await; + tokio::time::delay_for(Duration::from_secs(2)).await; Ok::<_, std::convert::Infallible>("Hello") })); @@ -77,7 +77,7 @@ fn timeout_closes_connection() { let server = server::http(move |_req| { async { // delay returning the response - tokio::timer::delay(Instant::now() + Duration::from_secs(2)).await; + tokio::time::delay_for(Duration::from_secs(2)).await; http::Response::default() } }); @@ -106,7 +106,7 @@ fn write_timeout_large_body() { let server = server::http(move |_req| { async { // delay returning the response - tokio::timer::delay(Instant::now() + Duration::from_secs(2)).await; + tokio::time::delay_for(Duration::from_secs(2)).await; http::Response::default() } });