Merge pull request #1391 from JohnTitor/new-awc

Release awc v2.0.0-alpha.1

CHANGES.md

@@ -10,10 +10,10 @@
 ### Changed
 
 * Use `sha-1` crate instead of unmaintained `sha1` crate
-
 * Skip empty chunks when returning response from a `Stream` #1308
-
 * Update the `time` dependency to 0.2.7
+* Update `actix-tls` dependency to 2.0.0-alpha.1
+* Update `rustls` dependency to 0.17
 
 ## [2.0.0] - 2019-12-25
 

Cargo.toml

@@ -68,11 +68,11 @@ actix-server = "1.0.0"
 actix-testing = "1.0.0"
 actix-macros = "0.1.0"
 actix-threadpool = "0.3.1"
-actix-tls = "1.0.0"
+actix-tls = "2.0.0-alpha.1"
 
 actix-web-codegen = "0.2.0"
-actix-http = "2.0.0-alpha.1"
-awc = { version = "1.0.1", default-features = false }
+actix-http = "2.0.0-alpha.2"
+awc = { version = "2.0.0-alpha.1", default-features = false }
 
 bytes = "0.5.3"
 derive_more = "0.99.2"
@@ -90,7 +90,7 @@ serde_urlencoded = "0.6.1"
 time = { version = "0.2.7", default-features = false, features = ["std"] }
 url = "2.1"
 open-ssl = { version="0.10", package = "openssl", optional = true }
-rust-tls = { version = "0.16.0", package = "rustls", optional = true }
+rust-tls = { version = "0.17.0", package = "rustls", optional = true }
 
 [dev-dependencies]
 actix = "0.10.0-alpha.1"

actix-files/Cargo.toml

@@ -19,7 +19,7 @@ path = "src/lib.rs"
 
 [dependencies]
 actix-web = { version = "2.0.0-rc", default-features = false }
-actix-http = "2.0.0-alpha.1"
+actix-http = "2.0.0-alpha.2"
 actix-service = "1.0.1"
 bitflags = "1"
 bytes = "0.5.3"

actix-framed/Cargo.toml

@@ -23,7 +23,7 @@ actix-codec = "0.2.0"
 actix-service = "1.0.1"
 actix-router = "0.2.1"
 actix-rt = "1.0.0"
-actix-http = "2.0.0-alpha.1"
+actix-http = "2.0.0-alpha.2"
 
 bytes = "0.5.3"
 futures = "0.3.1"
@@ -32,6 +32,6 @@ log = "0.4"
 
 [dev-dependencies]
 actix-server = "1.0.0"
-actix-connect = { version = "1.0.0", features=["openssl"] }
+actix-connect = { version = "2.0.0-alpha.2", features=["openssl"] }
 actix-http-test = { version = "1.0.0", features=["openssl"] }
 actix-utils = "1.0.3"

actix-framed/tests/test_server.rs

@@ -47,7 +47,8 @@ async fn test_simple() {
             )
             .finish(|_| future::ok::<_, Error>(Response::NotFound()))
             .tcp()
-    });
+    })
+    .await;
 
     assert!(srv.ws_at("/test").await.is_err());
 
@@ -108,7 +109,8 @@ async fn test_service() {
                     .map_err(|_| ()),
             ),
         )
-    });
+    })
+    .await;
 
     // non ws request
     let res = srv.get("/index.html").send().await.unwrap();

actix-http/CHANGES.md

@@ -1,15 +1,28 @@
 # Changes
 
