"]
description = "Actix Web is a powerful, pragmatic, and extremely fast web framework for Rust"
readme = "README.md"
@@ -66,7 +66,7 @@ required-features = ["compress"]
[[test]]
name = "test_server"
-required-features = ["compress"]
+required-features = ["compress", "cookies"]
[[example]]
name = "on_connect"
@@ -80,15 +80,15 @@ required-features = ["rustls"]
actix-codec = "0.4.0-beta.1"
actix-macros = "0.2.0"
actix-router = "0.2.7"
-actix-rt = "2"
+actix-rt = "2.1"
actix-server = "2.0.0-beta.3"
actix-service = "2.0.0-beta.4"
actix-utils = "3.0.0-beta.2"
-actix-tls = { version = "3.0.0-beta.3", default-features = false, optional = true }
+actix-tls = { version = "3.0.0-beta.4", default-features = false, optional = true }
-actix-web-codegen = "0.5.0-beta.1"
-actix-http = "3.0.0-beta.3"
-awc = { version = "3.0.0-beta.2", default-features = false }
+actix-web-codegen = "0.5.0-beta.2"
+actix-http = "3.0.0-beta.4"
+awc = { version = "3.0.0-beta.3", default-features = false }
ahash = "0.7"
bytes = "1"
@@ -105,18 +105,12 @@ serde = { version = "1.0", features = ["derive"] }
serde_json = "1.0"
serde_urlencoded = "0.7"
smallvec = "1.6"
-socket2 = "0.3.16"
+socket2 = "0.4.0"
time = { version = "0.2.23", default-features = false, features = ["std"] }
tls-openssl = { package = "openssl", version = "0.10.9", optional = true }
tls-rustls = { package = "rustls", version = "0.19.0", optional = true }
url = "2.1"
-[target.'cfg(windows)'.dependencies.tls-openssl]
-version = "0.10.9"
-package = "openssl"
-features = ["vendored"]
-optional = true
-
[dev-dependencies]
brotli2 = "0.3.2"
criterion = "0.3"
diff --git a/README.md b/README.md
index bf68e7961..64fd7d08d 100644
--- a/README.md
+++ b/README.md
@@ -6,10 +6,10 @@
[](https://crates.io/crates/actix-web)
-[](https://docs.rs/actix-web/4.0.0-beta.2)
+[](https://docs.rs/actix-web/4.0.0-beta.4)
[](https://blog.rust-lang.org/2020/03/12/Rust-1.46.html)
-[](https://deps.rs/crate/actix-web/4.0.0-beta.2)
+[](https://deps.rs/crate/actix-web/4.0.0-beta.4)
[](https://github.com/actix/actix-web/actions)
[](https://codecov.io/gh/actix/actix-web)
@@ -71,18 +71,18 @@ async fn main() -> std::io::Result<()> {
### More examples
-* [Basic Setup](https://github.com/actix/examples/tree/master/basics/)
-* [Application State](https://github.com/actix/examples/tree/master/state/)
-* [JSON Handling](https://github.com/actix/examples/tree/master/json/)
-* [Multipart Streams](https://github.com/actix/examples/tree/master/multipart/)
-* [Diesel Integration](https://github.com/actix/examples/tree/master/diesel/)
-* [r2d2 Integration](https://github.com/actix/examples/tree/master/r2d2/)
-* [Simple WebSocket](https://github.com/actix/examples/tree/master/websocket/)
-* [Tera Templates](https://github.com/actix/examples/tree/master/template_tera/)
-* [Askama Templates](https://github.com/actix/examples/tree/master/template_askama/)
-* [HTTPS using Rustls](https://github.com/actix/examples/tree/master/rustls/)
-* [HTTPS using OpenSSL](https://github.com/actix/examples/tree/master/openssl/)
-* [WebSocket Chat](https://github.com/actix/examples/tree/master/websocket-chat/)
+* [Basic Setup](https://github.com/actix/examples/tree/master/basics/basics/)
+* [Application State](https://github.com/actix/examples/tree/master/basics/state/)
+* [JSON Handling](https://github.com/actix/examples/tree/master/json/json/)
+* [Multipart Streams](https://github.com/actix/examples/tree/master/forms/multipart/)
+* [Diesel Integration](https://github.com/actix/examples/tree/master/database_interactions/diesel/)
+* [r2d2 Integration](https://github.com/actix/examples/tree/master/database_interactions/r2d2/)
+* [Simple WebSocket](https://github.com/actix/examples/tree/master/websockets/websocket/)
+* [Tera Templates](https://github.com/actix/examples/tree/master/template_engines/tera/)
+* [Askama Templates](https://github.com/actix/examples/tree/master/template_engines/askama/)
+* [HTTPS using Rustls](https://github.com/actix/examples/tree/master/security/rustls/)
+* [HTTPS using OpenSSL](https://github.com/actix/examples/tree/master/security/openssl/)
+* [WebSocket Chat](https://github.com/actix/examples/tree/master/websockets/chat/)
You may consider checking out
[this directory](https://github.com/actix/examples/tree/master/) for more examples.
diff --git a/actix-files/CHANGES.md b/actix-files/CHANGES.md
index 6e2a241ac..c035d5afe 100644
--- a/actix-files/CHANGES.md
+++ b/actix-files/CHANGES.md
@@ -3,6 +3,10 @@
## Unreleased - 2021-xx-xx
+## 0.6.0-beta.3 - 2021-03-09
+* No notable changes.
+
+
## 0.6.0-beta.2 - 2021-02-10
* Fix If-Modified-Since and If-Unmodified-Since to not compare using sub-second timestamps. [#1887]
* Replace `v_htmlescape` with `askama_escape`. [#1953]
diff --git a/actix-files/Cargo.toml b/actix-files/Cargo.toml
index 08b7b36fc..49cd6966c 100644
--- a/actix-files/Cargo.toml
+++ b/actix-files/Cargo.toml
@@ -1,6 +1,6 @@
[package]
name = "actix-files"
-version = "0.6.0-beta.2"
+version = "0.6.0-beta.3"
authors = ["Nikolay Kim "]
description = "Static file serving for Actix Web"
readme = "README.md"
@@ -17,7 +17,7 @@ name = "actix_files"
path = "src/lib.rs"
[dependencies]
-actix-web = { version = "4.0.0-beta.3", default-features = false }
+actix-web = { version = "4.0.0-beta.4", default-features = false }
actix-service = "2.0.0-beta.4"
askama_escape = "0.10"
@@ -33,5 +33,5 @@ mime_guess = "2.0.1"
percent-encoding = "2.1"
[dev-dependencies]
-actix-rt = "2"
-actix-web = "4.0.0-beta.3"
+actix-rt = "2.1"
+actix-web = "4.0.0-beta.4"
diff --git a/actix-files/README.md b/actix-files/README.md
index 463f20224..c7b7424ec 100644
--- a/actix-files/README.md
+++ b/actix-files/README.md
@@ -3,17 +3,17 @@
> Static file serving for Actix Web
[](https://crates.io/crates/actix-files)
-[](https://docs.rs/actix-files/0.5.0)
+[](https://docs.rs/actix-files/0.6.0-beta.3)
[](https://blog.rust-lang.org/2020/03/12/Rust-1.46.html)
-[](https://deps.rs/crate/actix-files/0.5.0)
+[](https://deps.rs/crate/actix-files/0.6.0-beta.3)
[](https://crates.io/crates/actix-files)
[](https://gitter.im/actix/actix?utm_source=badge&utm_medium=badge&utm_campaign=pr-badge&utm_content=badge)
## Documentation & Resources
- [API Documentation](https://docs.rs/actix-files/)
-- [Example Project](https://github.com/actix/examples/tree/master/static_index)
+- [Example Project](https://github.com/actix/examples/tree/master/basics/static_index)
- [Chat on Gitter](https://gitter.im/actix/actix-web)
- Minimum supported Rust version: 1.46 or later
diff --git a/actix-files/src/files.rs b/actix-files/src/files.rs
index 6f8b28bbf..292e3fdf3 100644
--- a/actix-files/src/files.rs
+++ b/actix-files/src/files.rs
@@ -19,7 +19,7 @@ use crate::{
///
/// `Files` service must be registered with `App::service()` method.
///
-/// ```rust
+/// ```
/// use actix_web::App;
/// use actix_files::Files;
///
diff --git a/actix-files/src/lib.rs b/actix-files/src/lib.rs
index 04dd9f07f..018079b21 100644
--- a/actix-files/src/lib.rs
+++ b/actix-files/src/lib.rs
@@ -3,7 +3,7 @@
//! Provides a non-blocking service for serving static files from disk.
//!
//! # Example
-//! ```rust
+//! ```
//! use actix_web::App;
//! use actix_files::Files;
//!
@@ -662,8 +662,12 @@ mod tests {
#[actix_rt::test]
async fn test_static_files_bad_directory() {
- let _st: Files = Files::new("/", "missing");
- let _st: Files = Files::new("/", "Cargo.toml");
+ let service = Files::new("/", "./missing").new_service(()).await.unwrap();
+
+ let req = TestRequest::with_uri("/").to_srv_request();
+ let resp = test::call_service(&service, req).await;
+
+ assert_eq!(resp.status(), StatusCode::NOT_FOUND);
}
#[actix_rt::test]
@@ -676,75 +680,34 @@ mod tests {
.await
.unwrap();
let req = TestRequest::with_uri("/missing").to_srv_request();
-
let resp = test::call_service(&st, req).await;
+
assert_eq!(resp.status(), StatusCode::OK);
let bytes = test::read_body(resp).await;
assert_eq!(bytes, web::Bytes::from_static(b"default content"));
}
- // #[actix_rt::test]
- // async fn test_serve_index() {
- // let st = Files::new(".").index_file("test.binary");
- // let req = TestRequest::default().uri("/tests").finish();
+ #[actix_rt::test]
+ async fn test_serve_index_nested() {
+ let service = Files::new(".", ".")
+ .index_file("lib.rs")
+ .new_service(())
+ .await
+ .unwrap();
- // let resp = st.handle(&req).respond_to(&req).unwrap();
- // let resp = resp.as_msg();
- // assert_eq!(resp.status(), StatusCode::OK);
- // assert_eq!(
- // resp.headers()
- // .get(header::CONTENT_TYPE)
- // .expect("content type"),
- // "application/octet-stream"
- // );
- // assert_eq!(
- // resp.headers()
- // .get(header::CONTENT_DISPOSITION)
- // .expect("content disposition"),
- // "attachment; filename=\"test.binary\""
- // );
+ let req = TestRequest::default().uri("/src").to_srv_request();
+ let resp = test::call_service(&service, req).await;
- // let req = TestRequest::default().uri("/tests/").finish();
- // let resp = st.handle(&req).respond_to(&req).unwrap();
- // let resp = resp.as_msg();
- // assert_eq!(resp.status(), StatusCode::OK);
- // assert_eq!(
- // resp.headers().get(header::CONTENT_TYPE).unwrap(),
- // "application/octet-stream"
- // );
- // assert_eq!(
- // resp.headers().get(header::CONTENT_DISPOSITION).unwrap(),
- // "attachment; filename=\"test.binary\""
- // );
-
- // // nonexistent index file
- // let req = TestRequest::default().uri("/tests/unknown").finish();
- // let resp = st.handle(&req).respond_to(&req).unwrap();
- // let resp = resp.as_msg();
- // assert_eq!(resp.status(), StatusCode::NOT_FOUND);
-
- // let req = TestRequest::default().uri("/tests/unknown/").finish();
- // let resp = st.handle(&req).respond_to(&req).unwrap();
- // let resp = resp.as_msg();
- // assert_eq!(resp.status(), StatusCode::NOT_FOUND);
- // }
-
- // #[actix_rt::test]
- // async fn test_serve_index_nested() {
- // let st = Files::new(".").index_file("mod.rs");
- // let req = TestRequest::default().uri("/src/client").finish();
- // let resp = st.handle(&req).respond_to(&req).unwrap();
- // let resp = resp.as_msg();
- // assert_eq!(resp.status(), StatusCode::OK);
- // assert_eq!(
- // resp.headers().get(header::CONTENT_TYPE).unwrap(),
- // "text/x-rust"
- // );
- // assert_eq!(
- // resp.headers().get(header::CONTENT_DISPOSITION).unwrap(),
- // "inline; filename=\"mod.rs\""
- // );
- // }
+ assert_eq!(resp.status(), StatusCode::OK);
+ assert_eq!(
+ resp.headers().get(header::CONTENT_TYPE).unwrap(),
+ "text/x-rust"
+ );
+ assert_eq!(
+ resp.headers().get(header::CONTENT_DISPOSITION).unwrap(),
+ "inline; filename=\"lib.rs\""
+ );
+ }
#[actix_rt::test]
async fn integration_serve_index() {
diff --git a/actix-files/src/named.rs b/actix-files/src/named.rs
index a688b2e6c..2846646a2 100644
--- a/actix-files/src/named.rs
+++ b/actix-files/src/named.rs
@@ -60,7 +60,7 @@ impl NamedFile {
///
/// # Examples
///
- /// ```rust
+ /// ```
/// use actix_files::NamedFile;
/// use std::io::{self, Write};
/// use std::env;
@@ -137,7 +137,7 @@ impl NamedFile {
///
/// # Examples
///
- /// ```rust
+ /// ```
/// use actix_files::NamedFile;
///
/// let file = NamedFile::open("foo.txt");
@@ -156,7 +156,7 @@ impl NamedFile {
///
/// # Examples
///
- /// ```rust
+ /// ```
/// # use std::io;
/// use actix_files::NamedFile;
///
diff --git a/actix-files/src/service.rs b/actix-files/src/service.rs
index 14eea6ebc..3214963ed 100644
--- a/actix-files/src/service.rs
+++ b/actix-files/src/service.rs
@@ -1,4 +1,4 @@
-use std::{fmt, io, path::PathBuf, rc::Rc, task::Poll};
+use std::{fmt, io, path::PathBuf, rc::Rc};
use actix_service::Service;
use actix_web::{
diff --git a/actix-http-test/CHANGES.md b/actix-http-test/CHANGES.md
index 2f47d700d..d6a2cdd9b 100644
--- a/actix-http-test/CHANGES.md
+++ b/actix-http-test/CHANGES.md
@@ -3,6 +3,10 @@
## Unreleased - 2021-xx-xx
+## 3.0.0-beta.3 - 2021-03-09
+* No notable changes.
+
+
## 3.0.0-beta.2 - 2021-02-10
* No notable changes.
diff --git a/actix-http-test/Cargo.toml b/actix-http-test/Cargo.toml
index 6dcf73637..a7efc5310 100644
--- a/actix-http-test/Cargo.toml
+++ b/actix-http-test/Cargo.toml
@@ -1,6 +1,6 @@
[package]
name = "actix-http-test"
-version = "3.0.0-beta.2"
+version = "3.0.0-beta.3"
authors = ["Nikolay Kim "]
description = "Various helpers for Actix applications to use during testing"
readme = "README.md"
@@ -31,18 +31,18 @@ openssl = ["tls-openssl", "awc/openssl"]
[dependencies]
actix-service = "2.0.0-beta.4"
actix-codec = "0.4.0-beta.1"
-actix-tls = "3.0.0-beta.3"
+actix-tls = "3.0.0-beta.4"
actix-utils = "3.0.0-beta.2"
-actix-rt = "2"
+actix-rt = "2.1"
actix-server = "2.0.0-beta.3"
-awc = "3.0.0-beta.2"
+awc = { version = "3.0.0-beta.3", default-features = false }
base64 = "0.13"
bytes = "1"
futures-core = { version = "0.3.7", default-features = false }
http = "0.2.2"
log = "0.4"
-socket2 = "0.3"
+socket2 = "0.4"
serde = "1.0"
serde_json = "1.0"
slab = "0.4"
@@ -50,12 +50,6 @@ serde_urlencoded = "0.7"
time = { version = "0.2.23", default-features = false, features = ["std"] }
tls-openssl = { version = "0.10.9", package = "openssl", optional = true }
-[target.'cfg(windows)'.dependencies.tls-openssl]
-version = "0.10.9"
-package = "openssl"
-features = ["vendored"]
-optional = true
-
[dev-dependencies]
-actix-web = "4.0.0-beta.3"
-actix-http = "3.0.0-beta.3"
+actix-web = { version = "4.0.0-beta.4", default-features = false, features = ["cookies"] }
+actix-http = "3.0.0-beta.4"
diff --git a/actix-http-test/README.md b/actix-http-test/README.md
index 66f15979d..8cec94808 100644
--- a/actix-http-test/README.md
+++ b/actix-http-test/README.md
@@ -3,9 +3,9 @@
> Various helpers for Actix applications to use during testing.
[](https://crates.io/crates/actix-http-test)
-[](https://docs.rs/actix-http-test/2.1.0)
+[](https://docs.rs/actix-http-test/3.0.0-beta.3)
-[](https://deps.rs/crate/actix-http-test/2.1.0)
+[](https://deps.rs/crate/actix-http-test/3.0.0-beta.3)
[](https://gitter.im/actix/actix-web?utm_source=badge&utm_medium=badge&utm_campaign=pr-badge&utm_content=badge)
## Documentation & Resources
diff --git a/actix-http-test/src/lib.rs b/actix-http-test/src/lib.rs
index 8de07c8d3..3749b78ca 100644
--- a/actix-http-test/src/lib.rs
+++ b/actix-http-test/src/lib.rs
@@ -26,7 +26,7 @@ use socket2::{Domain, Protocol, Socket, Type};
///
/// # Examples
///
-/// ```rust
+/// ```
/// use actix_http::HttpService;
/// use actix_http_test::TestServer;
/// use actix_web::{web, App, HttpResponse, Error};
@@ -118,10 +118,10 @@ pub async fn test_server_with_addr>(
/// Get first available unused address
pub fn unused_addr() -> net::SocketAddr {
let addr: net::SocketAddr = "127.0.0.1:0".parse().unwrap();
- let socket = Socket::new(Domain::ipv4(), Type::stream(), Some(Protocol::tcp())).unwrap();
+ let socket = Socket::new(Domain::IPV4, Type::STREAM, Some(Protocol::TCP)).unwrap();
socket.bind(&addr.into()).unwrap();
socket.set_reuse_address(true).unwrap();
- let tcp = socket.into_tcp_listener();
+ let tcp = net::TcpListener::from(socket);
tcp.local_addr().unwrap()
}
diff --git a/actix-http/CHANGES.md b/actix-http/CHANGES.md
index 6ba111eb3..c4e0aec89 100644
--- a/actix-http/CHANGES.md
+++ b/actix-http/CHANGES.md
@@ -1,15 +1,32 @@
# Changes
## Unreleased - 2021-xx-xx
+### Added
+* `client::Connector::handshake_timeout` method for customize tls connection handshake timeout. [#2081]
+* `client::ConnectorService` as `client::Connector::finish` method's return type [#2081]
+* `client::ConnectionIo` trait alias [#2081]
+
+### Chaged
+* `client::Connector` type now only have one generic type for `actix_service::Service`. [#2063]
+
+[#2063]: https://github.com/actix/actix-web/pull/2063
+[#2081]: https://github.com/actix/actix-web/pull/2081
+
+
+## 3.0.0-beta.4 - 2021-03-08
### Changed
* Feature `cookies` is now optional and disabled by default. [#1981]
+* `ws::hash_key` now returns array. [#2035]
+* `ResponseBuilder::json` now takes `impl Serialize`. [#2052]
### Removed
-* re-export of `futures_channel::oneshot::Canceled` is removed from `error` mod. [#1994]
+* Re-export of `futures_channel::oneshot::Canceled` is removed from `error` mod. [#1994]
* `ResponseError` impl for `futures_channel::oneshot::Canceled` is removed. [#1994]
[#1981]: https://github.com/actix/actix-web/pull/1981
[#1994]: https://github.com/actix/actix-web/pull/1994
+[#2035]: https://github.com/actix/actix-web/pull/2035
+[#2052]: https://github.com/actix/actix-web/pull/2052
## 3.0.0-beta.3 - 2021-02-10
diff --git a/actix-http/Cargo.toml b/actix-http/Cargo.toml
index 78fb55079..e1aebb76b 100644
--- a/actix-http/Cargo.toml
+++ b/actix-http/Cargo.toml
@@ -1,6 +1,6 @@
[package]
name = "actix-http"
-version = "3.0.0-beta.3"
+version = "3.0.0-beta.4"
authors = ["Nikolay Kim "]
description = "HTTP primitives for the Actix ecosystem"
readme = "README.md"
@@ -47,8 +47,8 @@ trust-dns = ["trust-dns-resolver"]
actix-service = "2.0.0-beta.4"
actix-codec = "0.4.0-beta.1"
actix-utils = "3.0.0-beta.2"
-actix-rt = "2"
-actix-tls = "3.0.0-beta.2"
+actix-rt = "2.1"
+actix-tls = "3.0.0-beta.4"
ahash = "0.7"
base64 = "0.13"
@@ -61,12 +61,12 @@ derive_more = "0.99.5"
encoding_rs = "0.8"
futures-core = { version = "0.3.7", default-features = false, features = ["alloc"] }
futures-util = { version = "0.3.7", default-features = false, features = ["alloc", "sink"] }
-h2 = "0.3.0"
+h2 = "0.3.1"
http = "0.2.2"
httparse = "1.3"
itoa = "0.4"
language-tags = "0.2"
-lazy_static = "1.4"
+once_cell = "1.5"
log = "0.4"
mime = "0.3"
percent-encoding = "2.1"
@@ -89,8 +89,8 @@ trust-dns-resolver = { version = "0.20.0", optional = true }
[dev-dependencies]
actix-server = "2.0.0-beta.3"
-actix-http-test = { version = "3.0.0-beta.2", features = ["openssl"] }
-actix-tls = { version = "3.0.0-beta.2", features = ["openssl"] }
+actix-http-test = { version = "3.0.0-beta.3", features = ["openssl"] }
+actix-tls = { version = "3.0.0-beta.4", features = ["openssl"] }
criterion = "0.3"
env_logger = "0.8"
rcgen = "0.8"
@@ -98,10 +98,9 @@ serde_derive = "1.0"
tls-openssl = { version = "0.10", package = "openssl" }
tls-rustls = { version = "0.19", package = "rustls" }
-[target.'cfg(windows)'.dev-dependencies.tls-openssl]
-version = "0.10.9"
-package = "openssl"
-features = ["vendored"]
+[[example]]
+name = "ws"
+required-features = ["rustls"]
[[bench]]
name = "write-camel-case"
diff --git a/actix-http/README.md b/actix-http/README.md
index 881fbc8c5..53fedd40e 100644
--- a/actix-http/README.md
+++ b/actix-http/README.md
@@ -3,11 +3,11 @@
> HTTP primitives for the Actix ecosystem.
[](https://crates.io/crates/actix-http)
-[](https://docs.rs/actix-http/3.0.0-beta.3)
+[](https://docs.rs/actix-http/3.0.0-beta.4)
[](https://blog.rust-lang.org/2020/03/12/Rust-1.46.html)
-[](https://deps.rs/crate/actix-http/3.0.0-beta.3)
+[](https://deps.rs/crate/actix-http/3.0.0-beta.4)
[](https://crates.io/crates/actix-http)
[](https://gitter.im/actix/actix?utm_source=badge&utm_medium=badge&utm_campaign=pr-badge&utm_content=badge)
diff --git a/actix-http/examples/echo.rs b/actix-http/examples/echo.rs
index 90d768cbe..176ac5c2b 100644
--- a/actix-http/examples/echo.rs
+++ b/actix-http/examples/echo.rs
@@ -3,7 +3,7 @@ use std::{env, io};
use actix_http::{Error, HttpService, Request, Response};
use actix_server::Server;
use bytes::BytesMut;
-use futures_util::StreamExt;
+use futures_util::StreamExt as _;
use http::header::HeaderValue;
use log::info;
diff --git a/actix-http/examples/echo2.rs b/actix-http/examples/echo2.rs
index bc932ce8f..408a40114 100644
--- a/actix-http/examples/echo2.rs
+++ b/actix-http/examples/echo2.rs
@@ -4,7 +4,7 @@ use actix_http::http::HeaderValue;
use actix_http::{Error, HttpService, Request, Response};
use actix_server::Server;
use bytes::BytesMut;
-use futures_util::StreamExt;
+use futures_util::StreamExt as _;
use log::info;
async fn handle_request(mut req: Request) -> Result {
diff --git a/actix-http/examples/ws.rs b/actix-http/examples/ws.rs
new file mode 100644
index 000000000..4e03aa8ab
--- /dev/null
+++ b/actix-http/examples/ws.rs
@@ -0,0 +1,107 @@
+//! Sets up a WebSocket server over TCP and TLS.
+//! Sends a heartbeat message every 4 seconds but does not respond to any incoming frames.
+
+extern crate tls_rustls as rustls;
+
+use std::{
+ env, io,
+ pin::Pin,
+ task::{Context, Poll},
+ time::Duration,
+};
+
+use actix_codec::Encoder;
+use actix_http::{error::Error, ws, HttpService, Request, Response};
+use actix_rt::time::{interval, Interval};
+use actix_server::Server;
+use bytes::{Bytes, BytesMut};
+use bytestring::ByteString;
+use futures_core::{ready, Stream};
+
+#[actix_rt::main]
+async fn main() -> io::Result<()> {
+ env::set_var("RUST_LOG", "actix=info,h2_ws=info");
+ env_logger::init();
+
+ Server::build()
+ .bind("tcp", ("127.0.0.1", 8080), || {
+ HttpService::build().h1(handler).tcp()
+ })?
+ .bind("tls", ("127.0.0.1", 8443), || {
+ HttpService::build().finish(handler).rustls(tls_config())
+ })?
+ .run()
+ .await
+}
+
+async fn handler(req: Request) -> Result {
+ log::info!("handshaking");
+ let mut res = ws::handshake(req.head())?;
+
+ // handshake will always fail under HTTP/2
+
+ log::info!("responding");
+ Ok(res.streaming(Heartbeat::new(ws::Codec::new())))
+}
+
+struct Heartbeat {
+ codec: ws::Codec,
+ interval: Interval,
+}
+
+impl Heartbeat {
+ fn new(codec: ws::Codec) -> Self {
+ Self {
+ codec,
+ interval: interval(Duration::from_secs(4)),
+ }
+ }
+}
+
+impl Stream for Heartbeat {
+ type Item = Result;
+
+ fn poll_next(
+ mut self: Pin<&mut Self>,
+ cx: &mut Context<'_>,
+ ) -> Poll> {
+ log::trace!("poll");
+
+ ready!(self.as_mut().interval.poll_tick(cx));
+
+ let mut buffer = BytesMut::new();
+
+ self.as_mut()
+ .codec
+ .encode(
+ ws::Message::Text(ByteString::from_static("hello world")),
+ &mut buffer,
+ )
+ .unwrap();
+
+ Poll::Ready(Some(Ok(buffer.freeze())))
+ }
+}
+
+fn tls_config() -> rustls::ServerConfig {
+ use std::io::BufReader;
+
+ use rustls::{
+ internal::pemfile::{certs, pkcs8_private_keys},
+ NoClientAuth, ServerConfig,
+ };
+
+ let cert = rcgen::generate_simple_self_signed(vec!["localhost".to_owned()]).unwrap();
+ let cert_file = cert.serialize_pem().unwrap();
+ let key_file = cert.serialize_private_key_pem();
+
+ let mut config = ServerConfig::new(NoClientAuth::new());
+ let cert_file = &mut BufReader::new(cert_file.as_bytes());
+ let key_file = &mut BufReader::new(key_file.as_bytes());
+
+ let cert_chain = certs(cert_file).unwrap();
+ let mut keys = pkcs8_private_keys(key_file).unwrap();
+ config.set_single_cert(cert_chain, keys.remove(0)).unwrap();
+
+ config
+}
diff --git a/actix-http/src/client/config.rs b/actix-http/src/client/config.rs
index fad902d04..1c0405cbc 100644
--- a/actix-http/src/client/config.rs
+++ b/actix-http/src/client/config.rs
@@ -1,3 +1,4 @@
+use std::net::IpAddr;
use std::time::Duration;
const DEFAULT_H2_CONN_WINDOW: u32 = 1024 * 1024 * 2; // 2MB
@@ -7,24 +8,28 @@ const DEFAULT_H2_STREAM_WINDOW: u32 = 1024 * 1024; // 1MB
#[derive(Clone)]
pub(crate) struct ConnectorConfig {
pub(crate) timeout: Duration,
+ pub(crate) handshake_timeout: Duration,
pub(crate) conn_lifetime: Duration,
pub(crate) conn_keep_alive: Duration,
pub(crate) disconnect_timeout: Option,
pub(crate) limit: usize,
pub(crate) conn_window_size: u32,
pub(crate) stream_window_size: u32,
+ pub(crate) local_address: Option,
}
impl Default for ConnectorConfig {
fn default() -> Self {
Self {
timeout: Duration::from_secs(5),
+ handshake_timeout: Duration::from_secs(5),
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,
+ local_address: None,
}
}
}
diff --git a/actix-http/src/client/connection.rs b/actix-http/src/client/connection.rs
index 778083a1c..78101397d 100644
--- a/actix-http/src/client/connection.rs
+++ b/actix-http/src/client/connection.rs
@@ -1,14 +1,16 @@
-use std::ops::{Deref, DerefMut};
-use std::pin::Pin;
-use std::task::{Context, Poll};
-use std::{fmt, io, time};
+use std::{
+ io,
+ ops::{Deref, DerefMut},
+ pin::Pin,
+ task::{Context, Poll},
+ time,
+};
use actix_codec::{AsyncRead, AsyncWrite, Framed, ReadBuf};
use actix_rt::task::JoinHandle;
use bytes::Bytes;
use futures_core::future::LocalBoxFuture;
use h2::client::SendRequest;
-use pin_project::pin_project;
use crate::body::MessageBody;
use crate::h1::ClientCodec;
@@ -19,28 +21,148 @@ use super::error::SendRequestError;
use super::pool::Acquired;
use super::{h1proto, h2proto};
-pub(crate) enum ConnectionType {
- H1(Io),
- H2(H2Connection),
+/// Trait alias for types impl [tokio::io::AsyncRead] and [tokio::io::AsyncWrite].
+pub trait ConnectionIo: AsyncRead + AsyncWrite + Unpin + 'static {}
+
+impl ConnectionIo for T {}
+
+/// HTTP client connection
+pub struct H1Connection {
+ io: Option,
+ created: time::Instant,
+ acquired: Acquired,
}
-/// `H2Connection` has two parts: `SendRequest` and `Connection`.
+impl H1Connection {
+ /// close or release the connection to pool based on flag input
+ pub(super) fn on_release(&mut self, keep_alive: bool) {
+ if keep_alive {
+ self.release();
+ } else {
+ self.close();
+ }
+ }
+
+ /// Close connection
+ fn close(&mut self) {
+ let io = self.io.take().unwrap();
+ self.acquired.close(ConnectionInnerType::H1(io));
+ }
+
+ /// Release this connection to the connection pool
+ fn release(&mut self) {
+ let io = self.io.take().unwrap();
+ self.acquired
+ .release(ConnectionInnerType::H1(io), self.created);
+ }
+
+ fn io_pin_mut(self: Pin<&mut Self>) -> Pin<&mut Io> {
+ Pin::new(self.get_mut().io.as_mut().unwrap())
+ }
+}
+
+impl AsyncRead for H1Connection {
+ fn poll_read(
+ self: Pin<&mut Self>,
+ cx: &mut Context<'_>,
+ buf: &mut ReadBuf<'_>,
+ ) -> Poll> {
+ self.io_pin_mut().poll_read(cx, buf)
+ }
+}
+
+impl AsyncWrite for H1Connection {
+ fn poll_write(
+ self: Pin<&mut Self>,
+ cx: &mut Context<'_>,
+ buf: &[u8],
+ ) -> Poll> {
+ self.io_pin_mut().poll_write(cx, buf)
+ }
+
+ fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> {
+ self.io_pin_mut().poll_flush(cx)
+ }
+
+ fn poll_shutdown(
+ self: Pin<&mut Self>,
+ cx: &mut Context<'_>,
+ ) -> Poll> {
+ self.io_pin_mut().poll_shutdown(cx)
+ }
+
+ fn poll_write_vectored(
+ self: Pin<&mut Self>,
+ cx: &mut Context<'_>,
+ bufs: &[io::IoSlice<'_>],
+ ) -> Poll> {
+ self.io_pin_mut().poll_write_vectored(cx, bufs)
+ }
+
+ fn is_write_vectored(&self) -> bool {
+ self.io.as_ref().unwrap().is_write_vectored()
+ }
+}
+
+/// HTTP2 client connection
+pub struct H2Connection {
+ io: Option,
+ created: time::Instant,
+ acquired: Acquired,
+}
+
+impl Deref for H2Connection {
+ type Target = SendRequest;
+
+ fn deref(&self) -> &Self::Target {
+ &self.io.as_ref().unwrap().sender
+ }
+}
+
+impl DerefMut for H2Connection {
+ fn deref_mut(&mut self) -> &mut Self::Target {
+ &mut self.io.as_mut().unwrap().sender
+ }
+}
+
+impl H2Connection {
+ /// close or release the connection to pool based on flag input
+ pub(super) fn on_release(&mut self, close: bool) {
+ if close {
+ self.close();
+ } else {
+ self.release();
+ }
+ }
+
+ /// Close connection
+ fn close(&mut self) {
+ let io = self.io.take().unwrap();
+ self.acquired.close(ConnectionInnerType::H2(io));
+ }
+
+ /// Release this connection to the connection pool
+ fn release(&mut self) {
+ let io = self.io.take().unwrap();
+ self.acquired
+ .release(ConnectionInnerType::H2(io), self.created);
+ }
+}
+
+/// `H2ConnectionInner` has two parts: `SendRequest` and `Connection`.
///
-/// `Connection` is spawned as an async task on runtime and `H2Connection` holds a handle for
-/// this task. Therefore, it can wake up and quit the task when SendRequest is dropped.
-pub(crate) struct H2Connection {
+/// `Connection` is spawned as an async task on runtime and `H2ConnectionInner` holds a handle
+/// for this task. Therefore, it can wake up and quit the task when SendRequest is dropped.
+pub(super) struct H2ConnectionInner {
handle: JoinHandle<()>,
sender: SendRequest,
}
-impl H2Connection {
- pub(crate) fn new(
+impl H2ConnectionInner {
+ pub(super) fn new(
sender: SendRequest,
connection: h2::client::Connection,
- ) -> Self
- where
- Io: AsyncRead + AsyncWrite + Unpin + 'static,
- {
+ ) -> Self {
let handle = actix_rt::spawn(async move {
let _ = connection.await;
});
@@ -49,248 +171,188 @@ impl H2Connection {
}
}
-// cancel spawned connection task on drop.
-impl Drop for H2Connection {
+/// Cancel spawned connection task on drop.
+impl Drop for H2ConnectionInner {
fn drop(&mut self) {
self.handle.abort();
}
}
-// only expose sender type to public.
-impl Deref for H2Connection {
- type Target = SendRequest;
-
- fn deref(&self) -> &Self::Target {
- &self.sender
- }
-}
-
-impl DerefMut for H2Connection {
- fn deref_mut(&mut self) -> &mut Self::Target {
- &mut self.sender
- }
-}
-
-pub trait Connection {
- type Io: AsyncRead + AsyncWrite + Unpin;
-
- /// Send request and body
- fn send_request>(
- self,
- head: H,
- body: B,
- ) -> LocalBoxFuture<'static, Result<(ResponseHead, Payload), SendRequestError>>;
-
- /// Send request, returns Response and Framed
- fn open_tunnel + 'static>(
- self,
- head: H,
- ) -> LocalBoxFuture<
- 'static,
- Result<(ResponseHead, Framed), SendRequestError>,
- >;
-}
-
-pub(crate) trait ConnectionLifetime: AsyncRead + AsyncWrite + 'static {
- /// Close connection
- fn close(self: Pin<&mut Self>);
-
- /// Release connection to the connection pool
- fn release(self: Pin<&mut Self>);
-}
-
-#[doc(hidden)]
-/// HTTP client connection
-pub struct IoConnection
-where
- T: AsyncWrite + Unpin + 'static,
-{
- io: Option>,
- created: time::Instant,
- pool: Option>,
-}
-
-impl fmt::Debug for IoConnection
-where
- T: AsyncWrite + Unpin + fmt::Debug + 'static,
-{
- fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
- match self.io {
- Some(ConnectionType::H1(ref io)) => write!(f, "H1Connection({:?})", io),
- Some(ConnectionType::H2(_)) => write!(f, "H2Connection"),
- None => write!(f, "Connection(Empty)"),
- }
- }
-}
-
-impl IoConnection {
- pub(crate) fn new(
- io: ConnectionType,
- created: time::Instant,
- pool: Option>,
- ) -> Self {
- IoConnection {
- pool,
- created,
- io: Some(io),
- }
- }
-
- pub(crate) fn into_inner(self) -> (ConnectionType, time::Instant) {
- (self.io.unwrap(), self.created)
- }
-
- #[cfg(test)]
- pub(crate) fn into_parts(self) -> (ConnectionType, time::Instant, Acquired) {
- (self.io.unwrap(), self.created, self.pool.unwrap())
- }
-}
-
-impl Connection for IoConnection
-where
- T: AsyncRead + AsyncWrite + Unpin + 'static,
-{
- type Io = T;
-
- fn send_request>(
- mut self,
- head: H,
- body: B,
- ) -> LocalBoxFuture<'static, Result<(ResponseHead, Payload), SendRequestError>> {
- match self.io.take().unwrap() {
- ConnectionType::H1(io) => Box::pin(h1proto::send_request(
- io,
- head.into(),
- body,
- self.created,
- self.pool,
- )),
- ConnectionType::H2(io) => Box::pin(h2proto::send_request(
- io,
- head.into(),
- body,
- self.created,
- self.pool,
- )),
- }
- }
-
- /// Send request, returns Response and Framed
- fn open_tunnel>(
- mut self,
- head: H,
- ) -> LocalBoxFuture<
- 'static,
- Result<(ResponseHead, Framed), SendRequestError>,
- > {
- match self.io.take().unwrap() {
- ConnectionType::H1(io) => Box::pin(h1proto::open_tunnel(io, head.into())),
- ConnectionType::H2(io) => {
- if let Some(mut pool) = self.pool.take() {
- pool.release(IoConnection::new(
- ConnectionType::H2(io),
- self.created,
- None,
- ));
- }
- Box::pin(async { Err(SendRequestError::TunnelNotSupported) })
- }
- }
- }
-}
-
#[allow(dead_code)]
-pub(crate) enum EitherIoConnection
+/// Unified connection type cover Http1 Plain/Tls and Http2 protocols
+pub enum Connection>
where
- A: AsyncRead + AsyncWrite + Unpin + 'static,
- B: AsyncRead + AsyncWrite + Unpin + 'static,
+ A: ConnectionIo,
+ B: ConnectionIo,
{
- A(IoConnection),
- B(IoConnection),
+ Tcp(ConnectionType),
+ Tls(ConnectionType),
}
-impl Connection for EitherIoConnection
-where
- A: AsyncRead + AsyncWrite + Unpin + 'static,
- B: AsyncRead + AsyncWrite + Unpin + 'static,
-{
- type Io = EitherIo;
+/// Unified connection type cover Http1/2 protocols
+pub enum ConnectionType {
+ H1(H1Connection),
+ H2(H2Connection),
+}
- fn send_request>(
+/// Helper type for storing connection types in pool.
+pub(super) enum ConnectionInnerType {
+ H1(Io),
+ H2(H2ConnectionInner),
+}
+
+impl ConnectionType {
+ pub(super) fn from_pool(
+ inner: ConnectionInnerType,
+ created: time::Instant,
+ acquired: Acquired,
+ ) -> Self {
+ match inner {
+ ConnectionInnerType::H1(io) => Self::from_h1(io, created, acquired),
+ ConnectionInnerType::H2(io) => Self::from_h2(io, created, acquired),
+ }
+ }
+
+ pub(super) fn from_h1(
+ io: Io,
+ created: time::Instant,
+ acquired: Acquired,
+ ) -> Self {
+ Self::H1(H1Connection {
+ io: Some(io),
+ created,
+ acquired,
+ })
+ }
+
+ pub(super) fn from_h2(
+ io: H2ConnectionInner,
+ created: time::Instant,
+ acquired: Acquired,
+ ) -> Self {
+ Self::H2(H2Connection {
+ io: Some(io),
+ created,
+ acquired,
+ })
+ }
+}
+
+impl Connection
+where
+ A: ConnectionIo,
+ B: ConnectionIo,
+{
+ /// Send a request through connection.
+ pub fn send_request(
self,
head: H,
body: RB,
- ) -> LocalBoxFuture<'static, Result<(ResponseHead, Payload), SendRequestError>> {
- match self {
- EitherIoConnection::A(con) => con.send_request(head, body),
- EitherIoConnection::B(con) => con.send_request(head, body),
- }
+ ) -> LocalBoxFuture<'static, Result<(ResponseHead, Payload), SendRequestError>>
+ where
+ RB: MessageBody + 'static,
+ H: Into + 'static,
+ {
+ Box::pin(async move {
+ match self {
+ Connection::Tcp(ConnectionType::H1(conn)) => {
+ h1proto::send_request(conn, head.into(), body).await
+ }
+ Connection::Tls(ConnectionType::H1(conn)) => {
+ h1proto::send_request(conn, head.into(), body).await
+ }
+ Connection::Tls(ConnectionType::H2(conn)) => {
+ h2proto::send_request(conn, head.into(), body).await
+ }
+ _ => unreachable!(
+ "Plain Tcp connection can be used only in Http1 protocol"
+ ),
+ }
+ })
}
- /// Send request, returns Response and Framed
- fn open_tunnel + 'static>(
+ /// Send request, returns Response and Framed tunnel.
+ pub fn open_tunnel + 'static>(
self,
head: H,
) -> LocalBoxFuture<
'static,
- Result<(ResponseHead, Framed), SendRequestError>,
+ Result<(ResponseHead, Framed, ClientCodec>), SendRequestError>,
> {
- match self {
- EitherIoConnection::A(con) => Box::pin(async {
- let (head, framed) = con.open_tunnel(head).await?;
- Ok((head, framed.into_map_io(EitherIo::A)))
- }),
- EitherIoConnection::B(con) => Box::pin(async {
- let (head, framed) = con.open_tunnel(head).await?;
- Ok((head, framed.into_map_io(EitherIo::B)))
- }),
- }
+ Box::pin(async move {
+ match self {
+ Connection::Tcp(ConnectionType::H1(ref _conn)) => {
+ let (head, framed) = h1proto::open_tunnel(self, head.into()).await?;
+ Ok((head, framed))
+ }
+ Connection::Tls(ConnectionType::H1(ref _conn)) => {
+ let (head, framed) = h1proto::open_tunnel(self, head.into()).await?;
+ Ok((head, framed))
+ }
+ Connection::Tls(ConnectionType::H2(mut conn)) => {
+ conn.release();
+ Err(SendRequestError::TunnelNotSupported)
+ }
+ Connection::Tcp(ConnectionType::H2(_)) => {
+ unreachable!(
+ "Plain Tcp connection can be used only in Http1 protocol"
+ )
+ }
+ }
+ })
}
}
-#[pin_project(project = EitherIoProj)]
-pub enum EitherIo {
- A(#[pin] A),
- B(#[pin] B),
-}
-
-impl AsyncRead for EitherIo
+impl AsyncRead for Connection
where
- A: AsyncRead,
- B: AsyncRead,
+ A: ConnectionIo,
+ B: ConnectionIo,
{
fn poll_read(
self: Pin<&mut Self>,
cx: &mut Context<'_>,
buf: &mut ReadBuf<'_>,
) -> Poll> {
- match self.project() {
- EitherIoProj::A(val) => val.poll_read(cx, buf),
- EitherIoProj::B(val) => val.poll_read(cx, buf),
+ match self.get_mut() {
+ Connection::Tcp(ConnectionType::H1(conn)) => {
+ Pin::new(conn).poll_read(cx, buf)
+ }
+ Connection::Tls(ConnectionType::H1(conn)) => {
+ Pin::new(conn).poll_read(cx, buf)
+ }
+ _ => unreachable!("H2Connection can not impl AsyncRead trait"),
}
}
}
-impl AsyncWrite for EitherIo
+const H2_UNREACHABLE_WRITE: &str = "H2Connection can not impl AsyncWrite trait";
+
+impl AsyncWrite for Connection
where
- A: AsyncWrite,
- B: AsyncWrite,
+ A: ConnectionIo,
+ B: ConnectionIo,
{
fn poll_write(
self: Pin<&mut Self>,
cx: &mut Context<'_>,
buf: &[u8],
) -> Poll> {
- match self.project() {
- EitherIoProj::A(val) => val.poll_write(cx, buf),
- EitherIoProj::B(val) => val.poll_write(cx, buf),
+ match self.get_mut() {
+ Connection::Tcp(ConnectionType::H1(conn)) => {
+ Pin::new(conn).poll_write(cx, buf)
+ }
+ Connection::Tls(ConnectionType::H1(conn)) => {
+ Pin::new(conn).poll_write(cx, buf)
+ }
+ _ => unreachable!(H2_UNREACHABLE_WRITE),
}
}
fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> {
- match self.project() {
- EitherIoProj::A(val) => val.poll_flush(cx),
- EitherIoProj::B(val) => val.poll_flush(cx),
+ match self.get_mut() {
+ Connection::Tcp(ConnectionType::H1(conn)) => Pin::new(conn).poll_flush(cx),
+ Connection::Tls(ConnectionType::H1(conn)) => Pin::new(conn).poll_flush(cx),
+ _ => unreachable!(H2_UNREACHABLE_WRITE),
}
}
@@ -298,9 +360,38 @@ where
self: Pin<&mut Self>,
cx: &mut Context<'_>,
) -> Poll> {
- match self.project() {
- EitherIoProj::A(val) => val.poll_shutdown(cx),
- EitherIoProj::B(val) => val.poll_shutdown(cx),
+ match self.get_mut() {
+ Connection::Tcp(ConnectionType::H1(conn)) => {
+ Pin::new(conn).poll_shutdown(cx)
+ }
+ Connection::Tls(ConnectionType::H1(conn)) => {
+ Pin::new(conn).poll_shutdown(cx)
+ }
+ _ => unreachable!(H2_UNREACHABLE_WRITE),
+ }
+ }
+
+ fn poll_write_vectored(
+ self: Pin<&mut Self>,
+ cx: &mut Context<'_>,
+ bufs: &[io::IoSlice<'_>],
+ ) -> Poll> {
+ match self.get_mut() {
+ Connection::Tcp(ConnectionType::H1(conn)) => {
+ Pin::new(conn).poll_write_vectored(cx, bufs)
+ }
+ Connection::Tls(ConnectionType::H1(conn)) => {
+ Pin::new(conn).poll_write_vectored(cx, bufs)
+ }
+ _ => unreachable!(H2_UNREACHABLE_WRITE),
+ }
+ }
+
+ fn is_write_vectored(&self) -> bool {
+ match *self {
+ Connection::Tcp(ConnectionType::H1(ref conn)) => conn.is_write_vectored(),
+ Connection::Tls(ConnectionType::H1(ref conn)) => conn.is_write_vectored(),
+ _ => unreachable!(H2_UNREACHABLE_WRITE),
}
}
}
@@ -323,10 +414,13 @@ mod test {
let tcp = TcpStream::connect(local).await.unwrap();
let (sender, connection) = h2::client::handshake(tcp).await.unwrap();
- let conn = H2Connection::new(sender.clone(), connection);
+ let conn = H2ConnectionInner::new(sender.clone(), connection);
assert!(sender.clone().ready().await.is_ok());
- assert!(h2::client::SendRequest::clone(&*conn).ready().await.is_ok());
+ assert!(h2::client::SendRequest::clone(&conn.sender)
+ .ready()
+ .await
+ .is_ok());
drop(conn);
diff --git a/actix-http/src/client/connector.rs b/actix-http/src/client/connector.rs
index 65536f257..6996677d2 100644
--- a/actix-http/src/client/connector.rs
+++ b/actix-http/src/client/connector.rs
@@ -1,45 +1,53 @@
-use std::fmt;
-use std::marker::PhantomData;
-use std::time::Duration;
-
-use actix_codec::{AsyncRead, AsyncWrite};
-use actix_rt::net::TcpStream;
-use actix_service::{apply_fn, Service, ServiceExt};
-use actix_tls::connect::{
- new_connector, Connect as TcpConnect, Connection as TcpConnection, Resolver,
+use std::{
+ fmt,
+ future::Future,
+ net::IpAddr,
+ pin::Pin,
+ rc::Rc,
+ task::{Context, Poll},
+ time::Duration,
};
-use actix_utils::timeout::{TimeoutError, TimeoutService};
+
+use actix_rt::{
+ net::TcpStream,
+ time::{sleep, Sleep},
+};
+use actix_service::Service;
+use actix_tls::connect::{
+ new_connector, Connect as TcpConnect, ConnectError as TcpConnectError,
+ Connection as TcpConnection, Resolver,
+};
+use futures_core::{future::LocalBoxFuture, ready};
use http::Uri;
+use pin_project::pin_project;
use super::config::ConnectorConfig;
-use super::connection::Connection;
+use super::connection::{Connection, ConnectionIo};
use super::error::ConnectError;
-use super::pool::{ConnectionPool, Protocol};
+use super::pool::ConnectionPool;
use super::Connect;
+use super::Protocol;
#[cfg(feature = "openssl")]
use actix_tls::connect::ssl::openssl::SslConnector as OpensslConnector;
#[cfg(feature = "rustls")]
use actix_tls::connect::ssl::rustls::ClientConfig;
-#[cfg(feature = "rustls")]
-use std::sync::Arc;
-#[cfg(any(feature = "openssl", feature = "rustls"))]
enum SslConnector {
+ #[allow(dead_code)]
+ None,
#[cfg(feature = "openssl")]
Openssl(OpensslConnector),
#[cfg(feature = "rustls")]
- Rustls(Arc),
+ Rustls(std::sync::Arc),
}
-#[cfg(not(any(feature = "openssl", feature = "rustls")))]
-type SslConnector = ();
/// Manages HTTP client network connectivity.
///
/// The `Connector` type uses a builder-like combinator pattern for service
/// construction that finishes by calling the `.finish()` method.
///
-/// ```rust,ignore
+/// ```ignore
/// use std::time::Duration;
/// use actix_http::client::Connector;
///
@@ -47,18 +55,14 @@ type SslConnector = ();
/// .timeout(Duration::from_secs(5))
/// .finish();
/// ```
-pub struct Connector {
+pub struct Connector {
connector: T,
config: ConnectorConfig,
#[allow(dead_code)]
ssl: SslConnector,
- _phantom: PhantomData,
}
-trait Io: AsyncRead + AsyncWrite + Unpin {}
-impl Io for T {}
-
-impl Connector<(), ()> {
+impl Connector<()> {
#[allow(clippy::new_ret_no_self, clippy::let_unit_value)]
pub fn new() -> Connector<
impl Service<
@@ -66,13 +70,11 @@ impl Connector<(), ()> {
Response = TcpConnection,
Error = actix_tls::connect::ConnectError,
> + Clone,
- TcpStream,
> {
Connector {
ssl: Self::build_ssl(vec![b"h2".to_vec(), b"http/1.1".to_vec()]),
connector: new_connector(resolver::resolver()),
config: ConnectorConfig::default(),
- _phantom: PhantomData,
}
}
@@ -103,51 +105,59 @@ impl Connector<(), ()> {
config.root_store.add_server_trust_anchors(
&actix_tls::connect::ssl::rustls::TLS_SERVER_ROOTS,
);
- SslConnector::Rustls(Arc::new(config))
+ SslConnector::Rustls(std::sync::Arc::new(config))
}
// ssl turned off, provides empty ssl connector
#[cfg(not(any(feature = "openssl", feature = "rustls")))]
- fn build_ssl(_: Vec>) -> SslConnector {}
+ fn build_ssl(_: Vec>) -> SslConnector {
+ SslConnector::None
+ }
}
-impl Connector {
+impl Connector {
/// Use custom connector.
- pub fn connector(self, connector: T1) -> Connector
+ pub fn connector(self, connector: S1) -> Connector
where
- U1: AsyncRead + AsyncWrite + Unpin + fmt::Debug,
- T1: Service<
+ Io1: ConnectionIo + fmt::Debug,
+ S1: Service<
TcpConnect,
- Response = TcpConnection,
- Error = actix_tls::connect::ConnectError,
+ Response = TcpConnection,
+ Error = TcpConnectError,
> + Clone,
{
Connector {
connector,
config: self.config,
ssl: self.ssl,
- _phantom: PhantomData,
}
}
}
-impl Connector
+impl Connector
where
- U: AsyncRead + AsyncWrite + Unpin + fmt::Debug + 'static,
- T: Service<
+ Io: ConnectionIo + fmt::Debug,
+ S: Service<
TcpConnect,
- Response = TcpConnection,
- Error = actix_tls::connect::ConnectError,
+ Response = TcpConnection,
+ Error = TcpConnectError,
> + Clone
+ 'static,
{
- /// Connection timeout, i.e. max time to connect to remote host including dns name resolution.
- /// Set to 1 second by default.
+ /// Tcp connection timeout, i.e. max time to connect to remote host including dns name
+ /// resolution. Set to 5 second by default.
pub fn timeout(mut self, timeout: Duration) -> Self {
self.config.timeout = timeout;
self
}
+ /// Tls handshake timeout, i.e. max time to do tls handshake with remote host after tcp
+ /// connection established. Set to 5 second by default.
+ pub fn handshake_timeout(mut self, timeout: Duration) -> Self {
+ self.config.handshake_timeout = timeout;
+ self
+ }
+
#[cfg(feature = "openssl")]
/// Use custom `SslConnector` instance.
pub fn ssl(mut self, connector: OpensslConnector) -> Self {
@@ -156,7 +166,8 @@ where
}
#[cfg(feature = "rustls")]
- pub fn rustls(mut self, connector: Arc) -> Self {
+ /// Use custom `SslConnector` instance.
+ pub fn rustls(mut self, connector: std::sync::Arc) -> Self {
self.ssl = SslConnector::Rustls(connector);
self
}
@@ -237,263 +248,434 @@ where
self
}
+ /// Set local IP Address the connector would use for establishing connection.
+ pub fn local_address(mut self, addr: IpAddr) -> Self {
+ self.config.local_address = Some(addr);
+ self
+ }
+
/// Finish configuration process and create connector service.
/// The Connector builder always concludes by calling `finish()` last in
/// its combinator chain.
- pub fn finish(
- self,
- ) -> impl Service + Clone
- {
- #[cfg(not(any(feature = "openssl", feature = "rustls")))]
- {
- let connector = TimeoutService::new(
- self.config.timeout,
- apply_fn(self.connector, |msg: Connect, srv| {
- srv.call(TcpConnect::new(msg.uri).set_addr(msg.addr))
- })
- .map_err(ConnectError::from)
- .map(|stream| (stream.into_parts().0, Protocol::Http1)),
- )
- .map_err(|e| match e {
- TimeoutError::Service(e) => e,
- TimeoutError::Timeout => ConnectError::Timeout,
- });
+ pub fn finish(self) -> ConnectorService {
+ let local_address = self.config.local_address;
+ let timeout = self.config.timeout;
- connect_impl::InnerConnector {
- tcp_pool: ConnectionPool::new(
- connector,
- self.config.no_disconnect_timeout(),
- ),
- }
- }
- #[cfg(any(feature = "openssl", feature = "rustls"))]
- {
- const H2: &[u8] = b"h2";
- use actix_service::{boxed::service, pipeline};
+ let tcp_service_inner =
+ TcpConnectorInnerService::new(self.connector, timeout, local_address);
+
+ #[allow(clippy::redundant_clone)]
+ let tcp_service = TcpConnectorService {
+ service: tcp_service_inner.clone(),
+ };
+
+ let tls_service = match self.ssl {
+ SslConnector::None => None,
#[cfg(feature = "openssl")]
- use actix_tls::connect::ssl::openssl::OpensslConnector;
+ SslConnector::Openssl(tls) => {
+ const H2: &[u8] = b"h2";
+
+ use actix_tls::connect::ssl::openssl::{OpensslConnector, SslStream};
+
+ impl IntoConnectionIo for TcpConnection> {
+ fn into_connection_io(self) -> (Box, Protocol) {
+ let sock = self.into_parts().0;
+ let h2 = sock
+ .ssl()
+ .selected_alpn_protocol()
+ .map(|protos| protos.windows(2).any(|w| w == H2))
+ .unwrap_or(false);
+ if h2 {
+ (Box::new(sock), Protocol::Http2)
+ } else {
+ (Box::new(sock), Protocol::Http1)
+ }
+ }
+ }
+
+ let handshake_timeout = self.config.handshake_timeout;
+
+ let tls_service = TlsConnectorService {
+ tcp_service: tcp_service_inner,
+ tls_service: OpensslConnector::service(tls),
+ timeout: handshake_timeout,
+ };
+
+ Some(actix_service::boxed::rc_service(tls_service))
+ }
#[cfg(feature = "rustls")]
- use actix_tls::connect::ssl::rustls::{RustlsConnector, Session};
+ SslConnector::Rustls(tls) => {
+ const H2: &[u8] = b"h2";
- let ssl_service = TimeoutService::new(
- self.config.timeout,
- pipeline(
- apply_fn(self.connector.clone(), |msg: Connect, srv| {
- srv.call(TcpConnect::new(msg.uri).set_addr(msg.addr))
- })
- .map_err(ConnectError::from),
- )
- .and_then(match self.ssl {
- #[cfg(feature = "openssl")]
- SslConnector::Openssl(ssl) => service(
- OpensslConnector::service(ssl)
- .map(|stream| {
- let sock = stream.into_parts().0;
- let h2 = sock
- .ssl()
- .selected_alpn_protocol()
- .map(|protos| protos.windows(2).any(|w| w == H2))
- .unwrap_or(false);
- if h2 {
- (Box::new(sock) as Box, Protocol::Http2)
- } else {
- (Box::new(sock) as Box, Protocol::Http1)
- }
- })
- .map_err(ConnectError::from),
- ),
- #[cfg(feature = "rustls")]
- SslConnector::Rustls(ssl) => service(
- RustlsConnector::service(ssl)
- .map_err(ConnectError::from)
- .map(|stream| {
- let sock = stream.into_parts().0;
- let h2 = sock
- .get_ref()
- .1
- .get_alpn_protocol()
- .map(|protos| protos.windows(2).any(|w| w == H2))
- .unwrap_or(false);
- if h2 {
- (Box::new(sock) as Box, Protocol::Http2)
- } else {
- (Box::new(sock) as Box, Protocol::Http1)
- }
- }),
- ),
- }),
- )
- .map_err(|e| match e {
- TimeoutError::Service(e) => e,
- TimeoutError::Timeout => ConnectError::Timeout,
- });
+ use actix_tls::connect::ssl::rustls::{
+ RustlsConnector, Session, TlsStream,
+ };
- let tcp_service = TimeoutService::new(
- self.config.timeout,
- apply_fn(self.connector, |msg: Connect, srv| {
- srv.call(TcpConnect::new(msg.uri).set_addr(msg.addr))
- })
- .map_err(ConnectError::from)
- .map(|stream| (stream.into_parts().0, Protocol::Http1)),
- )
- .map_err(|e| match e {
- TimeoutError::Service(e) => e,
- TimeoutError::Timeout => ConnectError::Timeout,
- });
+ impl IntoConnectionIo for TcpConnection> {
+ fn into_connection_io(self) -> (Box, Protocol) {
+ let sock = self.into_parts().0;
+ let h2 = sock
+ .get_ref()
+ .1
+ .get_alpn_protocol()
+ .map(|protos| protos.windows(2).any(|w| w == H2))
+ .unwrap_or(false);
+ if h2 {
+ (Box::new(sock), Protocol::Http2)
+ } else {
+ (Box::new(sock), Protocol::Http1)
+ }
+ }
+ }
- connect_impl::InnerConnector {
- tcp_pool: ConnectionPool::new(
- tcp_service,
- self.config.no_disconnect_timeout(),
- ),
- ssl_pool: ConnectionPool::new(ssl_service, self.config),
+ let handshake_timeout = self.config.handshake_timeout;
+
+ let tls_service = TlsConnectorService {
+ tcp_service: tcp_service_inner,
+ tls_service: RustlsConnector::service(tls),
+ timeout: handshake_timeout,
+ };
+
+ Some(actix_service::boxed::rc_service(tls_service))
}
+ };
+
+ let tcp_config = self.config.no_disconnect_timeout();
+
+ let tcp_pool = ConnectionPool::new(tcp_service, tcp_config);
+
+ let tls_config = self.config;
+ let tls_pool = tls_service
+ .map(move |tls_service| ConnectionPool::new(tls_service, tls_config));
+
+ ConnectorServicePriv { tcp_pool, tls_pool }
+ }
+}
+
+/// tcp service for map `TcpConnection` type to `(Io, Protocol)`
+#[derive(Clone)]
+pub struct TcpConnectorService {
+ service: S,
+}
+
+impl Service for TcpConnectorService
+where
+ S: Service, Error = ConnectError>
+ + Clone
+ + 'static,
+{
+ type Response = (Io, Protocol);
+ type Error = ConnectError;
+ type Future = TcpConnectorFuture;
+
+ actix_service::forward_ready!(service);
+
+ fn call(&self, req: Connect) -> Self::Future {
+ TcpConnectorFuture {
+ fut: self.service.call(req),
}
}
}
-#[cfg(not(any(feature = "openssl", feature = "rustls")))]
-mod connect_impl {
- use std::task::{Context, Poll};
+#[pin_project]
+pub struct TcpConnectorFuture {
+ #[pin]
+ fut: Fut,
+}
- use futures_core::future::LocalBoxFuture;
+impl Future for TcpConnectorFuture
+where
+ Fut: Future