actix/actix-net
1
0
Fork 0
mirror of https://github.com/fafhrd91/actix-net synced 2024-11-27 18:02:58 +01:00
Code Issues Releases Wiki Activity

remove pin-project; update Unpin consrtaint

Browse Source
This commit is contained in:
Nikolay Kim 2019-11-18 18:28:54 +06:00
parent 7404d82a9b
commit 1354946460
22 changed files with 225 additions and 161 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

17
actix-codec/src/framed.rs
View File

@ -225,6 +225,15 @@ impl<T, U> Framed<T, U> {
self.inner.get_mut().force_send(item)
}
pub fn is_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), U::Error>>
where
T: AsyncWrite + Unpin,
U: Encoder + Unpin,
U::Error: From<io::Error>,
{
Pin::new(&mut self.inner.get_mut()).poll_ready(cx)
}
pub fn next_item(&mut self, cx: &mut Context<'_>) -> Poll<Option<Result<U::Item, U::Error>>>
where
T: AsyncRead + Unpin,
@ -240,6 +249,14 @@ impl<T, U> Framed<T, U> {
{
Pin::new(self.inner.get_mut()).poll_flush(cx)
}
pub fn close(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), U::Error>>
where
T: AsyncWrite + Unpin,
U: Encoder + Unpin,
{
Pin::new(&mut self.inner.get_mut()).poll_close(cx)
}
}
impl<T, U> Stream for Framed<T, U>

1
actix-connect/Cargo.toml
View File

@ -40,7 +40,6 @@ actix-rt = "1.0.0-alpha.1"
derive_more = "0.15"
either = "1.5.2"
futures = "0.3.1"
pin-project = "0.4.5"
http = { version = "0.1.17", optional = true }
log = "0.4"
tokio-net = "=0.2.0-alpha.6"

6
actix-connect/src/connector.rs
View File

@ -8,7 +8,6 @@ use std::task::{Context, Poll};
use actix_service::{Service, ServiceFactory};
use futures::future::{err, ok, BoxFuture, Either, FutureExt, Ready};
use pin_project::pin_project;
use tokio_net::tcp::TcpStream;
use super::connect::{Address, Connect, Connection};
@ -94,7 +93,6 @@ impl<T: Address> Service for TcpConnector<T> {
}
}
#[pin_project]
#[doc(hidden)]
/// Tcp stream connector response future
pub struct TcpConnectorResponse<T> {
@ -137,7 +135,7 @@ impl<T: Address> Future for TcpConnectorResponse<T> {
type Output = Result<Connection<T, TcpStream>, ConnectError>;
fn poll(self: Pin<&mut Self>, cx: &mut Context) -> Poll<Self::Output> {
let this = self.project();
let this = self.get_mut();
// connect
loop {
@ -167,7 +165,7 @@ impl<T: Address> Future for TcpConnectorResponse<T> {
// try to connect
let addr = this.addrs.as_mut().unwrap().pop_front().unwrap();
*this.stream = Some(TcpStream::connect(addr).boxed());
this.stream = Some(TcpStream::connect(addr).boxed());
}
}
}

7
actix-connect/src/resolver.rs
View File

@ -6,7 +6,6 @@ use std::task::{Context, Poll};
use actix_service::{Service, ServiceFactory};
use futures::future::{ok, Either, Ready};
use pin_project::pin_project;
use trust_dns_resolver::lookup_ip::LookupIpFuture;
use trust_dns_resolver::{AsyncResolver, Background};
@ -129,12 +128,10 @@ impl<T: Address> Service for Resolver<T> {
}
}
#[pin_project]
#[doc(hidden)]
/// Resolver future
pub struct ResolverFuture<T: Address> {
req: Option<Connect<T>>,
#[pin]
lookup: Background<LookupIpFuture>,
}
@ -157,9 +154,9 @@ impl<T: Address> Future for ResolverFuture<T> {
type Output = Result<Connect<T>, ConnectError>;
fn poll(self: Pin<&mut Self>, cx: &mut Context) -> Poll<Self::Output> {
let this = self.project();
let this = self.get_mut();
match this.lookup.poll(cx) {
match Pin::new(&mut this.lookup).poll(cx) {
Poll::Pending => Poll::Pending,
Poll::Ready(Ok(ips)) => {
let req = this.req.take().unwrap();

1
actix-ioframe/Cargo.toml
View File

@ -24,7 +24,6 @@ actix-utils = "0.5.0-alpha.1"
bytes = "0.4"
either = "1.5.2"
futures = "0.3.1"
pin-project = "0.4.5"
tokio-executor = "=0.2.0-alpha.6"
log = "0.4"

20
actix-ioframe/src/connect.rs
View File

@ -5,7 +5,6 @@ use std::task::{Context, Poll};
use actix_codec::{AsyncRead, AsyncWrite, Decoder, Encoder, Framed};
use actix_utils::mpsc;
use futures::Stream;
use pin_project::pin_project;
use crate::dispatcher::FramedMessage;
use crate::sink::Sink;
@ -42,10 +41,8 @@ where
}
}
#[pin_project]
pub struct ConnectResult<Io, St, Codec: Encoder + Decoder> {
pub(crate) state: St,
#[pin]
pub(crate) framed: Framed<Io, Codec>,
pub(crate) rx: mpsc::Receiver<FramedMessage<<Codec as Encoder>::Item>>,
pub(crate) sink: Sink<<Codec as Encoder>::Item>,
@ -78,6 +75,13 @@ impl<Io, St, Codec: Encoder + Decoder> ConnectResult<Io, St, Codec> {
}
}
impl<Io, St, Codec> Unpin for ConnectResult<Io, St, Codec>
where
Io: AsyncRead + AsyncWrite + Unpin,
Codec: Encoder + Decoder + Unpin,
{
}
impl<Io, St, Codec> Stream for ConnectResult<Io, St, Codec>
where
Io: AsyncRead + AsyncWrite + Unpin,
@ -86,7 +90,7 @@ where
type Item = Result<<Codec as Decoder>::Item, <Codec as Decoder>::Error>;
fn poll_next(self: Pin<&mut Self>, cx: &mut Context) -> Poll<Option<Self::Item>> {
self.project().framed.poll_next(cx)
self.get_mut().framed.next_item(cx)
}
}
@ -98,21 +102,21 @@ where
type Error = <Codec as Encoder>::Error;
fn poll_ready(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
self.project().framed.poll_ready(cx)
self.get_mut().framed.is_ready(cx)
}
fn start_send(
self: Pin<&mut Self>,
item: <Codec as Encoder>::Item,
) -> Result<(), Self::Error> {
self.project().framed.start_send(item)
self.get_mut().framed.write(item)
}
fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
self.project().framed.poll_flush(cx)
self.get_mut().framed.flush(cx)
}
fn poll_close(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
self.project().framed.poll_close(cx)
self.get_mut().framed.close(cx)
}
}

47
actix-ioframe/src/dispatcher.rs
View File

@ -1,6 +1,5 @@
//! Framed dispatcher service and related utilities
use std::collections::VecDeque;
use std::future::Future;
use std::mem;
use std::pin::Pin;
use std::rc::Rc;
@ -31,7 +30,6 @@ pub(crate) enum FramedMessage<T> {
/// FramedTransport - is a future that reads frames from Framed object
/// and pass then to the service.
#[pin_project::pin_project]
pub(crate) struct FramedDispatcher<St, S, T, U>
where
S: Service<Request = Request<St, U>, Response = Option<Response<U>>>,
@ -123,32 +121,45 @@ struct FramedDispatcherInner<I, E> {
task: LocalWaker,
}
impl<St, S, T, U> Future for FramedDispatcher<St, S, T, U>
impl<St, S, T, U> Unpin for FramedDispatcher<St, S, T, U>
where
S: Service<Request = Request<St, U>, Response = Option<Response<U>>>,
S::Error: 'static,
S::Future: 'static,
S: Service<Request = Request<St, U>, Response = Option<Response<U>>> + Unpin,
S::Error: Unpin + 'static,
S::Future: Unpin + 'static,
T: AsyncRead + AsyncWrite + Unpin,
U: Decoder + Encoder + Unpin,
<U as Encoder>::Item: 'static,
<U as Encoder>::Error: std::fmt::Debug,
{
type Output = Result<(), ServiceError<S::Error, U>>;
}
fn poll(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll<Self::Output> {
unsafe { self.inner.get_ref().task.register(cx.waker()) };
impl<St, S, T, U> FramedDispatcher<St, S, T, U>
where
S: Service<Request = Request<St, U>, Response = Option<Response<U>>> + Unpin,
S::Error: 'static,
S::Future: Unpin + 'static,
T: AsyncRead + AsyncWrite + Unpin,
U: Decoder + Encoder + Unpin,
<U as Encoder>::Item: 'static,
<U as Encoder>::Error: std::fmt::Debug,
{
pub(crate) fn poll(
&mut self,
cx: &mut Context,
) -> Poll<Result<(), ServiceError<S::Error, U>>> {
let this = self;
unsafe { this.inner.get_ref().task.register(cx.waker()) };
let this = self.project();
poll(
cx,
this.service,
this.state,
this.sink,
this.framed,
this.dispatch_state,
this.rx,
this.inner,
this.disconnect,
&mut this.service,
&mut this.state,
&mut this.sink,
&mut this.framed,
&mut this.dispatch_state,
&mut this.rx,
&mut this.inner,
&mut this.disconnect,
)
}
}

99
actix-ioframe/src/service.rs
View File

@ -8,7 +8,6 @@ use actix_codec::{AsyncRead, AsyncWrite, Decoder, Encoder};
use actix_service::{IntoService, IntoServiceFactory, Service, ServiceFactory};
use either::Either;
use futures::future::{FutureExt, LocalBoxFuture};
use pin_project::{pin_project, project};
use crate::connect::{Connect, ConnectResult};
use crate::dispatcher::FramedDispatcher;
@ -77,6 +76,7 @@ where
Io: AsyncRead + AsyncWrite + Unpin,
C: Service<Request = Connect<Io>, Response = ConnectResult<Io, St, Codec>>,
C::Error: 'static,
C::Future: Unpin,
Codec: Decoder + Encoder + Unpin,
<Codec as Encoder>::Item: 'static,
<Codec as Encoder>::Error: std::fmt::Debug,
@ -93,10 +93,7 @@ where
}
/// Provide stream items handler service and construct service factory.
pub fn finish<F, T>(
self,
service: F,
) -> impl Service<Request = Io, Response = (), Error = ServiceError<C::Error, Codec>>
pub fn finish<F, T>(self, service: F) -> FramedServiceImpl<St, C, T, Io, Codec>
where
F: IntoServiceFactory<T>,
T: ServiceFactory<
@ -106,6 +103,9 @@ where
Error = C::Error,
InitError = C::Error,
> + 'static,
T::Future: Unpin,
T::Service: Unpin,
<T::Service as Service>::Future: Unpin + 'static,
{
FramedServiceImpl {
connect: self.connect,
@ -132,7 +132,8 @@ where
Response = ConnectResult<Io, St, Codec>,
>,
C::Error: 'static,
C::Future: 'static,
C::Future: Unpin + 'static,
<C::Service as Service>::Future: Unpin,
Codec: Decoder + Encoder + Unpin,
<Codec as Encoder>::Item: 'static,
<Codec as Encoder>::Error: std::fmt::Debug,
@ -148,15 +149,7 @@ where
self
}
pub fn finish<F, T, Cfg>(
self,
service: F,
) -> impl ServiceFactory<
Config = Cfg,
Request = Io,
Response = (),
Error = ServiceError<C::Error, Codec>,
>
pub fn finish<F, T, Cfg>(self, service: F) -> FramedService<St, C, T, Io, Codec, Cfg>
where
F: IntoServiceFactory<T>,
T: ServiceFactory<
@ -166,6 +159,9 @@ where
Error = C::Error,
InitError = C::Error,
> + 'static,
T::Future: Unpin,
T::Service: Unpin,
<T::Service as Service>::Future: Unpin + 'static,
{
FramedService {
connect: self.connect,
@ -176,7 +172,7 @@ where
}
}
pub(crate) struct FramedService<St, C, T, Io, Codec, Cfg> {
pub struct FramedService<St, C, T, Io, Codec, Cfg> {
connect: C,
handler: Rc<T>,
disconnect: Option<Rc<dyn Fn(&mut St, bool)>>,
@ -193,7 +189,8 @@ where
Response = ConnectResult<Io, St, Codec>,
>,
C::Error: 'static,
C::Future: 'static,
C::Future: Unpin + 'static,
<C::Service as Service>::Future: Unpin,
T: ServiceFactory<
Config = St,
Request = RequestItem<St, Codec>,
@ -201,6 +198,9 @@ where
Error = C::Error,
InitError = C::Error,
> + 'static,
T::Future: Unpin,
T::Service: Unpin,
<T::Service as Service>::Future: Unpin + 'static,
Codec: Decoder + Encoder + Unpin,
<Codec as Encoder>::Item: 'static,
<Codec as Encoder>::Error: std::fmt::Debug,
@ -244,6 +244,7 @@ where
Io: AsyncRead + AsyncWrite + Unpin,
C: Service<Request = Connect<Io>, Response = ConnectResult<Io, St, Codec>>,
C::Error: 'static,
C::Future: Unpin,
T: ServiceFactory<
Config = St,
Request = RequestItem<St, Codec>,
@ -251,7 +252,9 @@ where
Error = C::Error,
InitError = C::Error,
>,
<<T as ServiceFactory>::Service as Service>::Future: 'static,
T::Future: Unpin,
T::Service: Unpin,
<T::Service as Service>::Future: Unpin + 'static,
Codec: Decoder + Encoder + Unpin,
<Codec as Encoder>::Item: 'static,
<Codec as Encoder>::Error: std::fmt::Debug,
@ -276,11 +279,11 @@ where
}
}
#[pin_project]
pub struct FramedServiceImplResponse<St, Io, Codec, C, T>
where
C: Service<Request = Connect<Io>, Response = ConnectResult<Io, St, Codec>>,
C::Error: 'static,
C::Future: Unpin,
T: ServiceFactory<
Config = St,
Request = RequestItem<St, Codec>,
@ -288,7 +291,9 @@ where
Error = C::Error,
InitError = C::Error,
>,
<<T as ServiceFactory>::Service as Service>::Future: 'static,
T::Future: Unpin,
T::Service: Unpin,
<T::Service as Service>::Future: Unpin + 'static,
Io: AsyncRead + AsyncWrite + Unpin,
Codec: Encoder + Decoder + Unpin,
<Codec as Encoder>::Item: 'static,
@ -297,9 +302,10 @@ where
inner: FramedServiceImplResponseInner<St, Io, Codec, C, T>,
}
impl<St, Io, Codec, C, T> Future for FramedServiceImplResponse<St, Io, Codec, C, T>
impl<St, Io, Codec, C, T> Unpin for FramedServiceImplResponse<St, Io, Codec, C, T>
where
C: Service<Request = Connect<Io>, Response = ConnectResult<Io, St, Codec>>,
C::Future: Unpin,
C::Error: 'static,
T: ServiceFactory<
Config = St,
@ -308,7 +314,31 @@ where
Error = C::Error,
InitError = C::Error,
>,
<<T as ServiceFactory>::Service as Service>::Future: 'static,
T::Future: Unpin,
T::Service: Unpin,
<T::Service as Service>::Future: Unpin + 'static,
Io: AsyncRead + AsyncWrite + Unpin,
Codec: Encoder + Decoder + Unpin,
<Codec as Encoder>::Item: 'static,
<Codec as Encoder>::Error: std::fmt::Debug,
{
}
impl<St, Io, Codec, C, T> Future for FramedServiceImplResponse<St, Io, Codec, C, T>
where
C: Service<Request = Connect<Io>, Response = ConnectResult<Io, St, Codec>>,
C::Future: Unpin,
C::Error: 'static,
T: ServiceFactory<
Config = St,
Request = RequestItem<St, Codec>,
Response = ResponseItem<Codec>,
Error = C::Error,
InitError = C::Error,
>,
T::Future: Unpin,
T::Service: Unpin,
<T::Service as Service>::Future: Unpin + 'static,
Io: AsyncRead + AsyncWrite + Unpin,
Codec: Encoder + Decoder + Unpin,
<Codec as Encoder>::Item: 'static,
@ -320,7 +350,7 @@ where
let this = self.get_mut();
loop {
match unsafe { Pin::new_unchecked(&mut this.inner) }.poll(cx) {
match this.inner.poll(cx) {
Either::Left(new) => this.inner = new,
Either::Right(poll) => return poll,
};
@ -328,10 +358,10 @@ where
}
}
#[pin_project]
enum FramedServiceImplResponseInner<St, Io, Codec, C, T>
where
C: Service<Request = Connect<Io>, Response = ConnectResult<Io, St, Codec>>,
C::Future: Unpin,
C::Error: 'static,
T: ServiceFactory<
Config = St,
@ -340,15 +370,17 @@ where
Error = C::Error,
InitError = C::Error,
>,
<<T as ServiceFactory>::Service as Service>::Future: 'static,
T::Future: Unpin,
T::Service: Unpin,
<T::Service as Service>::Future: Unpin + 'static,
Io: AsyncRead + AsyncWrite + Unpin,
Codec: Encoder + Decoder + Unpin,
<Codec as Encoder>::Item: 'static,
<Codec as Encoder>::Error: std::fmt::Debug,
{
Connect(#[pin] C::Future, Rc<T>, Option<Rc<dyn Fn(&mut St, bool)>>),
Connect(C::Future, Rc<T>, Option<Rc<dyn Fn(&mut St, bool)>>),
Handler(
#[pin] T::Future,
T::Future,
Option<ConnectResult<Io, St, Codec>>,
Option<Rc<dyn Fn(&mut St, bool)>>,
),
@ -358,6 +390,7 @@ where
impl<St, Io, Codec, C, T> FramedServiceImplResponseInner<St, Io, Codec, C, T>
where
C: Service<Request = Connect<Io>, Response = ConnectResult<Io, St, Codec>>,
C::Future: Unpin,
C::Error: 'static,
T: ServiceFactory<
Config = St,
@ -366,22 +399,22 @@ where
Error = C::Error,
InitError = C::Error,
>,
<<T as ServiceFactory>::Service as Service>::Future: 'static,
T::Future: Unpin,
T::Service: Unpin,
<T::Service as Service>::Future: Unpin + 'static,
Io: AsyncRead + AsyncWrite + Unpin,
Codec: Encoder + Decoder + Unpin,
<Codec as Encoder>::Item: 'static,
<Codec as Encoder>::Error: std::fmt::Debug,
{
#[project]
fn poll(
self: Pin<&mut Self>,
&mut self,
cx: &mut Context,
) -> Either<
FramedServiceImplResponseInner<St, Io, Codec, C, T>,
Poll<Result<(), ServiceError<C::Error, Codec>>>,
> {
#[project]
match self.project() {
match self {
FramedServiceImplResponseInner::Connect(
ref mut fut,
ref handler,
@ -417,7 +450,7 @@ where
Poll::Ready(Err(e)) => Either::Right(Poll::Ready(Err(e.into()))),
},
FramedServiceImplResponseInner::Dispatcher(ref mut fut) => {
Either::Right(Pin::new(fut).poll(cx))
Either::Right(fut.poll(cx))
}
}
}

1
actix-server/Cargo.toml
View File

@ -33,7 +33,6 @@ actix-server-config = "0.3.0-alpha.1"
log = "0.4"
num_cpus = "1.0"
pin-project = "0.4.5"
mio = "0.6.19"
net2 = "0.2"
futures = "0.3.1"

18
actix-server/src/ssl/openssl.rs
View File

@ -6,7 +6,6 @@ use std::task::{Context, Poll};
use actix_service::{Service, ServiceFactory};
use futures::future::{ok, FutureExt, LocalBoxFuture, Ready};
use open_ssl::ssl::SslAcceptor;
use pin_project::pin_project;
use tokio_io::{AsyncRead, AsyncWrite};
use tokio_openssl::{HandshakeError, SslStream};
@ -41,7 +40,9 @@ impl<T: AsyncRead + AsyncWrite, P> Clone for OpensslAcceptor<T, P> {
}
}
impl<T: AsyncRead + AsyncWrite + Unpin + 'static, P> ServiceFactory for OpensslAcceptor<T, P> {
impl<T: AsyncRead + AsyncWrite + Unpin + 'static, P: Unpin> ServiceFactory
for OpensslAcceptor<T, P>
{
type Request = Io<T, P>;
type Response = Io<SslStream<T>, P>;
type Error = HandshakeError<T>;
@ -69,7 +70,9 @@ pub struct OpensslAcceptorService<T, P> {
io: PhantomData<(T, P)>,
}
impl<T: AsyncRead + AsyncWrite + Unpin + 'static, P> Service for OpensslAcceptorService<T, P> {
impl<T: AsyncRead + AsyncWrite + Unpin + 'static, P: Unpin> Service
for OpensslAcceptorService<T, P>
{
type Request = Io<T, P>;
type Response = Io<SslStream<T>, P>;
type Error = HandshakeError<T>;
@ -98,24 +101,23 @@ impl<T: AsyncRead + AsyncWrite + Unpin + 'static, P> Service for OpensslAcceptor
}
}
#[pin_project]
pub struct OpensslAcceptorServiceFut<T, P>
where
P: Unpin,
T: AsyncRead + AsyncWrite,
{
#[pin]
fut: LocalBoxFuture<'static, Result<SslStream<T>, HandshakeError<T>>>,
params: Option<P>,
_guard: CounterGuard,
}
impl<T: AsyncRead + AsyncWrite, P> Future for OpensslAcceptorServiceFut<T, P> {
impl<T: AsyncRead + AsyncWrite, P: Unpin> Future for OpensslAcceptorServiceFut<T, P> {
type Output = Result<Io<SslStream<T>, P>, HandshakeError<T>>;
fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
let this = self.project();
let this = self.get_mut();
let io = futures::ready!(this.fut.poll(cx))?;
let io = futures::ready!(Pin::new(&mut this.fut).poll(cx))?;
let proto = if let Some(protos) = io.ssl().selected_alpn_protocol() {
const H2: &[u8] = b"\x02h2";
const HTTP10: &[u8] = b"\x08http/1.0";

14
actix-server/src/ssl/rustls.rs
View File

@ -7,7 +7,6 @@ use std::task::{Context, Poll};
use actix_service::{Service, ServiceFactory};
use futures::future::{ok, Ready};
use pin_project::pin_project;
use rust_tls::ServerConfig;
use tokio_io::{AsyncRead, AsyncWrite};
use tokio_rustls::{server::TlsStream, Accept, TlsAcceptor};
@ -43,7 +42,7 @@ impl<T, P> Clone for RustlsAcceptor<T, P> {
}
}
impl<T: AsyncRead + AsyncWrite + Unpin, P> ServiceFactory for RustlsAcceptor<T, P> {
impl<T: AsyncRead + AsyncWrite + Unpin, P: Unpin> ServiceFactory for RustlsAcceptor<T, P> {
type Request = Io<T, P>;
type Response = Io<TlsStream<T>, P>;
type Error = io::Error;
@ -72,7 +71,7 @@ pub struct RustlsAcceptorService<T, P> {
conns: Counter,
}
impl<T: AsyncRead + AsyncWrite + Unpin, P> Service for RustlsAcceptorService<T, P> {
impl<T: AsyncRead + AsyncWrite + Unpin, P: Unpin> Service for RustlsAcceptorService<T, P> {
type Request = Io<T, P>;
type Response = Io<TlsStream<T>, P>;
type Error = io::Error;
@ -96,25 +95,24 @@ impl<T: AsyncRead + AsyncWrite + Unpin, P> Service for RustlsAcceptorService<T,
}
}
#[pin_project]
pub struct RustlsAcceptorServiceFut<T, P>
where
T: AsyncRead + AsyncWrite + Unpin,
P: Unpin,
{
#[pin]
fut: Accept<T>,
params: Option<P>,
_guard: CounterGuard,
}
impl<T: AsyncRead + AsyncWrite + Unpin, P> Future for RustlsAcceptorServiceFut<T, P> {
impl<T: AsyncRead + AsyncWrite + Unpin, P: Unpin> Future for RustlsAcceptorServiceFut<T, P> {
type Output = Result<Io<TlsStream<T>, P>, io::Error>;
fn poll(self: Pin<&mut Self>, cx: &mut Context) -> Poll<Self::Output> {
let this = self.project();
let this = self.get_mut();
let params = this.params.take().unwrap();
Poll::Ready(
futures::ready!(this.fut.poll(cx))
futures::ready!(Pin::new(&mut this.fut).poll(cx))
.map(move |io| Io::from_parts(io, params, Protocol::Unknown)),
)
}

17
actix-service/src/and_then.rs
View File

@ -154,10 +154,8 @@ where
InitError = A::InitError,
>,
A::Future: Unpin,
A::Service: Unpin,
<A::Service as Service>::Future: Unpin,
B::Future: Unpin,
B::Service: Unpin,
<B::Service as Service>::Future: Unpin,
{
type Request = A::Request;
@ -204,10 +202,8 @@ where
A: ServiceFactory,
B: ServiceFactory<Request = A::Response>,
A::Future: Unpin,
A::Service: Unpin,
<A::Service as Service>::Future: Unpin,
B::Future: Unpin,
B::Service: Unpin,
<B::Service as Service>::Future: Unpin,
{
fn new(fut_a: A::Future, fut_b: B::Future) -> Self {
@ -220,15 +216,24 @@ where
}
}
impl<A, B> Unpin for AndThenServiceFactoryResponse<A, B>
where
A: ServiceFactory,
B: ServiceFactory<Request = A::Response, Error = A::Error, InitError = A::InitError>,
A::Future: Unpin,
<A::Service as Service>::Future: Unpin,
B::Future: Unpin,
<B::Service as Service>::Future: Unpin,
{
}
impl<A, B> Future for AndThenServiceFactoryResponse<A, B>
where
A: ServiceFactory,
B: ServiceFactory<Request = A::Response, Error = A::Error, InitError = A::InitError>,
A::Future: Unpin,
A::Service: Unpin,
<A::Service as Service>::Future: Unpin,
B::Future: Unpin,
B::Service: Unpin,
<B::Service as Service>::Future: Unpin,
{
type Output = Result<AndThenService<A::Service, B::Service>, A::InitError>;

3
actix-service/src/apply.rs
View File

@ -23,7 +23,8 @@ pub fn apply_fn_factory<T, F, R, In, Out, Err, U>(
) -> ApplyServiceFactory<T, F, R, In, Out, Err>
where
T: ServiceFactory<Error = Err>,
F: FnMut(In, &mut T::Service) -> R + Clone,
T::Future: Unpin,
F: FnMut(In, &mut T::Service) -> R + Unpin + Clone,
R: Future<Output = Result<Out, Err>>,
U: IntoServiceFactory<T>,
{

6
actix-service/src/lib.rs
View File

@ -73,7 +73,6 @@ pub trait Service {
fn map<F, R>(self, f: F) -> crate::dev::Map<Self, F, R>
where
Self: Sized,
Self::Future: Unpin,
F: FnMut(Self::Response) -> R + Unpin,
{
crate::dev::Map::new(self, f)
@ -90,7 +89,6 @@ pub trait Service {
fn map_err<F, E>(self, f: F) -> crate::dev::MapErr<Self, F, E>
where
Self: Sized,
Self::Future: Unpin,
F: Fn(Self::Error) -> E,
{
crate::dev::MapErr::new(self, f)
@ -140,7 +138,6 @@ pub trait ServiceFactory {
fn map<F, R>(self, f: F) -> crate::map::MapServiceFactory<Self, F, R>
where
Self: Sized,
<Self::Service as Service>::Future: Unpin,
F: FnMut(Self::Response) -> R + Unpin + Clone,
{
crate::map::MapServiceFactory::new(self, f)
@ -150,7 +147,6 @@ pub trait ServiceFactory {
fn map_err<F, E>(self, f: F) -> crate::map_err::MapErrServiceFactory<Self, F, E>
where
Self: Sized,
<Self::Service as Service>::Future: Unpin,
F: Fn(Self::Error) -> E + Unpin + Clone,
{
crate::map_err::MapErrServiceFactory::new(self, f)
@ -160,7 +156,6 @@ pub trait ServiceFactory {
fn map_init_err<F, E>(self, f: F) -> crate::map_init_err::MapInitErr<Self, F, E>
where
Self: Sized,
<Self::Service as Service>::Future: Unpin,
F: Fn(Self::InitError) -> E + Unpin + Clone,
{
crate::map_init_err::MapInitErr::new(self, f)
@ -299,6 +294,7 @@ pub mod dev {
FnService, FnServiceConfig, FnServiceFactory, FnServiceNoConfig,
};
pub use crate::map::{Map, MapServiceFactory};
pub use crate::map_config::{MapConfig, UnitConfig};
pub use crate::map_err::{MapErr, MapErrServiceFactory};
pub use crate::map_init_err::MapInitErr;
pub use crate::then::{ThenService, ThenServiceFactory};

31
actix-service/src/map_config.rs
View File

@ -8,16 +8,7 @@ pub enum MappedConfig<'a, T> {
}
/// Adapt external config to a config for provided new service
pub fn map_config<T, F, C>(
factory: T,
f: F,
) -> impl ServiceFactory<
Config = C,
Request = T::Request,
Response = T::Response,
Error = T::Error,
InitError = T::InitError,
>
pub fn map_config<T, F, C>(factory: T, f: F) -> MapConfig<T, F, C>
where
T: ServiceFactory,
F: Fn(&C) -> MappedConfig<T::Config>,
@ -26,23 +17,15 @@ where
}
/// Replace config with unit
pub fn unit_config<T, C>(
new_service: T,
) -> impl ServiceFactory<
Config = C,
Request = T::Request,
Response = T::Response,
Error = T::Error,
InitError = T::InitError,
>
pub fn unit_config<T, C>(new_service: T) -> UnitConfig<T, C>
where
T: ServiceFactory<Config = ()>,
{
UnitConfig::new(new_service)
}
/// `MapInitErr` service combinator
pub(crate) struct MapConfig<A, F, C> {
/// `.map_config()` service combinator
pub struct MapConfig<A, F, C> {
a: A,
f: F,
e: PhantomData<C>,
@ -50,7 +33,7 @@ pub(crate) struct MapConfig<A, F, C> {
impl<A, F, C> MapConfig<A, F, C> {
/// Create new `MapConfig` combinator
pub fn new(a: A, f: F) -> Self
pub(crate) fn new(a: A, f: F) -> Self
where
A: ServiceFactory,
F: Fn(&C) -> MappedConfig<A::Config>,
@ -99,8 +82,8 @@ where
}
}
/// `MapInitErr` service combinator
pub(crate) struct UnitConfig<A, C> {
/// `unit_config()` config combinator
pub struct UnitConfig<A, C> {
a: A,
e: PhantomData<C>,
}

2
actix-service/src/pipeline.rs
View File

@ -43,7 +43,7 @@ impl<T: Service> Pipeline<T> {
where
Self: Sized,
F: IntoService<U>,
U: Service<Request = T::Response, Error = T::Error> + Unpin,
U: Service<Request = T::Response, Error = T::Error>,
{
Pipeline {
service: AndThenService::new(self.service, service.into_service()),

22
actix-service/src/then.rs
View File

@ -149,10 +149,8 @@ where
InitError = A::InitError,
>,
A::Future: Unpin,
A::Service: Unpin,
<A::Service as Service>::Future: Unpin,
B::Future: Unpin,
B::Service: Unpin,
<B::Service as Service>::Future: Unpin,
{
type Request = A::Request;
@ -208,10 +206,8 @@ where
InitError = A::InitError,
>,
A::Future: Unpin,
A::Service: Unpin,
<A::Service as Service>::Future: Unpin,
B::Future: Unpin,
B::Service: Unpin,
<B::Service as Service>::Future: Unpin,
{
fn new(fut_a: A::Future, fut_b: B::Future) -> Self {
@ -224,6 +220,22 @@ where
}
}
impl<A, B> Unpin for ThenServiceFactoryResponse<A, B>
where
A: ServiceFactory,
B: ServiceFactory<
Config = A::Config,
Request = Result<A::Response, A::Error>,
Error = A::Error,
InitError = A::InitError,
>,
A::Future: Unpin,
<A::Service as Service>::Future: Unpin,
B::Future: Unpin,
<B::Service as Service>::Future: Unpin,
{
}
impl<A, B> Future for ThenServiceFactoryResponse<A, B>
where
A: ServiceFactory,
@ -234,10 +246,8 @@ where
InitError = A::InitError,
>,
A::Future: Unpin,
A::Service: Unpin,
<A::Service as Service>::Future: Unpin,
B::Future: Unpin,
B::Service: Unpin,
<B::Service as Service>::Future: Unpin,
{
type Output = Result<ThenService<A::Service, B::Service>, A::InitError>;

1
actix-utils/Cargo.toml
View File

@ -23,7 +23,6 @@ actix-codec = "0.2.0-alpha.1"
bytes = "0.4"
either = "1.5.2"
futures = "0.3.1"
pin-project = "0.4.5"
tokio-timer = "0.3.0-alpha.6"
tokio-executor = { version="=0.2.0-alpha.6", features=["current-thread"] }
log = "0.4"

23
actix-utils/src/either.rs
View File

@ -4,7 +4,6 @@ use std::task::{Context, Poll};
use actix_service::{Service, ServiceFactory};
use futures::{future, ready, Future};
use pin_project::pin_project;
/// Combine two different service types into a single type.
///
@ -84,6 +83,8 @@ where
Error = A::Error,
InitError = A::InitError,
>,
A::Future: Unpin,
B::Future: Unpin,
{
type Request = either::Either<A::Request, B::Request>;
type Response = A::Response;
@ -112,32 +113,40 @@ impl<A: Clone, B: Clone> Clone for Either<A, B> {
}
}
#[pin_project]
#[doc(hidden)]
pub struct EitherNewService<A: ServiceFactory, B: ServiceFactory> {
left: Option<A::Service>,
right: Option<B::Service>,
#[pin]
left_fut: A::Future,
#[pin]
right_fut: B::Future,
}
impl<A, B> Unpin for EitherNewService<A, B>
where
A: ServiceFactory,
B: ServiceFactory<Response = A::Response, Error = A::Error, InitError = A::InitError>,
A::Future: Unpin,
B::Future: Unpin,
{
}
impl<A, B> Future for EitherNewService<A, B>
where
A: ServiceFactory,
B: ServiceFactory<Response = A::Response, Error = A::Error, InitError = A::InitError>,
A::Future: Unpin,
B::Future: Unpin,
{
type Output = Result<EitherService<A::Service, B::Service>, A::InitError>;
fn poll(self: Pin<&mut Self>, cx: &mut Context) -> Poll<Self::Output> {
let this = self.project();
let this = self.get_mut();
if this.left.is_none() {
*this.left = Some(ready!(this.left_fut.poll(cx))?);
this.left = Some(ready!(Pin::new(&mut this.left_fut).poll(cx))?);
}
if this.right.is_none() {
*this.right = Some(ready!(this.right_fut.poll(cx))?);
this.right = Some(ready!(Pin::new(&mut this.right_fut).poll(cx))?);
}
if this.left.is_some() && this.right.is_some() {

23
actix-utils/src/framed.rs
View File

@ -10,7 +10,6 @@ use actix_service::{IntoService, Service};
use futures::future::{ready, FutureExt};
use futures::{Future, Sink, Stream};
use log::debug;
use pin_project::pin_project;
use crate::cell::Cell;
use crate::mpsc;
@ -78,7 +77,6 @@ type Inner<S: Service, U> = Cell<FramedTransportInner<<U as Encoder>::Item, S::E
/// FramedTransport - is a future that reads frames from Framed object
/// and pass then to the service.
#[pin_project]
pub struct FramedTransport<S, T, U>
where
S: Service<Request = Request<U>, Response = Response<U>>,
@ -111,7 +109,7 @@ struct FramedTransportInner<I, E> {
impl<S, T, U> FramedTransport<S, T, U>
where
S: Service<Request = Request<U>, Response = Response<U>>,
S: Service<Request = Request<U>, Response = Response<U>> + Unpin,
S::Error: 'static,
S::Future: 'static,
T: AsyncRead + AsyncWrite + Unpin,
@ -167,27 +165,28 @@ where
impl<S, T, U> Future for FramedTransport<S, T, U>
where
S: Service<Request = Request<U>, Response = Response<U>>,
S::Error: 'static,
S: Service<Request = Request<U>, Response = Response<U>> + Unpin,
S::Error: Unpin + 'static,
S::Future: 'static,
T: AsyncRead + AsyncWrite + Unpin,
U: Decoder + Encoder + Unpin,
<U as Encoder>::Item: 'static,
<U as Encoder>::Error: std::fmt::Debug,
<U as Encoder>::Error: Unpin + std::fmt::Debug,
<U as Decoder>::Error: Unpin + std::fmt::Debug,
{
type Output = Result<(), FramedTransportError<S::Error, U>>;
fn poll(self: Pin<&mut Self>, cx: &mut Context) -> Poll<Self::Output> {
self.inner.get_ref().task.register(cx.waker());
let this = self.project();
let this = self.get_mut();
poll(
cx,
this.service,
this.state,
this.framed,
this.rx,
this.inner,
&mut this.service,
&mut this.state,
&mut this.framed,
&mut this.rx,
&mut this.inner,
)
}
}

17
actix-utils/src/inflight.rs
View File

@ -5,7 +5,6 @@ use std::task::{Context, Poll};
use actix_service::{IntoService, Service, Transform};
use futures::future::{ok, Ready};
use pin_project::pin_project;
use super::counter::{Counter, CounterGuard};
@ -29,7 +28,11 @@ impl Default for InFlight {
}
}
impl<S: Service> Transform<S> for InFlight {
impl<S> Transform<S> for InFlight
where
S: Service,
S::Future: Unpin,
{
type Request = S::Request;
type Response = S::Response;
type Error = S::Error;
@ -65,6 +68,7 @@ where
impl<T> Service for InFlightService<T>
where
T: Service,
T::Future: Unpin,
{
type Request = T::Request;
type Response = T::Response;
@ -90,19 +94,20 @@ where
}
}
#[pin_project]
#[doc(hidden)]
pub struct InFlightServiceResponse<T: Service> {
#[pin]
fut: T::Future,
_guard: CounterGuard,
}
impl<T: Service> Future for InFlightServiceResponse<T> {
impl<T: Service> Future for InFlightServiceResponse<T>
where
T::Future: Unpin,
{
type Output = Result<T::Response, T::Error>;
fn poll(self: Pin<&mut Self>, cx: &mut Context) -> Poll<Self::Output> {
self.project().fut.poll(cx)
Pin::new(&mut self.get_mut().fut).poll(cx)
}
}

10
actix-utils/src/timeout.rs
View File

@ -10,7 +10,6 @@ use std::{fmt, time};
use actix_service::{IntoService, Service, Transform};
use futures::future::{ok, Ready};
use pin_project::pin_project;
use tokio_timer::{clock, delay, Delay};
/// Applies a timeout to requests.
@ -85,6 +84,7 @@ impl<E> Clone for Timeout<E> {
impl<S, E> Transform<S> for Timeout<E>
where
S: Service,
S::Future: Unpin,
{
type Request = S::Request;
type Response = S::Response;
@ -126,6 +126,7 @@ where
impl<S> Service for TimeoutService<S>
where
S: Service,
S::Future: Unpin,
{
type Request = S::Request;
type Response = S::Response;
@ -145,10 +146,8 @@ where
}
/// `TimeoutService` response future
#[pin_project]
#[derive(Debug)]
pub struct TimeoutServiceResponse<T: Service> {
#[pin]
fut: T::Future,
sleep: Delay,
}
@ -156,14 +155,15 @@ pub struct TimeoutServiceResponse<T: Service> {
impl<T> Future for TimeoutServiceResponse<T>
where
T: Service,
T::Future: Unpin,
{
type Output = Result<T::Response, TimeoutError<T::Error>>;
fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
let mut this = self.project();
let this = self.get_mut();
// First, try polling the future
match this.fut.poll(cx) {
match Pin::new(&mut this.fut).poll(cx) {
Poll::Ready(Ok(v)) => return Poll::Ready(Ok(v)),
Poll::Ready(Err(e)) => return Poll::Ready(Err(TimeoutError::Service(e))),
Poll::Pending => {}