+## [2.0.0-alpha.2] - 2020-03-07
+
+### Changed
+
+* Update `actix-connect` and `actix-tls` dependency to 2.0.0-alpha.1. [#1395]
+
+* Change default initial window size and connection window size for HTTP2 to 2MB and 1MB respectively
+  to improve download speed for awc when downloading large objects. [#1394]
+
+* client::Connector accepts initial_window_size and initial_connection_window_size HTTP2 configuration. [#1394]
+
+* client::Connector allowing to set max_http_version to limit HTTP version to be used. [#1394]
+
+[#1394]: https://github.com/actix/actix-web/pull/1394
+[#1395]: https://github.com/actix/actix-web/pull/1395
+
 ## [2.0.0-alpha.1] - 2020-02-27
 
 ### Changed
 
 * Update the `time` dependency to 0.2.7.
-
 * Moved actors messages support from actix crate, enabled with feature `actors`.
-
 * Breaking change: trait MessageBody requires Unpin and accepting Pin<&mut Self> instead of &mut self in the poll_next().
-
 * MessageBody is not implemented for &'static [u8] anymore.
 
 ### Fixed

actix-http/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "actix-http"
-version = "2.0.0-alpha.1"
+version = "2.0.0-alpha.2"
 authors = ["Nikolay Kim <fafhrd91@gmail.com>"]
 description = "Actix http primitives"
 readme = "README.md"
@@ -45,11 +45,11 @@ actors = ["actix"]
 [dependencies]
 actix-service = "1.0.5"
 actix-codec = "0.2.0"
-actix-connect = "1.0.2"
+actix-connect = "2.0.0-alpha.2"
 actix-utils = "1.0.6"
 actix-rt = "1.0.0"
 actix-threadpool = "0.3.1"
-actix-tls = { version = "1.0.0", optional = true }
+actix-tls = { version = "2.0.0-alpha.1", optional = true }
 actix = { version = "0.10.0-alpha.1", optional = true }
 
 base64 = "0.11"
@@ -94,15 +94,15 @@ fail-ure = { version = "0.1.5", package="failure", optional = true }
 
 [dev-dependencies]
 actix-server = "1.0.1"
-actix-connect = { version = "1.0.2", features=["openssl"] }
+actix-connect = { version = "2.0.0-alpha.2", features=["openssl"] }
 actix-http-test = { version = "1.0.0", features=["openssl"] }
-actix-tls = { version = "1.0.0", features=["openssl"] }
+actix-tls = { version = "2.0.0-alpha.1", features=["openssl"] }
 criterion = "0.3"
 futures = "0.3.1"
 env_logger = "0.7"
 serde_derive = "1.0"
 open-ssl = { version="0.10", package = "openssl" }
-rust-tls = { version="0.16", package = "rustls" }
+rust-tls = { version="0.17", package = "rustls" }
 
 [[bench]]
 name = "content-length"

actix-http/README.md

@@ -14,19 +14,34 @@ Actix http
 
 ```rust
 // see examples/framed_hello.rs for complete list of used crates.
-extern crate actix_http;
-use actix_http::{h1, Response, ServiceConfig};
+use std::{env, io};
 
-fn main() {
-    Server::new().bind("framed_hello", "127.0.0.1:8080", || {
-        IntoFramed::new(|| h1::Codec::new(ServiceConfig::default()))	// <- create h1 codec
-            .and_then(TakeItem::new().map_err(|_| ()))	                // <- read one request
-            .and_then(|(_req, _framed): (_, Framed<_, _>)| {	        // <- send response and close conn
-                SendResponse::send(_framed, Response::Ok().body("Hello world!"))
-                    .map_err(|_| ())
-                    .map(|_| ())
-            })
-    }).unwrap().run();
+use actix_http::{HttpService, Response};
+use actix_server::Server;
+use futures::future;
+use http::header::HeaderValue;
+use log::info;
+
+#[actix_rt::main]
+async fn main() -> io::Result<()> {
+    env::set_var("RUST_LOG", "hello_world=info");
+    env_logger::init();
+
+    Server::build()
+        .bind("hello-world", "127.0.0.1:8080", || {
+            HttpService::build()
+                .client_timeout(1000)
+                .client_disconnect(1000)
+                .finish(|_req| {
+                    info!("{:?}", _req);
+                    let mut res = Response::Ok();
+                    res.header("x-head", HeaderValue::from_static("dummy value!"));
+                    future::ok::<_, ()>(res.body("Hello world!"))
+                })
+                .tcp()
+        })?
+        .run()
+        .await
 }
 ```
 

actix-http/src/client/config.rs

@@ -0,0 +1,39 @@
+use std::time::Duration;
+
+// These values are taken from hyper/src/proto/h2/client.rs
+const DEFAULT_H2_CONN_WINDOW: u32 = 1024 * 1024 * 2; // 2mb
+const DEFAULT_H2_STREAM_WINDOW: u32 = 1024 * 1024; // 1mb
+
+/// Connector configuration
+#[derive(Clone)]
+pub(crate) struct ConnectorConfig {
+    pub(crate) timeout: Duration,
+    pub(crate) conn_lifetime: Duration,
+    pub(crate) conn_keep_alive: Duration,
+    pub(crate) disconnect_timeout: Option<Duration>,
+    pub(crate) limit: usize,
+    pub(crate) conn_window_size: u32,
+    pub(crate) stream_window_size: u32,
+}
+
+impl Default for ConnectorConfig {
+    fn default() -> Self {
+        Self {
+            timeout: Duration::from_secs(1),
+            conn_lifetime: Duration::from_secs(75),
+            conn_keep_alive: Duration::from_secs(15),
+            disconnect_timeout: Some(Duration::from_millis(3000)),
+            limit: 100,
+            conn_window_size: DEFAULT_H2_CONN_WINDOW,
+            stream_window_size: DEFAULT_H2_STREAM_WINDOW,
+        }
+    }
+}
+
+impl ConnectorConfig {
+    pub(crate) fn no_disconnect_timeout(&self) -> Self {
+        let mut res = self.clone();
+        res.disconnect_timeout = None;
+        res
+    }
+}

actix-http/src/client/connector.rs

@@ -11,6 +11,7 @@ use actix_service::{apply_fn, Service};
 use actix_utils::timeout::{TimeoutError, TimeoutService};
 use http::Uri;
 
+use super::config::ConnectorConfig;
 use super::connection::Connection;
 use super::error::ConnectError;
 use super::pool::{ConnectionPool, Protocol};
@@ -48,11 +49,7 @@ type SslConnector = ();
 /// ```
 pub struct Connector<T, U> {
     connector: T,
-    timeout: Duration,
-    conn_lifetime: Duration,
-    conn_keep_alive: Duration,
-    disconnect_timeout: Duration,
-    limit: usize,
+    config: ConnectorConfig,
     #[allow(dead_code)]
     ssl: SslConnector,
     _t: PhantomData<U>,
@@ -71,42 +68,47 @@ impl Connector<(), ()> {
             > + Clone,
         TcpStream,
     > {
-        let ssl = {
-            #[cfg(feature = "openssl")]
-            {
-                use actix_connect::ssl::openssl::SslMethod;
-
-                let mut ssl = OpensslConnector::builder(SslMethod::tls()).unwrap();
-                let _ = ssl
-                    .set_alpn_protos(b"\x02h2\x08http/1.1")
-                    .map_err(|e| error!("Can not set alpn protocol: {:?}", e));
-                SslConnector::Openssl(ssl.build())
-            }
-            #[cfg(all(not(feature = "openssl"), feature = "rustls"))]
-            {
-                let protos = vec![b"h2".to_vec(), b"http/1.1".to_vec()];
-                let mut config = ClientConfig::new();
-                config.set_protocols(&protos);
-                config
-                    .root_store
-                    .add_server_trust_anchors(&actix_tls::rustls::TLS_SERVER_ROOTS);
-                SslConnector::Rustls(Arc::new(config))
-            }
-            #[cfg(not(any(feature = "openssl", feature = "rustls")))]
-            {}
-        };
-
         Connector {
-            ssl,
+            ssl: Self::build_ssl(vec![b"h2".to_vec(), b"http/1.1".to_vec()]),
             connector: default_connector(),
-            timeout: Duration::from_secs(1),
-            conn_lifetime: Duration::from_secs(75),
-            conn_keep_alive: Duration::from_secs(15),
-            disconnect_timeout: Duration::from_millis(3000),
-            limit: 100,
+            config: ConnectorConfig::default(),
             _t: PhantomData,
         }
     }
+
+    // Build Ssl connector with openssl, based on supplied alpn protocols
+    #[cfg(feature = "openssl")]
+    fn build_ssl(protocols: Vec<Vec<u8>>) -> SslConnector {
+        use actix_connect::ssl::openssl::SslMethod;
+        use bytes::{BufMut, BytesMut};
+
+        let mut alpn = BytesMut::with_capacity(20);
+        for proto in protocols.iter() {
+            alpn.put_u8(proto.len() as u8);
+            alpn.put(proto.as_slice());
+        }
+
+        let mut ssl = OpensslConnector::builder(SslMethod::tls()).unwrap();
+        let _ = ssl
+            .set_alpn_protos(&alpn)
+            .map_err(|e| error!("Can not set alpn protocol: {:?}", e));
+        SslConnector::Openssl(ssl.build())
+    }
+
+    // Build Ssl connector with rustls, based on supplied alpn protocols
+    #[cfg(all(not(feature = "openssl"), feature = "rustls"))]
+    fn build_ssl(protocols: Vec<Vec<u8>>) -> SslConnector {
+        let mut config = ClientConfig::new();
+        config.set_protocols(&protocols);
+        config
+            .root_store
+            .add_server_trust_anchors(&actix_tls::rustls::TLS_SERVER_ROOTS);
+        SslConnector::Rustls(Arc::new(config))
+    }
+
+    // ssl turned off, provides empty ssl connector
+    #[cfg(not(any(feature = "openssl", feature = "rustls")))]
+    fn build_ssl(_: Vec<Vec<u8>>) -> SslConnector {}
 }
 
 impl<T, U> Connector<T, U> {
@@ -122,11 +124,7 @@ impl<T, U> Connector<T, U> {
     {
         Connector {
             connector,
-            timeout: self.timeout,
-            conn_lifetime: self.conn_lifetime,
-            conn_keep_alive: self.conn_keep_alive,
-            disconnect_timeout: self.disconnect_timeout,
-            limit: self.limit,
+            config: self.config,
             ssl: self.ssl,
             _t: PhantomData,
         }
@@ -146,7 +144,7 @@ where
     /// Connection timeout, i.e. max time to connect to remote host including dns name resolution.
     /// Set to 1 second by default.
     pub fn timeout(mut self, timeout: Duration) -> Self {
-        self.timeout = timeout;
+        self.config.timeout = timeout;
         self
     }
 
@@ -163,12 +161,44 @@ where
         self
     }
 
+    /// Maximum supported http major version
+    /// Supported versions http/1.1, http/2
+    pub fn max_http_version(mut self, val: http::Version) -> Self {
+        let versions = match val {
+            http::Version::HTTP_11 => vec![b"http/1.1".to_vec()],
+            http::Version::HTTP_2 => vec![b"h2".to_vec(), b"http/1.1".to_vec()],
+            _ => {
+                unimplemented!("actix-http:client: supported versions http/1.1, http/2")
+            }
+        };
+        self.ssl = Connector::build_ssl(versions);
+        self
+    }
+
+    /// Indicates the initial window size (in octets) for
+    /// HTTP2 stream-level flow control for received data.
+    ///
+    /// The default value is 65,535 and is good for APIs, but not for big objects.
+    pub fn initial_window_size(mut self, size: u32) -> Self {
+        self.config.stream_window_size = size;
+        self
+    }
+
+    /// Indicates the initial window size (in octets) for
+    /// HTTP2 connection-level flow control for received data.
+    ///
+    /// The default value is 65,535 and is good for APIs, but not for big objects.
+    pub fn initial_connection_window_size(mut self, size: u32) -> Self {
+        self.config.conn_window_size = size;
+        self
+    }
+
     /// Set total number of simultaneous connections per type of scheme.
     ///
     /// If limit is 0, the connector has no limit.
     /// The default limit size is 100.
     pub fn limit(mut self, limit: usize) -> Self {
-        self.limit = limit;
+        self.config.limit = limit;
         self
     }
 
@@ -179,7 +209,7 @@ where
     /// exceeds this period, the connection is closed.
     /// Default keep-alive period is 15 seconds.
     pub fn conn_keep_alive(mut self, dur: Duration) -> Self {
-        self.conn_keep_alive = dur;
+        self.config.conn_keep_alive = dur;
         self
     }
 
@@ -189,7 +219,7 @@ where
     /// until it is closed regardless of keep-alive period.
     /// Default lifetime period is 75 seconds.
     pub fn conn_lifetime(mut self, dur: Duration) -> Self {
-        self.conn_lifetime = dur;
+        self.config.conn_lifetime = dur;
         self
     }
 
@@ -202,7 +232,7 @@ where
     ///
     /// By default disconnect timeout is set to 3000 milliseconds.
     pub fn disconnect_timeout(mut self, dur: Duration) -> Self {
-        self.disconnect_timeout = dur;
+        self.config.disconnect_timeout = Some(dur);
         self
     }
 
@@ -216,7 +246,7 @@ where
         #[cfg(not(any(feature = "openssl", feature = "rustls")))]
         {
             let connector = TimeoutService::new(
-                self.timeout,
+                self.config.timeout,
                 apply_fn(self.connector, |msg: Connect, srv| {
                     srv.call(TcpConnect::new(msg.uri).set_addr(msg.addr))
                 })
@@ -231,10 +261,7 @@ where
             connect_impl::InnerConnector {
                 tcp_pool: ConnectionPool::new(
                     connector,
-                    self.conn_lifetime,
-                    self.conn_keep_alive,
-                    None,
-                    self.limit,
+                    self.config.no_disconnect_timeout(),
                 ),
             }
         }
@@ -248,7 +275,7 @@ where
             use actix_service::{boxed::service, pipeline};
 
             let ssl_service = TimeoutService::new(
-                self.timeout,
+                self.config.timeout,
                 pipeline(
                     apply_fn(self.connector.clone(), |msg: Connect, srv| {
                         srv.call(TcpConnect::new(msg.uri).set_addr(msg.addr))
@@ -301,7 +328,7 @@ where
             });
 
             let tcp_service = TimeoutService::new(
-                self.timeout,
+                self.config.timeout,
                 apply_fn(self.connector, |msg: Connect, srv| {
                     srv.call(TcpConnect::new(msg.uri).set_addr(msg.addr))
                 })
@@ -316,18 +343,9 @@ where
             connect_impl::InnerConnector {
                 tcp_pool: ConnectionPool::new(
                     tcp_service,
-                    self.conn_lifetime,
-                    self.conn_keep_alive,
-                    None,
-                    self.limit,
-                ),
-                ssl_pool: ConnectionPool::new(
-                    ssl_service,
-                    self.conn_lifetime,
-                    self.conn_keep_alive,
-                    Some(self.disconnect_timeout),
-                    self.limit,
+                    self.config.no_disconnect_timeout(),
                 ),
+                ssl_pool: ConnectionPool::new(ssl_service, self.config),
             }
         }
     }

actix-http/src/client/h2proto.rs

@@ -1,11 +1,15 @@
 use std::convert::TryFrom;
+use std::future::Future;
 use std::time;
 
 use actix_codec::{AsyncRead, AsyncWrite};
 use bytes::Bytes;
 use futures_util::future::poll_fn;
 use futures_util::pin_mut;
-use h2::{client::SendRequest, SendStream};
+use h2::{
+    client::{Builder, Connection, SendRequest},
+    SendStream,
+};
 use http::header::{HeaderValue, CONNECTION, CONTENT_LENGTH, TRANSFER_ENCODING};
 use http::{request::Request, Method, Version};
 
@@ -14,6 +18,7 @@ use crate::header::HeaderMap;
 use crate::message::{RequestHeadType, ResponseHead};
 use crate::payload::Payload;
 
+use super::config::ConnectorConfig;
 use super::connection::{ConnectionType, IoConnection};
 use super::error::SendRequestError;
 use super::pool::Acquired;
@@ -185,3 +190,18 @@ fn release<T: AsyncRead + AsyncWrite + Unpin + 'static>(
         }
     }
 }
+
+pub(crate) fn handshake<Io>(
+    io: Io,
+    config: &ConnectorConfig,
+) -> impl Future<Output = Result<(SendRequest<Bytes>, Connection<Io, Bytes>), h2::Error>>
+where
+    Io: AsyncRead + AsyncWrite + Unpin + 'static,
+{
+    let mut builder = Builder::new();
+    builder
+        .initial_window_size(config.stream_window_size)
+        .initial_connection_window_size(config.conn_window_size)
+        .enable_push(false);
+    builder.handshake(io)
+}

actix-http/src/client/mod.rs

@@ -1,6 +1,7 @@
 //! Http client api
 use http::Uri;
 
+mod config;
 mod connection;
 mod connector;
 mod error;

actix-http/src/client/pool.rs

@@ -13,14 +13,16 @@ use actix_utils::{oneshot, task::LocalWaker};
 use bytes::Bytes;
 use futures_util::future::{poll_fn, FutureExt, LocalBoxFuture};
 use fxhash::FxHashMap;
-use h2::client::{handshake, Connection, SendRequest};
+use h2::client::{Connection, SendRequest};
 use http::uri::Authority;
 use indexmap::IndexSet;
 use pin_project::pin_project;
 use slab::Slab;
 
+use super::config::ConnectorConfig;
 use super::connection::{ConnectionType, IoConnection};
 use super::error::ConnectError;
+use super::h2proto::handshake;
 use super::Connect;
 
 #[derive(Clone, Copy, PartialEq)]
@@ -50,20 +52,11 @@ where
     T: Service<Request = Connect, Response = (Io, Protocol), Error = ConnectError>
         + 'static,
 {
-    pub(crate) fn new(
-        connector: T,
-        conn_lifetime: Duration,
-        conn_keep_alive: Duration,
-        disconnect_timeout: Option<Duration>,
-        limit: usize,
-    ) -> Self {
+    pub(crate) fn new(connector: T, config: ConnectorConfig) -> Self {
         ConnectionPool(
             Rc::new(RefCell::new(connector)),
             Rc::new(RefCell::new(Inner {
-                conn_lifetime,
-                conn_keep_alive,
-                disconnect_timeout,
-                limit,
+                config,
                 acquired: 0,
                 waiters: Slab::new(),
                 waiters_queue: IndexSet::new(),
@@ -129,6 +122,8 @@ where
                     // open tcp connection
                     let (io, proto) = connector.call(req).await?;
 
+                    let config = inner.borrow().config.clone();
+
                     let guard = OpenGuard::new(key, inner);
 
                     if proto == Protocol::Http1 {
@@ -138,7 +133,7 @@ where
                             Some(guard.consume()),
                         ))
                     } else {
-                        let (snd, connection) = handshake(io).await?;
+                        let (snd, connection) = handshake(io, &config).await?;
                         actix_rt::spawn(connection.map(|_| ()));
                         Ok(IoConnection::new(
                             ConnectionType::H2(snd),
@@ -255,10 +250,7 @@ struct AvailableConnection<Io> {
 }
 
 pub(crate) struct Inner<Io> {
-    conn_lifetime: Duration,
-    conn_keep_alive: Duration,
-    disconnect_timeout: Option<Duration>,
-    limit: usize,
+    config: ConnectorConfig,
     acquired: usize,
     available: FxHashMap<Key, VecDeque<AvailableConnection<Io>>>,
     waiters: Slab<
@@ -311,7 +303,7 @@ where
 
     fn acquire(&mut self, key: &Key, cx: &mut Context<'_>) -> Acquire<Io> {
         // check limits
-        if self.limit > 0 && self.acquired >= self.limit {
+        if self.config.limit > 0 && self.acquired >= self.config.limit {
             return Acquire::NotAvailable;
         }
 
@@ -323,10 +315,10 @@ where
             let now = Instant::now();
             while let Some(conn) = connections.pop_back() {
                 // check if it still usable
-                if (now - conn.used) > self.conn_keep_alive
-                    || (now - conn.created) > self.conn_lifetime
+                if (now - conn.used) > self.config.conn_keep_alive
+                    || (now - conn.created) > self.config.conn_lifetime
                 {
-                    if let Some(timeout) = self.disconnect_timeout {
+                    if let Some(timeout) = self.config.disconnect_timeout {
                         if let ConnectionType::H1(io) = conn.io {
                             actix_rt::spawn(CloseConnection::new(io, timeout))
                         }
@@ -338,7 +330,7 @@ where
                         match Pin::new(s).poll_read(cx, &mut buf) {
                             Poll::Pending => (),
                             Poll::Ready(Ok(n)) if n > 0 => {
-                                if let Some(timeout) = self.disconnect_timeout {
+                                if let Some(timeout) = self.config.disconnect_timeout {
                                     if let ConnectionType::H1(io) = io {
                                         actix_rt::spawn(CloseConnection::new(
                                             io, timeout,
@@ -372,7 +364,7 @@ where
 
     fn release_close(&mut self, io: ConnectionType<Io>) {
         self.acquired -= 1;
-        if let Some(timeout) = self.disconnect_timeout {
+        if let Some(timeout) = self.config.disconnect_timeout {
             if let ConnectionType::H1(io) = io {
                 actix_rt::spawn(CloseConnection::new(io, timeout))
             }
@@ -381,7 +373,7 @@ where
     }
 
     fn check_availibility(&self) {
-        if !self.waiters_queue.is_empty() && self.acquired < self.limit {
+        if !self.waiters_queue.is_empty() && self.acquired < self.config.limit {
             self.waker.wake();
         }
     }
@@ -480,6 +472,7 @@ where
                         tx,
                         this.inner.clone(),
                         this.connector.call(connect),
+                        inner.config.clone(),
                     );
                 }
             }
@@ -506,6 +499,7 @@ where
     >,
     rx: Option<oneshot::Sender<Result<IoConnection<Io>, ConnectError>>>,
     inner: Option<Rc<RefCell<Inner<Io>>>>,
+    config: ConnectorConfig,
 }
 
 impl<F, Io> OpenWaitingConnection<F, Io>
@@ -518,6 +512,7 @@ where
         rx: oneshot::Sender<Result<IoConnection<Io>, ConnectError>>,
         inner: Rc<RefCell<Inner<Io>>>,
         fut: F,
+        config: ConnectorConfig,
     ) {
         actix_rt::spawn(OpenWaitingConnection {
             key,
@@ -525,6 +520,7 @@ where
             h2: None,
             rx: Some(rx),
             inner: Some(inner),
+            config,
         })
     }
 }
@@ -594,7 +590,7 @@ where
                     )));
                     Poll::Ready(())
                 } else {
-                    *this.h2 = Some(handshake(io).boxed_local());
+                    *this.h2 = Some(handshake(io, this.config).boxed_local());
                     self.poll(cx)
                 }
             }

actix-http/src/cookie/mod.rs

@@ -47,7 +47,7 @@
 //! ```
 
 #![doc(html_root_url = "https://docs.rs/cookie/0.11")]
-#![deny(missing_docs)]
+#![warn(missing_docs)]
 
 mod builder;
 mod delta;

actix-http/src/h1/dispatcher.rs

@@ -73,7 +73,7 @@ where
     U::Error: fmt::Display,
 {
     Normal(#[pin] InnerDispatcher<T, S, B, X, U>),
-    Upgrade(#[pin] U::Future),
+    Upgrade(Pin<Box<U::Future>>),
 }
 
 #[pin_project]
@@ -123,8 +123,8 @@ where
     B: MessageBody,
 {
     None,
-    ExpectCall(#[pin] X::Future),
-    ServiceCall(#[pin] S::Future),
+    ExpectCall(Pin<Box<X::Future>>),
+    ServiceCall(Pin<Box<S::Future>>),
     SendPayload(#[pin] ResponseBody<B>),
 }
 
@@ -391,11 +391,12 @@ where
                     }
                     None => None,
                 },
-                State::ExpectCall(fut) => match fut.poll(cx) {
+                State::ExpectCall(fut) => match fut.as_mut().poll(cx) {
                     Poll::Ready(Ok(req)) => {
                         self.as_mut().send_continue();
                         this = self.as_mut().project();
-                        this.state.set(State::ServiceCall(this.service.call(req)));
+                        this.state
+                            .set(State::ServiceCall(Box::pin(this.service.call(req))));
                         continue;
                     }
                     Poll::Ready(Err(e)) => {
@@ -405,7 +406,7 @@ where
                     }
                     Poll::Pending => None,
                 },
-                State::ServiceCall(fut) => match fut.poll(cx) {
+                State::ServiceCall(fut) => match fut.as_mut().poll(cx) {
                     Poll::Ready(Ok(res)) => {
                         let (res, body) = res.into().replace_body(());
                         let state = self.as_mut().send_response(res, body)?;
@@ -485,8 +486,8 @@ where
     ) -> Result<State<S, B, X>, DispatchError> {
         // Handle `EXPECT: 100-Continue` header
         let req = if req.head().expect() {
-            let mut task = self.as_mut().project().expect.call(req);
-            match unsafe { Pin::new_unchecked(&mut task) }.poll(cx) {
+            let mut task = Box::pin(self.as_mut().project().expect.call(req));
+            match task.as_mut().poll(cx) {
                 Poll::Ready(Ok(req)) => {
                     self.as_mut().send_continue();
                     req
@@ -504,8 +505,8 @@ where
         };
 
         // Call service
-        let mut task = self.as_mut().project().service.call(req);
-        match unsafe { Pin::new_unchecked(&mut task) }.poll(cx) {
+        let mut task = Box::pin(self.as_mut().project().service.call(req));
+        match task.as_mut().poll(cx) {
             Poll::Ready(Ok(res)) => {
                 let (res, body) = res.into().replace_body(());
                 self.send_response(res, body)
@@ -807,7 +808,7 @@ where
                             self.as_mut()
                                 .project()
                                 .inner
-                                .set(DispatcherState::Upgrade(upgrade));
+                                .set(DispatcherState::Upgrade(Box::pin(upgrade)));
                             return self.poll(cx);
                         }
 
@@ -855,7 +856,7 @@ where
                     }
                 }
             }
-            DispatcherState::Upgrade(fut) => fut.poll(cx).map_err(|e| {
+            DispatcherState::Upgrade(fut) => fut.as_mut().poll(cx).map_err(|e| {
                 error!("Upgrade handler error: {}", e);
                 DispatchError::Upgrade
             }),

actix-http/src/lib.rs

@@ -1,5 +1,5 @@
 //! Basic http primitives for actix-net framework.
-#![deny(rust_2018_idioms, warnings)]
+#![warn(rust_2018_idioms, warnings)]
 #![allow(
     clippy::type_complexity,
     clippy::too_many_arguments,

actix-http/tests/test_client.rs

@@ -33,7 +33,8 @@ async fn test_h1_v2() {
         HttpService::build()
             .finish(|_| future::ok::<_, ()>(Response::Ok().body(STR)))
             .tcp()
-    });
+    })
+    .await;
 
     let response = srv.get("/").send().await.unwrap();
     assert!(response.status().is_success());
@@ -61,7 +62,8 @@ async fn test_connection_close() {
             .finish(|_| ok::<_, ()>(Response::Ok().body(STR)))
             .tcp()
             .map(|_| ())
-    });
+    })
+    .await;
 
     let response = srv.get("/").force_close().send().await.unwrap();
     assert!(response.status().is_success());
@@ -80,7 +82,8 @@ async fn test_with_query_parameter() {
             })
             .tcp()
             .map(|_| ())
-    });
+    })
+    .await;
 
     let request = srv.request(http::Method::GET, srv.url("/?qp=5"));
     let response = request.send().await.unwrap();

actix-http/tests/test_openssl.rs

@@ -67,7 +67,8 @@ async fn test_h2() -> io::Result<()> {
             .h2(|_| ok::<_, Error>(Response::Ok().finish()))
             .openssl(ssl_acceptor())
             .map_err(|_| ())
-    });
+    })
+    .await;
 
     let response = srv.sget("/").send().await.unwrap();
     assert!(response.status().is_success());
@@ -85,7 +86,8 @@ async fn test_h2_1() -> io::Result<()> {
             })
             .openssl(ssl_acceptor())
             .map_err(|_| ())
-    });
+    })
+    .await;
 
     let response = srv.sget("/").send().await.unwrap();
     assert!(response.status().is_success());
@@ -103,7 +105,8 @@ async fn test_h2_body() -> io::Result<()> {
             })
             .openssl(ssl_acceptor())
             .map_err(|_| ())
-    });
+    })
+    .await;
 
     let response = srv.sget("/").send_body(data.clone()).await.unwrap();
     assert!(response.status().is_success());
@@ -131,7 +134,8 @@ async fn test_h2_content_length() {
             })
             .openssl(ssl_acceptor())
             .map_err(|_| ())
-    });
+    })
+    .await;
 
     let header = HeaderName::from_static("content-length");
     let value = HeaderValue::from_static("0");
@@ -192,7 +196,7 @@ async fn test_h2_headers() {
         })
             .openssl(ssl_acceptor())
                     .map_err(|_| ())
-    });
+    }).await;
 
     let response = srv.sget("/").send().await.unwrap();
     assert!(response.status().is_success());
