mirror of https://github.com/fafhrd91/actix-web
Merge branch 'master' into feature/test_websocket_with_config
This commit is contained in:
commit
e78715f4cd
|
@ -0,0 +1,3 @@
|
||||||
|
[alias]
|
||||||
|
chk = "hack check --workspace --tests --examples"
|
||||||
|
lint = "hack --clean-per-run clippy --workspace --tests --examples"
|
|
@ -23,6 +23,9 @@ jobs:
|
||||||
name: ${{ matrix.target.name }} / ${{ matrix.version }}
|
name: ${{ matrix.target.name }} / ${{ matrix.version }}
|
||||||
runs-on: ${{ matrix.target.os }}
|
runs-on: ${{ matrix.target.os }}
|
||||||
|
|
||||||
|
env:
|
||||||
|
VCPKGRS_DYNAMIC: 1
|
||||||
|
|
||||||
steps:
|
steps:
|
||||||
- uses: actions/checkout@v2
|
- uses: actions/checkout@v2
|
||||||
|
|
||||||
|
@ -65,7 +68,13 @@ jobs:
|
||||||
uses: actions-rs/cargo@v1
|
uses: actions-rs/cargo@v1
|
||||||
with:
|
with:
|
||||||
command: hack
|
command: hack
|
||||||
args: --clean-per-run check --workspace --no-default-features --tests
|
args: check --workspace --no-default-features
|
||||||
|
|
||||||
|
- name: check minimal + tests
|
||||||
|
uses: actions-rs/cargo@v1
|
||||||
|
with:
|
||||||
|
command: hack
|
||||||
|
args: check --workspace --no-default-features --tests --examples
|
||||||
|
|
||||||
- name: check full
|
- name: check full
|
||||||
uses: actions-rs/cargo@v1
|
uses: actions-rs/cargo@v1
|
||||||
|
|
|
@ -80,11 +80,11 @@ required-features = ["rustls"]
|
||||||
actix-codec = "0.4.0-beta.1"
|
actix-codec = "0.4.0-beta.1"
|
||||||
actix-macros = "0.2.0"
|
actix-macros = "0.2.0"
|
||||||
actix-router = "0.2.7"
|
actix-router = "0.2.7"
|
||||||
actix-rt = "2.1"
|
actix-rt = "2.2"
|
||||||
actix-server = "2.0.0-beta.3"
|
actix-server = "2.0.0-beta.3"
|
||||||
actix-service = "2.0.0-beta.4"
|
actix-service = "2.0.0-beta.4"
|
||||||
actix-utils = "3.0.0-beta.2"
|
actix-utils = "3.0.0-beta.2"
|
||||||
actix-tls = { version = "3.0.0-beta.4", default-features = false, optional = true }
|
actix-tls = { version = "3.0.0-beta.5", default-features = false, optional = true }
|
||||||
|
|
||||||
actix-web-codegen = "0.5.0-beta.2"
|
actix-web-codegen = "0.5.0-beta.2"
|
||||||
actix-http = "3.0.0-beta.4"
|
actix-http = "3.0.0-beta.4"
|
||||||
|
|
|
@ -33,5 +33,5 @@ mime_guess = "2.0.1"
|
||||||
percent-encoding = "2.1"
|
percent-encoding = "2.1"
|
||||||
|
|
||||||
[dev-dependencies]
|
[dev-dependencies]
|
||||||
actix-rt = "2.1"
|
actix-rt = "2.2"
|
||||||
actix-web = "4.0.0-beta.4"
|
actix-web = "4.0.0-beta.4"
|
||||||
|
|
|
@ -31,9 +31,9 @@ openssl = ["tls-openssl", "awc/openssl"]
|
||||||
[dependencies]
|
[dependencies]
|
||||||
actix-service = "2.0.0-beta.4"
|
actix-service = "2.0.0-beta.4"
|
||||||
actix-codec = "0.4.0-beta.1"
|
actix-codec = "0.4.0-beta.1"
|
||||||
actix-tls = "3.0.0-beta.4"
|
actix-tls = "3.0.0-beta.5"
|
||||||
actix-utils = "3.0.0-beta.2"
|
actix-utils = "3.0.0-beta.2"
|
||||||
actix-rt = "2.1"
|
actix-rt = "2.2"
|
||||||
actix-server = "2.0.0-beta.3"
|
actix-server = "2.0.0-beta.3"
|
||||||
awc = { version = "3.0.0-beta.3", default-features = false }
|
awc = { version = "3.0.0-beta.3", default-features = false }
|
||||||
|
|
||||||
|
|
|
@ -6,7 +6,7 @@
|
||||||
* `client::ConnectorService` as `client::Connector::finish` method's return type [#2081]
|
* `client::ConnectorService` as `client::Connector::finish` method's return type [#2081]
|
||||||
* `client::ConnectionIo` trait alias [#2081]
|
* `client::ConnectionIo` trait alias [#2081]
|
||||||
|
|
||||||
### Chaged
|
### Changed
|
||||||
* `client::Connector` type now only have one generic type for `actix_service::Service`. [#2063]
|
* `client::Connector` type now only have one generic type for `actix_service::Service`. [#2063]
|
||||||
|
|
||||||
[#2063]: https://github.com/actix/actix-web/pull/2063
|
[#2063]: https://github.com/actix/actix-web/pull/2063
|
||||||
|
|
|
@ -47,8 +47,8 @@ trust-dns = ["trust-dns-resolver"]
|
||||||
actix-service = "2.0.0-beta.4"
|
actix-service = "2.0.0-beta.4"
|
||||||
actix-codec = "0.4.0-beta.1"
|
actix-codec = "0.4.0-beta.1"
|
||||||
actix-utils = "3.0.0-beta.2"
|
actix-utils = "3.0.0-beta.2"
|
||||||
actix-rt = "2.1"
|
actix-rt = "2.2"
|
||||||
actix-tls = "3.0.0-beta.4"
|
actix-tls = "3.0.0-beta.5"
|
||||||
|
|
||||||
ahash = "0.7"
|
ahash = "0.7"
|
||||||
base64 = "0.13"
|
base64 = "0.13"
|
||||||
|
@ -89,7 +89,7 @@ trust-dns-resolver = { version = "0.20.0", optional = true }
|
||||||
[dev-dependencies]
|
[dev-dependencies]
|
||||||
actix-server = "2.0.0-beta.3"
|
actix-server = "2.0.0-beta.3"
|
||||||
actix-http-test = { version = "3.0.0-beta.3", features = ["openssl"] }
|
actix-http-test = { version = "3.0.0-beta.3", features = ["openssl"] }
|
||||||
actix-tls = { version = "3.0.0-beta.4", features = ["openssl"] }
|
actix-tls = { version = "3.0.0-beta.5", features = ["openssl"] }
|
||||||
criterion = "0.3"
|
criterion = "0.3"
|
||||||
env_logger = "0.8"
|
env_logger = "0.8"
|
||||||
rcgen = "0.8"
|
rcgen = "0.8"
|
||||||
|
|
|
@ -63,11 +63,9 @@ where
|
||||||
X: ServiceFactory<Request, Config = (), Response = Request>,
|
X: ServiceFactory<Request, Config = (), Response = Request>,
|
||||||
X::Error: Into<Error>,
|
X::Error: Into<Error>,
|
||||||
X::InitError: fmt::Debug,
|
X::InitError: fmt::Debug,
|
||||||
<X::Service as Service<Request>>::Future: 'static,
|
|
||||||
U: ServiceFactory<(Request, Framed<T, Codec>), Config = (), Response = ()>,
|
U: ServiceFactory<(Request, Framed<T, Codec>), Config = (), Response = ()>,
|
||||||
U::Error: fmt::Display,
|
U::Error: fmt::Display,
|
||||||
U::InitError: fmt::Debug,
|
U::InitError: fmt::Debug,
|
||||||
<U::Service as Service<(Request, Framed<T, Codec>)>>::Future: 'static,
|
|
||||||
{
|
{
|
||||||
/// Set server keep-alive setting.
|
/// Set server keep-alive setting.
|
||||||
///
|
///
|
||||||
|
@ -127,7 +125,6 @@ where
|
||||||
X1: ServiceFactory<Request, Config = (), Response = Request>,
|
X1: ServiceFactory<Request, Config = (), Response = Request>,
|
||||||
X1::Error: Into<Error>,
|
X1::Error: Into<Error>,
|
||||||
X1::InitError: fmt::Debug,
|
X1::InitError: fmt::Debug,
|
||||||
<X1::Service as Service<Request>>::Future: 'static,
|
|
||||||
{
|
{
|
||||||
HttpServiceBuilder {
|
HttpServiceBuilder {
|
||||||
keep_alive: self.keep_alive,
|
keep_alive: self.keep_alive,
|
||||||
|
@ -152,7 +149,6 @@ where
|
||||||
U1: ServiceFactory<(Request, Framed<T, Codec>), Config = (), Response = ()>,
|
U1: ServiceFactory<(Request, Framed<T, Codec>), Config = (), Response = ()>,
|
||||||
U1::Error: fmt::Display,
|
U1::Error: fmt::Display,
|
||||||
U1::InitError: fmt::Debug,
|
U1::InitError: fmt::Debug,
|
||||||
<U1::Service as Service<(Request, Framed<T, Codec>)>>::Future: 'static,
|
|
||||||
{
|
{
|
||||||
HttpServiceBuilder {
|
HttpServiceBuilder {
|
||||||
keep_alive: self.keep_alive,
|
keep_alive: self.keep_alive,
|
||||||
|
@ -211,7 +207,6 @@ where
|
||||||
S::Error: Into<Error> + 'static,
|
S::Error: Into<Error> + 'static,
|
||||||
S::InitError: fmt::Debug,
|
S::InitError: fmt::Debug,
|
||||||
S::Response: Into<Response<B>> + 'static,
|
S::Response: Into<Response<B>> + 'static,
|
||||||
<S::Service as Service<Request>>::Future: 'static,
|
|
||||||
{
|
{
|
||||||
let cfg = ServiceConfig::new(
|
let cfg = ServiceConfig::new(
|
||||||
self.keep_alive,
|
self.keep_alive,
|
||||||
|
@ -233,7 +228,6 @@ where
|
||||||
S::Error: Into<Error> + 'static,
|
S::Error: Into<Error> + 'static,
|
||||||
S::InitError: fmt::Debug,
|
S::InitError: fmt::Debug,
|
||||||
S::Response: Into<Response<B>> + 'static,
|
S::Response: Into<Response<B>> + 'static,
|
||||||
<S::Service as Service<Request>>::Future: 'static,
|
|
||||||
{
|
{
|
||||||
let cfg = ServiceConfig::new(
|
let cfg = ServiceConfig::new(
|
||||||
self.keep_alive,
|
self.keep_alive,
|
||||||
|
|
|
@ -9,7 +9,7 @@ use std::{
|
||||||
};
|
};
|
||||||
|
|
||||||
use actix_rt::{
|
use actix_rt::{
|
||||||
net::TcpStream,
|
net::{ActixStream, TcpStream},
|
||||||
time::{sleep, Sleep},
|
time::{sleep, Sleep},
|
||||||
};
|
};
|
||||||
use actix_service::Service;
|
use actix_service::Service;
|
||||||
|
@ -119,7 +119,7 @@ impl<S> Connector<S> {
|
||||||
/// Use custom connector.
|
/// Use custom connector.
|
||||||
pub fn connector<S1, Io1>(self, connector: S1) -> Connector<S1>
|
pub fn connector<S1, Io1>(self, connector: S1) -> Connector<S1>
|
||||||
where
|
where
|
||||||
Io1: ConnectionIo + fmt::Debug,
|
Io1: ActixStream + fmt::Debug + 'static,
|
||||||
S1: Service<
|
S1: Service<
|
||||||
TcpConnect<Uri>,
|
TcpConnect<Uri>,
|
||||||
Response = TcpConnection<Uri, Io1>,
|
Response = TcpConnection<Uri, Io1>,
|
||||||
|
@ -136,7 +136,14 @@ impl<S> Connector<S> {
|
||||||
|
|
||||||
impl<S, Io> Connector<S>
|
impl<S, Io> Connector<S>
|
||||||
where
|
where
|
||||||
Io: ConnectionIo + fmt::Debug,
|
// Note:
|
||||||
|
// Input Io type is bound to ActixStream trait but internally in client module they
|
||||||
|
// are bound to ConnectionIo trait alias. And latter is the trait exposed to public
|
||||||
|
// in the form of Box<dyn ConnectionIo> type.
|
||||||
|
//
|
||||||
|
// This remap is to hide ActixStream's trait methods. They are not meant to be called
|
||||||
|
// from user code.
|
||||||
|
Io: ActixStream + fmt::Debug + 'static,
|
||||||
S: Service<
|
S: Service<
|
||||||
TcpConnect<Uri>,
|
TcpConnect<Uri>,
|
||||||
Response = TcpConnection<Uri, Io>,
|
Response = TcpConnection<Uri, Io>,
|
||||||
|
@ -407,16 +414,14 @@ struct TlsConnectorService<S, St> {
|
||||||
timeout: Duration,
|
timeout: Duration,
|
||||||
}
|
}
|
||||||
|
|
||||||
impl<S, St, Io, Res> Service<Connect> for TlsConnectorService<S, St>
|
impl<S, St, Io> Service<Connect> for TlsConnectorService<S, St>
|
||||||
where
|
where
|
||||||
S: Service<Connect, Response = TcpConnection<Uri, Io>, Error = ConnectError>
|
S: Service<Connect, Response = TcpConnection<Uri, Io>, Error = ConnectError>
|
||||||
+ Clone
|
+ Clone
|
||||||
+ 'static,
|
+ 'static,
|
||||||
St: Service<TcpConnection<Uri, Io>, Response = Res, Error = std::io::Error>
|
St: Service<TcpConnection<Uri, Io>, Error = std::io::Error> + Clone + 'static,
|
||||||
+ Clone
|
|
||||||
+ 'static,
|
|
||||||
Io: ConnectionIo,
|
Io: ConnectionIo,
|
||||||
Res: IntoConnectionIo,
|
St::Response: IntoConnectionIo,
|
||||||
{
|
{
|
||||||
type Response = (Box<dyn ConnectionIo>, Protocol);
|
type Response = (Box<dyn ConnectionIo>, Protocol);
|
||||||
type Error = ConnectError;
|
type Error = ConnectError;
|
||||||
|
@ -471,10 +476,10 @@ where
|
||||||
Error = std::io::Error,
|
Error = std::io::Error,
|
||||||
Future = Fut2,
|
Future = Fut2,
|
||||||
>,
|
>,
|
||||||
|
S::Response: IntoConnectionIo,
|
||||||
Fut1: Future<Output = Result<TcpConnection<Uri, Io>, ConnectError>>,
|
Fut1: Future<Output = Result<TcpConnection<Uri, Io>, ConnectError>>,
|
||||||
Fut2: Future<Output = Result<S::Response, S::Error>>,
|
Fut2: Future<Output = Result<S::Response, S::Error>>,
|
||||||
Io: ConnectionIo,
|
Io: ConnectionIo,
|
||||||
Res: IntoConnectionIo,
|
|
||||||
{
|
{
|
||||||
type Output = Result<(Box<dyn ConnectionIo>, Protocol), ConnectError>;
|
type Output = Result<(Box<dyn ConnectionIo>, Protocol), ConnectError>;
|
||||||
|
|
||||||
|
|
|
@ -7,7 +7,7 @@ use std::{
|
||||||
use actix_codec::Framed;
|
use actix_codec::Framed;
|
||||||
use bytes::buf::BufMut;
|
use bytes::buf::BufMut;
|
||||||
use bytes::{Bytes, BytesMut};
|
use bytes::{Bytes, BytesMut};
|
||||||
use futures_core::Stream;
|
use futures_core::{ready, Stream};
|
||||||
use futures_util::{future::poll_fn, SinkExt as _};
|
use futures_util::{future::poll_fn, SinkExt as _};
|
||||||
|
|
||||||
use crate::error::PayloadError;
|
use crate::error::PayloadError;
|
||||||
|
@ -17,7 +17,7 @@ use crate::http::{
|
||||||
StatusCode,
|
StatusCode,
|
||||||
};
|
};
|
||||||
use crate::message::{RequestHeadType, ResponseHead};
|
use crate::message::{RequestHeadType, ResponseHead};
|
||||||
use crate::payload::{Payload, PayloadStream};
|
use crate::payload::Payload;
|
||||||
|
|
||||||
use super::connection::{ConnectionIo, H1Connection};
|
use super::connection::{ConnectionIo, H1Connection};
|
||||||
use super::error::{ConnectError, SendRequestError};
|
use super::error::{ConnectError, SendRequestError};
|
||||||
|
@ -122,10 +122,7 @@ where
|
||||||
|
|
||||||
Ok((head, Payload::None))
|
Ok((head, Payload::None))
|
||||||
}
|
}
|
||||||
_ => {
|
_ => Ok((head, Payload::Stream(Box::pin(PlStream::new(framed))))),
|
||||||
let pl: PayloadStream = Box::pin(PlStream::new(framed));
|
|
||||||
Ok((head, pl.into()))
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ -194,21 +191,16 @@ where
|
||||||
}
|
}
|
||||||
|
|
||||||
#[pin_project::pin_project]
|
#[pin_project::pin_project]
|
||||||
pub(crate) struct PlStream<Io: ConnectionIo>
|
pub(crate) struct PlStream<Io: ConnectionIo> {
|
||||||
where
|
|
||||||
Io: ConnectionIo,
|
|
||||||
{
|
|
||||||
#[pin]
|
#[pin]
|
||||||
framed: Option<Framed<H1Connection<Io>, h1::ClientPayloadCodec>>,
|
framed: Framed<H1Connection<Io>, h1::ClientPayloadCodec>,
|
||||||
}
|
}
|
||||||
|
|
||||||
impl<Io: ConnectionIo> PlStream<Io> {
|
impl<Io: ConnectionIo> PlStream<Io> {
|
||||||
fn new(framed: Framed<H1Connection<Io>, h1::ClientCodec>) -> Self {
|
fn new(framed: Framed<H1Connection<Io>, h1::ClientCodec>) -> Self {
|
||||||
let framed = framed.into_map_codec(|codec| codec.into_payload_codec());
|
let framed = framed.into_map_codec(|codec| codec.into_payload_codec());
|
||||||
|
|
||||||
PlStream {
|
PlStream { framed }
|
||||||
framed: Some(framed),
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ -219,20 +211,16 @@ impl<Io: ConnectionIo> Stream for PlStream<Io> {
|
||||||
self: Pin<&mut Self>,
|
self: Pin<&mut Self>,
|
||||||
cx: &mut Context<'_>,
|
cx: &mut Context<'_>,
|
||||||
) -> Poll<Option<Self::Item>> {
|
) -> Poll<Option<Self::Item>> {
|
||||||
let mut framed = self.project().framed.as_pin_mut().unwrap();
|
let mut this = self.project();
|
||||||
|
|
||||||
match framed.as_mut().next_item(cx)? {
|
match ready!(this.framed.as_mut().next_item(cx)?) {
|
||||||
Poll::Pending => Poll::Pending,
|
Some(Some(chunk)) => Poll::Ready(Some(Ok(chunk))),
|
||||||
Poll::Ready(Some(chunk)) => {
|
Some(None) => {
|
||||||
if let Some(chunk) = chunk {
|
let keep_alive = this.framed.codec_ref().keepalive();
|
||||||
Poll::Ready(Some(Ok(chunk)))
|
this.framed.io_mut().on_release(keep_alive);
|
||||||
} else {
|
Poll::Ready(None)
|
||||||
let keep_alive = framed.codec_ref().keepalive();
|
|
||||||
framed.io_mut().on_release(keep_alive);
|
|
||||||
Poll::Ready(None)
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
Poll::Ready(None) => Poll::Ready(None),
|
None => Poll::Ready(None),
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
|
@ -1,6 +1,4 @@
|
||||||
use std::future::Future;
|
|
||||||
use std::marker::PhantomData;
|
use std::marker::PhantomData;
|
||||||
use std::pin::Pin;
|
|
||||||
use std::rc::Rc;
|
use std::rc::Rc;
|
||||||
use std::task::{Context, Poll};
|
use std::task::{Context, Poll};
|
||||||
use std::{fmt, net};
|
use std::{fmt, net};
|
||||||
|
@ -8,7 +6,7 @@ use std::{fmt, net};
|
||||||
use actix_codec::{AsyncRead, AsyncWrite, Framed};
|
use actix_codec::{AsyncRead, AsyncWrite, Framed};
|
||||||
use actix_rt::net::TcpStream;
|
use actix_rt::net::TcpStream;
|
||||||
use actix_service::{pipeline_factory, IntoServiceFactory, Service, ServiceFactory};
|
use actix_service::{pipeline_factory, IntoServiceFactory, Service, ServiceFactory};
|
||||||
use futures_core::ready;
|
use futures_core::future::LocalBoxFuture;
|
||||||
use futures_util::future::ready;
|
use futures_util::future::ready;
|
||||||
|
|
||||||
use crate::body::MessageBody;
|
use crate::body::MessageBody;
|
||||||
|
@ -16,7 +14,7 @@ use crate::config::ServiceConfig;
|
||||||
use crate::error::{DispatchError, Error};
|
use crate::error::{DispatchError, Error};
|
||||||
use crate::request::Request;
|
use crate::request::Request;
|
||||||
use crate::response::Response;
|
use crate::response::Response;
|
||||||
use crate::service::HttpFlow;
|
use crate::service::HttpServiceHandler;
|
||||||
use crate::{ConnectCallback, OnConnectData};
|
use crate::{ConnectCallback, OnConnectData};
|
||||||
|
|
||||||
use super::codec::Codec;
|
use super::codec::Codec;
|
||||||
|
@ -60,14 +58,17 @@ where
|
||||||
impl<S, B, X, U> H1Service<TcpStream, S, B, X, U>
|
impl<S, B, X, U> H1Service<TcpStream, S, B, X, U>
|
||||||
where
|
where
|
||||||
S: ServiceFactory<Request, Config = ()>,
|
S: ServiceFactory<Request, Config = ()>,
|
||||||
|
S::Future: 'static,
|
||||||
S::Error: Into<Error>,
|
S::Error: Into<Error>,
|
||||||
S::InitError: fmt::Debug,
|
S::InitError: fmt::Debug,
|
||||||
S::Response: Into<Response<B>>,
|
S::Response: Into<Response<B>>,
|
||||||
B: MessageBody,
|
B: MessageBody,
|
||||||
X: ServiceFactory<Request, Config = (), Response = Request>,
|
X: ServiceFactory<Request, Config = (), Response = Request>,
|
||||||
|
X::Future: 'static,
|
||||||
X::Error: Into<Error>,
|
X::Error: Into<Error>,
|
||||||
X::InitError: fmt::Debug,
|
X::InitError: fmt::Debug,
|
||||||
U: ServiceFactory<(Request, Framed<TcpStream, Codec>), Config = (), Response = ()>,
|
U: ServiceFactory<(Request, Framed<TcpStream, Codec>), Config = (), Response = ()>,
|
||||||
|
U::Future: 'static,
|
||||||
U::Error: fmt::Display + Into<Error>,
|
U::Error: fmt::Display + Into<Error>,
|
||||||
U::InitError: fmt::Debug,
|
U::InitError: fmt::Debug,
|
||||||
{
|
{
|
||||||
|
@ -94,17 +95,21 @@ mod openssl {
|
||||||
use super::*;
|
use super::*;
|
||||||
|
|
||||||
use actix_service::ServiceFactoryExt;
|
use actix_service::ServiceFactoryExt;
|
||||||
use actix_tls::accept::openssl::{Acceptor, SslAcceptor, SslError, TlsStream};
|
use actix_tls::accept::{
|
||||||
use actix_tls::accept::TlsError;
|
openssl::{Acceptor, SslAcceptor, SslError, TlsStream},
|
||||||
|
TlsError,
|
||||||
|
};
|
||||||
|
|
||||||
impl<S, B, X, U> H1Service<TlsStream<TcpStream>, S, B, X, U>
|
impl<S, B, X, U> H1Service<TlsStream<TcpStream>, S, B, X, U>
|
||||||
where
|
where
|
||||||
S: ServiceFactory<Request, Config = ()>,
|
S: ServiceFactory<Request, Config = ()>,
|
||||||
|
S::Future: 'static,
|
||||||
S::Error: Into<Error>,
|
S::Error: Into<Error>,
|
||||||
S::InitError: fmt::Debug,
|
S::InitError: fmt::Debug,
|
||||||
S::Response: Into<Response<B>>,
|
S::Response: Into<Response<B>>,
|
||||||
B: MessageBody,
|
B: MessageBody,
|
||||||
X: ServiceFactory<Request, Config = (), Response = Request>,
|
X: ServiceFactory<Request, Config = (), Response = Request>,
|
||||||
|
X::Future: 'static,
|
||||||
X::Error: Into<Error>,
|
X::Error: Into<Error>,
|
||||||
X::InitError: fmt::Debug,
|
X::InitError: fmt::Debug,
|
||||||
U: ServiceFactory<
|
U: ServiceFactory<
|
||||||
|
@ -112,6 +117,7 @@ mod openssl {
|
||||||
Config = (),
|
Config = (),
|
||||||
Response = (),
|
Response = (),
|
||||||
>,
|
>,
|
||||||
|
U::Future: 'static,
|
||||||
U::Error: fmt::Display + Into<Error>,
|
U::Error: fmt::Display + Into<Error>,
|
||||||
U::InitError: fmt::Debug,
|
U::InitError: fmt::Debug,
|
||||||
{
|
{
|
||||||
|
@ -143,19 +149,25 @@ mod openssl {
|
||||||
#[cfg(feature = "rustls")]
|
#[cfg(feature = "rustls")]
|
||||||
mod rustls {
|
mod rustls {
|
||||||
use super::*;
|
use super::*;
|
||||||
|
|
||||||
|
use std::io;
|
||||||
|
|
||||||
use actix_service::ServiceFactoryExt;
|
use actix_service::ServiceFactoryExt;
|
||||||
use actix_tls::accept::rustls::{Acceptor, ServerConfig, TlsStream};
|
use actix_tls::accept::{
|
||||||
use actix_tls::accept::TlsError;
|
rustls::{Acceptor, ServerConfig, TlsStream},
|
||||||
use std::{fmt, io};
|
TlsError,
|
||||||
|
};
|
||||||
|
|
||||||
impl<S, B, X, U> H1Service<TlsStream<TcpStream>, S, B, X, U>
|
impl<S, B, X, U> H1Service<TlsStream<TcpStream>, S, B, X, U>
|
||||||
where
|
where
|
||||||
S: ServiceFactory<Request, Config = ()>,
|
S: ServiceFactory<Request, Config = ()>,
|
||||||
|
S::Future: 'static,
|
||||||
S::Error: Into<Error>,
|
S::Error: Into<Error>,
|
||||||
S::InitError: fmt::Debug,
|
S::InitError: fmt::Debug,
|
||||||
S::Response: Into<Response<B>>,
|
S::Response: Into<Response<B>>,
|
||||||
B: MessageBody,
|
B: MessageBody,
|
||||||
X: ServiceFactory<Request, Config = (), Response = Request>,
|
X: ServiceFactory<Request, Config = (), Response = Request>,
|
||||||
|
X::Future: 'static,
|
||||||
X::Error: Into<Error>,
|
X::Error: Into<Error>,
|
||||||
X::InitError: fmt::Debug,
|
X::InitError: fmt::Debug,
|
||||||
U: ServiceFactory<
|
U: ServiceFactory<
|
||||||
|
@ -163,6 +175,7 @@ mod rustls {
|
||||||
Config = (),
|
Config = (),
|
||||||
Response = (),
|
Response = (),
|
||||||
>,
|
>,
|
||||||
|
U::Future: 'static,
|
||||||
U::Error: fmt::Display + Into<Error>,
|
U::Error: fmt::Display + Into<Error>,
|
||||||
U::InitError: fmt::Debug,
|
U::InitError: fmt::Debug,
|
||||||
{
|
{
|
||||||
|
@ -241,16 +254,19 @@ where
|
||||||
impl<T, S, B, X, U> ServiceFactory<(T, Option<net::SocketAddr>)>
|
impl<T, S, B, X, U> ServiceFactory<(T, Option<net::SocketAddr>)>
|
||||||
for H1Service<T, S, B, X, U>
|
for H1Service<T, S, B, X, U>
|
||||||
where
|
where
|
||||||
T: AsyncRead + AsyncWrite + Unpin,
|
T: AsyncRead + AsyncWrite + Unpin + 'static,
|
||||||
S: ServiceFactory<Request, Config = ()>,
|
S: ServiceFactory<Request, Config = ()>,
|
||||||
|
S::Future: 'static,
|
||||||
S::Error: Into<Error>,
|
S::Error: Into<Error>,
|
||||||
S::Response: Into<Response<B>>,
|
S::Response: Into<Response<B>>,
|
||||||
S::InitError: fmt::Debug,
|
S::InitError: fmt::Debug,
|
||||||
B: MessageBody,
|
B: MessageBody,
|
||||||
X: ServiceFactory<Request, Config = (), Response = Request>,
|
X: ServiceFactory<Request, Config = (), Response = Request>,
|
||||||
|
X::Future: 'static,
|
||||||
X::Error: Into<Error>,
|
X::Error: Into<Error>,
|
||||||
X::InitError: fmt::Debug,
|
X::InitError: fmt::Debug,
|
||||||
U: ServiceFactory<(Request, Framed<T, Codec>), Config = (), Response = ()>,
|
U: ServiceFactory<(Request, Framed<T, Codec>), Config = (), Response = ()>,
|
||||||
|
U::Future: 'static,
|
||||||
U::Error: fmt::Display + Into<Error>,
|
U::Error: fmt::Display + Into<Error>,
|
||||||
U::InitError: fmt::Debug,
|
U::InitError: fmt::Debug,
|
||||||
{
|
{
|
||||||
|
@ -259,148 +275,50 @@ where
|
||||||
type Config = ();
|
type Config = ();
|
||||||
type Service = H1ServiceHandler<T, S::Service, B, X::Service, U::Service>;
|
type Service = H1ServiceHandler<T, S::Service, B, X::Service, U::Service>;
|
||||||
type InitError = ();
|
type InitError = ();
|
||||||
type Future = H1ServiceResponse<T, S, B, X, U>;
|
type Future = LocalBoxFuture<'static, Result<Self::Service, Self::InitError>>;
|
||||||
|
|
||||||
fn new_service(&self, _: ()) -> Self::Future {
|
fn new_service(&self, _: ()) -> Self::Future {
|
||||||
H1ServiceResponse {
|
let service = self.srv.new_service(());
|
||||||
fut: self.srv.new_service(()),
|
let expect = self.expect.new_service(());
|
||||||
fut_ex: Some(self.expect.new_service(())),
|
let upgrade = self.upgrade.as_ref().map(|s| s.new_service(()));
|
||||||
fut_upg: self.upgrade.as_ref().map(|f| f.new_service(())),
|
let on_connect_ext = self.on_connect_ext.clone();
|
||||||
expect: None,
|
let cfg = self.cfg.clone();
|
||||||
upgrade: None,
|
|
||||||
on_connect_ext: self.on_connect_ext.clone(),
|
|
||||||
cfg: Some(self.cfg.clone()),
|
|
||||||
_phantom: PhantomData,
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
#[doc(hidden)]
|
Box::pin(async move {
|
||||||
#[pin_project::pin_project]
|
let expect = expect
|
||||||
pub struct H1ServiceResponse<T, S, B, X, U>
|
.await
|
||||||
where
|
.map_err(|e| log::error!("Init http expect service error: {:?}", e))?;
|
||||||
S: ServiceFactory<Request>,
|
|
||||||
S::Error: Into<Error>,
|
|
||||||
S::InitError: fmt::Debug,
|
|
||||||
X: ServiceFactory<Request, Response = Request>,
|
|
||||||
X::Error: Into<Error>,
|
|
||||||
X::InitError: fmt::Debug,
|
|
||||||
U: ServiceFactory<(Request, Framed<T, Codec>), Response = ()>,
|
|
||||||
U::Error: fmt::Display,
|
|
||||||
U::InitError: fmt::Debug,
|
|
||||||
{
|
|
||||||
#[pin]
|
|
||||||
fut: S::Future,
|
|
||||||
#[pin]
|
|
||||||
fut_ex: Option<X::Future>,
|
|
||||||
#[pin]
|
|
||||||
fut_upg: Option<U::Future>,
|
|
||||||
expect: Option<X::Service>,
|
|
||||||
upgrade: Option<U::Service>,
|
|
||||||
on_connect_ext: Option<Rc<ConnectCallback<T>>>,
|
|
||||||
cfg: Option<ServiceConfig>,
|
|
||||||
_phantom: PhantomData<B>,
|
|
||||||
}
|
|
||||||
|
|
||||||
impl<T, S, B, X, U> Future for H1ServiceResponse<T, S, B, X, U>
|
let upgrade = match upgrade {
|
||||||
where
|
Some(upgrade) => {
|
||||||
T: AsyncRead + AsyncWrite + Unpin,
|
let upgrade = upgrade.await.map_err(|e| {
|
||||||
S: ServiceFactory<Request>,
|
log::error!("Init http upgrade service error: {:?}", e)
|
||||||
S::Error: Into<Error>,
|
})?;
|
||||||
S::Response: Into<Response<B>>,
|
Some(upgrade)
|
||||||
S::InitError: fmt::Debug,
|
}
|
||||||
B: MessageBody,
|
None => None,
|
||||||
X: ServiceFactory<Request, Response = Request>,
|
};
|
||||||
X::Error: Into<Error>,
|
|
||||||
X::InitError: fmt::Debug,
|
|
||||||
U: ServiceFactory<(Request, Framed<T, Codec>), Response = ()>,
|
|
||||||
U::Error: fmt::Display,
|
|
||||||
U::InitError: fmt::Debug,
|
|
||||||
{
|
|
||||||
type Output = Result<H1ServiceHandler<T, S::Service, B, X::Service, U::Service>, ()>;
|
|
||||||
|
|
||||||
fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
|
let service = service
|
||||||
let mut this = self.as_mut().project();
|
.await
|
||||||
|
.map_err(|e| log::error!("Init http service error: {:?}", e))?;
|
||||||
|
|
||||||
if let Some(fut) = this.fut_ex.as_pin_mut() {
|
Ok(H1ServiceHandler::new(
|
||||||
let expect = ready!(fut
|
cfg,
|
||||||
.poll(cx)
|
|
||||||
.map_err(|e| log::error!("Init http service error: {:?}", e)))?;
|
|
||||||
this = self.as_mut().project();
|
|
||||||
*this.expect = Some(expect);
|
|
||||||
this.fut_ex.set(None);
|
|
||||||
}
|
|
||||||
|
|
||||||
if let Some(fut) = this.fut_upg.as_pin_mut() {
|
|
||||||
let upgrade = ready!(fut
|
|
||||||
.poll(cx)
|
|
||||||
.map_err(|e| log::error!("Init http service error: {:?}", e)))?;
|
|
||||||
this = self.as_mut().project();
|
|
||||||
*this.upgrade = Some(upgrade);
|
|
||||||
this.fut_upg.set(None);
|
|
||||||
}
|
|
||||||
|
|
||||||
let result = ready!(this
|
|
||||||
.fut
|
|
||||||
.poll(cx)
|
|
||||||
.map_err(|e| log::error!("Init http service error: {:?}", e)));
|
|
||||||
|
|
||||||
Poll::Ready(result.map(|service| {
|
|
||||||
let this = self.as_mut().project();
|
|
||||||
|
|
||||||
H1ServiceHandler::new(
|
|
||||||
this.cfg.take().unwrap(),
|
|
||||||
service,
|
service,
|
||||||
this.expect.take().unwrap(),
|
expect,
|
||||||
this.upgrade.take(),
|
upgrade,
|
||||||
this.on_connect_ext.clone(),
|
on_connect_ext,
|
||||||
)
|
))
|
||||||
}))
|
})
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
/// `Service` implementation for HTTP/1 transport
|
/// `Service` implementation for HTTP/1 transport
|
||||||
pub struct H1ServiceHandler<T, S, B, X, U>
|
pub type H1ServiceHandler<T, S, B, X, U> = HttpServiceHandler<T, S, B, X, U>;
|
||||||
where
|
|
||||||
S: Service<Request>,
|
|
||||||
X: Service<Request>,
|
|
||||||
U: Service<(Request, Framed<T, Codec>)>,
|
|
||||||
{
|
|
||||||
flow: Rc<HttpFlow<S, X, U>>,
|
|
||||||
on_connect_ext: Option<Rc<ConnectCallback<T>>>,
|
|
||||||
cfg: ServiceConfig,
|
|
||||||
_phantom: PhantomData<B>,
|
|
||||||
}
|
|
||||||
|
|
||||||
impl<T, S, B, X, U> H1ServiceHandler<T, S, B, X, U>
|
|
||||||
where
|
|
||||||
S: Service<Request>,
|
|
||||||
S::Error: Into<Error>,
|
|
||||||
S::Response: Into<Response<B>>,
|
|
||||||
B: MessageBody,
|
|
||||||
X: Service<Request, Response = Request>,
|
|
||||||
X::Error: Into<Error>,
|
|
||||||
U: Service<(Request, Framed<T, Codec>), Response = ()>,
|
|
||||||
U::Error: fmt::Display,
|
|
||||||
{
|
|
||||||
fn new(
|
|
||||||
cfg: ServiceConfig,
|
|
||||||
service: S,
|
|
||||||
expect: X,
|
|
||||||
upgrade: Option<U>,
|
|
||||||
on_connect_ext: Option<Rc<ConnectCallback<T>>>,
|
|
||||||
) -> H1ServiceHandler<T, S, B, X, U> {
|
|
||||||
H1ServiceHandler {
|
|
||||||
flow: HttpFlow::new(service, expect, upgrade),
|
|
||||||
cfg,
|
|
||||||
on_connect_ext,
|
|
||||||
_phantom: PhantomData,
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
impl<T, S, B, X, U> Service<(T, Option<net::SocketAddr>)>
|
impl<T, S, B, X, U> Service<(T, Option<net::SocketAddr>)>
|
||||||
for H1ServiceHandler<T, S, B, X, U>
|
for HttpServiceHandler<T, S, B, X, U>
|
||||||
where
|
where
|
||||||
T: AsyncRead + AsyncWrite + Unpin,
|
T: AsyncRead + AsyncWrite + Unpin,
|
||||||
S: Service<Request>,
|
S: Service<Request>,
|
||||||
|
@ -417,47 +335,10 @@ where
|
||||||
type Future = Dispatcher<T, S, B, X, U>;
|
type Future = Dispatcher<T, S, B, X, U>;
|
||||||
|
|
||||||
fn poll_ready(&self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
|
fn poll_ready(&self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
|
||||||
let ready = self
|
self._poll_ready(cx).map_err(|e| {
|
||||||
.flow
|
log::error!("HTTP/1 service readiness error: {:?}", e);
|
||||||
.expect
|
DispatchError::Service(e)
|
||||||
.poll_ready(cx)
|
})
|
||||||
.map_err(|e| {
|
|
||||||
let e = e.into();
|
|
||||||
log::error!("Http service readiness error: {:?}", e);
|
|
||||||
DispatchError::Service(e)
|
|
||||||
})?
|
|
||||||
.is_ready();
|
|
||||||
|
|
||||||
let ready = self
|
|
||||||
.flow
|
|
||||||
.service
|
|
||||||
.poll_ready(cx)
|
|
||||||
.map_err(|e| {
|
|
||||||
let e = e.into();
|
|
||||||
log::error!("Http service readiness error: {:?}", e);
|
|
||||||
DispatchError::Service(e)
|
|
||||||
})?
|
|
||||||
.is_ready()
|
|
||||||
&& ready;
|
|
||||||
|
|
||||||
let ready = if let Some(ref upg) = self.flow.upgrade {
|
|
||||||
upg.poll_ready(cx)
|
|
||||||
.map_err(|e| {
|
|
||||||
let e = e.into();
|
|
||||||
log::error!("Http service readiness error: {:?}", e);
|
|
||||||
DispatchError::Service(e)
|
|
||||||
})?
|
|
||||||
.is_ready()
|
|
||||||
&& ready
|
|
||||||
} else {
|
|
||||||
ready
|
|
||||||
};
|
|
||||||
|
|
||||||
if ready {
|
|
||||||
Poll::Ready(Ok(()))
|
|
||||||
} else {
|
|
||||||
Poll::Pending
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|
||||||
fn call(&self, (io, addr): (T, Option<net::SocketAddr>)) -> Self::Future {
|
fn call(&self, (io, addr): (T, Option<net::SocketAddr>)) -> Self::Future {
|
||||||
|
|
|
@ -11,8 +11,8 @@ use actix_service::{
|
||||||
ServiceFactory,
|
ServiceFactory,
|
||||||
};
|
};
|
||||||
use bytes::Bytes;
|
use bytes::Bytes;
|
||||||
use futures_core::ready;
|
use futures_core::{future::LocalBoxFuture, ready};
|
||||||
use futures_util::future::ok;
|
use futures_util::future::ready;
|
||||||
use h2::server::{handshake, Handshake};
|
use h2::server::{handshake, Handshake};
|
||||||
use log::error;
|
use log::error;
|
||||||
|
|
||||||
|
@ -65,6 +65,7 @@ where
|
||||||
impl<S, B> H2Service<TcpStream, S, B>
|
impl<S, B> H2Service<TcpStream, S, B>
|
||||||
where
|
where
|
||||||
S: ServiceFactory<Request, Config = ()>,
|
S: ServiceFactory<Request, Config = ()>,
|
||||||
|
S::Future: 'static,
|
||||||
S::Error: Into<Error> + 'static,
|
S::Error: Into<Error> + 'static,
|
||||||
S::Response: Into<Response<B>> + 'static,
|
S::Response: Into<Response<B>> + 'static,
|
||||||
<S::Service as Service<Request>>::Future: 'static,
|
<S::Service as Service<Request>>::Future: 'static,
|
||||||
|
@ -80,11 +81,11 @@ where
|
||||||
Error = DispatchError,
|
Error = DispatchError,
|
||||||
InitError = S::InitError,
|
InitError = S::InitError,
|
||||||
> {
|
> {
|
||||||
pipeline_factory(fn_factory(|| async {
|
pipeline_factory(fn_factory(|| {
|
||||||
Ok::<_, S::InitError>(fn_service(|io: TcpStream| {
|
ready(Ok::<_, S::InitError>(fn_service(|io: TcpStream| {
|
||||||
let peer_addr = io.peer_addr().ok();
|
let peer_addr = io.peer_addr().ok();
|
||||||
ok::<_, DispatchError>((io, peer_addr))
|
ready(Ok::<_, DispatchError>((io, peer_addr)))
|
||||||
}))
|
})))
|
||||||
}))
|
}))
|
||||||
.and_then(self)
|
.and_then(self)
|
||||||
}
|
}
|
||||||
|
@ -101,6 +102,7 @@ mod openssl {
|
||||||
impl<S, B> H2Service<TlsStream<TcpStream>, S, B>
|
impl<S, B> H2Service<TlsStream<TcpStream>, S, B>
|
||||||
where
|
where
|
||||||
S: ServiceFactory<Request, Config = ()>,
|
S: ServiceFactory<Request, Config = ()>,
|
||||||
|
S::Future: 'static,
|
||||||
S::Error: Into<Error> + 'static,
|
S::Error: Into<Error> + 'static,
|
||||||
S::Response: Into<Response<B>> + 'static,
|
S::Response: Into<Response<B>> + 'static,
|
||||||
<S::Service as Service<Request>>::Future: 'static,
|
<S::Service as Service<Request>>::Future: 'static,
|
||||||
|
@ -123,10 +125,12 @@ mod openssl {
|
||||||
.map_init_err(|_| panic!()),
|
.map_init_err(|_| panic!()),
|
||||||
)
|
)
|
||||||
.and_then(fn_factory(|| {
|
.and_then(fn_factory(|| {
|
||||||
ok::<_, S::InitError>(fn_service(|io: TlsStream<TcpStream>| {
|
ready(Ok::<_, S::InitError>(fn_service(
|
||||||
let peer_addr = io.get_ref().peer_addr().ok();
|
|io: TlsStream<TcpStream>| {
|
||||||
ok((io, peer_addr))
|
let peer_addr = io.get_ref().peer_addr().ok();
|
||||||
}))
|
ready(Ok((io, peer_addr)))
|
||||||
|
},
|
||||||
|
)))
|
||||||
}))
|
}))
|
||||||
.and_then(self.map_err(TlsError::Service))
|
.and_then(self.map_err(TlsError::Service))
|
||||||
}
|
}
|
||||||
|
@ -144,6 +148,7 @@ mod rustls {
|
||||||
impl<S, B> H2Service<TlsStream<TcpStream>, S, B>
|
impl<S, B> H2Service<TlsStream<TcpStream>, S, B>
|
||||||
where
|
where
|
||||||
S: ServiceFactory<Request, Config = ()>,
|
S: ServiceFactory<Request, Config = ()>,
|
||||||
|
S::Future: 'static,
|
||||||
S::Error: Into<Error> + 'static,
|
S::Error: Into<Error> + 'static,
|
||||||
S::Response: Into<Response<B>> + 'static,
|
S::Response: Into<Response<B>> + 'static,
|
||||||
<S::Service as Service<Request>>::Future: 'static,
|
<S::Service as Service<Request>>::Future: 'static,
|
||||||
|
@ -169,10 +174,12 @@ mod rustls {
|
||||||
.map_init_err(|_| panic!()),
|
.map_init_err(|_| panic!()),
|
||||||
)
|
)
|
||||||
.and_then(fn_factory(|| {
|
.and_then(fn_factory(|| {
|
||||||
ok::<_, S::InitError>(fn_service(|io: TlsStream<TcpStream>| {
|
ready(Ok::<_, S::InitError>(fn_service(
|
||||||
let peer_addr = io.get_ref().0.peer_addr().ok();
|
|io: TlsStream<TcpStream>| {
|
||||||
ok((io, peer_addr))
|
let peer_addr = io.get_ref().0.peer_addr().ok();
|
||||||
}))
|
ready(Ok((io, peer_addr)))
|
||||||
|
},
|
||||||
|
)))
|
||||||
}))
|
}))
|
||||||
.and_then(self.map_err(TlsError::Service))
|
.and_then(self.map_err(TlsError::Service))
|
||||||
}
|
}
|
||||||
|
@ -181,8 +188,9 @@ mod rustls {
|
||||||
|
|
||||||
impl<T, S, B> ServiceFactory<(T, Option<net::SocketAddr>)> for H2Service<T, S, B>
|
impl<T, S, B> ServiceFactory<(T, Option<net::SocketAddr>)> for H2Service<T, S, B>
|
||||||
where
|
where
|
||||||
T: AsyncRead + AsyncWrite + Unpin,
|
T: AsyncRead + AsyncWrite + Unpin + 'static,
|
||||||
S: ServiceFactory<Request, Config = ()>,
|
S: ServiceFactory<Request, Config = ()>,
|
||||||
|
S::Future: 'static,
|
||||||
S::Error: Into<Error> + 'static,
|
S::Error: Into<Error> + 'static,
|
||||||
S::Response: Into<Response<B>> + 'static,
|
S::Response: Into<Response<B>> + 'static,
|
||||||
<S::Service as Service<Request>>::Future: 'static,
|
<S::Service as Service<Request>>::Future: 'static,
|
||||||
|
@ -193,52 +201,16 @@ where
|
||||||
type Config = ();
|
type Config = ();
|
||||||
type Service = H2ServiceHandler<T, S::Service, B>;
|
type Service = H2ServiceHandler<T, S::Service, B>;
|
||||||
type InitError = S::InitError;
|
type InitError = S::InitError;
|
||||||
type Future = H2ServiceResponse<T, S, B>;
|
type Future = LocalBoxFuture<'static, Result<Self::Service, Self::InitError>>;
|
||||||
|
|
||||||
fn new_service(&self, _: ()) -> Self::Future {
|
fn new_service(&self, _: ()) -> Self::Future {
|
||||||
H2ServiceResponse {
|
let service = self.srv.new_service(());
|
||||||
fut: self.srv.new_service(()),
|
let cfg = self.cfg.clone();
|
||||||
cfg: Some(self.cfg.clone()),
|
let on_connect_ext = self.on_connect_ext.clone();
|
||||||
on_connect_ext: self.on_connect_ext.clone(),
|
|
||||||
_phantom: PhantomData,
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
#[doc(hidden)]
|
Box::pin(async move {
|
||||||
#[pin_project::pin_project]
|
let service = service.await?;
|
||||||
pub struct H2ServiceResponse<T, S, B>
|
Ok(H2ServiceHandler::new(cfg, on_connect_ext, service))
|
||||||
where
|
|
||||||
S: ServiceFactory<Request>,
|
|
||||||
{
|
|
||||||
#[pin]
|
|
||||||
fut: S::Future,
|
|
||||||
cfg: Option<ServiceConfig>,
|
|
||||||
on_connect_ext: Option<Rc<ConnectCallback<T>>>,
|
|
||||||
_phantom: PhantomData<B>,
|
|
||||||
}
|
|
||||||
|
|
||||||
impl<T, S, B> Future for H2ServiceResponse<T, S, B>
|
|
||||||
where
|
|
||||||
T: AsyncRead + AsyncWrite + Unpin,
|
|
||||||
S: ServiceFactory<Request, Config = ()>,
|
|
||||||
S::Error: Into<Error> + 'static,
|
|
||||||
S::Response: Into<Response<B>> + 'static,
|
|
||||||
<S::Service as Service<Request>>::Future: 'static,
|
|
||||||
B: MessageBody + 'static,
|
|
||||||
{
|
|
||||||
type Output = Result<H2ServiceHandler<T, S::Service, B>, S::InitError>;
|
|
||||||
|
|
||||||
fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
|
|
||||||
let this = self.as_mut().project();
|
|
||||||
|
|
||||||
this.fut.poll(cx).map_ok(|service| {
|
|
||||||
let this = self.as_mut().project();
|
|
||||||
H2ServiceHandler::new(
|
|
||||||
this.cfg.take().unwrap(),
|
|
||||||
this.on_connect_ext.clone(),
|
|
||||||
service,
|
|
||||||
)
|
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
|
@ -1,18 +1,20 @@
|
||||||
use std::fmt::{self, Display};
|
use std::{
|
||||||
use std::io::Write;
|
fmt,
|
||||||
use std::str::FromStr;
|
io::Write,
|
||||||
use std::time::{SystemTime, UNIX_EPOCH};
|
str::FromStr,
|
||||||
|
time::{SystemTime, UNIX_EPOCH},
|
||||||
|
};
|
||||||
|
|
||||||
use bytes::buf::BufMut;
|
use bytes::buf::BufMut;
|
||||||
use bytes::BytesMut;
|
use bytes::BytesMut;
|
||||||
use http::header::{HeaderValue, InvalidHeaderValue};
|
use http::header::{HeaderValue, InvalidHeaderValue};
|
||||||
use time::{offset, OffsetDateTime, PrimitiveDateTime};
|
use time::{OffsetDateTime, PrimitiveDateTime, UtcOffset};
|
||||||
|
|
||||||
use crate::error::ParseError;
|
use crate::error::ParseError;
|
||||||
use crate::header::IntoHeaderValue;
|
use crate::header::IntoHeaderValue;
|
||||||
use crate::time_parser;
|
use crate::time_parser;
|
||||||
|
|
||||||
/// A timestamp with HTTP formatting and parsing
|
/// A timestamp with HTTP formatting and parsing.
|
||||||
#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord)]
|
#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord)]
|
||||||
pub struct HttpDate(OffsetDateTime);
|
pub struct HttpDate(OffsetDateTime);
|
||||||
|
|
||||||
|
@ -27,18 +29,12 @@ impl FromStr for HttpDate {
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
impl Display for HttpDate {
|
impl fmt::Display for HttpDate {
|
||||||
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
|
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
|
||||||
fmt::Display::fmt(&self.0.format("%a, %d %b %Y %H:%M:%S GMT"), f)
|
fmt::Display::fmt(&self.0.format("%a, %d %b %Y %H:%M:%S GMT"), f)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
impl From<OffsetDateTime> for HttpDate {
|
|
||||||
fn from(dt: OffsetDateTime) -> HttpDate {
|
|
||||||
HttpDate(dt)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
impl From<SystemTime> for HttpDate {
|
impl From<SystemTime> for HttpDate {
|
||||||
fn from(sys: SystemTime) -> HttpDate {
|
fn from(sys: SystemTime) -> HttpDate {
|
||||||
HttpDate(PrimitiveDateTime::from(sys).assume_utc())
|
HttpDate(PrimitiveDateTime::from(sys).assume_utc())
|
||||||
|
@ -54,7 +50,7 @@ impl IntoHeaderValue for HttpDate {
|
||||||
wrt,
|
wrt,
|
||||||
"{}",
|
"{}",
|
||||||
self.0
|
self.0
|
||||||
.to_offset(offset!(UTC))
|
.to_offset(UtcOffset::UTC)
|
||||||
.format("%a, %d %b %Y %H:%M:%S GMT")
|
.format("%a, %d %b %Y %H:%M:%S GMT")
|
||||||
)
|
)
|
||||||
.unwrap();
|
.unwrap();
|
||||||
|
|
|
@ -12,7 +12,7 @@ use actix_codec::{AsyncRead, AsyncWrite, Framed};
|
||||||
use actix_rt::net::TcpStream;
|
use actix_rt::net::TcpStream;
|
||||||
use actix_service::{pipeline_factory, IntoServiceFactory, Service, ServiceFactory};
|
use actix_service::{pipeline_factory, IntoServiceFactory, Service, ServiceFactory};
|
||||||
use bytes::Bytes;
|
use bytes::Bytes;
|
||||||
use futures_core::ready;
|
use futures_core::{future::LocalBoxFuture, ready};
|
||||||
use h2::server::{handshake, Handshake};
|
use h2::server::{handshake, Handshake};
|
||||||
use pin_project::pin_project;
|
use pin_project::pin_project;
|
||||||
|
|
||||||
|
@ -107,7 +107,6 @@ where
|
||||||
X1: ServiceFactory<Request, Config = (), Response = Request>,
|
X1: ServiceFactory<Request, Config = (), Response = Request>,
|
||||||
X1::Error: Into<Error>,
|
X1::Error: Into<Error>,
|
||||||
X1::InitError: fmt::Debug,
|
X1::InitError: fmt::Debug,
|
||||||
<X1::Service as Service<Request>>::Future: 'static,
|
|
||||||
{
|
{
|
||||||
HttpService {
|
HttpService {
|
||||||
expect,
|
expect,
|
||||||
|
@ -128,7 +127,6 @@ where
|
||||||
U1: ServiceFactory<(Request, Framed<T, h1::Codec>), Config = (), Response = ()>,
|
U1: ServiceFactory<(Request, Framed<T, h1::Codec>), Config = (), Response = ()>,
|
||||||
U1::Error: fmt::Display,
|
U1::Error: fmt::Display,
|
||||||
U1::InitError: fmt::Debug,
|
U1::InitError: fmt::Debug,
|
||||||
<U1::Service as Service<(Request, Framed<T, h1::Codec>)>>::Future: 'static,
|
|
||||||
{
|
{
|
||||||
HttpService {
|
HttpService {
|
||||||
upgrade,
|
upgrade,
|
||||||
|
@ -150,23 +148,24 @@ where
|
||||||
impl<S, B, X, U> HttpService<TcpStream, S, B, X, U>
|
impl<S, B, X, U> HttpService<TcpStream, S, B, X, U>
|
||||||
where
|
where
|
||||||
S: ServiceFactory<Request, Config = ()>,
|
S: ServiceFactory<Request, Config = ()>,
|
||||||
|
S::Future: 'static,
|
||||||
S::Error: Into<Error> + 'static,
|
S::Error: Into<Error> + 'static,
|
||||||
S::InitError: fmt::Debug,
|
S::InitError: fmt::Debug,
|
||||||
S::Response: Into<Response<B>> + 'static,
|
S::Response: Into<Response<B>> + 'static,
|
||||||
<S::Service as Service<Request>>::Future: 'static,
|
<S::Service as Service<Request>>::Future: 'static,
|
||||||
B: MessageBody + 'static,
|
B: MessageBody + 'static,
|
||||||
X: ServiceFactory<Request, Config = (), Response = Request>,
|
X: ServiceFactory<Request, Config = (), Response = Request>,
|
||||||
|
X::Future: 'static,
|
||||||
X::Error: Into<Error>,
|
X::Error: Into<Error>,
|
||||||
X::InitError: fmt::Debug,
|
X::InitError: fmt::Debug,
|
||||||
<X::Service as Service<Request>>::Future: 'static,
|
|
||||||
U: ServiceFactory<
|
U: ServiceFactory<
|
||||||
(Request, Framed<TcpStream, h1::Codec>),
|
(Request, Framed<TcpStream, h1::Codec>),
|
||||||
Config = (),
|
Config = (),
|
||||||
Response = (),
|
Response = (),
|
||||||
>,
|
>,
|
||||||
|
U::Future: 'static,
|
||||||
U::Error: fmt::Display + Into<Error>,
|
U::Error: fmt::Display + Into<Error>,
|
||||||
U::InitError: fmt::Debug,
|
U::InitError: fmt::Debug,
|
||||||
<U::Service as Service<(Request, Framed<TcpStream, h1::Codec>)>>::Future: 'static,
|
|
||||||
{
|
{
|
||||||
/// Create simple tcp stream service
|
/// Create simple tcp stream service
|
||||||
pub fn tcp(
|
pub fn tcp(
|
||||||
|
@ -196,23 +195,24 @@ mod openssl {
|
||||||
impl<S, B, X, U> HttpService<TlsStream<TcpStream>, S, B, X, U>
|
impl<S, B, X, U> HttpService<TlsStream<TcpStream>, S, B, X, U>
|
||||||
where
|
where
|
||||||
S: ServiceFactory<Request, Config = ()>,
|
S: ServiceFactory<Request, Config = ()>,
|
||||||
|
S::Future: 'static,
|
||||||
S::Error: Into<Error> + 'static,
|
S::Error: Into<Error> + 'static,
|
||||||
S::InitError: fmt::Debug,
|
S::InitError: fmt::Debug,
|
||||||
S::Response: Into<Response<B>> + 'static,
|
S::Response: Into<Response<B>> + 'static,
|
||||||
<S::Service as Service<Request>>::Future: 'static,
|
<S::Service as Service<Request>>::Future: 'static,
|
||||||
B: MessageBody + 'static,
|
B: MessageBody + 'static,
|
||||||
X: ServiceFactory<Request, Config = (), Response = Request>,
|
X: ServiceFactory<Request, Config = (), Response = Request>,
|
||||||
|
X::Future: 'static,
|
||||||
X::Error: Into<Error>,
|
X::Error: Into<Error>,
|
||||||
X::InitError: fmt::Debug,
|
X::InitError: fmt::Debug,
|
||||||
<X::Service as Service<Request>>::Future: 'static,
|
|
||||||
U: ServiceFactory<
|
U: ServiceFactory<
|
||||||
(Request, Framed<TlsStream<TcpStream>, h1::Codec>),
|
(Request, Framed<TlsStream<TcpStream>, h1::Codec>),
|
||||||
Config = (),
|
Config = (),
|
||||||
Response = (),
|
Response = (),
|
||||||
>,
|
>,
|
||||||
|
U::Future: 'static,
|
||||||
U::Error: fmt::Display + Into<Error>,
|
U::Error: fmt::Display + Into<Error>,
|
||||||
U::InitError: fmt::Debug,
|
U::InitError: fmt::Debug,
|
||||||
<U::Service as Service<(Request, Framed<TlsStream<TcpStream>, h1::Codec>)>>::Future: 'static,
|
|
||||||
{
|
{
|
||||||
/// Create openssl based service
|
/// Create openssl based service
|
||||||
pub fn openssl(
|
pub fn openssl(
|
||||||
|
@ -261,23 +261,24 @@ mod rustls {
|
||||||
impl<S, B, X, U> HttpService<TlsStream<TcpStream>, S, B, X, U>
|
impl<S, B, X, U> HttpService<TlsStream<TcpStream>, S, B, X, U>
|
||||||
where
|
where
|
||||||
S: ServiceFactory<Request, Config = ()>,
|
S: ServiceFactory<Request, Config = ()>,
|
||||||
|
S::Future: 'static,
|
||||||
S::Error: Into<Error> + 'static,
|
S::Error: Into<Error> + 'static,
|
||||||
S::InitError: fmt::Debug,
|
S::InitError: fmt::Debug,
|
||||||
S::Response: Into<Response<B>> + 'static,
|
S::Response: Into<Response<B>> + 'static,
|
||||||
<S::Service as Service<Request>>::Future: 'static,
|
<S::Service as Service<Request>>::Future: 'static,
|
||||||
B: MessageBody + 'static,
|
B: MessageBody + 'static,
|
||||||
X: ServiceFactory<Request, Config = (), Response = Request>,
|
X: ServiceFactory<Request, Config = (), Response = Request>,
|
||||||
|
X::Future: 'static,
|
||||||
X::Error: Into<Error>,
|
X::Error: Into<Error>,
|
||||||
X::InitError: fmt::Debug,
|
X::InitError: fmt::Debug,
|
||||||
<X::Service as Service<Request>>::Future: 'static,
|
|
||||||
U: ServiceFactory<
|
U: ServiceFactory<
|
||||||
(Request, Framed<TlsStream<TcpStream>, h1::Codec>),
|
(Request, Framed<TlsStream<TcpStream>, h1::Codec>),
|
||||||
Config = (),
|
Config = (),
|
||||||
Response = (),
|
Response = (),
|
||||||
>,
|
>,
|
||||||
|
U::Future: 'static,
|
||||||
U::Error: fmt::Display + Into<Error>,
|
U::Error: fmt::Display + Into<Error>,
|
||||||
U::InitError: fmt::Debug,
|
U::InitError: fmt::Debug,
|
||||||
<U::Service as Service<(Request, Framed<TlsStream<TcpStream>, h1::Codec>)>>::Future: 'static,
|
|
||||||
{
|
{
|
||||||
/// Create openssl based service
|
/// Create openssl based service
|
||||||
pub fn rustls(
|
pub fn rustls(
|
||||||
|
@ -319,137 +320,117 @@ mod rustls {
|
||||||
impl<T, S, B, X, U> ServiceFactory<(T, Protocol, Option<net::SocketAddr>)>
|
impl<T, S, B, X, U> ServiceFactory<(T, Protocol, Option<net::SocketAddr>)>
|
||||||
for HttpService<T, S, B, X, U>
|
for HttpService<T, S, B, X, U>
|
||||||
where
|
where
|
||||||
T: AsyncRead + AsyncWrite + Unpin,
|
T: AsyncRead + AsyncWrite + Unpin + 'static,
|
||||||
S: ServiceFactory<Request, Config = ()>,
|
S: ServiceFactory<Request, Config = ()>,
|
||||||
|
S::Future: 'static,
|
||||||
S::Error: Into<Error> + 'static,
|
S::Error: Into<Error> + 'static,
|
||||||
S::InitError: fmt::Debug,
|
S::InitError: fmt::Debug,
|
||||||
S::Response: Into<Response<B>> + 'static,
|
S::Response: Into<Response<B>> + 'static,
|
||||||
<S::Service as Service<Request>>::Future: 'static,
|
<S::Service as Service<Request>>::Future: 'static,
|
||||||
B: MessageBody + 'static,
|
B: MessageBody + 'static,
|
||||||
X: ServiceFactory<Request, Config = (), Response = Request>,
|
X: ServiceFactory<Request, Config = (), Response = Request>,
|
||||||
|
X::Future: 'static,
|
||||||
X::Error: Into<Error>,
|
X::Error: Into<Error>,
|
||||||
X::InitError: fmt::Debug,
|
X::InitError: fmt::Debug,
|
||||||
<X::Service as Service<Request>>::Future: 'static,
|
|
||||||
U: ServiceFactory<(Request, Framed<T, h1::Codec>), Config = (), Response = ()>,
|
U: ServiceFactory<(Request, Framed<T, h1::Codec>), Config = (), Response = ()>,
|
||||||
|
U::Future: 'static,
|
||||||
U::Error: fmt::Display + Into<Error>,
|
U::Error: fmt::Display + Into<Error>,
|
||||||
U::InitError: fmt::Debug,
|
U::InitError: fmt::Debug,
|
||||||
<U::Service as Service<(Request, Framed<T, h1::Codec>)>>::Future: 'static,
|
|
||||||
{
|
{
|
||||||
type Response = ();
|
type Response = ();
|
||||||
type Error = DispatchError;
|
type Error = DispatchError;
|
||||||
type Config = ();
|
type Config = ();
|
||||||
type Service = HttpServiceHandler<T, S::Service, B, X::Service, U::Service>;
|
type Service = HttpServiceHandler<T, S::Service, B, X::Service, U::Service>;
|
||||||
type InitError = ();
|
type InitError = ();
|
||||||
type Future = HttpServiceResponse<T, S, B, X, U>;
|
type Future = LocalBoxFuture<'static, Result<Self::Service, Self::InitError>>;
|
||||||
|
|
||||||
fn new_service(&self, _: ()) -> Self::Future {
|
fn new_service(&self, _: ()) -> Self::Future {
|
||||||
HttpServiceResponse {
|
let service = self.srv.new_service(());
|
||||||
fut: self.srv.new_service(()),
|
let expect = self.expect.new_service(());
|
||||||
fut_ex: Some(self.expect.new_service(())),
|
let upgrade = self.upgrade.as_ref().map(|s| s.new_service(()));
|
||||||
fut_upg: self.upgrade.as_ref().map(|f| f.new_service(())),
|
let on_connect_ext = self.on_connect_ext.clone();
|
||||||
expect: None,
|
let cfg = self.cfg.clone();
|
||||||
upgrade: None,
|
|
||||||
on_connect_ext: self.on_connect_ext.clone(),
|
|
||||||
cfg: self.cfg.clone(),
|
|
||||||
_phantom: PhantomData,
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
#[doc(hidden)]
|
Box::pin(async move {
|
||||||
#[pin_project]
|
let expect = expect
|
||||||
pub struct HttpServiceResponse<T, S, B, X, U>
|
.await
|
||||||
where
|
.map_err(|e| log::error!("Init http expect service error: {:?}", e))?;
|
||||||
S: ServiceFactory<Request>,
|
|
||||||
X: ServiceFactory<Request>,
|
|
||||||
U: ServiceFactory<(Request, Framed<T, h1::Codec>)>,
|
|
||||||
{
|
|
||||||
#[pin]
|
|
||||||
fut: S::Future,
|
|
||||||
#[pin]
|
|
||||||
fut_ex: Option<X::Future>,
|
|
||||||
#[pin]
|
|
||||||
fut_upg: Option<U::Future>,
|
|
||||||
expect: Option<X::Service>,
|
|
||||||
upgrade: Option<U::Service>,
|
|
||||||
on_connect_ext: Option<Rc<ConnectCallback<T>>>,
|
|
||||||
cfg: ServiceConfig,
|
|
||||||
_phantom: PhantomData<B>,
|
|
||||||
}
|
|
||||||
|
|
||||||
impl<T, S, B, X, U> Future for HttpServiceResponse<T, S, B, X, U>
|
let upgrade = match upgrade {
|
||||||
where
|
Some(upgrade) => {
|
||||||
T: AsyncRead + AsyncWrite + Unpin,
|
let upgrade = upgrade.await.map_err(|e| {
|
||||||
S: ServiceFactory<Request>,
|
log::error!("Init http upgrade service error: {:?}", e)
|
||||||
S::Error: Into<Error> + 'static,
|
})?;
|
||||||
S::InitError: fmt::Debug,
|
Some(upgrade)
|
||||||
S::Response: Into<Response<B>> + 'static,
|
}
|
||||||
<S::Service as Service<Request>>::Future: 'static,
|
None => None,
|
||||||
B: MessageBody + 'static,
|
};
|
||||||
X: ServiceFactory<Request, Response = Request>,
|
|
||||||
X::Error: Into<Error>,
|
|
||||||
X::InitError: fmt::Debug,
|
|
||||||
<X::Service as Service<Request>>::Future: 'static,
|
|
||||||
U: ServiceFactory<(Request, Framed<T, h1::Codec>), Response = ()>,
|
|
||||||
U::Error: fmt::Display,
|
|
||||||
U::InitError: fmt::Debug,
|
|
||||||
<U::Service as Service<(Request, Framed<T, h1::Codec>)>>::Future: 'static,
|
|
||||||
{
|
|
||||||
type Output =
|
|
||||||
Result<HttpServiceHandler<T, S::Service, B, X::Service, U::Service>, ()>;
|
|
||||||
|
|
||||||
fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
|
let service = service
|
||||||
let mut this = self.as_mut().project();
|
.await
|
||||||
|
.map_err(|e| log::error!("Init http service error: {:?}", e))?;
|
||||||
|
|
||||||
if let Some(fut) = this.fut_ex.as_pin_mut() {
|
Ok(HttpServiceHandler::new(
|
||||||
let expect = ready!(fut
|
cfg,
|
||||||
.poll(cx)
|
|
||||||
.map_err(|e| log::error!("Init http service error: {:?}", e)))?;
|
|
||||||
this = self.as_mut().project();
|
|
||||||
*this.expect = Some(expect);
|
|
||||||
this.fut_ex.set(None);
|
|
||||||
}
|
|
||||||
|
|
||||||
if let Some(fut) = this.fut_upg.as_pin_mut() {
|
|
||||||
let upgrade = ready!(fut
|
|
||||||
.poll(cx)
|
|
||||||
.map_err(|e| log::error!("Init http service error: {:?}", e)))?;
|
|
||||||
this = self.as_mut().project();
|
|
||||||
*this.upgrade = Some(upgrade);
|
|
||||||
this.fut_upg.set(None);
|
|
||||||
}
|
|
||||||
|
|
||||||
let result = ready!(this
|
|
||||||
.fut
|
|
||||||
.poll(cx)
|
|
||||||
.map_err(|e| log::error!("Init http service error: {:?}", e)));
|
|
||||||
|
|
||||||
Poll::Ready(result.map(|service| {
|
|
||||||
let this = self.as_mut().project();
|
|
||||||
HttpServiceHandler::new(
|
|
||||||
this.cfg.clone(),
|
|
||||||
service,
|
service,
|
||||||
this.expect.take().unwrap(),
|
expect,
|
||||||
this.upgrade.take(),
|
upgrade,
|
||||||
this.on_connect_ext.clone(),
|
on_connect_ext,
|
||||||
)
|
))
|
||||||
}))
|
})
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
/// `Service` implementation for HTTP transport
|
/// `Service` implementation for HTTP/1 and HTTP/2 transport
|
||||||
pub struct HttpServiceHandler<T, S, B, X, U>
|
pub struct HttpServiceHandler<T, S, B, X, U>
|
||||||
where
|
where
|
||||||
S: Service<Request>,
|
S: Service<Request>,
|
||||||
X: Service<Request>,
|
X: Service<Request>,
|
||||||
U: Service<(Request, Framed<T, h1::Codec>)>,
|
U: Service<(Request, Framed<T, h1::Codec>)>,
|
||||||
{
|
{
|
||||||
flow: Rc<HttpFlow<S, X, U>>,
|
pub(super) flow: Rc<HttpFlow<S, X, U>>,
|
||||||
cfg: ServiceConfig,
|
pub(super) cfg: ServiceConfig,
|
||||||
on_connect_ext: Option<Rc<ConnectCallback<T>>>,
|
pub(super) on_connect_ext: Option<Rc<ConnectCallback<T>>>,
|
||||||
_phantom: PhantomData<B>,
|
_phantom: PhantomData<B>,
|
||||||
}
|
}
|
||||||
|
|
||||||
|
impl<T, S, B, X, U> HttpServiceHandler<T, S, B, X, U>
|
||||||
|
where
|
||||||
|
S: Service<Request>,
|
||||||
|
S::Error: Into<Error>,
|
||||||
|
X: Service<Request>,
|
||||||
|
X::Error: Into<Error>,
|
||||||
|
U: Service<(Request, Framed<T, h1::Codec>)>,
|
||||||
|
U::Error: Into<Error>,
|
||||||
|
{
|
||||||
|
pub(super) fn new(
|
||||||
|
cfg: ServiceConfig,
|
||||||
|
service: S,
|
||||||
|
expect: X,
|
||||||
|
upgrade: Option<U>,
|
||||||
|
on_connect_ext: Option<Rc<ConnectCallback<T>>>,
|
||||||
|
) -> HttpServiceHandler<T, S, B, X, U> {
|
||||||
|
HttpServiceHandler {
|
||||||
|
cfg,
|
||||||
|
on_connect_ext,
|
||||||
|
flow: HttpFlow::new(service, expect, upgrade),
|
||||||
|
_phantom: PhantomData,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
pub(super) fn _poll_ready(&self, cx: &mut Context<'_>) -> Poll<Result<(), Error>> {
|
||||||
|
ready!(self.flow.expect.poll_ready(cx).map_err(Into::into))?;
|
||||||
|
|
||||||
|
ready!(self.flow.service.poll_ready(cx).map_err(Into::into))?;
|
||||||
|
|
||||||
|
if let Some(ref upg) = self.flow.upgrade {
|
||||||
|
ready!(upg.poll_ready(cx).map_err(Into::into))?;
|
||||||
|
};
|
||||||
|
|
||||||
|
Poll::Ready(Ok(()))
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
/// A collection of services that describe an HTTP request flow.
|
/// A collection of services that describe an HTTP request flow.
|
||||||
pub(super) struct HttpFlow<S, X, U> {
|
pub(super) struct HttpFlow<S, X, U> {
|
||||||
pub(super) service: S,
|
pub(super) service: S,
|
||||||
|
@ -467,34 +448,6 @@ impl<S, X, U> HttpFlow<S, X, U> {
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
impl<T, S, B, X, U> HttpServiceHandler<T, S, B, X, U>
|
|
||||||
where
|
|
||||||
S: Service<Request>,
|
|
||||||
S::Error: Into<Error> + 'static,
|
|
||||||
S::Future: 'static,
|
|
||||||
S::Response: Into<Response<B>> + 'static,
|
|
||||||
B: MessageBody + 'static,
|
|
||||||
X: Service<Request, Response = Request>,
|
|
||||||
X::Error: Into<Error>,
|
|
||||||
U: Service<(Request, Framed<T, h1::Codec>), Response = ()>,
|
|
||||||
U::Error: fmt::Display,
|
|
||||||
{
|
|
||||||
fn new(
|
|
||||||
cfg: ServiceConfig,
|
|
||||||
service: S,
|
|
||||||
expect: X,
|
|
||||||
upgrade: Option<U>,
|
|
||||||
on_connect_ext: Option<Rc<ConnectCallback<T>>>,
|
|
||||||
) -> HttpServiceHandler<T, S, B, X, U> {
|
|
||||||
HttpServiceHandler {
|
|
||||||
cfg,
|
|
||||||
on_connect_ext,
|
|
||||||
flow: HttpFlow::new(service, expect, upgrade),
|
|
||||||
_phantom: PhantomData,
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
impl<T, S, B, X, U> Service<(T, Protocol, Option<net::SocketAddr>)>
|
impl<T, S, B, X, U> Service<(T, Protocol, Option<net::SocketAddr>)>
|
||||||
for HttpServiceHandler<T, S, B, X, U>
|
for HttpServiceHandler<T, S, B, X, U>
|
||||||
where
|
where
|
||||||
|
@ -514,47 +467,10 @@ where
|
||||||
type Future = HttpServiceHandlerResponse<T, S, B, X, U>;
|
type Future = HttpServiceHandlerResponse<T, S, B, X, U>;
|
||||||
|
|
||||||
fn poll_ready(&self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
|
fn poll_ready(&self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
|
||||||
let ready = self
|
self._poll_ready(cx).map_err(|e| {
|
||||||
.flow
|
log::error!("HTTP service readiness error: {:?}", e);
|
||||||
.expect
|
DispatchError::Service(e)
|
||||||
.poll_ready(cx)
|
})
|
||||||
.map_err(|e| {
|
|
||||||
let e = e.into();
|
|
||||||
log::error!("Http service readiness error: {:?}", e);
|
|
||||||
DispatchError::Service(e)
|
|
||||||
})?
|
|
||||||
.is_ready();
|
|
||||||
|
|
||||||
let ready = self
|
|
||||||
.flow
|
|
||||||
.service
|
|
||||||
.poll_ready(cx)
|
|
||||||
.map_err(|e| {
|
|
||||||
let e = e.into();
|
|
||||||
log::error!("Http service readiness error: {:?}", e);
|
|
||||||
DispatchError::Service(e)
|
|
||||||
})?
|
|
||||||
.is_ready()
|
|
||||||
&& ready;
|
|
||||||
|
|
||||||
let ready = if let Some(ref upg) = self.flow.upgrade {
|
|
||||||
upg.poll_ready(cx)
|
|
||||||
.map_err(|e| {
|
|
||||||
let e = e.into();
|
|
||||||
log::error!("Http service readiness error: {:?}", e);
|
|
||||||
DispatchError::Service(e)
|
|
||||||
})?
|
|
||||||
.is_ready()
|
|
||||||
&& ready
|
|
||||||
} else {
|
|
||||||
ready
|
|
||||||
};
|
|
||||||
|
|
||||||
if ready {
|
|
||||||
Poll::Ready(Ok(()))
|
|
||||||
} else {
|
|
||||||
Poll::Pending
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|
||||||
fn call(
|
fn call(
|
||||||
|
|
|
@ -3,6 +3,7 @@ use std::future::Future;
|
||||||
use std::marker::PhantomData;
|
use std::marker::PhantomData;
|
||||||
use std::pin::Pin;
|
use std::pin::Pin;
|
||||||
use std::sync::{Arc, Mutex};
|
use std::sync::{Arc, Mutex};
|
||||||
|
use std::task::{Context, Poll};
|
||||||
|
|
||||||
use actix_codec::{AsyncRead, AsyncWrite, Framed};
|
use actix_codec::{AsyncRead, AsyncWrite, Framed};
|
||||||
use actix_http::{body, h1, ws, Error, HttpService, Request, Response};
|
use actix_http::{body, h1, ws, Error, HttpService, Request, Response};
|
||||||
|
@ -11,7 +12,6 @@ use actix_service::{fn_factory, Service};
|
||||||
use actix_utils::dispatcher::Dispatcher;
|
use actix_utils::dispatcher::Dispatcher;
|
||||||
use bytes::Bytes;
|
use bytes::Bytes;
|
||||||
use futures_util::future;
|
use futures_util::future;
|
||||||
use futures_util::task::{Context, Poll};
|
|
||||||
use futures_util::{SinkExt as _, StreamExt as _};
|
use futures_util::{SinkExt as _, StreamExt as _};
|
||||||
|
|
||||||
struct WsService<T>(Arc<Mutex<(PhantomData<T>, Cell<bool>)>>);
|
struct WsService<T>(Arc<Mutex<(PhantomData<T>, Cell<bool>)>>);
|
||||||
|
|
|
@ -28,7 +28,7 @@ mime = "0.3"
|
||||||
twoway = "0.2"
|
twoway = "0.2"
|
||||||
|
|
||||||
[dev-dependencies]
|
[dev-dependencies]
|
||||||
actix-rt = "2.1"
|
actix-rt = "2.2"
|
||||||
actix-http = "3.0.0-beta.4"
|
actix-http = "3.0.0-beta.4"
|
||||||
tokio = { version = "1", features = ["sync"] }
|
tokio = { version = "1", features = ["sync"] }
|
||||||
tokio-stream = "0.1"
|
tokio-stream = "0.1"
|
||||||
|
|
|
@ -28,7 +28,7 @@ pin-project = "1.0.0"
|
||||||
tokio = { version = "1", features = ["sync"] }
|
tokio = { version = "1", features = ["sync"] }
|
||||||
|
|
||||||
[dev-dependencies]
|
[dev-dependencies]
|
||||||
actix-rt = "2.1"
|
actix-rt = "2.2"
|
||||||
awc = { version = "3.0.0-beta.3", default-features = false }
|
awc = { version = "3.0.0-beta.3", default-features = false }
|
||||||
env_logger = "0.8"
|
env_logger = "0.8"
|
||||||
futures-util = { version = "0.3.7", default-features = false }
|
futures-util = { version = "0.3.7", default-features = false }
|
||||||
|
|
|
@ -19,7 +19,7 @@ syn = { version = "1", features = ["full", "parsing"] }
|
||||||
proc-macro2 = "1"
|
proc-macro2 = "1"
|
||||||
|
|
||||||
[dev-dependencies]
|
[dev-dependencies]
|
||||||
actix-rt = "2.1"
|
actix-rt = "2.2"
|
||||||
actix-web = "4.0.0-beta.4"
|
actix-web = "4.0.0-beta.4"
|
||||||
futures-util = { version = "0.3.7", default-features = false }
|
futures-util = { version = "0.3.7", default-features = false }
|
||||||
trybuild = "1"
|
trybuild = "1"
|
||||||
|
|
|
@ -71,7 +71,7 @@ actix-http = { version = "3.0.0-beta.4", features = ["openssl"] }
|
||||||
actix-http-test = { version = "3.0.0-beta.3", features = ["openssl"] }
|
actix-http-test = { version = "3.0.0-beta.3", features = ["openssl"] }
|
||||||
actix-utils = "3.0.0-beta.1"
|
actix-utils = "3.0.0-beta.1"
|
||||||
actix-server = "2.0.0-beta.3"
|
actix-server = "2.0.0-beta.3"
|
||||||
actix-tls = { version = "3.0.0-beta.4", features = ["openssl", "rustls"] }
|
actix-tls = { version = "3.0.0-beta.5", features = ["openssl", "rustls"] }
|
||||||
|
|
||||||
brotli2 = "0.3.2"
|
brotli2 = "0.3.2"
|
||||||
env_logger = "0.8"
|
env_logger = "0.8"
|
||||||
|
|
|
@ -4,12 +4,11 @@ use std::net::IpAddr;
|
||||||
use std::rc::Rc;
|
use std::rc::Rc;
|
||||||
use std::time::Duration;
|
use std::time::Duration;
|
||||||
|
|
||||||
use actix_codec::{AsyncRead, AsyncWrite};
|
|
||||||
use actix_http::{
|
use actix_http::{
|
||||||
client::{Connector, ConnectorService, TcpConnect, TcpConnectError, TcpConnection},
|
client::{Connector, ConnectorService, TcpConnect, TcpConnectError, TcpConnection},
|
||||||
http::{self, header, Error as HttpError, HeaderMap, HeaderName, Uri},
|
http::{self, header, Error as HttpError, HeaderMap, HeaderName, Uri},
|
||||||
};
|
};
|
||||||
use actix_rt::net::TcpStream;
|
use actix_rt::net::{ActixStream, TcpStream};
|
||||||
use actix_service::{boxed, Service};
|
use actix_service::{boxed, Service};
|
||||||
|
|
||||||
use crate::connect::DefaultConnector;
|
use crate::connect::DefaultConnector;
|
||||||
|
@ -64,7 +63,7 @@ where
|
||||||
S: Service<TcpConnect<Uri>, Response = TcpConnection<Uri, Io>, Error = TcpConnectError>
|
S: Service<TcpConnect<Uri>, Response = TcpConnection<Uri, Io>, Error = TcpConnectError>
|
||||||
+ Clone
|
+ Clone
|
||||||
+ 'static,
|
+ 'static,
|
||||||
Io: AsyncRead + AsyncWrite + Unpin + fmt::Debug + 'static,
|
Io: ActixStream + fmt::Debug + 'static,
|
||||||
{
|
{
|
||||||
/// Use custom connector service.
|
/// Use custom connector service.
|
||||||
pub fn connector<S1, Io1>(self, connector: Connector<S1>) -> ClientBuilder<S1, M>
|
pub fn connector<S1, Io1>(self, connector: Connector<S1>) -> ClientBuilder<S1, M>
|
||||||
|
@ -75,7 +74,7 @@ where
|
||||||
Error = TcpConnectError,
|
Error = TcpConnectError,
|
||||||
> + Clone
|
> + Clone
|
||||||
+ 'static,
|
+ 'static,
|
||||||
Io1: AsyncRead + AsyncWrite + Unpin + fmt::Debug + 'static,
|
Io1: ActixStream + fmt::Debug + 'static,
|
||||||
{
|
{
|
||||||
ClientBuilder {
|
ClientBuilder {
|
||||||
middleware: self.middleware,
|
middleware: self.middleware,
|
||||||
|
|
|
@ -1,21 +1,19 @@
|
||||||
digraph {
|
digraph {
|
||||||
subgraph cluster_net {
|
subgraph cluster_net {
|
||||||
label="actix/actix-net";
|
label="actix/actix-net";
|
||||||
"actix-codec"
|
"actix-codec" "actix-macros" "actix-rt" "actix-server" "actix-service"
|
||||||
"actix-macros"
|
"actix-tls" "actix-tracing" "actix-utils" "actix-router"
|
||||||
"actix-rt"
|
"local-channel" "local-waker"
|
||||||
"actix-server"
|
|
||||||
"actix-service"
|
|
||||||
"actix-threadpool"
|
|
||||||
"actix-tls"
|
|
||||||
"actix-tracing"
|
|
||||||
"actix-utils"
|
|
||||||
"actix-router"
|
|
||||||
}
|
}
|
||||||
|
|
||||||
"actix-utils" -> { "actix-service" "actix-rt" "actix-codec" }
|
"actix-codec" -> { "actix-rt" "actix-service" "local-channel" "tokio" }
|
||||||
|
"actix-utils" -> { "local-waker" }
|
||||||
"actix-tracing" -> { "actix-service" }
|
"actix-tracing" -> { "actix-service" }
|
||||||
"actix-tls" -> { "actix-service" "actix-codec" "actix-utils" "actix-rt" }
|
"actix-tls" -> { "actix-service" "actix-codec" "actix-utils" "actix-rt" }
|
||||||
"actix-server" -> { "actix-service" "actix-rt" "actix-codec" "actix-utils" }
|
"actix-server" -> { "actix-service" "actix-rt" "actix-codec" "actix-utils" "tokio" }
|
||||||
"actix-rt" -> { "actix-macros" "actix-threadpool" }
|
"actix-rt" -> { "actix-macros" "tokio" }
|
||||||
|
|
||||||
|
"local-channel" -> { "local-waker" }
|
||||||
|
|
||||||
|
"tokio" [fontcolor = darkgreen]
|
||||||
}
|
}
|
||||||
|
|
Loading…
Reference in New Issue