actix/actix-web
1
0
Fork 0
mirror of https://github.com/fafhrd91/actix-web synced 2025-06-25 14:49:20 +02:00
Code Issues Releases Wiki Activity

Tokio 1.0 (#1813)

Browse Source 
Co-authored-by: Rob Ede <robjtede@icloud.com>
This commit is contained in:
fakeshadow
2021-01-04 07:47:04 +08:00
committed by GitHub
parent 1f202d40e4
commit 32de9f8840
75 changed files with 788 additions and 826 deletions
Download Patch File Download Diff File Expand all files Collapse all files

48
actix-http/src/builder.rs
View File

@ -19,7 +19,7 @@ use crate::{ConnectCallback, Extensions};
///
/// This type can be used to construct an instance of [`HttpService`] through a
/// builder-like pattern.
pub struct HttpServiceBuilder<T, S, X = ExpectHandler, U = UpgradeHandler<T>> {
pub struct HttpServiceBuilder<T, S, X = ExpectHandler, U = UpgradeHandler> {
keep_alive: KeepAlive,
client_timeout: u64,
client_disconnect: u64,
@ -28,15 +28,15 @@ pub struct HttpServiceBuilder<T, S, X = ExpectHandler, U = UpgradeHandler<T>> {
expect: X,
upgrade: Option<U>,
on_connect_ext: Option<Rc<ConnectCallback<T>>>,
_t: PhantomData<(T, S)>,
_t: PhantomData<S>,
}
impl<T, S> HttpServiceBuilder<T, S, ExpectHandler, UpgradeHandler<T>>
impl<T, S> HttpServiceBuilder<T, S, ExpectHandler, UpgradeHandler>
where
S: ServiceFactory<Config = (), Request = Request>,
S: ServiceFactory<Request, Config = ()>,
S::Error: Into<Error> + 'static,
S::InitError: fmt::Debug,
<S::Service as Service>::Future: 'static,
<S::Service as Service<Request>>::Future: 'static,
{
/// Create instance of `ServiceConfigBuilder`
pub fn new() -> Self {
@ -56,18 +56,18 @@ where
impl<T, S, X, U> HttpServiceBuilder<T, S, X, U>
where
S: ServiceFactory<Config = (), Request = Request>,
S: ServiceFactory<Request, Config = ()>,
S::Error: Into<Error> + 'static,
S::InitError: fmt::Debug,
<S::Service as Service>::Future: 'static,
X: ServiceFactory<Config = (), Request = Request, Response = Request>,
<S::Service as Service<Request>>::Future: 'static,
X: ServiceFactory<Request, Config = (), Response = Request>,
X::Error: Into<Error>,
X::InitError: fmt::Debug,
<X::Service as Service>::Future: 'static,
U: ServiceFactory<Config = (), Request = (Request, Framed<T, Codec>), Response = ()>,
<X::Service as Service<Request>>::Future: 'static,
U: ServiceFactory<(Request, Framed<T, Codec>), Config = (), Response = ()>,
U::Error: fmt::Display,
U::InitError: fmt::Debug,
<U::Service as Service>::Future: 'static,
<U::Service as Service<(Request, Framed<T, Codec>)>>::Future: 'static,
{
/// Set server keep-alive setting.
///
@ -123,11 +123,11 @@ where
/// request will be forwarded to main service.
pub fn expect<F, X1>(self, expect: F) -> HttpServiceBuilder<T, S, X1, U>
where
F: IntoServiceFactory<X1>,
X1: ServiceFactory<Config = (), Request = Request, Response = Request>,
F: IntoServiceFactory<X1, Request>,
X1: ServiceFactory<Request, Config = (), Response = Request>,
X1::Error: Into<Error>,
X1::InitError: fmt::Debug,
<X1::Service as Service>::Future: 'static,
<X1::Service as Service<Request>>::Future: 'static,
{
HttpServiceBuilder {
keep_alive: self.keep_alive,
@ -148,15 +148,11 @@ where
/// and this service get called with original request and framed object.
pub fn upgrade<F, U1>(self, upgrade: F) -> HttpServiceBuilder<T, S, X, U1>
where
F: IntoServiceFactory<U1>,
U1: ServiceFactory<
Config = (),
Request = (Request, Framed<T, Codec>),
Response = (),
>,
F: IntoServiceFactory<U1, (Request, Framed<T, Codec>)>,
U1: ServiceFactory<(Request, Framed<T, Codec>), Config = (), Response = ()>,
U1::Error: fmt::Display,
U1::InitError: fmt::Debug,
<U1::Service as Service>::Future: 'static,
<U1::Service as Service<(Request, Framed<T, Codec>)>>::Future: 'static,
{
HttpServiceBuilder {
keep_alive: self.keep_alive,
@ -188,7 +184,7 @@ where
pub fn h1<F, B>(self, service: F) -> H1Service<T, S, B, X, U>
where
B: MessageBody,
F: IntoServiceFactory<S>,
F: IntoServiceFactory<S, Request>,
S::Error: Into<Error>,
S::InitError: fmt::Debug,
S::Response: Into<Response<B>>,
@ -211,11 +207,11 @@ where
pub fn h2<F, B>(self, service: F) -> H2Service<T, S, B>
where
B: MessageBody + 'static,
F: IntoServiceFactory<S>,
F: IntoServiceFactory<S, Request>,
S::Error: Into<Error> + 'static,
S::InitError: fmt::Debug,
S::Response: Into<Response<B>> + 'static,
<S::Service as Service>::Future: 'static,
<S::Service as Service<Request>>::Future: 'static,
{
let cfg = ServiceConfig::new(
self.keep_alive,
@ -233,11 +229,11 @@ where
pub fn finish<F, B>(self, service: F) -> HttpService<T, S, B, X, U>
where
B: MessageBody + 'static,
F: IntoServiceFactory<S>,
F: IntoServiceFactory<S, Request>,
S::Error: Into<Error> + 'static,
S::InitError: fmt::Debug,
S::Response: Into<Response<B>> + 'static,
<S::Service as Service>::Future: 'static,
<S::Service as Service<Request>>::Future: 'static,
{
let cfg = ServiceConfig::new(
self.keep_alive,

34
actix-http/src/client/connection.rs
View File

@ -1,10 +1,10 @@
use std::future::Future;
use std::pin::Pin;
use std::task::{Context, Poll};
use std::{fmt, io, mem, time};
use std::{fmt, io, time};
use actix_codec::{AsyncRead, AsyncWrite, Framed};
use bytes::{Buf, Bytes};
use actix_codec::{AsyncRead, AsyncWrite, Framed, ReadBuf};
use bytes::Bytes;
use futures_util::future::{err, Either, FutureExt, LocalBoxFuture, Ready};
use h2::client::SendRequest;
use pin_project::pin_project;
@ -223,23 +223,13 @@ where
fn poll_read(
self: Pin<&mut Self>,
cx: &mut Context<'_>,
buf: &mut [u8],
) -> Poll<io::Result<usize>> {
buf: &mut ReadBuf<'_>,
) -> Poll<io::Result<()>> {
match self.project() {
EitherIoProj::A(val) => val.poll_read(cx, buf),
EitherIoProj::B(val) => val.poll_read(cx, buf),
}
}
unsafe fn prepare_uninitialized_buffer(
&self,
buf: &mut [mem::MaybeUninit<u8>],
) -> bool {
match self {
EitherIo::A(ref val) => val.prepare_uninitialized_buffer(buf),
EitherIo::B(ref val) => val.prepare_uninitialized_buffer(buf),
}
}
}
impl<A, B> AsyncWrite for EitherIo<A, B>
@ -274,18 +264,4 @@ where
EitherIoProj::B(val) => val.poll_shutdown(cx),
}
}
fn poll_write_buf<U: Buf>(
self: Pin<&mut Self>,
cx: &mut Context<'_>,
buf: &mut U,
) -> Poll<Result<usize, io::Error>>
where
Self: Sized,
{
match self.project() {
EitherIoProj::A(val) => val.poll_write_buf(cx, buf),
EitherIoProj::B(val) => val.poll_write_buf(cx, buf),
}
}
}

87
actix-http/src/client/connector.rs
View File

@ -3,11 +3,11 @@ use std::marker::PhantomData;
use std::time::Duration;
use actix_codec::{AsyncRead, AsyncWrite};
use actix_connect::{
use actix_rt::net::TcpStream;
use actix_service::{apply_fn, Service, ServiceExt};
use actix_tls::connect::{
default_connector, Connect as TcpConnect, Connection as TcpConnection,
};
use actix_rt::net::TcpStream;
use actix_service::{apply_fn, Service};
use actix_utils::timeout::{TimeoutError, TimeoutService};
use http::Uri;
@ -18,10 +18,10 @@ use super::pool::{ConnectionPool, Protocol};
use super::Connect;
#[cfg(feature = "openssl")]
use actix_connect::ssl::openssl::SslConnector as OpensslConnector;
use actix_tls::connect::ssl::openssl::SslConnector as OpensslConnector;
#[cfg(feature = "rustls")]
use actix_connect::ssl::rustls::ClientConfig;
use actix_tls::connect::ssl::rustls::ClientConfig;
#[cfg(feature = "rustls")]
use std::sync::Arc;
@ -62,9 +62,9 @@ impl Connector<(), ()> {
#[allow(clippy::new_ret_no_self, clippy::let_unit_value)]
pub fn new() -> Connector<
impl Service<
Request = TcpConnect<Uri>,
TcpConnect<Uri>,
Response = TcpConnection<Uri, TcpStream>,
Error = actix_connect::ConnectError,
Error = actix_tls::connect::ConnectError,
> + Clone,
TcpStream,
> {
@ -79,7 +79,7 @@ impl Connector<(), ()> {
// Build Ssl connector with openssl, based on supplied alpn protocols
#[cfg(feature = "openssl")]
fn build_ssl(protocols: Vec<Vec<u8>>) -> SslConnector {
use actix_connect::ssl::openssl::SslMethod;
use actix_tls::connect::ssl::openssl::SslMethod;
use bytes::{BufMut, BytesMut};
let mut alpn = BytesMut::with_capacity(20);
@ -102,7 +102,7 @@ impl Connector<(), ()> {
config.set_protocols(&protocols);
config
.root_store
.add_server_trust_anchors(&actix_tls::rustls::TLS_SERVER_ROOTS);
.add_server_trust_anchors(&actix_tls::accept::rustls::TLS_SERVER_ROOTS);
SslConnector::Rustls(Arc::new(config))
}
@ -117,9 +117,9 @@ impl<T, U> Connector<T, U> {
where
U1: AsyncRead + AsyncWrite + Unpin + fmt::Debug,
T1: Service<
Request = TcpConnect<Uri>,
TcpConnect<Uri>,
Response = TcpConnection<Uri, U1>,
Error = actix_connect::ConnectError,
Error = actix_tls::connect::ConnectError,
> + Clone,
{
Connector {
@ -135,9 +135,9 @@ impl<T, U> Connector<T, U>
where
U: AsyncRead + AsyncWrite + Unpin + fmt::Debug + 'static,
T: Service<
Request = TcpConnect<Uri>,
TcpConnect<Uri>,
Response = TcpConnection<Uri, U>,
Error = actix_connect::ConnectError,
Error = actix_tls::connect::ConnectError,
> + Clone
+ 'static,
{
@ -241,8 +241,8 @@ where
/// its combinator chain.
pub fn finish(
self,
) -> impl Service<Request = Connect, Response = impl Connection, Error = ConnectError>
+ Clone {
) -> impl Service<Connect, Response = impl Connection, Error = ConnectError> + Clone
{
#[cfg(not(any(feature = "openssl", feature = "rustls")))]
{
let connector = TimeoutService::new(
@ -268,11 +268,11 @@ where
#[cfg(any(feature = "openssl", feature = "rustls"))]
{
const H2: &[u8] = b"h2";
#[cfg(feature = "openssl")]
use actix_connect::ssl::openssl::OpensslConnector;
#[cfg(feature = "rustls")]
use actix_connect::ssl::rustls::{RustlsConnector, Session};
use actix_service::{boxed::service, pipeline};
#[cfg(feature = "openssl")]
use actix_tls::connect::ssl::openssl::OpensslConnector;
#[cfg(feature = "rustls")]
use actix_tls::connect::ssl::rustls::{RustlsConnector, Session};
let ssl_service = TimeoutService::new(
self.config.timeout,
@ -363,8 +363,7 @@ mod connect_impl {
pub(crate) struct InnerConnector<T, Io>
where
Io: AsyncRead + AsyncWrite + Unpin + 'static,
T: Service<Request = Connect, Response = (Io, Protocol), Error = ConnectError>
+ 'static,
T: Service<Connect, Response = (Io, Protocol), Error = ConnectError> + 'static,
{
pub(crate) tcp_pool: ConnectionPool<T, Io>,
}
@ -372,8 +371,7 @@ mod connect_impl {
impl<T, Io> Clone for InnerConnector<T, Io>
where
Io: AsyncRead + AsyncWrite + Unpin + 'static,
T: Service<Request = Connect, Response = (Io, Protocol), Error = ConnectError>
+ 'static,
T: Service<Connect, Response = (Io, Protocol), Error = ConnectError> + 'static,
{
fn clone(&self) -> Self {
InnerConnector {
@ -382,17 +380,15 @@ mod connect_impl {
}
}
impl<T, Io> Service for InnerConnector<T, Io>
impl<T, Io> Service<Connect> for InnerConnector<T, Io>
where
Io: AsyncRead + AsyncWrite + Unpin + 'static,
T: Service<Request = Connect, Response = (Io, Protocol), Error = ConnectError>
+ 'static,
T: Service<Connect, Response = (Io, Protocol), Error = ConnectError> + 'static,
{
type Request = Connect;
type Response = IoConnection<Io>;
type Error = ConnectError;
type Future = Either<
<ConnectionPool<T, Io> as Service>::Future,
<ConnectionPool<T, Io> as Service<Connect>>::Future,
Ready<Result<IoConnection<Io>, ConnectError>>,
>;
@ -428,8 +424,8 @@ mod connect_impl {
where
Io1: AsyncRead + AsyncWrite + Unpin + 'static,
Io2: AsyncRead + AsyncWrite + Unpin + 'static,
T1: Service<Request = Connect, Response = (Io1, Protocol), Error = ConnectError>,
T2: Service<Request = Connect, Response = (Io2, Protocol), Error = ConnectError>,
T1: Service<Connect, Response = (Io1, Protocol), Error = ConnectError>,
T2: Service<Connect, Response = (Io2, Protocol), Error = ConnectError>,
{
pub(crate) tcp_pool: ConnectionPool<T1, Io1>,
pub(crate) ssl_pool: ConnectionPool<T2, Io2>,
@ -439,10 +435,8 @@ mod connect_impl {
where
Io1: AsyncRead + AsyncWrite + Unpin + 'static,
Io2: AsyncRead + AsyncWrite + Unpin + 'static,
T1: Service<Request = Connect, Response = (Io1, Protocol), Error = ConnectError>
+ 'static,
T2: Service<Request = Connect, Response = (Io2, Protocol), Error = ConnectError>
+ 'static,
T1: Service<Connect, Response = (Io1, Protocol), Error = ConnectError> + 'static,
T2: Service<Connect, Response = (Io2, Protocol), Error = ConnectError> + 'static,
{
fn clone(&self) -> Self {
InnerConnector {
@ -452,16 +446,13 @@ mod connect_impl {
}
}
impl<T1, T2, Io1, Io2> Service for InnerConnector<T1, T2, Io1, Io2>
impl<T1, T2, Io1, Io2> Service<Connect> for InnerConnector<T1, T2, Io1, Io2>
where
Io1: AsyncRead + AsyncWrite + Unpin + 'static,
Io2: AsyncRead + AsyncWrite + Unpin + 'static,
T1: Service<Request = Connect, Response = (Io1, Protocol), Error = ConnectError>
+ 'static,
T2: Service<Request = Connect, Response = (Io2, Protocol), Error = ConnectError>
+ 'static,
T1: Service<Connect, Response = (Io1, Protocol), Error = ConnectError> + 'static,
T2: Service<Connect, Response = (Io2, Protocol), Error = ConnectError> + 'static,
{
type Request = Connect;
type Response = EitherConnection<Io1, Io2>;
type Error = ConnectError;
type Future = Either<
@ -491,18 +482,16 @@ mod connect_impl {
pub(crate) struct InnerConnectorResponseA<T, Io1, Io2>
where
Io1: AsyncRead + AsyncWrite + Unpin + 'static,
T: Service<Request = Connect, Response = (Io1, Protocol), Error = ConnectError>
+ 'static,
T: Service<Connect, Response = (Io1, Protocol), Error = ConnectError> + 'static,
{
#[pin]
fut: <ConnectionPool<T, Io1> as Service>::Future,
fut: <ConnectionPool<T, Io1> as Service<Connect>>::Future,
_t: PhantomData<Io2>,
}
impl<T, Io1, Io2> Future for InnerConnectorResponseA<T, Io1, Io2>
where
T: Service<Request = Connect, Response = (Io1, Protocol), Error = ConnectError>
+ 'static,
T: Service<Connect, Response = (Io1, Protocol), Error = ConnectError> + 'static,
Io1: AsyncRead + AsyncWrite + Unpin + 'static,
Io2: AsyncRead + AsyncWrite + Unpin + 'static,
{
@ -520,18 +509,16 @@ mod connect_impl {
pub(crate) struct InnerConnectorResponseB<T, Io1, Io2>
where
Io2: AsyncRead + AsyncWrite + Unpin + 'static,
T: Service<Request = Connect, Response = (Io2, Protocol), Error = ConnectError>
+ 'static,
T: Service<Connect, Response = (Io2, Protocol), Error = ConnectError> + 'static,
{
#[pin]
fut: <ConnectionPool<T, Io2> as Service>::Future,
fut: <ConnectionPool<T, Io2> as Service<Connect>>::Future,
_t: PhantomData<Io1>,
}
impl<T, Io1, Io2> Future for InnerConnectorResponseB<T, Io1, Io2>
where
T: Service<Request = Connect, Response = (Io2, Protocol), Error = ConnectError>
+ 'static,
T: Service<Connect, Response = (Io2, Protocol), Error = ConnectError> + 'static,
Io1: AsyncRead + AsyncWrite + Unpin + 'static,
Io2: AsyncRead + AsyncWrite + Unpin + 'static,
{

30
actix-http/src/client/error.rs
View File

@ -1,10 +1,10 @@
use std::io;
use actix_connect::resolver::ResolveError;
use actix_tls::connect::resolver::ResolveError;
use derive_more::{Display, From};
#[cfg(feature = "openssl")]
use actix_connect::ssl::openssl::{HandshakeError, SslError};
use actix_tls::accept::openssl::SslError;
use crate::error::{Error, ParseError, ResponseError};
use crate::http::{Error as HttpError, StatusCode};
@ -21,11 +21,6 @@ pub enum ConnectError {
#[display(fmt = "{}", _0)]
SslError(SslError),
/// SSL Handshake error
#[cfg(feature = "openssl")]
#[display(fmt = "{}", _0)]
SslHandshakeError(String),
/// Failed to resolve the hostname
#[display(fmt = "Failed resolving hostname: {}", _0)]
Resolver(ResolveError),
@ -57,25 +52,18 @@ pub enum ConnectError {
impl std::error::Error for ConnectError {}
impl From<actix_connect::ConnectError> for ConnectError {
fn from(err: actix_connect::ConnectError) -> ConnectError {
impl From<actix_tls::connect::ConnectError> for ConnectError {
fn from(err: actix_tls::connect::ConnectError) -> ConnectError {
match err {
actix_connect::ConnectError::Resolver(e) => ConnectError::Resolver(e),
actix_connect::ConnectError::NoRecords => ConnectError::NoRecords,
actix_connect::ConnectError::InvalidInput => panic!(),
actix_connect::ConnectError::Unresolved => ConnectError::Unresolved,
actix_connect::ConnectError::Io(e) => ConnectError::Io(e),
actix_tls::connect::ConnectError::Resolver(e) => ConnectError::Resolver(e),
actix_tls::connect::ConnectError::NoRecords => ConnectError::NoRecords,
actix_tls::connect::ConnectError::InvalidInput => panic!(),
actix_tls::connect::ConnectError::Unresolved => ConnectError::Unresolved,
actix_tls::connect::ConnectError::Io(e) => ConnectError::Io(e),
}
}
}
#[cfg(feature = "openssl")]
impl<T: std::fmt::Debug> From<HandshakeError<T>> for ConnectError {
fn from(err: HandshakeError<T>) -> ConnectError {
ConnectError::SslHandshakeError(format!("{:?}", err))
}
}
#[derive(Debug, Display, From)]
pub enum InvalidUrl {
#[display(fmt = "Missing url scheme")]

17
actix-http/src/client/h1proto.rs
View File

@ -1,10 +1,10 @@
use std::io::Write;
use std::pin::Pin;
use std::task::{Context, Poll};
use std::{io, mem, time};
use std::{io, time};
use actix_codec::{AsyncRead, AsyncWrite, Framed};
use bytes::buf::BufMutExt;
use actix_codec::{AsyncRead, AsyncWrite, Framed, ReadBuf};
use bytes::buf::BufMut;
use bytes::{Bytes, BytesMut};
use futures_core::Stream;
use futures_util::future::poll_fn;
@ -204,18 +204,11 @@ where
}
impl<T: AsyncRead + AsyncWrite + Unpin + 'static> AsyncRead for H1Connection<T> {
unsafe fn prepare_uninitialized_buffer(
&self,
buf: &mut [mem::MaybeUninit<u8>],
) -> bool {
self.io.as_ref().unwrap().prepare_uninitialized_buffer(buf)
}
fn poll_read(
mut self: Pin<&mut Self>,
cx: &mut Context<'_>,
buf: &mut [u8],
) -> Poll<io::Result<usize>> {
buf: &mut ReadBuf<'_>,
) -> Poll<io::Result<()>> {
Pin::new(&mut self.io.as_mut().unwrap()).poll_read(cx, buf)
}
}

32
actix-http/src/client/pool.rs
View File

@ -6,8 +6,8 @@ use std::rc::Rc;
use std::task::{Context, Poll};
use std::time::{Duration, Instant};
use actix_codec::{AsyncRead, AsyncWrite};
use actix_rt::time::{delay_for, Delay};
use actix_codec::{AsyncRead, AsyncWrite, ReadBuf};
use actix_rt::time::{sleep, Sleep};
use actix_service::Service;
use actix_utils::task::LocalWaker;
use bytes::Bytes;
@ -50,8 +50,7 @@ pub(crate) struct ConnectionPool<T, Io: 'static>(Rc<RefCell<T>>, Rc<RefCell<Inne
impl<T, Io> ConnectionPool<T, Io>
where
Io: AsyncRead + AsyncWrite + Unpin + 'static,
T: Service<Request = Connect, Response = (Io, Protocol), Error = ConnectError>
+ 'static,
T: Service<Connect, Response = (Io, Protocol), Error = ConnectError> + 'static,
{
pub(crate) fn new(connector: T, config: ConnectorConfig) -> Self {
let connector_rc = Rc::new(RefCell::new(connector));
@ -90,13 +89,11 @@ impl<T, Io> Drop for ConnectionPool<T, Io> {
}
}
impl<T, Io> Service for ConnectionPool<T, Io>
impl<T, Io> Service<Connect> for ConnectionPool<T, Io>
where
Io: AsyncRead + AsyncWrite + Unpin + 'static,
T: Service<Request = Connect, Response = (Io, Protocol), Error = ConnectError>
+ 'static,
T: Service<Connect, Response = (Io, Protocol), Error = ConnectError> + 'static,
{
type Request = Connect;
type Response = IoConnection<Io>;
type Error = ConnectError;
type Future = LocalBoxFuture<'static, Result<IoConnection<Io>, ConnectError>>;
@ -334,10 +331,11 @@ where
} else {
let mut io = conn.io;
let mut buf = [0; 2];
let mut read_buf = ReadBuf::new(&mut buf);
if let ConnectionType::H1(ref mut s) = io {
match Pin::new(s).poll_read(cx, &mut buf) {
match Pin::new(s).poll_read(cx, &mut read_buf) {
Poll::Pending => (),
Poll::Ready(Ok(n)) if n > 0 => {
Poll::Ready(Ok(())) if !read_buf.filled().is_empty() => {
if let Some(timeout) = self.config.disconnect_timeout {
if let ConnectionType::H1(io) = io {
actix_rt::spawn(CloseConnection::new(
@ -387,9 +385,11 @@ where
}
}
#[pin_project::pin_project]
struct CloseConnection<T> {
io: T,
timeout: Delay,
#[pin]
timeout: Sleep,
}
impl<T> CloseConnection<T>
@ -399,7 +399,7 @@ where
fn new(io: T, timeout: Duration) -> Self {
CloseConnection {
io,
timeout: delay_for(timeout),
timeout: sleep(timeout),
}
}
}
@ -411,11 +411,11 @@ where
type Output = ();
fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<()> {
let this = self.get_mut();
let this = self.project();
match Pin::new(&mut this.timeout).poll(cx) {
match this.timeout.poll(cx) {
Poll::Ready(_) => Poll::Ready(()),
Poll::Pending => match Pin::new(&mut this.io).poll_shutdown(cx) {
Poll::Pending => match Pin::new(this.io).poll_shutdown(cx) {
Poll::Ready(_) => Poll::Ready(()),
Poll::Pending => Poll::Pending,
},
@ -435,7 +435,7 @@ where
impl<T, Io> Future for ConnectorPoolSupport<T, Io>
where
Io: AsyncRead + AsyncWrite + Unpin + 'static,
T: Service<Request = Connect, Response = (Io, Protocol), Error = ConnectError>,
T: Service<Connect, Response = (Io, Protocol), Error = ConnectError>,
T::Future: 'static,
{
type Output = ();

11
actix-http/src/cloneable.rs
View File

@ -10,22 +10,21 @@ use actix_service::Service;
/// CloneableService might panic with some creative use of thread local storage.
/// See https://github.com/actix/actix-web/issues/1295 for example
#[doc(hidden)]
pub(crate) struct CloneableService<T: Service>(Rc<RefCell<T>>);
pub(crate) struct CloneableService<T>(Rc<RefCell<T>>);
impl<T: Service> CloneableService<T> {
impl<T> CloneableService<T> {
pub(crate) fn new(service: T) -> Self {
Self(Rc::new(RefCell::new(service)))
}
}
impl<T: Service> Clone for CloneableService<T> {
impl<T> Clone for CloneableService<T> {
fn clone(&self) -> Self {
Self(self.0.clone())
}
}
impl<T: Service> Service for CloneableService<T> {
type Request = T::Request;
impl<T: Service<Req>, Req> Service<Req> for CloneableService<T> {
type Response = T::Response;
type Error = T::Error;
type Future = T::Future;
@ -34,7 +33,7 @@ impl<T: Service> Service for CloneableService<T> {
self.0.borrow_mut().poll_ready(cx)
}
fn call(&mut self, req: T::Request) -> Self::Future {
fn call(&mut self, req: Req) -> Self::Future {
self.0.borrow_mut().call(req)
}
}

12
actix-http/src/config.rs
View File

@ -4,7 +4,7 @@ use std::rc::Rc;
use std::time::Duration;
use std::{fmt, net};
use actix_rt::time::{delay_for, delay_until, Delay, Instant};
use actix_rt::time::{sleep, sleep_until, Instant, Sleep};
use bytes::BytesMut;
use futures_util::{future, FutureExt};
use time::OffsetDateTime;
@ -121,10 +121,10 @@ impl ServiceConfig {
#[inline]
/// Client timeout for first request.
pub fn client_timer(&self) -> Option<Delay> {
pub fn client_timer(&self) -> Option<Sleep> {
let delay_time = self.0.client_timeout;
if delay_time != 0 {
Some(delay_until(
Some(sleep_until(
self.0.timer.now() + Duration::from_millis(delay_time),
))
} else {
@ -154,9 +154,9 @@ impl ServiceConfig {
#[inline]
/// Return keep-alive timer delay is configured.
pub fn keep_alive_timer(&self) -> Option<Delay> {
pub fn keep_alive_timer(&self) -> Option<Sleep> {
if let Some(ka) = self.0.keep_alive {
Some(delay_until(self.0.timer.now() + ka))
Some(sleep_until(self.0.timer.now() + ka))
} else {
None
}
@ -266,7 +266,7 @@ impl DateService {
// periodic date update
let s = self.clone();
actix_rt::spawn(delay_for(Duration::from_millis(500)).then(move |_| {
actix_rt::spawn(sleep(Duration::from_millis(500)).then(move |_| {
s.0.reset();
future::ready(())
}));

11
actix-http/src/error.rs
View File

@ -178,11 +178,7 @@ impl ResponseError for FormError {}
#[cfg(feature = "openssl")]
/// `InternalServerError` for `openssl::ssl::Error`
impl ResponseError for actix_connect::ssl::openssl::SslError {}
#[cfg(feature = "openssl")]
/// `InternalServerError` for `openssl::ssl::HandshakeError`
impl<T: std::fmt::Debug> ResponseError for actix_tls::openssl::HandshakeError<T> {}
impl ResponseError for actix_tls::accept::openssl::SslError {}
/// Return `BAD_REQUEST` for `de::value::Error`
impl ResponseError for DeError {
@ -956,11 +952,6 @@ where
/// This is supported on feature=`actors` only
impl ResponseError for actix::MailboxError {}
#[cfg(feature = "actors")]
/// `InternalServerError` for `actix::ResolverError`
/// This is supported on feature=`actors` only
impl ResponseError for actix::actors::resolver::ResolverError {}
#[cfg(test)]
mod tests {
use super::*;

108
actix-http/src/h1/dispatcher.rs
View File

@ -8,7 +8,7 @@ use std::{
};
use actix_codec::{AsyncRead, AsyncWrite, Decoder, Encoder, Framed, FramedParts};
use actix_rt::time::{delay_until, Delay, Instant};
use actix_rt::time::{sleep_until, Instant, Sleep};
use actix_service::Service;
use bitflags::bitflags;
use bytes::{Buf, BytesMut};
@ -51,12 +51,12 @@ bitflags! {
/// Dispatcher for HTTP/1.1 protocol
pub struct Dispatcher<T, S, B, X, U>
where
S: Service<Request = Request>,
S: Service<Request>,
S::Error: Into<Error>,
B: MessageBody,
X: Service<Request = Request, Response = Request>,
X: Service<Request, Response = Request>,
X::Error: Into<Error>,
U: Service<Request = (Request, Framed<T, Codec>), Response = ()>,
U: Service<(Request, Framed<T, Codec>), Response = ()>,
U::Error: fmt::Display,
{
#[pin]
@ -69,12 +69,12 @@ where
#[pin_project(project = DispatcherStateProj)]
enum DispatcherState<T, S, B, X, U>
where
S: Service<Request = Request>,
S: Service<Request>,
S::Error: Into<Error>,
B: MessageBody,
X: Service<Request = Request, Response = Request>,
X: Service<Request, Response = Request>,
X::Error: Into<Error>,
U: Service<Request = (Request, Framed<T, Codec>), Response = ()>,
U: Service<(Request, Framed<T, Codec>), Response = ()>,
U::Error: fmt::Display,
{
Normal(#[pin] InnerDispatcher<T, S, B, X, U>),
@ -84,12 +84,12 @@ where
#[pin_project(project = InnerDispatcherProj)]
struct InnerDispatcher<T, S, B, X, U>
where
S: Service<Request = Request>,
S: Service<Request>,
S::Error: Into<Error>,
B: MessageBody,
X: Service<Request = Request, Response = Request>,
X: Service<Request, Response = Request>,
X::Error: Into<Error>,
U: Service<Request = (Request, Framed<T, Codec>), Response = ()>,
U: Service<(Request, Framed<T, Codec>), Response = ()>,
U::Error: fmt::Display,
{
service: CloneableService<S>,
@ -106,7 +106,8 @@ where
messages: VecDeque<DispatcherMessage>,
ka_expire: Instant,
ka_timer: Option<Delay>,
#[pin]
ka_timer: Option<Sleep>,
io: Option<T>,
read_buf: BytesMut,
@ -123,8 +124,8 @@ enum DispatcherMessage {
#[pin_project(project = StateProj)]
enum State<S, B, X>
where
S: Service<Request = Request>,
X: Service<Request = Request, Response = Request>,
S: Service<Request>,
X: Service<Request, Response = Request>,
B: MessageBody,
{
None,
@ -135,8 +136,8 @@ where
impl<S, B, X> State<S, B, X>
where
S: Service<Request = Request>,
X: Service<Request = Request, Response = Request>,
S: Service<Request>,
X: Service<Request, Response = Request>,
B: MessageBody,
{
fn is_empty(&self) -> bool {
@ -166,13 +167,13 @@ impl PartialEq for PollResponse {
impl<T, S, B, X, U> Dispatcher<T, S, B, X, U>
where
T: AsyncRead + AsyncWrite + Unpin,
S: Service<Request = Request>,
S: Service<Request>,
S::Error: Into<Error>,
S::Response: Into<Response<B>>,
B: MessageBody,
X: Service<Request = Request, Response = Request>,
X: Service<Request, Response = Request>,
X::Error: Into<Error>,
U: Service<Request = (Request, Framed<T, Codec>), Response = ()>,
U: Service<(Request, Framed<T, Codec>), Response = ()>,
U::Error: fmt::Display,
{
/// Create HTTP/1 dispatcher.
@ -205,7 +206,7 @@ where
codec: Codec,
config: ServiceConfig,
read_buf: BytesMut,
timeout: Option<Delay>,
timeout: Option<Sleep>,
service: CloneableService<S>,
expect: CloneableService<X>,
upgrade: Option<CloneableService<U>>,
@ -257,13 +258,13 @@ where
impl<T, S, B, X, U> InnerDispatcher<T, S, B, X, U>
where
T: AsyncRead + AsyncWrite + Unpin,
S: Service<Request = Request>,
S: Service<Request>,
S::Error: Into<Error>,
S::Response: Into<Response<B>>,
B: MessageBody,
X: Service<Request = Request, Response = Request>,
X: Service<Request, Response = Request>,
X::Error: Into<Error>,
U: Service<Request = (Request, Framed<T, Codec>), Response = ()>,
U: Service<(Request, Framed<T, Codec>), Response = ()>,
U::Error: fmt::Display,
{
fn can_read(&self, cx: &mut Context<'_>) -> bool {
@ -660,7 +661,7 @@ where
// shutdown timeout
if this.flags.contains(Flags::SHUTDOWN) {
if let Some(interval) = this.codec.config().client_disconnect_timer() {
*this.ka_timer = Some(delay_until(interval));
this.ka_timer.set(Some(sleep_until(interval)));
} else {
this.flags.insert(Flags::READ_DISCONNECT);
if let Some(mut payload) = this.payload.take() {
@ -673,12 +674,14 @@ where
}
}
match Pin::new(&mut this.ka_timer.as_mut().unwrap()).poll(cx) {
match this.ka_timer.as_mut().as_pin_mut().unwrap().poll(cx) {
Poll::Ready(()) => {
// if we get timeout during shutdown, drop connection
if this.flags.contains(Flags::SHUTDOWN) {
return Err(DispatchError::DisconnectTimeout);
} else if this.ka_timer.as_mut().unwrap().deadline() >= *this.ka_expire {
} else if this.ka_timer.as_mut().as_pin_mut().unwrap().deadline()
>= *this.ka_expire
{
// check for any outstanding tasks
if this.state.is_empty() && this.write_buf.is_empty() {
if this.flags.contains(Flags::STARTED) {
@ -689,9 +692,15 @@ where
if let Some(deadline) =
this.codec.config().client_disconnect_timer()
{
if let Some(mut timer) = this.ka_timer.as_mut() {
if let Some(timer) = this.ka_timer.as_mut().as_pin_mut()
{
timer.reset(deadline);
let _ = Pin::new(&mut timer).poll(cx);
let _ = this
.ka_timer
.as_mut()
.as_pin_mut()
.unwrap()
.poll(cx);
}
} else {
// no shutdown timeout, drop socket
@ -716,14 +725,15 @@ where
} else if let Some(deadline) =
this.codec.config().keep_alive_expire()
{
if let Some(mut timer) = this.ka_timer.as_mut() {
if let Some(timer) = this.ka_timer.as_mut().as_pin_mut() {
timer.reset(deadline);
let _ = Pin::new(&mut timer).poll(cx);
let _ =
this.ka_timer.as_mut().as_pin_mut().unwrap().poll(cx);
}
}
} else if let Some(mut timer) = this.ka_timer.as_mut() {
} else if let Some(timer) = this.ka_timer.as_mut().as_pin_mut() {
timer.reset(*this.ka_expire);
let _ = Pin::new(&mut timer).poll(cx);
let _ = this.ka_timer.as_mut().as_pin_mut().unwrap().poll(cx);
}
}
Poll::Pending => (),
@ -736,13 +746,13 @@ where
impl<T, S, B, X, U> Future for Dispatcher<T, S, B, X, U>
where
T: AsyncRead + AsyncWrite + Unpin,
S: Service<Request = Request>,
S: Service<Request>,
S::Error: Into<Error>,
S::Response: Into<Response<B>>,
B: MessageBody,
X: Service<Request = Request, Response = Request>,
X: Service<Request, Response = Request>,
X::Error: Into<Error>,
U: Service<Request = (Request, Framed<T, Codec>), Response = ()>,
U: Service<(Request, Framed<T, Codec>), Response = ()>,
U::Error: fmt::Display,
{
type Output = Result<(), DispatchError>;
@ -951,12 +961,12 @@ fn read<T>(
where
T: AsyncRead + Unpin,
{
Pin::new(io).poll_read_buf(cx, buf)
actix_codec::poll_read_buf(Pin::new(io), cx, buf)
}
#[cfg(test)]
mod tests {
use std::{marker::PhantomData, str};
use std::str;
use actix_service::fn_service;
use futures_util::future::{lazy, ready};
@ -985,21 +995,19 @@ mod tests {
}
}
fn ok_service() -> impl Service<Request = Request, Response = Response, Error = Error>
{
fn ok_service() -> impl Service<Request, Response = Response, Error = Error> {
fn_service(|_req: Request| ready(Ok::<_, Error>(Response::Ok().finish())))
}
fn echo_path_service(
) -> impl Service<Request = Request, Response = Response, Error = Error> {
fn echo_path_service() -> impl Service<Request, Response = Response, Error = Error> {
fn_service(|req: Request| {
let path = req.path().as_bytes();
ready(Ok::<_, Error>(Response::Ok().body(Body::from_slice(path))))
})
}
fn echo_payload_service(
) -> impl Service<Request = Request, Response = Response, Error = Error> {
fn echo_payload_service() -> impl Service<Request, Response = Response, Error = Error>
{
fn_service(|mut req: Request| {
Box::pin(async move {
use futures_util::stream::StreamExt as _;
@ -1007,7 +1015,7 @@ mod tests {
let mut pl = req.take_payload();
let mut body = BytesMut::new();
while let Some(chunk) = pl.next().await {
body.extend_from_slice(chunk.unwrap().bytes())
body.extend_from_slice(chunk.unwrap().chunk())
}
Ok::<_, Error>(Response::Ok().body(body))
@ -1020,7 +1028,7 @@ mod tests {
lazy(|cx| {
let buf = TestBuffer::new("GET /test HTTP/1\r\n\r\n");
let h1 = Dispatcher::<_, _, _, _, UpgradeHandler<TestBuffer>>::new(
let h1 = Dispatcher::<_, _, _, _, UpgradeHandler>::new(
buf,
ServiceConfig::default(),
CloneableService::new(ok_service()),
@ -1060,7 +1068,7 @@ mod tests {
let cfg = ServiceConfig::new(KeepAlive::Disabled, 1, 1, false, None);
let h1 = Dispatcher::<_, _, _, _, UpgradeHandler<TestBuffer>>::new(
let h1 = Dispatcher::<_, _, _, _, UpgradeHandler>::new(
buf,
cfg,
CloneableService::new(echo_path_service()),
@ -1114,7 +1122,7 @@ mod tests {
let cfg = ServiceConfig::new(KeepAlive::Disabled, 1, 1, false, None);
let h1 = Dispatcher::<_, _, _, _, UpgradeHandler<TestBuffer>>::new(
let h1 = Dispatcher::<_, _, _, _, UpgradeHandler>::new(
buf,
cfg,
CloneableService::new(echo_path_service()),
@ -1163,7 +1171,7 @@ mod tests {
lazy(|cx| {
let mut buf = TestSeqBuffer::empty();
let cfg = ServiceConfig::new(KeepAlive::Disabled, 0, 0, false, None);
let h1 = Dispatcher::<_, _, _, _, UpgradeHandler<_>>::new(
let h1 = Dispatcher::<_, _, _, _, UpgradeHandler>::new(
buf.clone(),
cfg,
CloneableService::new(echo_payload_service()),
@ -1234,7 +1242,7 @@ mod tests {
lazy(|cx| {
let mut buf = TestSeqBuffer::empty();
let cfg = ServiceConfig::new(KeepAlive::Disabled, 0, 0, false, None);
let h1 = Dispatcher::<_, _, _, _, UpgradeHandler<_>>::new(
let h1 = Dispatcher::<_, _, _, _, UpgradeHandler>::new(
buf.clone(),
cfg,
CloneableService::new(echo_path_service()),
@ -1293,12 +1301,12 @@ mod tests {
lazy(|cx| {
let mut buf = TestSeqBuffer::empty();
let cfg = ServiceConfig::new(KeepAlive::Disabled, 0, 0, false, None);
let h1 = Dispatcher::<_, _, _, _, UpgradeHandler<_>>::new(
let h1 = Dispatcher::<_, _, _, _, UpgradeHandler>::new(
buf.clone(),
cfg,
CloneableService::new(ok_service()),
CloneableService::new(ExpectHandler),
Some(CloneableService::new(UpgradeHandler(PhantomData))),
Some(CloneableService::new(UpgradeHandler)),
Extensions::new(),
None,
);

6
actix-http/src/h1/encoder.rs
View File

@ -135,7 +135,7 @@ pub(crate) trait MessageType: Sized {
let mut has_date = false;
let mut buf = dst.bytes_mut().as_mut_ptr() as *mut u8;
let mut buf = dst.chunk_mut().as_mut_ptr() as *mut u8;
let mut remaining = dst.capacity() - dst.len();
// tracks bytes written since last buffer resize
@ -177,7 +177,7 @@ pub(crate) trait MessageType: Sized {
// re-assign buf raw pointer since it's possible that the buffer was
// reallocated and/or resized
buf = dst.bytes_mut().as_mut_ptr() as *mut u8;
buf = dst.chunk_mut().as_mut_ptr() as *mut u8;
}
// SAFETY: on each write, it is enough to ensure that the advancement of the
@ -224,7 +224,7 @@ pub(crate) trait MessageType: Sized {
// re-assign buf raw pointer since it's possible that the buffer was
// reallocated and/or resized
buf = dst.bytes_mut().as_mut_ptr() as *mut u8;
buf = dst.chunk_mut().as_mut_ptr() as *mut u8;
}
// SAFETY: on each write, it is enough to ensure that the advancement of

8
actix-http/src/h1/expect.rs
View File

@ -8,11 +8,10 @@ use crate::request::Request;
pub struct ExpectHandler;
impl ServiceFactory for ExpectHandler {
type Config = ();
type Request = Request;
impl ServiceFactory<Request> for ExpectHandler {
type Response = Request;
type Error = Error;
type Config = ();
type Service = ExpectHandler;
type InitError = Error;
type Future = Ready<Result<Self::Service, Self::InitError>>;
@ -22,8 +21,7 @@ impl ServiceFactory for ExpectHandler {
}
}
impl Service for ExpectHandler {
type Request = Request;
impl Service<Request> for ExpectHandler {
type Response = Request;
type Error = Error;
type Future = Ready<Result<Self::Response, Self::Error>>;

107
actix-http/src/h1/service.rs
View File

@ -24,25 +24,25 @@ use super::dispatcher::Dispatcher;
use super::{ExpectHandler, UpgradeHandler};
/// `ServiceFactory` implementation for HTTP1 transport
pub struct H1Service<T, S, B, X = ExpectHandler, U = UpgradeHandler<T>> {
pub struct H1Service<T, S, B, X = ExpectHandler, U = UpgradeHandler> {
srv: S,
cfg: ServiceConfig,
expect: X,
upgrade: Option<U>,
on_connect_ext: Option<Rc<ConnectCallback<T>>>,
_t: PhantomData<(T, B)>,
_t: PhantomData<B>,
}
impl<T, S, B> H1Service<T, S, B>
where
S: ServiceFactory<Config = (), Request = Request>,
S: ServiceFactory<Request, Config = ()>,
S::Error: Into<Error>,
S::InitError: fmt::Debug,
S::Response: Into<Response<B>>,
B: MessageBody,
{
/// Create new `HttpService` instance with config.
pub(crate) fn with_config<F: IntoServiceFactory<S>>(
pub(crate) fn with_config<F: IntoServiceFactory<S, Request>>(
cfg: ServiceConfig,
service: F,
) -> Self {
@ -59,19 +59,15 @@ where
impl<S, B, X, U> H1Service<TcpStream, S, B, X, U>
where
S: ServiceFactory<Config = (), Request = Request>,
S: ServiceFactory<Request, Config = ()>,
S::Error: Into<Error>,
S::InitError: fmt::Debug,
S::Response: Into<Response<B>>,
B: MessageBody,
X: ServiceFactory<Config = (), Request = Request, Response = Request>,
X: ServiceFactory<Request, Config = (), Response = Request>,
X::Error: Into<Error>,
X::InitError: fmt::Debug,
U: ServiceFactory<
Config = (),
Request = (Request, Framed<TcpStream, Codec>),
Response = (),
>,
U: ServiceFactory<(Request, Framed<TcpStream, Codec>), Config = (), Response = ()>,
U::Error: fmt::Display + Into<Error>,
U::InitError: fmt::Debug,
{
@ -79,8 +75,8 @@ where
pub fn tcp(
self,
) -> impl ServiceFactory<
TcpStream,
Config = (),
Request = TcpStream,
Response = (),
Error = DispatchError,
InitError = (),
@ -97,22 +93,23 @@ where
mod openssl {
use super::*;
use actix_tls::openssl::{Acceptor, SslAcceptor, SslStream};
use actix_tls::{openssl::HandshakeError, TlsError};
use actix_service::ServiceFactoryExt;
use actix_tls::accept::openssl::{Acceptor, SslAcceptor, SslError, SslStream};
use actix_tls::accept::TlsError;
impl<S, B, X, U> H1Service<SslStream<TcpStream>, S, B, X, U>
where
S: ServiceFactory<Config = (), Request = Request>,
S: ServiceFactory<Request, Config = ()>,
S::Error: Into<Error>,
S::InitError: fmt::Debug,
S::Response: Into<Response<B>>,
B: MessageBody,
X: ServiceFactory<Config = (), Request = Request, Response = Request>,
X: ServiceFactory<Request, Config = (), Response = Request>,
X::Error: Into<Error>,
X::InitError: fmt::Debug,
U: ServiceFactory<
(Request, Framed<SslStream<TcpStream>, Codec>),
Config = (),
Request = (Request, Framed<SslStream<TcpStream>, Codec>),
Response = (),
>,
U::Error: fmt::Display + Into<Error>,
@ -123,10 +120,10 @@ mod openssl {
self,
acceptor: SslAcceptor,
) -> impl ServiceFactory<
TcpStream,
Config = (),
Request = TcpStream,
Response = (),
Error = TlsError<HandshakeError<TcpStream>, DispatchError>,
Error = TlsError<SslError, DispatchError>,
InitError = (),
> {
pipeline_factory(
@ -146,23 +143,24 @@ mod openssl {
#[cfg(feature = "rustls")]
mod rustls {
use super::*;
use actix_tls::rustls::{Acceptor, ServerConfig, TlsStream};
use actix_tls::TlsError;
use actix_service::ServiceFactoryExt;
use actix_tls::accept::rustls::{Acceptor, ServerConfig, TlsStream};
use actix_tls::accept::TlsError;
use std::{fmt, io};
impl<S, B, X, U> H1Service<TlsStream<TcpStream>, S, B, X, U>
where
S: ServiceFactory<Config = (), Request = Request>,
S: ServiceFactory<Request, Config = ()>,
S::Error: Into<Error>,
S::InitError: fmt::Debug,
S::Response: Into<Response<B>>,
B: MessageBody,
X: ServiceFactory<Config = (), Request = Request, Response = Request>,
X: ServiceFactory<Request, Config = (), Response = Request>,
X::Error: Into<Error>,
X::InitError: fmt::Debug,
U: ServiceFactory<
(Request, Framed<TlsStream<TcpStream>, Codec>),
Config = (),
Request = (Request, Framed<TlsStream<TcpStream>, Codec>),
Response = (),
>,
U::Error: fmt::Display + Into<Error>,
@ -173,8 +171,8 @@ mod rustls {
self,
config: ServerConfig,
) -> impl ServiceFactory<
TcpStream,
Config = (),
Request = TcpStream,
Response = (),
Error = TlsError<io::Error, DispatchError>,
InitError = (),
@ -195,7 +193,7 @@ mod rustls {
impl<T, S, B, X, U> H1Service<T, S, B, X, U>
where
S: ServiceFactory<Config = (), Request = Request>,
S: ServiceFactory<Request, Config = ()>,
S::Error: Into<Error>,
S::Response: Into<Response<B>>,
S::InitError: fmt::Debug,
@ -203,7 +201,7 @@ where
{
pub fn expect<X1>(self, expect: X1) -> H1Service<T, S, B, X1, U>
where
X1: ServiceFactory<Request = Request, Response = Request>,
X1: ServiceFactory<Request, Response = Request>,
X1::Error: Into<Error>,
X1::InitError: fmt::Debug,
{
@ -219,7 +217,7 @@ where
pub fn upgrade<U1>(self, upgrade: Option<U1>) -> H1Service<T, S, B, X, U1>
where
U1: ServiceFactory<Request = (Request, Framed<T, Codec>), Response = ()>,
U1: ServiceFactory<(Request, Framed<T, Codec>), Response = ()>,
U1::Error: fmt::Display,
U1::InitError: fmt::Debug,
{
@ -240,27 +238,27 @@ where
}
}
impl<T, S, B, X, U> ServiceFactory for H1Service<T, S, B, X, U>
impl<T, S, B, X, U> ServiceFactory<(T, Option<net::SocketAddr>)>
for H1Service<T, S, B, X, U>
where
T: AsyncRead + AsyncWrite + Unpin,
S: ServiceFactory<Config = (), Request = Request>,
S: ServiceFactory<Request, Config = ()>,
S::Error: Into<Error>,
S::Response: Into<Response<B>>,
S::InitError: fmt::Debug,
B: MessageBody,
X: ServiceFactory<Config = (), Request = Request, Response = Request>,
X: ServiceFactory<Request, Config = (), Response = Request>,
X::Error: Into<Error>,
X::InitError: fmt::Debug,
U: ServiceFactory<Config = (), Request = (Request, Framed<T, Codec>), Response = ()>,
U: ServiceFactory<(Request, Framed<T, Codec>), Config = (), Response = ()>,
U::Error: fmt::Display + Into<Error>,
U::InitError: fmt::Debug,
{
type Config = ();
type Request = (T, Option<net::SocketAddr>);
type Response = ();
type Error = DispatchError;
type InitError = ();
type Config = ();
type Service = H1ServiceHandler<T, S::Service, B, X::Service, U::Service>;
type InitError = ();
type Future = H1ServiceResponse<T, S, B, X, U>;
fn new_service(&self, _: ()) -> Self::Future {
@ -281,13 +279,13 @@ where
#[pin_project::pin_project]
pub struct H1ServiceResponse<T, S, B, X, U>
where
S: ServiceFactory<Request = Request>,
S: ServiceFactory<Request>,
S::Error: Into<Error>,
S::InitError: fmt::Debug,
X: ServiceFactory<Request = Request, Response = Request>,
X: ServiceFactory<Request, Response = Request>,
X::Error: Into<Error>,
X::InitError: fmt::Debug,
U: ServiceFactory<Request = (Request, Framed<T, Codec>), Response = ()>,
U: ServiceFactory<(Request, Framed<T, Codec>), Response = ()>,
U::Error: fmt::Display,
U::InitError: fmt::Debug,
{
@ -307,15 +305,15 @@ where
impl<T, S, B, X, U> Future for H1ServiceResponse<T, S, B, X, U>
where
T: AsyncRead + AsyncWrite + Unpin,
S: ServiceFactory<Request = Request>,
S: ServiceFactory<Request>,
S::Error: Into<Error>,
S::Response: Into<Response<B>>,
S::InitError: fmt::Debug,
B: MessageBody,
X: ServiceFactory<Request = Request, Response = Request>,
X: ServiceFactory<Request, Response = Request>,
X::Error: Into<Error>,
X::InitError: fmt::Debug,
U: ServiceFactory<Request = (Request, Framed<T, Codec>), Response = ()>,
U: ServiceFactory<(Request, Framed<T, Codec>), Response = ()>,
U::Error: fmt::Display,
U::InitError: fmt::Debug,
{
@ -362,24 +360,29 @@ where
}
/// `Service` implementation for HTTP/1 transport
pub struct H1ServiceHandler<T, S: Service, B, X: Service, U: Service> {
pub struct H1ServiceHandler<T, S, B, X, U>
where
S: Service<Request>,
X: Service<Request>,
U: Service<(Request, Framed<T, Codec>)>,
{
srv: CloneableService<S>,
expect: CloneableService<X>,
upgrade: Option<CloneableService<U>>,
on_connect_ext: Option<Rc<ConnectCallback<T>>>,
cfg: ServiceConfig,
_t: PhantomData<(T, B)>,
_t: PhantomData<B>,
}
impl<T, S, B, X, U> H1ServiceHandler<T, S, B, X, U>
where
S: Service<Request = Request>,
S: Service<Request>,
S::Error: Into<Error>,
S::Response: Into<Response<B>>,
B: MessageBody,
X: Service<Request = Request, Response = Request>,
X: Service<Request, Response = Request>,
X::Error: Into<Error>,
U: Service<Request = (Request, Framed<T, Codec>), Response = ()>,
U: Service<(Request, Framed<T, Codec>), Response = ()>,
U::Error: fmt::Display,
{
fn new(
@ -400,19 +403,19 @@ where
}
}
impl<T, S, B, X, U> Service for H1ServiceHandler<T, S, B, X, U>
impl<T, S, B, X, U> Service<(T, Option<net::SocketAddr>)>
for H1ServiceHandler<T, S, B, X, U>
where
T: AsyncRead + AsyncWrite + Unpin,
S: Service<Request = Request>,
S: Service<Request>,
S::Error: Into<Error>,
S::Response: Into<Response<B>>,
B: MessageBody,
X: Service<Request = Request, Response = Request>,
X: Service<Request, Response = Request>,
X::Error: Into<Error>,
U: Service<Request = (Request, Framed<T, Codec>), Response = ()>,
U: Service<(Request, Framed<T, Codec>), Response = ()>,
U::Error: fmt::Display + Into<Error>,
{
type Request = (T, Option<net::SocketAddr>);
type Response = ();
type Error = DispatchError;
type Future = Dispatcher<T, S, B, X, U>;
@ -459,7 +462,7 @@ where
}
}
fn call(&mut self, (io, addr): Self::Request) -> Self::Future {
fn call(&mut self, (io, addr): (T, Option<net::SocketAddr>)) -> Self::Future {
let mut connect_extensions = Extensions::new();
if let Some(ref handler) = self.on_connect_ext {
// run on_connect_ext callback, populating connect extensions

15
actix-http/src/h1/upgrade.rs
View File

@ -1,4 +1,3 @@
use std::marker::PhantomData;
use std::task::{Context, Poll};
use actix_codec::Framed;
@ -9,14 +8,13 @@ use crate::error::Error;
use crate::h1::Codec;
use crate::request::Request;
pub struct UpgradeHandler<T>(pub(crate) PhantomData<T>);
pub struct UpgradeHandler;
impl<T> ServiceFactory for UpgradeHandler<T> {
type Config = ();
type Request = (Request, Framed<T, Codec>);
impl<T> ServiceFactory<(Request, Framed<T, Codec>)> for UpgradeHandler {
type Response = ();
type Error = Error;
type Service = UpgradeHandler<T>;
type Config = ();
type Service = UpgradeHandler;
type InitError = Error;
type Future = Ready<Result<Self::Service, Self::InitError>>;
@ -25,8 +23,7 @@ impl<T> ServiceFactory for UpgradeHandler<T> {
}
}
impl<T> Service for UpgradeHandler<T> {
type Request = (Request, Framed<T, Codec>);
impl<T> Service<(Request, Framed<T, Codec>)> for UpgradeHandler {
type Response = ();
type Error = Error;
type Future = Ready<Result<Self::Response, Self::Error>>;
@ -35,7 +32,7 @@ impl<T> Service for UpgradeHandler<T> {
Poll::Ready(Ok(()))
}
fn call(&mut self, _: Self::Request) -> Self::Future {
fn call(&mut self, _: (Request, Framed<T, Codec>)) -> Self::Future {
ready(Ok(()))
}
}

14
actix-http/src/h2/dispatcher.rs
View File

@ -6,7 +6,7 @@ use std::pin::Pin;
use std::task::{Context, Poll};
use actix_codec::{AsyncRead, AsyncWrite};
use actix_rt::time::{Delay, Instant};
use actix_rt::time::{Instant, Sleep};
use actix_service::Service;
use bytes::{Bytes, BytesMut};
use h2::server::{Connection, SendResponse};
@ -29,9 +29,11 @@ const CHUNK_SIZE: usize = 16_384;
/// Dispatcher for HTTP/2 protocol
#[pin_project::pin_project]
pub struct Dispatcher<T, S: Service<Request = Request>, B: MessageBody>
pub struct Dispatcher<T, S, B>
where
T: AsyncRead + AsyncWrite + Unpin,
S: Service<Request>,
B: MessageBody,
{
service: CloneableService<S>,
connection: Connection<T, Bytes>,
@ -39,14 +41,14 @@ where
config: ServiceConfig,
peer_addr: Option<net::SocketAddr>,
ka_expire: Instant,
ka_timer: Option<Delay>,
ka_timer: Option<Sleep>,
_t: PhantomData<B>,
}
impl<T, S, B> Dispatcher<T, S, B>
where
T: AsyncRead + AsyncWrite + Unpin,
S: Service<Request = Request>,
S: Service<Request>,
S::Error: Into<Error>,
// S::Future: 'static,
S::Response: Into<Response<B>>,
@ -57,7 +59,7 @@ where
connection: Connection<T, Bytes>,
on_connect_data: Extensions,
config: ServiceConfig,
timeout: Option<Delay>,
timeout: Option<Sleep>,
peer_addr: Option<net::SocketAddr>,
) -> Self {
// let keepalive = config.keep_alive_enabled();
@ -92,7 +94,7 @@ where
impl<T, S, B> Future for Dispatcher<T, S, B>
where
T: AsyncRead + AsyncWrite + Unpin,
S: Service<Request = Request>,
S: Service<Request>,
S::Error: Into<Error> + 'static,
S::Future: 'static,
S::Response: Into<Response<B>> + 'static,

81
actix-http/src/h2/service.rs
View File

@ -36,14 +36,14 @@ pub struct H2Service<T, S, B> {
impl<T, S, B> H2Service<T, S, B>
where
S: ServiceFactory<Config = (), Request = Request>,
S: ServiceFactory<Request, Config = ()>,
S::Error: Into<Error> + 'static,
S::Response: Into<Response<B>> + 'static,
<S::Service as Service>::Future: 'static,
<S::Service as Service<Request>>::Future: 'static,
B: MessageBody + 'static,
{
/// Create new `HttpService` instance with config.
pub(crate) fn with_config<F: IntoServiceFactory<S>>(
pub(crate) fn with_config<F: IntoServiceFactory<S, Request>>(
cfg: ServiceConfig,
service: F,
) -> Self {
@ -64,18 +64,18 @@ where
impl<S, B> H2Service<TcpStream, S, B>
where
S: ServiceFactory<Config = (), Request = Request>,
S: ServiceFactory<Request, Config = ()>,
S::Error: Into<Error> + 'static,
S::Response: Into<Response<B>> + 'static,
<S::Service as Service>::Future: 'static,
<S::Service as Service<Request>>::Future: 'static,
B: MessageBody + 'static,
{
/// Create simple tcp based service
pub fn tcp(
self,
) -> impl ServiceFactory<
TcpStream,
Config = (),
Request = TcpStream,
Response = (),
Error = DispatchError,
InitError = S::InitError,
@ -92,18 +92,18 @@ where
#[cfg(feature = "openssl")]
mod openssl {
use actix_service::{fn_factory, fn_service};
use actix_tls::openssl::{Acceptor, SslAcceptor, SslStream};
use actix_tls::{openssl::HandshakeError, TlsError};
use actix_service::{fn_factory, fn_service, ServiceFactoryExt};
use actix_tls::accept::openssl::{Acceptor, SslAcceptor, SslError, SslStream};
use actix_tls::accept::TlsError;
use super::*;
impl<S, B> H2Service<SslStream<TcpStream>, S, B>
where
S: ServiceFactory<Config = (), Request = Request>,
S: ServiceFactory<Request, Config = ()>,
S::Error: Into<Error> + 'static,
S::Response: Into<Response<B>> + 'static,
<S::Service as Service>::Future: 'static,
<S::Service as Service<Request>>::Future: 'static,
B: MessageBody + 'static,
{
/// Create ssl based service
@ -111,10 +111,10 @@ mod openssl {
self,
acceptor: SslAcceptor,
) -> impl ServiceFactory<
TcpStream,
Config = (),
Request = TcpStream,
Response = (),
Error = TlsError<HandshakeError<TcpStream>, DispatchError>,
Error = TlsError<SslError, DispatchError>,
InitError = S::InitError,
> {
pipeline_factory(
@ -136,16 +136,17 @@ mod openssl {
#[cfg(feature = "rustls")]
mod rustls {
use super::*;
use actix_tls::rustls::{Acceptor, ServerConfig, TlsStream};
use actix_tls::TlsError;
use actix_service::ServiceFactoryExt;
use actix_tls::accept::rustls::{Acceptor, ServerConfig, TlsStream};
use actix_tls::accept::TlsError;
use std::io;
impl<S, B> H2Service<TlsStream<TcpStream>, S, B>
where
S: ServiceFactory<Config = (), Request = Request>,
S: ServiceFactory<Request, Config = ()>,
S::Error: Into<Error> + 'static,
S::Response: Into<Response<B>> + 'static,
<S::Service as Service>::Future: 'static,
<S::Service as Service<Request>>::Future: 'static,
B: MessageBody + 'static,
{
/// Create openssl based service
@ -153,8 +154,8 @@ mod rustls {
self,
mut config: ServerConfig,
) -> impl ServiceFactory<
TcpStream,
Config = (),
Request = TcpStream,
Response = (),
Error = TlsError<io::Error, DispatchError>,
InitError = S::InitError,
@ -178,21 +179,20 @@ mod rustls {
}
}
impl<T, S, B> ServiceFactory for H2Service<T, S, B>
impl<T, S, B> ServiceFactory<(T, Option<net::SocketAddr>)> for H2Service<T, S, B>
where
T: AsyncRead + AsyncWrite + Unpin,
S: ServiceFactory<Config = (), Request = Request>,
S: ServiceFactory<Request, Config = ()>,
S::Error: Into<Error> + 'static,
S::Response: Into<Response<B>> + 'static,
<S::Service as Service>::Future: 'static,
<S::Service as Service<Request>>::Future: 'static,
B: MessageBody + 'static,
{
type Config = ();
type Request = (T, Option<net::SocketAddr>);
type Response = ();
type Error = DispatchError;
type InitError = S::InitError;
type Config = ();
type Service = H2ServiceHandler<T, S::Service, B>;
type InitError = S::InitError;
type Future = H2ServiceResponse<T, S, B>;
fn new_service(&self, _: ()) -> Self::Future {
@ -207,21 +207,24 @@ where
#[doc(hidden)]
#[pin_project::pin_project]
pub struct H2ServiceResponse<T, S: ServiceFactory, B> {
pub struct H2ServiceResponse<T, S, B>
where
S: ServiceFactory<Request>,
{
#[pin]
fut: S::Future,
cfg: Option<ServiceConfig>,
on_connect_ext: Option<Rc<ConnectCallback<T>>>,
_t: PhantomData<(T, B)>,
_t: PhantomData<B>,
}
impl<T, S, B> Future for H2ServiceResponse<T, S, B>
where
T: AsyncRead + AsyncWrite + Unpin,
S: ServiceFactory<Config = (), Request = Request>,
S: ServiceFactory<Request, Config = ()>,
S::Error: Into<Error> + 'static,
S::Response: Into<Response<B>> + 'static,
<S::Service as Service>::Future: 'static,
<S::Service as Service<Request>>::Future: 'static,
B: MessageBody + 'static,
{
type Output = Result<H2ServiceHandler<T, S::Service, B>, S::InitError>;
@ -241,16 +244,19 @@ where
}
/// `Service` implementation for http/2 transport
pub struct H2ServiceHandler<T, S: Service, B> {
pub struct H2ServiceHandler<T, S, B>
where
S: Service<Request>,
{
srv: CloneableService<S>,
cfg: ServiceConfig,
on_connect_ext: Option<Rc<ConnectCallback<T>>>,
_t: PhantomData<(T, B)>,
_t: PhantomData<B>,
}
impl<T, S, B> H2ServiceHandler<T, S, B>
where
S: Service<Request = Request>,
S: Service<Request>,
S::Error: Into<Error> + 'static,
S::Future: 'static,
S::Response: Into<Response<B>> + 'static,
@ -270,16 +276,15 @@ where
}
}
impl<T, S, B> Service for H2ServiceHandler<T, S, B>
impl<T, S, B> Service<(T, Option<net::SocketAddr>)> for H2ServiceHandler<T, S, B>
where
T: AsyncRead + AsyncWrite + Unpin,
S: Service<Request = Request>,
S: Service<Request>,
S::Error: Into<Error> + 'static,
S::Future: 'static,
S::Response: Into<Response<B>> + 'static,
B: MessageBody + 'static,
{
type Request = (T, Option<net::SocketAddr>);
type Response = ();
type Error = DispatchError;
type Future = H2ServiceHandlerResponse<T, S, B>;
@ -292,7 +297,7 @@ where
})
}
fn call(&mut self, (io, addr): Self::Request) -> Self::Future {
fn call(&mut self, (io, addr): (T, Option<net::SocketAddr>)) -> Self::Future {
let mut connect_extensions = Extensions::new();
if let Some(ref handler) = self.on_connect_ext {
// run on_connect_ext callback, populating connect extensions
@ -311,7 +316,7 @@ where
}
}
enum State<T, S: Service<Request = Request>, B: MessageBody>
enum State<T, S: Service<Request>, B: MessageBody>
where
T: AsyncRead + AsyncWrite + Unpin,
S::Future: 'static,
@ -329,7 +334,7 @@ where
pub struct H2ServiceHandlerResponse<T, S, B>
where
T: AsyncRead + AsyncWrite + Unpin,
S: Service<Request = Request>,
S: Service<Request>,
S::Error: Into<Error> + 'static,
S::Future: 'static,
S::Response: Into<Response<B>> + 'static,
@ -341,7 +346,7 @@ where
impl<T, S, B> Future for H2ServiceHandlerResponse<T, S, B>
where
T: AsyncRead + AsyncWrite + Unpin,
S: Service<Request = Request>,
S: Service<Request>,
S::Error: Into<Error> + 'static,
S::Future: 'static,
S::Response: Into<Response<B>> + 'static,

3
actix-http/src/header/shared/httpdate.rs
View File

@ -3,7 +3,8 @@ use std::io::Write;
use std::str::FromStr;
use std::time::{SystemTime, UNIX_EPOCH};
use bytes::{buf::BufMutExt, BytesMut};
use bytes::buf::BufMut;
use bytes::BytesMut;
use http::header::{HeaderValue, InvalidHeaderValue};
use time::{offset, OffsetDateTime, PrimitiveDateTime};

8
actix-http/src/request.rs
View File

@ -23,6 +23,10 @@ impl<P> HttpMessage for Request<P> {
&self.head().headers
}
fn take_payload(&mut self) -> Payload<P> {
std::mem::replace(&mut self.payload, Payload::None)
}
/// Request extensions
#[inline]
fn extensions(&self) -> Ref<'_, Extensions> {
@ -34,10 +38,6 @@ impl<P> HttpMessage for Request<P> {
fn extensions_mut(&self) -> RefMut<'_, Extensions> {
self.head.extensions_mut()
}
fn take_payload(&mut self) -> Payload<P> {
std::mem::replace(&mut self.payload, Payload::None)
}
}
impl From<Message<RequestHead>> for Request<PayloadStream> {

193
actix-http/src/service.rs
View File

@ -22,22 +22,22 @@ use crate::response::Response;
use crate::{h1, h2::Dispatcher, ConnectCallback, Extensions, Protocol};
/// A `ServiceFactory` for HTTP/1.1 or HTTP/2 protocol.
pub struct HttpService<T, S, B, X = h1::ExpectHandler, U = h1::UpgradeHandler<T>> {
pub struct HttpService<T, S, B, X = h1::ExpectHandler, U = h1::UpgradeHandler> {
srv: S,
cfg: ServiceConfig,
expect: X,
upgrade: Option<U>,
on_connect_ext: Option<Rc<ConnectCallback<T>>>,
_t: PhantomData<(T, B)>,
_t: PhantomData<B>,
}
impl<T, S, B> HttpService<T, S, B>
where
S: ServiceFactory<Config = (), Request = Request>,
S: ServiceFactory<Request, Config = ()>,
S::Error: Into<Error> + 'static,
S::InitError: fmt::Debug,
S::Response: Into<Response<B>> + 'static,
<S::Service as Service>::Future: 'static,
<S::Service as Service<Request>>::Future: 'static,
B: MessageBody + 'static,
{
/// Create builder for `HttpService` instance.
@ -48,15 +48,15 @@ where
impl<T, S, B> HttpService<T, S, B>
where
S: ServiceFactory<Config = (), Request = Request>,
S: ServiceFactory<Request, Config = ()>,
S::Error: Into<Error> + 'static,
S::InitError: fmt::Debug,
S::Response: Into<Response<B>> + 'static,
<S::Service as Service>::Future: 'static,
<S::Service as Service<Request>>::Future: 'static,
B: MessageBody + 'static,
{
/// Create new `HttpService` instance.
pub fn new<F: IntoServiceFactory<S>>(service: F) -> Self {
pub fn new<F: IntoServiceFactory<S, Request>>(service: F) -> Self {
let cfg = ServiceConfig::new(KeepAlive::Timeout(5), 5000, 0, false, None);
HttpService {
@ -70,7 +70,7 @@ where
}
/// Create new `HttpService` instance with config.
pub(crate) fn with_config<F: IntoServiceFactory<S>>(
pub(crate) fn with_config<F: IntoServiceFactory<S, Request>>(
cfg: ServiceConfig,
service: F,
) -> Self {
@ -87,11 +87,11 @@ where
impl<T, S, B, X, U> HttpService<T, S, B, X, U>
where
S: ServiceFactory<Config = (), Request = Request>,
S: ServiceFactory<Request, Config = ()>,
S::Error: Into<Error> + 'static,
S::InitError: fmt::Debug,
S::Response: Into<Response<B>> + 'static,
<S::Service as Service>::Future: 'static,
<S::Service as Service<Request>>::Future: 'static,
B: MessageBody,
{
/// Provide service for `EXPECT: 100-Continue` support.
@ -101,10 +101,10 @@ where
/// request will be forwarded to main service.
pub fn expect<X1>(self, expect: X1) -> HttpService<T, S, B, X1, U>
where
X1: ServiceFactory<Config = (), Request = Request, Response = Request>,
X1: ServiceFactory<Request, Config = (), Response = Request>,
X1::Error: Into<Error>,
X1::InitError: fmt::Debug,
<X1::Service as Service>::Future: 'static,
<X1::Service as Service<Request>>::Future: 'static,
{
HttpService {
expect,
@ -122,14 +122,10 @@ where
/// and this service get called with original request and framed object.
pub fn upgrade<U1>(self, upgrade: Option<U1>) -> HttpService<T, S, B, X, U1>
where
U1: ServiceFactory<
Config = (),
Request = (Request, Framed<T, h1::Codec>),
Response = (),
>,
U1: ServiceFactory<(Request, Framed<T, h1::Codec>), Config = (), Response = ()>,
U1::Error: fmt::Display,
U1::InitError: fmt::Debug,
<U1::Service as Service>::Future: 'static,
<U1::Service as Service<(Request, Framed<T, h1::Codec>)>>::Future: 'static,
{
HttpService {
upgrade,
@ -150,31 +146,31 @@ where
impl<S, B, X, U> HttpService<TcpStream, S, B, X, U>
where
S: ServiceFactory<Config = (), Request = Request>,
S: ServiceFactory<Request, Config = ()>,
S::Error: Into<Error> + 'static,
S::InitError: fmt::Debug,
S::Response: Into<Response<B>> + 'static,
<S::Service as Service>::Future: 'static,
<S::Service as Service<Request>>::Future: 'static,
B: MessageBody + 'static,
X: ServiceFactory<Config = (), Request = Request, Response = Request>,
X: ServiceFactory<Request, Config = (), Response = Request>,
X::Error: Into<Error>,
X::InitError: fmt::Debug,
<X::Service as Service>::Future: 'static,
<X::Service as Service<Request>>::Future: 'static,
U: ServiceFactory<
(Request, Framed<TcpStream, h1::Codec>),
Config = (),
Request = (Request, Framed<TcpStream, h1::Codec>),
Response = (),
>,
U::Error: fmt::Display + Into<Error>,
U::InitError: fmt::Debug,
<U::Service as Service>::Future: 'static,
<U::Service as Service<(Request, Framed<TcpStream, h1::Codec>)>>::Future: 'static,
{
/// Create simple tcp stream service
pub fn tcp(
self,
) -> impl ServiceFactory<
TcpStream,
Config = (),
Request = TcpStream,
Response = (),
Error = DispatchError,
InitError = (),
@ -190,39 +186,40 @@ where
#[cfg(feature = "openssl")]
mod openssl {
use super::*;
use actix_tls::openssl::{Acceptor, SslAcceptor, SslStream};
use actix_tls::{openssl::HandshakeError, TlsError};
use actix_service::ServiceFactoryExt;
use actix_tls::accept::openssl::{Acceptor, SslAcceptor, SslError, SslStream};
use actix_tls::accept::TlsError;
impl<S, B, X, U> HttpService<SslStream<TcpStream>, S, B, X, U>
where
S: ServiceFactory<Config = (), Request = Request>,
S: ServiceFactory<Request, Config = ()>,
S::Error: Into<Error> + 'static,
S::InitError: fmt::Debug,
S::Response: Into<Response<B>> + 'static,
<S::Service as Service>::Future: 'static,
<S::Service as Service<Request>>::Future: 'static,
B: MessageBody + 'static,
X: ServiceFactory<Config = (), Request = Request, Response = Request>,
X: ServiceFactory<Request, Config = (), Response = Request>,
X::Error: Into<Error>,
X::InitError: fmt::Debug,
<X::Service as Service>::Future: 'static,
<X::Service as Service<Request>>::Future: 'static,
U: ServiceFactory<
(Request, Framed<SslStream<TcpStream>, h1::Codec>),
Config = (),
Request = (Request, Framed<SslStream<TcpStream>, h1::Codec>),
Response = (),
>,
U::Error: fmt::Display + Into<Error>,
U::InitError: fmt::Debug,
<U::Service as Service>::Future: 'static,
<U::Service as Service<(Request, Framed<SslStream<TcpStream>, h1::Codec>)>>::Future: 'static,
{
/// Create openssl based service
pub fn openssl(
self,
acceptor: SslAcceptor,
) -> impl ServiceFactory<
TcpStream,
Config = (),
Request = TcpStream,
Response = (),
Error = TlsError<HandshakeError<TcpStream>, DispatchError>,
Error = TlsError<SslError, DispatchError>,
InitError = (),
> {
pipeline_factory(
@ -250,39 +247,42 @@ mod openssl {
#[cfg(feature = "rustls")]
mod rustls {
use super::*;
use actix_tls::rustls::{Acceptor, ServerConfig, Session, TlsStream};
use actix_tls::TlsError;
use std::io;
use actix_tls::accept::rustls::{Acceptor, ServerConfig, Session, TlsStream};
use actix_tls::accept::TlsError;
use super::*;
use actix_service::ServiceFactoryExt;
impl<S, B, X, U> HttpService<TlsStream<TcpStream>, S, B, X, U>
where
S: ServiceFactory<Config = (), Request = Request>,
S: ServiceFactory<Request, Config = ()>,
S::Error: Into<Error> + 'static,
S::InitError: fmt::Debug,
S::Response: Into<Response<B>> + 'static,
<S::Service as Service>::Future: 'static,
<S::Service as Service<Request>>::Future: 'static,
B: MessageBody + 'static,
X: ServiceFactory<Config = (), Request = Request, Response = Request>,
X: ServiceFactory<Request, Config = (), Response = Request>,
X::Error: Into<Error>,
X::InitError: fmt::Debug,
<X::Service as Service>::Future: 'static,
<X::Service as Service<Request>>::Future: 'static,
U: ServiceFactory<
(Request, Framed<TlsStream<TcpStream>, h1::Codec>),
Config = (),
Request = (Request, Framed<TlsStream<TcpStream>, h1::Codec>),
Response = (),
>,
U::Error: fmt::Display + Into<Error>,
U::InitError: fmt::Debug,
<U::Service as Service>::Future: 'static,
<U::Service as Service<(Request, Framed<TlsStream<TcpStream>, h1::Codec>)>>::Future: 'static,
{
/// Create openssl based service
pub fn rustls(
self,
mut config: ServerConfig,
) -> impl ServiceFactory<
TcpStream,
Config = (),
Request = TcpStream,
Response = (),
Error = TlsError<io::Error, DispatchError>,
InitError = (),
@ -313,34 +313,30 @@ mod rustls {
}
}
impl<T, S, B, X, U> ServiceFactory for HttpService<T, S, B, X, U>
impl<T, S, B, X, U> ServiceFactory<(T, Protocol, Option<net::SocketAddr>)>
for HttpService<T, S, B, X, U>
where
T: AsyncRead + AsyncWrite + Unpin,
S: ServiceFactory<Config = (), Request = Request>,
S: ServiceFactory<Request, Config = ()>,
S::Error: Into<Error> + 'static,
S::InitError: fmt::Debug,
S::Response: Into<Response<B>> + 'static,
<S::Service as Service>::Future: 'static,
<S::Service as Service<Request>>::Future: 'static,
B: MessageBody + 'static,
X: ServiceFactory<Config = (), Request = Request, Response = Request>,
X: ServiceFactory<Request, Config = (), Response = Request>,
X::Error: Into<Error>,
X::InitError: fmt::Debug,
<X::Service as Service>::Future: 'static,
U: ServiceFactory<
Config = (),
Request = (Request, Framed<T, h1::Codec>),
Response = (),
>,
<X::Service as Service<Request>>::Future: 'static,
U: ServiceFactory<(Request, Framed<T, h1::Codec>), Config = (), Response = ()>,
U::Error: fmt::Display + Into<Error>,
U::InitError: fmt::Debug,
<U::Service as Service>::Future: 'static,
<U::Service as Service<(Request, Framed<T, h1::Codec>)>>::Future: 'static,
{
type Config = ();
type Request = (T, Protocol, Option<net::SocketAddr>);
type Response = ();
type Error = DispatchError;
type InitError = ();
type Config = ();
type Service = HttpServiceHandler<T, S::Service, B, X::Service, U::Service>;
type InitError = ();
type Future = HttpServiceResponse<T, S, B, X, U>;
fn new_service(&self, _: ()) -> Self::Future {
@ -359,13 +355,12 @@ where
#[doc(hidden)]
#[pin_project]
pub struct HttpServiceResponse<
T,
S: ServiceFactory,
B,
X: ServiceFactory,
U: ServiceFactory,
> {
pub struct HttpServiceResponse<T, S, B, X, U>
where
S: ServiceFactory<Request>,
X: ServiceFactory<Request>,
U: ServiceFactory<(Request, Framed<T, h1::Codec>)>,
{
#[pin]
fut: S::Future,
#[pin]
@ -382,20 +377,20 @@ pub struct HttpServiceResponse<
impl<T, S, B, X, U> Future for HttpServiceResponse<T, S, B, X, U>
where
T: AsyncRead + AsyncWrite + Unpin,
S: ServiceFactory<Request = Request>,
S: ServiceFactory<Request>,
S::Error: Into<Error> + 'static,
S::InitError: fmt::Debug,
S::Response: Into<Response<B>> + 'static,
<S::Service as Service>::Future: 'static,
<S::Service as Service<Request>>::Future: 'static,
B: MessageBody + 'static,
X: ServiceFactory<Request = Request, Response = Request>,
X: ServiceFactory<Request, Response = Request>,
X::Error: Into<Error>,
X::InitError: fmt::Debug,
<X::Service as Service>::Future: 'static,
U: ServiceFactory<Request = (Request, Framed<T, h1::Codec>), Response = ()>,
<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>::Future: 'static,
<U::Service as Service<(Request, Framed<T, h1::Codec>)>>::Future: 'static,
{
type Output =
Result<HttpServiceHandler<T, S::Service, B, X::Service, U::Service>, ()>;
@ -440,25 +435,30 @@ where
}
/// `Service` implementation for http transport
pub struct HttpServiceHandler<T, S: Service, B, X: Service, U: Service> {
pub struct HttpServiceHandler<T, S, B, X, U>
where
S: Service<Request>,
X: Service<Request>,
U: Service<(Request, Framed<T, h1::Codec>)>,
{
srv: CloneableService<S>,
expect: CloneableService<X>,
upgrade: Option<CloneableService<U>>,
cfg: ServiceConfig,
on_connect_ext: Option<Rc<ConnectCallback<T>>>,
_t: PhantomData<(T, B, X)>,
_t: PhantomData<B>,
}
impl<T, S, B, X, U> HttpServiceHandler<T, S, B, X, U>
where
S: Service<Request = Request>,
S: Service<Request>,
S::Error: Into<Error> + 'static,
S::Future: 'static,
S::Response: Into<Response<B>> + 'static,
B: MessageBody + 'static,
X: Service<Request = Request, Response = Request>,
X: Service<Request, Response = Request>,
X::Error: Into<Error>,
U: Service<Request = (Request, Framed<T, h1::Codec>), Response = ()>,
U: Service<(Request, Framed<T, h1::Codec>), Response = ()>,
U::Error: fmt::Display,
{
fn new(
@ -479,20 +479,20 @@ where
}
}
impl<T, S, B, X, U> Service for HttpServiceHandler<T, S, B, X, U>
impl<T, S, B, X, U> Service<(T, Protocol, Option<net::SocketAddr>)>
for HttpServiceHandler<T, S, B, X, U>
where
T: AsyncRead + AsyncWrite + Unpin,
S: Service<Request = Request>,
S: Service<Request>,
S::Error: Into<Error> + 'static,
S::Future: 'static,
S::Response: Into<Response<B>> + 'static,
B: MessageBody + 'static,
X: Service<Request = Request, Response = Request>,
X: Service<Request, Response = Request>,
X::Error: Into<Error>,
U: Service<Request = (Request, Framed<T, h1::Codec>), Response = ()>,
U: Service<(Request, Framed<T, h1::Codec>), Response = ()>,
U::Error: fmt::Display + Into<Error>,
{
type Request = (T, Protocol, Option<net::SocketAddr>);
type Response = ();
type Error = DispatchError;
type Future = HttpServiceHandlerResponse<T, S, B, X, U>;
@ -539,7 +539,10 @@ where
}
}
fn call(&mut self, (io, proto, peer_addr): Self::Request) -> Self::Future {
fn call(
&mut self,
(io, proto, peer_addr): (T, Protocol, Option<net::SocketAddr>),
) -> Self::Future {
let mut connect_extensions = Extensions::new();
if let Some(ref handler) = self.on_connect_ext {
@ -575,14 +578,14 @@ where
#[pin_project(project = StateProj)]
enum State<T, S, B, X, U>
where
S: Service<Request = Request>,
S: Service<Request>,
S::Future: 'static,
S::Error: Into<Error>,
T: AsyncRead + AsyncWrite + Unpin,
B: MessageBody,
X: Service<Request = Request, Response = Request>,
X: Service<Request, Response = Request>,
X::Error: Into<Error>,
U: Service<Request = (Request, Framed<T, h1::Codec>), Response = ()>,
U: Service<(Request, Framed<T, h1::Codec>), Response = ()>,
U::Error: fmt::Display,
{
H1(#[pin] h1::Dispatcher<T, S, B, X, U>),
@ -602,14 +605,14 @@ where
pub struct HttpServiceHandlerResponse<T, S, B, X, U>
where
T: AsyncRead + AsyncWrite + Unpin,
S: Service<Request = Request>,
S: Service<Request>,
S::Error: Into<Error> + 'static,
S::Future: 'static,
S::Response: Into<Response<B>> + 'static,
B: MessageBody + 'static,
X: Service<Request = Request, Response = Request>,
X: Service<Request, Response = Request>,
X::Error: Into<Error>,
U: Service<Request = (Request, Framed<T, h1::Codec>), Response = ()>,
U: Service<(Request, Framed<T, h1::Codec>), Response = ()>,
U::Error: fmt::Display,
{
#[pin]
@ -619,14 +622,14 @@ where
impl<T, S, B, X, U> Future for HttpServiceHandlerResponse<T, S, B, X, U>
where
T: AsyncRead + AsyncWrite + Unpin,
S: Service<Request = Request>,
S: Service<Request>,
S::Error: Into<Error> + 'static,
S::Future: 'static,
S::Response: Into<Response<B>> + 'static,
B: MessageBody,
X: Service<Request = Request, Response = Request>,
X: Service<Request, Response = Request>,
X::Error: Into<Error>,
U: Service<Request = (Request, Framed<T, h1::Codec>), Response = ()>,
U: Service<(Request, Framed<T, h1::Codec>), Response = ()>,
U::Error: fmt::Display,
{
type Output = Result<(), DispatchError>;
@ -639,13 +642,13 @@ where
impl<T, S, B, X, U> State<T, S, B, X, U>
where
T: AsyncRead + AsyncWrite + Unpin,
S: Service<Request = Request>,
S: Service<Request>,
S::Error: Into<Error> + 'static,
S::Response: Into<Response<B>> + 'static,
B: MessageBody + 'static,
X: Service<Request = Request, Response = Request>,
X: Service<Request, Response = Request>,
X::Error: Into<Error>,
U: Service<Request = (Request, Framed<T, h1::Codec>), Response = ()>,
U: Service<(Request, Framed<T, h1::Codec>), Response = ()>,
U::Error: fmt::Display,
{
fn poll(

22
actix-http/src/test.rs
View File

@ -10,7 +10,7 @@ use std::{
task::{Context, Poll},
};
use actix_codec::{AsyncRead, AsyncWrite};
use actix_codec::{AsyncRead, AsyncWrite, ReadBuf};
use bytes::{Bytes, BytesMut};
use http::header::{self, HeaderName, HeaderValue};
use http::{Error as HttpError, Method, Uri, Version};
@ -251,9 +251,11 @@ impl AsyncRead for TestBuffer {
fn poll_read(
self: Pin<&mut Self>,
_: &mut Context<'_>,
buf: &mut [u8],
) -> Poll<io::Result<usize>> {
Poll::Ready(self.get_mut().read(buf))
buf: &mut ReadBuf<'_>,
) -> Poll<io::Result<()>> {
let dst = buf.initialize_unfilled();
let res = self.get_mut().read(dst).map(|n| buf.advance(n));
Poll::Ready(res)
}
}
@ -356,11 +358,15 @@ impl AsyncRead for TestSeqBuffer {
fn poll_read(
self: Pin<&mut Self>,
_: &mut Context<'_>,
buf: &mut [u8],
) -> Poll<io::Result<usize>> {
let r = self.get_mut().read(buf);
buf: &mut ReadBuf<'_>,
) -> Poll<io::Result<()>> {
let dst = buf.initialize_unfilled();
let r = self.get_mut().read(dst);
match r {
Ok(n) => Poll::Ready(Ok(n)),
Ok(n) => {
buf.advance(n);
Poll::Ready(Ok(()))
}
Err(err) if err.kind() == io::ErrorKind::WouldBlock => Poll::Pending,
Err(err) => Poll::Ready(Err(err)),
}

10
actix-http/src/ws/dispatcher.rs
View File

@ -11,7 +11,7 @@ use super::{Codec, Frame, Message};
#[pin_project::pin_project]
pub struct Dispatcher<S, T>
where
S: Service<Request = Frame, Response = Message> + 'static,
S: Service<Frame, Response = Message> + 'static,
T: AsyncRead + AsyncWrite,
{
#[pin]
@ -21,17 +21,17 @@ where
impl<S, T> Dispatcher<S, T>
where
T: AsyncRead + AsyncWrite,
S: Service<Request = Frame, Response = Message>,
S: Service<Frame, Response = Message>,
S::Future: 'static,
S::Error: 'static,
{
pub fn new<F: IntoService<S>>(io: T, service: F) -> Self {
pub fn new<F: IntoService<S, Frame>>(io: T, service: F) -> Self {
Dispatcher {
inner: InnerDispatcher::new(Framed::new(io, Codec::new()), service),
}
}
pub fn with<F: IntoService<S>>(framed: Framed<T, Codec>, service: F) -> Self {
pub fn with<F: IntoService<S, Frame>>(framed: Framed<T, Codec>, service: F) -> Self {
Dispatcher {
inner: InnerDispatcher::new(framed, service),
}
@ -41,7 +41,7 @@ where
impl<S, T> Future for Dispatcher<S, T>
where
T: AsyncRead + AsyncWrite,
S: Service<Request = Frame, Response = Message>,
S: Service<Frame, Response = Message>,
S::Future: 'static,
S::Error: 'static,
{