@@ -231,7 +235,8 @@ async fn test_h2_body2() {
             .h2(|_| ok::<_, ()>(Response::Ok().body(STR)))
             .openssl(ssl_acceptor())
             .map_err(|_| ())
-    });
+    })
+    .await;
 
     let response = srv.sget("/").send().await.unwrap();
     assert!(response.status().is_success());
@@ -248,7 +253,8 @@ async fn test_h2_head_empty() {
             .finish(|_| ok::<_, ()>(Response::Ok().body(STR)))
             .openssl(ssl_acceptor())
             .map_err(|_| ())
-    });
+    })
+    .await;
 
     let response = srv.shead("/").send().await.unwrap();
     assert!(response.status().is_success());
@@ -273,7 +279,8 @@ async fn test_h2_head_binary() {
             })
             .openssl(ssl_acceptor())
             .map_err(|_| ())
-    });
+    })
+    .await;
 
     let response = srv.shead("/").send().await.unwrap();
     assert!(response.status().is_success());
@@ -295,7 +302,8 @@ async fn test_h2_head_binary2() {
             .h2(|_| ok::<_, ()>(Response::Ok().body(STR)))
             .openssl(ssl_acceptor())
             .map_err(|_| ())
-    });
+    })
+    .await;
 
     let response = srv.shead("/").send().await.unwrap();
     assert!(response.status().is_success());
@@ -318,7 +326,8 @@ async fn test_h2_body_length() {
             })
             .openssl(ssl_acceptor())
             .map_err(|_| ())
-    });
+    })
+    .await;
 
     let response = srv.sget("/").send().await.unwrap();
     assert!(response.status().is_success());
@@ -342,7 +351,8 @@ async fn test_h2_body_chunked_explicit() {
             })
             .openssl(ssl_acceptor())
             .map_err(|_| ())
-    });
+    })
+    .await;
 
     let response = srv.sget("/").send().await.unwrap();
     assert!(response.status().is_success());
@@ -369,7 +379,8 @@ async fn test_h2_response_http_error_handling() {
             }))
             .openssl(ssl_acceptor())
             .map_err(|_| ())
-    });
+    })
+    .await;
 
     let response = srv.sget("/").send().await.unwrap();
     assert_eq!(response.status(), StatusCode::INTERNAL_SERVER_ERROR);
@@ -386,7 +397,8 @@ async fn test_h2_service_error() {
             .h2(|_| err::<Response, Error>(ErrorBadRequest("error")))
             .openssl(ssl_acceptor())
             .map_err(|_| ())
-    });
+    })
+    .await;
 
     let response = srv.sget("/").send().await.unwrap();
     assert_eq!(response.status(), StatusCode::BAD_REQUEST);
@@ -407,7 +419,8 @@ async fn test_h2_on_connect() {
             })
             .openssl(ssl_acceptor())
             .map_err(|_| ())
-    });
+    })
+    .await;
 
     let response = srv.sget("/").send().await.unwrap();
     assert!(response.status().is_success());

actix-http/tests/test_rustls.rs

@@ -45,7 +45,8 @@ async fn test_h1() -> io::Result<()> {
         HttpService::build()
             .h1(|_| future::ok::<_, Error>(Response::Ok().finish()))
             .rustls(ssl_acceptor())
-    });
+    })
+    .await;
 
     let response = srv.sget("/").send().await.unwrap();
     assert!(response.status().is_success());
@@ -58,7 +59,8 @@ async fn test_h2() -> io::Result<()> {
         HttpService::build()
             .h2(|_| future::ok::<_, Error>(Response::Ok().finish()))
             .rustls(ssl_acceptor())
-    });
+    })
+    .await;
 
     let response = srv.sget("/").send().await.unwrap();
     assert!(response.status().is_success());
@@ -75,7 +77,8 @@ async fn test_h1_1() -> io::Result<()> {
                 future::ok::<_, Error>(Response::Ok().finish())
             })
             .rustls(ssl_acceptor())
-    });
+    })
+    .await;
 
     let response = srv.sget("/").send().await.unwrap();
     assert!(response.status().is_success());
@@ -92,7 +95,8 @@ async fn test_h2_1() -> io::Result<()> {
                 future::ok::<_, Error>(Response::Ok().finish())
             })
             .rustls(ssl_acceptor())
-    });
+    })
+    .await;
 
     let response = srv.sget("/").send().await.unwrap();
     assert!(response.status().is_success());
@@ -109,7 +113,8 @@ async fn test_h2_body1() -> io::Result<()> {
                 Ok::<_, Error>(Response::Ok().body(body))
             })
             .rustls(ssl_acceptor())
-    });
+    })
+    .await;
 
     let response = srv.sget("/").send_body(data.clone()).await.unwrap();
     assert!(response.status().is_success());
@@ -136,7 +141,8 @@ async fn test_h2_content_length() {
                 future::ok::<_, ()>(Response::new(statuses[indx]))
             })
             .rustls(ssl_acceptor())
-    });
+    })
+    .await;
 
     let header = HeaderName::from_static("content-length");
     let value = HeaderValue::from_static("0");
@@ -195,7 +201,7 @@ async fn test_h2_headers() {
             future::ok::<_, ()>(config.body(data.clone()))
         })
             .rustls(ssl_acceptor())
-    });
+    }).await;
 
     let response = srv.sget("/").send().await.unwrap();
     assert!(response.status().is_success());
@@ -233,7 +239,8 @@ async fn test_h2_body2() {
         HttpService::build()
             .h2(|_| future::ok::<_, ()>(Response::Ok().body(STR)))
             .rustls(ssl_acceptor())
-    });
+    })
+    .await;
 
     let response = srv.sget("/").send().await.unwrap();
     assert!(response.status().is_success());
@@ -249,7 +256,8 @@ async fn test_h2_head_empty() {
         HttpService::build()
             .finish(|_| ok::<_, ()>(Response::Ok().body(STR)))
             .rustls(ssl_acceptor())
-    });
+    })
+    .await;
 
     let response = srv.shead("/").send().await.unwrap();
     assert!(response.status().is_success());
@@ -276,7 +284,8 @@ async fn test_h2_head_binary() {
                 ok::<_, ()>(Response::Ok().content_length(STR.len() as u64).body(STR))
             })
             .rustls(ssl_acceptor())
-    });
+    })
+    .await;
 
     let response = srv.shead("/").send().await.unwrap();
     assert!(response.status().is_success());
@@ -300,7 +309,8 @@ async fn test_h2_head_binary2() {
         HttpService::build()
             .h2(|_| ok::<_, ()>(Response::Ok().body(STR)))
             .rustls(ssl_acceptor())
-    });
+    })
+    .await;
 
     let response = srv.shead("/").send().await.unwrap();
     assert!(response.status().is_success());
@@ -325,7 +335,8 @@ async fn test_h2_body_length() {
                 )
             })
             .rustls(ssl_acceptor())
-    });
+    })
+    .await;
 
     let response = srv.sget("/").send().await.unwrap();
     assert!(response.status().is_success());
@@ -348,7 +359,8 @@ async fn test_h2_body_chunked_explicit() {
                 )
             })
             .rustls(ssl_acceptor())
-    });
+    })
+    .await;
 
     let response = srv.sget("/").send().await.unwrap();
     assert!(response.status().is_success());
@@ -376,7 +388,8 @@ async fn test_h2_response_http_error_handling() {
                 }))
             }))
             .rustls(ssl_acceptor())
-    });
+    })
+    .await;
 
     let response = srv.sget("/").send().await.unwrap();
     assert_eq!(response.status(), http::StatusCode::INTERNAL_SERVER_ERROR);
@@ -392,7 +405,8 @@ async fn test_h2_service_error() {
         HttpService::build()
             .h2(|_| err::<Response, Error>(error::ErrorBadRequest("error")))
             .rustls(ssl_acceptor())
-    });
+    })
+    .await;
 
     let response = srv.sget("/").send().await.unwrap();
     assert_eq!(response.status(), http::StatusCode::BAD_REQUEST);
@@ -408,7 +422,8 @@ async fn test_h1_service_error() {
         HttpService::build()
             .h1(|_| err::<Response, Error>(error::ErrorBadRequest("error")))
             .rustls(ssl_acceptor())
-    });
+    })
+    .await;
 
     let response = srv.sget("/").send().await.unwrap();
     assert_eq!(response.status(), http::StatusCode::BAD_REQUEST);

actix-http/tests/test_server.rs

@@ -27,7 +27,8 @@ async fn test_h1() {
                 future::ok::<_, ()>(Response::Ok().finish())
             })
             .tcp()
-    });
+    })
+    .await;
 
     let response = srv.get("/").send().await.unwrap();
     assert!(response.status().is_success());
@@ -46,7 +47,8 @@ async fn test_h1_2() {
                 future::ok::<_, ()>(Response::Ok().finish())
             })
             .tcp()
-    });
+    })
+    .await;
 
     let response = srv.get("/").send().await.unwrap();
     assert!(response.status().is_success());
@@ -65,7 +67,8 @@ async fn test_expect_continue() {
             }))
             .finish(|_| future::ok::<_, ()>(Response::Ok().finish()))
             .tcp()
-    });
+    })
+    .await;
 
     let mut stream = net::TcpStream::connect(srv.addr()).unwrap();
     let _ = stream.write_all(b"GET /test HTTP/1.1\r\nexpect: 100-continue\r\n\r\n");
@@ -95,7 +98,8 @@ async fn test_expect_continue_h1() {
             }))
             .h1(fn_service(|_| future::ok::<_, ()>(Response::Ok().finish())))
             .tcp()
-    });
+    })
+    .await;
 
     let mut stream = net::TcpStream::connect(srv.addr()).unwrap();
     let _ = stream.write_all(b"GET /test HTTP/1.1\r\nexpect: 100-continue\r\n\r\n");
@@ -130,7 +134,8 @@ async fn test_chunked_payload() {
                     })
             }))
             .tcp()
-    });
+    })
+    .await;
 
     let returned_size = {
         let mut stream = net::TcpStream::connect(srv.addr()).unwrap();
@@ -172,7 +177,8 @@ async fn test_slow_request() {
             .client_timeout(100)
             .finish(|_| future::ok::<_, ()>(Response::Ok().finish()))
             .tcp()
-    });
+    })
+    .await;
 
     let mut stream = net::TcpStream::connect(srv.addr()).unwrap();
     let _ = stream.write_all(b"GET /test/tests/test HTTP/1.1\r\n");
@@ -187,7 +193,8 @@ async fn test_http1_malformed_request() {
         HttpService::build()
             .h1(|_| future::ok::<_, ()>(Response::Ok().finish()))
             .tcp()
-    });
+    })
+    .await;
 
     let mut stream = net::TcpStream::connect(srv.addr()).unwrap();
     let _ = stream.write_all(b"GET /test/tests/test HTTP1.1\r\n");
@@ -202,7 +209,8 @@ async fn test_http1_keepalive() {
         HttpService::build()
             .h1(|_| future::ok::<_, ()>(Response::Ok().finish()))
             .tcp()
-    });
+    })
+    .await;
 
     let mut stream = net::TcpStream::connect(srv.addr()).unwrap();
     let _ = stream.write_all(b"GET /test/tests/test HTTP/1.1\r\n\r\n");
@@ -223,7 +231,8 @@ async fn test_http1_keepalive_timeout() {
             .keep_alive(1)
             .h1(|_| future::ok::<_, ()>(Response::Ok().finish()))
             .tcp()
-    });
+    })
+    .await;
 
     let mut stream = net::TcpStream::connect(srv.addr()).unwrap();
     let _ = stream.write_all(b"GET /test/tests/test HTTP/1.1\r\n\r\n");
@@ -243,7 +252,8 @@ async fn test_http1_keepalive_close() {
         HttpService::build()
             .h1(|_| future::ok::<_, ()>(Response::Ok().finish()))
             .tcp()
-    });
+    })
+    .await;
 
     let mut stream = net::TcpStream::connect(srv.addr()).unwrap();
     let _ =
@@ -263,7 +273,8 @@ async fn test_http10_keepalive_default_close() {
         HttpService::build()
             .h1(|_| future::ok::<_, ()>(Response::Ok().finish()))
             .tcp()
-    });
+    })
+    .await;
 
     let mut stream = net::TcpStream::connect(srv.addr()).unwrap();
     let _ = stream.write_all(b"GET /test/tests/test HTTP/1.0\r\n\r\n");
@@ -282,7 +293,8 @@ async fn test_http10_keepalive() {
         HttpService::build()
             .h1(|_| future::ok::<_, ()>(Response::Ok().finish()))
             .tcp()
-    });
+    })
+    .await;
 
     let mut stream = net::TcpStream::connect(srv.addr()).unwrap();
     let _ = stream
@@ -309,7 +321,8 @@ async fn test_http1_keepalive_disabled() {
             .keep_alive(KeepAlive::Disabled)
             .h1(|_| future::ok::<_, ()>(Response::Ok().finish()))
             .tcp()
-    });
+    })
+    .await;
 
     let mut stream = net::TcpStream::connect(srv.addr()).unwrap();
     let _ = stream.write_all(b"GET /test/tests/test HTTP/1.1\r\n\r\n");
@@ -344,7 +357,8 @@ async fn test_content_length() {
                 future::ok::<_, ()>(Response::new(statuses[indx]))
             })
             .tcp()
-    });
+    })
+    .await;
 
     let header = HeaderName::from_static("content-length");
     let value = HeaderValue::from_static("0");
@@ -397,7 +411,7 @@ async fn test_h1_headers() {
             }
             future::ok::<_, ()>(builder.body(data.clone()))
         }).tcp()
-    });
+    }).await;
 
     let response = srv.get("/").send().await.unwrap();
     assert!(response.status().is_success());
@@ -435,7 +449,8 @@ async fn test_h1_body() {
         HttpService::build()
             .h1(|_| ok::<_, ()>(Response::Ok().body(STR)))
             .tcp()
-    });
+    })
+    .await;
 
     let response = srv.get("/").send().await.unwrap();
     assert!(response.status().is_success());
@@ -451,7 +466,8 @@ async fn test_h1_head_empty() {
         HttpService::build()
             .h1(|_| ok::<_, ()>(Response::Ok().body(STR)))
             .tcp()
-    });
+    })
+    .await;
 
     let response = srv.head("/").send().await.unwrap();
     assert!(response.status().is_success());
@@ -477,7 +493,8 @@ async fn test_h1_head_binary() {
                 ok::<_, ()>(Response::Ok().content_length(STR.len() as u64).body(STR))
             })
             .tcp()
-    });
+    })
+    .await;
 
     let response = srv.head("/").send().await.unwrap();
     assert!(response.status().is_success());
@@ -501,7 +518,8 @@ async fn test_h1_head_binary2() {
         HttpService::build()
             .h1(|_| ok::<_, ()>(Response::Ok().body(STR)))
             .tcp()
-    });
+    })
+    .await;
 
     let response = srv.head("/").send().await.unwrap();
     assert!(response.status().is_success());
@@ -526,7 +544,8 @@ async fn test_h1_body_length() {
                 )
             })
             .tcp()
-    });
+    })
+    .await;
 
     let response = srv.get("/").send().await.unwrap();
     assert!(response.status().is_success());
@@ -549,7 +568,8 @@ async fn test_h1_body_chunked_explicit() {
                 )
             })
             .tcp()
-    });
+    })
+    .await;
 
     let response = srv.get("/").send().await.unwrap();
     assert!(response.status().is_success());
@@ -579,7 +599,8 @@ async fn test_h1_body_chunked_implicit() {
                 ok::<_, ()>(Response::Ok().streaming(body))
             })
             .tcp()
-    });
+    })
+    .await;
 
     let response = srv.get("/").send().await.unwrap();
     assert!(response.status().is_success());
@@ -611,7 +632,8 @@ async fn test_h1_response_http_error_handling() {
                 )
             }))
             .tcp()
-    });
+    })
+    .await;
 
     let response = srv.get("/").send().await.unwrap();
     assert_eq!(response.status(), http::StatusCode::INTERNAL_SERVER_ERROR);
@@ -627,7 +649,8 @@ async fn test_h1_service_error() {
         HttpService::build()
             .h1(|_| future::err::<Response, Error>(error::ErrorBadRequest("error")))
             .tcp()
-    });
+    })
+    .await;
 
     let response = srv.get("/").send().await.unwrap();
     assert_eq!(response.status(), http::StatusCode::BAD_REQUEST);
@@ -647,7 +670,8 @@ async fn test_h1_on_connect() {
                 future::ok::<_, ()>(Response::Ok().finish())
             })
             .tcp()
-    });
+    })
+    .await;
 
     let response = srv.get("/").send().await.unwrap();
     assert!(response.status().is_success());

actix-http/tests/test_ws.rs

@@ -93,7 +93,8 @@ async fn test_simple() {
                 .finish(|_| future::ok::<_, ()>(Response::NotFound()))
                 .tcp()
         }
-    });
+    })
+    .await;
 
     // client service
     let mut framed = srv.ws().await.unwrap();

actix-multipart/CHANGES.md

@@ -4,6 +4,8 @@
 
 * Remove the unused `time` dependency
 
+* Fix missing `std::error::Error` implement for `MultipartError`.
+
 ## [0.2.0] - 2019-12-20
 
 * Release

actix-multipart/Cargo.toml

@@ -29,4 +29,4 @@ twoway = "0.2"
 
 [dev-dependencies]
 actix-rt = "1.0.0"
-actix-http = "2.0.0-alpha.1"
+actix-http = "2.0.0-alpha.2"

actix-multipart/src/error.rs

@@ -33,6 +33,8 @@ pub enum MultipartError {
     NotConsumed,
 }
 
+impl std::error::Error for MultipartError {}
+
 /// Return `BadRequest` for `MultipartError`
 impl ResponseError for MultipartError {
     fn status_code(&self) -> StatusCode {

actix-web-actors/Cargo.toml

@@ -18,7 +18,7 @@ path = "src/lib.rs"
 [dependencies]
 actix = "0.10.0-alpha.1"
 actix-web = "2.0.0"
-actix-http = "2.0.0-alpha.1"
+actix-http = "2.0.0-alpha.2"
 actix-codec = "0.2.0"
 bytes = "0.5.2"
 futures = "0.3.1"

awc/CHANGES.md

@@ -1,5 +1,12 @@
 # Changes
 
+## [2.0.0-alpha.1] - 2020-03-11
+
+* Update `actix-http` dependency to 2.0.0-alpha.2
+* Update `rustls` dependency to 0.17
+* ClientBuilder accepts initial_window_size and initial_connection_window_size HTTP2 configuration
+* ClientBuilder allowing to set max_http_version to limit HTTP version to be used
+
 ## [1.0.1] - 2019-12-15
 
 * Fix compilation with default features off

awc/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "awc"
-version = "1.0.1"
+version = "2.0.0-alpha.1"
 authors = ["Nikolay Kim <fafhrd91@gmail.com>"]
 description = "Actix http client."
 readme = "README.md"
@@ -36,7 +36,7 @@ compress = ["actix-http/compress"]
 [dependencies]
 actix-codec = "0.2.0"
 actix-service = "1.0.1"
-actix-http = "2.0.0-alpha.1"
+actix-http = "2.0.0-alpha.2"
 actix-rt = "1.0.0"
 
 base64 = "0.11"
@@ -51,18 +51,18 @@ serde = "1.0"
 serde_json = "1.0"
 serde_urlencoded = "0.6.1"
 open-ssl = { version="0.10", package="openssl", optional = true }
-rust-tls = { version = "0.16.0", package="rustls", optional = true, features = ["dangerous_configuration"]  }
+rust-tls = { version = "0.17.0", package="rustls", optional = true, features = ["dangerous_configuration"]  }
 
 [dev-dependencies]
-actix-connect = { version = "1.0.1", features=["openssl"] }
+actix-connect = { version = "2.0.0-alpha.2", features=["openssl"] }
 actix-web = { version = "2.0.0", features=["openssl"] }
-actix-http = { version = "2.0.0-alpha.1", features=["openssl"] }
+actix-http = { version = "2.0.0-alpha.2", features=["openssl"] }
 actix-http-test = { version = "1.0.0", features=["openssl"] }
 actix-utils = "1.0.3"
 actix-server = "1.0.0"
-actix-tls = { version = "1.0.0", features=["openssl", "rustls"] }
+actix-tls = { version = "2.0.0-alpha.1", features=["openssl", "rustls"] }
 brotli2 = "0.3.2"
 flate2 = "1.0.13"
 futures = "0.3.1"
-env_logger = "0.6"
+env_logger = "0.7"
 webpki = "0.21"

awc/src/builder.rs

@@ -4,11 +4,11 @@ use std::fmt;
 use std::rc::Rc;
 use std::time::Duration;
 
-use actix_http::client::{Connect, ConnectError, Connection, Connector};
-use actix_http::http::{header, Error as HttpError, HeaderMap, HeaderName};
+use actix_http::client::{Connect as HttpConnect, ConnectError, Connection, Connector};
+use actix_http::http::{self, header, Error as HttpError, HeaderMap, HeaderName};
 use actix_service::Service;
 
-use crate::connect::ConnectorWrapper;
+use crate::connect::{Connect, ConnectorWrapper};
 use crate::{Client, ClientConfig};
 
 /// An HTTP Client builder
@@ -16,10 +16,15 @@ use crate::{Client, ClientConfig};
 /// This type can be used to construct an instance of `Client` through a
 /// builder-like pattern.
 pub struct ClientBuilder {
-    config: ClientConfig,
     default_headers: bool,
     allow_redirects: bool,
     max_redirects: usize,
+    max_http_version: Option<http::Version>,
+    stream_window_size: Option<u32>,
+    conn_window_size: Option<u32>,
+    headers: HeaderMap,
+    timeout: Option<Duration>,
+    connector: Option<RefCell<Box<dyn Connect>>>,
 }
 
 impl Default for ClientBuilder {
@@ -34,25 +39,24 @@ impl ClientBuilder {
             default_headers: true,
             allow_redirects: true,
             max_redirects: 10,
-            config: ClientConfig {
-                headers: HeaderMap::new(),
-                timeout: Some(Duration::from_secs(5)),
-                connector: RefCell::new(Box::new(ConnectorWrapper(
-                    Connector::new().finish(),
-                ))),
-            },
+            headers: HeaderMap::new(),
+            timeout: Some(Duration::from_secs(5)),
+            connector: None,
+            max_http_version: None,
+            stream_window_size: None,
+            conn_window_size: None,
         }
     }
 
     /// Use custom connector service.
     pub fn connector<T>(mut self, connector: T) -> Self
     where
-        T: Service<Request = Connect, Error = ConnectError> + 'static,
+        T: Service<Request = HttpConnect, Error = ConnectError> + 'static,
         T::Response: Connection,
         <T::Response as Connection>::Future: 'static,
         T::Future: 'static,
     {
-        self.config.connector = RefCell::new(Box::new(ConnectorWrapper(connector)));
+        self.connector = Some(RefCell::new(Box::new(ConnectorWrapper(connector))));
         self
     }
 
@@ -61,13 +65,13 @@ impl ClientBuilder {
     /// Request timeout is the total time before a response must be received.
     /// Default value is 5 seconds.
     pub fn timeout(mut self, timeout: Duration) -> Self {
-        self.config.timeout = Some(timeout);
+        self.timeout = Some(timeout);
         self
     }
 
     /// Disable request timeout.
     pub fn disable_timeout(mut self) -> Self {
-        self.config.timeout = None;
+        self.timeout = None;
         self
     }
 
@@ -79,6 +83,31 @@ impl ClientBuilder {
         self
     }
 
+    /// Maximum supported http major version
+    /// Supported versions http/1.1, http/2
+    pub fn max_http_version(mut self, val: http::Version) -> Self {
+        self.max_http_version = Some(val);
+        self
+    }
+
+    /// Indicates the initial window size (in octets) for
+    /// HTTP2 stream-level flow control for received data.
+    ///
+    /// The default value is 65,535 and is good for APIs, but not for big objects.
+    pub fn initial_window_size(mut self, size: u32) -> Self {
+        self.stream_window_size = Some(size);
+        self
+    }
+
+    /// Indicates the initial window size (in octets) for
+    /// HTTP2 connection-level flow control for received data.
+    ///
+    /// The default value is 65,535 and is good for APIs, but not for big objects.
+    pub fn initial_connection_window_size(mut self, size: u32) -> Self {
+        self.conn_window_size = Some(size);
+        self
+    }
+
     /// Set max number of redirects.
     ///
     /// Max redirects is set to 10 by default.
@@ -106,7 +135,7 @@ impl ClientBuilder {
         match HeaderName::try_from(key) {
             Ok(key) => match value.try_into() {
                 Ok(value) => {
-                    self.config.headers.append(key, value);
+                    self.headers.append(key, value);
                 }
                 Err(e) => log::error!("Header value error: {:?}", e),
             },
@@ -140,7 +169,29 @@ impl ClientBuilder {
 
     /// Finish build process and create `Client` instance.
     pub fn finish(self) -> Client {
-        Client(Rc::new(self.config))
+        let connector = if let Some(connector) = self.connector {
+            connector
+        } else {
+            let mut connector = Connector::new();
+            if let Some(val) = self.max_http_version {
+                connector = connector.max_http_version(val)
+            };
+            if let Some(val) = self.conn_window_size {
+                connector = connector.initial_connection_window_size(val)
+            };
+            if let Some(val) = self.stream_window_size {
+                connector = connector.initial_window_size(val)
+            };
+            RefCell::new(
+                Box::new(ConnectorWrapper(connector.finish())) as Box<dyn Connect>
+            )
+        };
+        let config = ClientConfig {
+            headers: self.headers,
+            timeout: self.timeout,
+            connector,
+        };
+        Client(Rc::new(config))
     }
 }
 
@@ -153,7 +204,6 @@ mod tests {
         let client = ClientBuilder::new().basic_auth("username", Some("password"));
         assert_eq!(
             client
-                .config
                 .headers
                 .get(header::AUTHORIZATION)
                 .unwrap()
@@ -165,7 +215,6 @@ mod tests {
         let client = ClientBuilder::new().basic_auth("username", None);
         assert_eq!(
             client
-                .config
                 .headers
                 .get(header::AUTHORIZATION)
                 .unwrap()
@@ -180,7 +229,6 @@ mod tests {
         let client = ClientBuilder::new().bearer_auth("someS3cr3tAutht0k3n");
         assert_eq!(
             client
-                .config
                 .headers
                 .get(header::AUTHORIZATION)
                 .unwrap()

awc/src/lib.rs

@@ -1,4 +1,4 @@
-#![deny(rust_2018_idioms, warnings)]
+#![warn(rust_2018_idioms, warnings)]
 #![allow(
     clippy::type_complexity,
     clippy::borrow_interior_mutable_const,

awc/tests/test_client.rs

@@ -107,17 +107,15 @@ async fn test_form() {
 #[actix_rt::test]
 async fn test_timeout() {
     let srv = test::start(|| {
-        App::new().service(web::resource("/").route(web::to(|| {
-            async {
-                actix_rt::time::delay_for(Duration::from_millis(200)).await;
-                Ok::<_, Error>(HttpResponse::Ok().body(STR))
-            }
+        App::new().service(web::resource("/").route(web::to(|| async {
+            actix_rt::time::delay_for(Duration::from_millis(200)).await;
+            Ok::<_, Error>(HttpResponse::Ok().body(STR))
         })))
     });
 
     let connector = awc::Connector::new()
         .connector(actix_connect::new_connector(
-            actix_connect::start_default_resolver(),
+            actix_connect::start_default_resolver().await.unwrap(),
         ))
         .timeout(Duration::from_secs(15))
         .finish();
@@ -137,11 +135,9 @@ async fn test_timeout() {
 #[actix_rt::test]
 async fn test_timeout_override() {
     let srv = test::start(|| {
-        App::new().service(web::resource("/").route(web::to(|| {
-            async {
-                actix_rt::time::delay_for(Duration::from_millis(200)).await;
-                Ok::<_, Error>(HttpResponse::Ok().body(STR))
-            }
+        App::new().service(web::resource("/").route(web::to(|| async {
+            actix_rt::time::delay_for(Duration::from_millis(200)).await;
+            Ok::<_, Error>(HttpResponse::Ok().body(STR))
         })))
     });
 
@@ -177,7 +173,8 @@ async fn test_connection_reuse() {
             ))
             .tcp(),
         )
-    });
+    })
+    .await;
 
     let client = awc::Client::default();
 
@@ -214,7 +211,8 @@ async fn test_connection_force_close() {
             ))
             .tcp(),
         )
-    });
+    })
+    .await;
 
     let client = awc::Client::default();
 
@@ -253,7 +251,8 @@ async fn test_connection_server_close() {
             ))
             .tcp(),
         )
-    });
+    })
+    .await;
 
     let client = awc::Client::default();
 
@@ -291,7 +290,8 @@ async fn test_connection_wait_queue() {
             ))
             .tcp(),
         )
-    });
+    })
+    .await;
 
     let client = awc::Client::build()
         .connector(awc::Connector::new().limit(1).finish())
@@ -339,7 +339,8 @@ async fn test_connection_wait_queue_force_close() {
             ))
             .tcp(),
         )
-    });
+    })
+    .await;
 
     let client = awc::Client::build()
         .connector(awc::Connector::new().limit(1).finish())

awc/tests/test_connector.rs

@@ -0,0 +1,61 @@
+#![cfg(feature = "openssl")]
+use actix_http::HttpService;
+use actix_http_test::test_server;
+use actix_service::{map_config, ServiceFactory};
+use actix_web::http::Version;
+use actix_web::{dev::AppConfig, web, App, HttpResponse};
+use open_ssl::ssl::{SslAcceptor, SslConnector, SslFiletype, SslMethod, SslVerifyMode};
+
+fn ssl_acceptor() -> SslAcceptor {
+    // load ssl keys
+    let mut builder = SslAcceptor::mozilla_intermediate(SslMethod::tls()).unwrap();
+    builder
+        .set_private_key_file("../tests/key.pem", SslFiletype::PEM)
+        .unwrap();
+    builder
+        .set_certificate_chain_file("../tests/cert.pem")
+        .unwrap();
+    builder.set_alpn_select_callback(|_, protos| {
+        const H2: &[u8] = b"\x02h2";
+        if protos.windows(3).any(|window| window == H2) {
+            Ok(b"h2")
+        } else {
+            Err(open_ssl::ssl::AlpnError::NOACK)
+        }
+    });
+    builder.set_alpn_protos(b"\x02h2").unwrap();
+    builder.build()
+}
+
+#[actix_rt::test]
+async fn test_connection_window_size() {
+    let srv = test_server(move || {
+        HttpService::build()
+            .h2(map_config(
+                App::new()
+                    .service(web::resource("/").route(web::to(|| HttpResponse::Ok()))),
+                |_| AppConfig::default(),
+            ))
+            .openssl(ssl_acceptor())
+            .map_err(|_| ())
+    })
+    .await;
+
+    // disable ssl verification
+    let mut builder = SslConnector::builder(SslMethod::tls()).unwrap();
+    builder.set_verify(SslVerifyMode::NONE);
+    let _ = builder
+        .set_alpn_protos(b"\x02h2\x08http/1.1")
+        .map_err(|e| log::error!("Can not set alpn protocol: {:?}", e));
+
+    let client = awc::Client::build()
+        .connector(awc::Connector::new().ssl(builder.build()).finish())
+        .initial_window_size(100)
+        .initial_connection_window_size(100)
+        .finish();
+
+    let request = client.get(srv.surl("/")).send();
+    let response = request.await.unwrap();
+    assert!(response.status().is_success());
+    assert_eq!(response.version(), Version::HTTP_2);
+}

awc/tests/test_rustls_client.rs

@@ -72,7 +72,8 @@ async fn _test_connection_reuse_h2() {
                 .openssl(ssl_acceptor())
                 .map_err(|_| ()),
         )
-    });
+    })
+    .await;
 
     // disable ssl verification
     let mut config = ClientConfig::new();

awc/tests/test_ssl_client.rs

@@ -53,7 +53,8 @@ async fn test_connection_reuse_h2() {
                 .openssl(ssl_acceptor())
                 .map_err(|_| ()),
         )
-    });
+    })
+    .await;
 
     // disable ssl verification
     let mut builder = SslConnector::builder(SslMethod::tls()).unwrap();

awc/tests/test_ws.rs

@@ -38,7 +38,8 @@ async fn test_simple() {
             })
             .finish(|_| ok::<_, Error>(Response::NotFound()))
             .tcp()
-    });
+    })
+    .await;
 
     // client service
     let mut framed = srv.ws().await.unwrap();

src/lib.rs

@@ -1,4 +1,4 @@
-#![deny(rust_2018_idioms, warnings)]
+#![warn(rust_2018_idioms, warnings)]
 #![allow(
     clippy::needless_doctest_main,
     clippy::type_complexity,

test-server/CHANGES.md

@@ -3,6 +3,8 @@
 ## [Unreleased] - 2020-xx-xx
 
 * Update the `time` dependency to 0.2.7
+* Update `actix-connect` dependency to 2.0.0-alpha.2
+* Make `test_server` `async` fn.
 
 ## [1.0.0] - 2019-12-13
 

test-server/Cargo.toml

@@ -32,12 +32,12 @@ openssl = ["open-ssl", "awc/openssl"]
 [dependencies]
 actix-service = "1.0.1"
 actix-codec = "0.2.0"
-actix-connect = "1.0.0"
+actix-connect = "2.0.0-alpha.2"
 actix-utils = "1.0.3"
 actix-rt = "1.0.0"
 actix-server = "1.0.0"
 actix-testing = "1.0.0"
-awc = "1.0.1"
+awc = "2.0.0-alpha.1"
 
 base64 = "0.11"
 bytes = "0.5.3"
@@ -56,4 +56,4 @@ open-ssl = { version="0.10", package="openssl", optional = true }
 
 [dev-dependencies]
 actix-web = "2.0.0"
-actix-http = "2.0.0-alpha.1"
+actix-http = "2.0.0-alpha.2"

test-server/src/lib.rs

@@ -43,7 +43,7 @@ pub use actix_testing::*;
 ///     assert!(response.status().is_success());
 /// }
 /// ```
-pub fn test_server<F: ServiceFactory<TcpStream>>(factory: F) -> TestServer {
+pub async fn test_server<F: ServiceFactory<TcpStream>>(factory: F) -> TestServer {
     let (tx, rx) = mpsc::channel();
 
     // run server in separate thread
@@ -92,7 +92,7 @@ pub fn test_server<F: ServiceFactory<TcpStream>>(factory: F) -> TestServer {
 
         Client::build().connector(connector).finish()
     };
-    actix_connect::start_default_resolver();
+    actix_connect::start_default_resolver().await.unwrap();
 
     TestServer {
         addr,