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

simplify FramedTransport

Browse Source
This commit is contained in:
Nikolay Kim 2018-09-24 20:40:31 -07:00
parent 7686011c1a
commit d4b772d454
1 changed files with 68 additions and 201 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

269
src/framed.rs
View File

@ -1,9 +1,9 @@
//! Framed dispatcher service and related utilities //! Framed dispatcher service and related utilities
use std::fmt;
use std::marker::PhantomData; use std::marker::PhantomData;
use std::mem;
use actix; use actix;
use futures::future::{ok, Either, FutureResult, Join}; use futures::future::{ok, Either, FutureResult};
use futures::unsync::mpsc; use futures::unsync::mpsc;
use futures::{Async, AsyncSink, Future, Poll, Sink, Stream}; use futures::{Async, AsyncSink, Future, Poll, Sink, Stream};
use tokio_codec::{Decoder, Encoder, Framed}; use tokio_codec::{Decoder, Encoder, Framed};
@ -14,110 +14,97 @@ use service::{IntoNewService, IntoService, NewService, Service};
type Request<U> = <U as Decoder>::Item; type Request<U> = <U as Decoder>::Item;
type Response<U> = <U as Encoder>::Item; type Response<U> = <U as Encoder>::Item;
pub struct FramedNewService<S, T, U, E> { pub struct FramedNewService<S, T, U> {
factory: S, factory: S,
error_handler: E,
_t: PhantomData<(T, U)>, _t: PhantomData<(T, U)>,
} }
impl<S, T, U> FramedNewService<S, T, U, DefaultErrorHandler<S, U, S::InitError>> impl<S, T, U> FramedNewService<S, T, U>
where where
T: AsyncRead + AsyncWrite, T: AsyncRead + AsyncWrite,
U: Decoder + Encoder, U: Decoder + Encoder,
S: NewService<Request = Request<U>, Response = Option<Response<U>>> + Clone, S: NewService<Request = Request<U>, Response = Option<Response<U>>> + Clone,
<<S as NewService>::Service as Service>::Future: 'static, <<S as NewService>::Service as Service>::Future: 'static,
<<S as NewService>::Service as Service>::Error: fmt::Debug + 'static, <<S as NewService>::Service as Service>::Error: 'static,
<U as Encoder>::Item: 'static, <U as Encoder>::Item: 'static,
<U as Encoder>::Error: fmt::Debug + 'static, <U as Encoder>::Error: 'static,
<U as Encoder>::Error: fmt::Debug + 'static,
{ {
pub fn new<F1: IntoNewService<S>>(factory: F1) -> Self { pub fn new<F1: IntoNewService<S>>(factory: F1) -> Self {
Self { Self {
factory: factory.into_new_service(), factory: factory.into_new_service(),
error_handler: DefaultErrorHandler(PhantomData),
_t: PhantomData, _t: PhantomData,
} }
} }
} }
impl<S, T, U, E> Clone for FramedNewService<S, T, U, E> impl<S, T, U> Clone for FramedNewService<S, T, U>
where where
S: Clone, S: Clone,
E: Clone,
{ {
fn clone(&self) -> Self { fn clone(&self) -> Self {
Self { Self {
factory: self.factory.clone(), factory: self.factory.clone(),
error_handler: self.error_handler.clone(),
_t: PhantomData, _t: PhantomData,
} }
} }
} }
impl<S, T, U, E> NewService for FramedNewService<S, T, U, E> impl<S, T, U> NewService for FramedNewService<S, T, U>
where where
T: AsyncRead + AsyncWrite, T: AsyncRead + AsyncWrite,
U: Decoder + Encoder, U: Decoder + Encoder,
S: NewService<Request = Request<U>, Response = Option<Response<U>>> + Clone, S: NewService<Request = Request<U>, Response = Option<Response<U>>> + Clone,
E: NewService<Request = TransportError<S::Service, U>, InitError = S::InitError> + Clone,
<<S as NewService>::Service as Service>::Future: 'static, <<S as NewService>::Service as Service>::Future: 'static,
<<S as NewService>::Service as Service>::Error: fmt::Debug + 'static, <<S as NewService>::Service as Service>::Error: 'static,
<U as Encoder>::Item: 'static, <U as Encoder>::Item: 'static,
<U as Decoder>::Error: fmt::Debug + 'static, <U as Encoder>::Error: 'static,
<U as Encoder>::Error: fmt::Debug + 'static,
{ {
type Request = Framed<T, U>; type Request = Framed<T, U>;
type Response = FramedTransport<S::Service, T, U, E::Service>; type Response = FramedTransport<S::Service, T, U>;
type Error = S::InitError; type Error = S::InitError;
type InitError = S::InitError; type InitError = S::InitError;
type Service = FramedService<S, T, U, E>; type Service = FramedService<S, T, U>;
type Future = FutureResult<Self::Service, Self::InitError>; type Future = FutureResult<Self::Service, Self::InitError>;
fn new_service(&self) -> Self::Future { fn new_service(&self) -> Self::Future {
ok(FramedService { ok(FramedService {
factory: self.factory.clone(), factory: self.factory.clone(),
error_service: self.error_handler.clone(),
_t: PhantomData, _t: PhantomData,
}) })
} }
} }
pub struct FramedService<S, T, U, E> { pub struct FramedService<S, T, U> {
factory: S, factory: S,
error_service: E,
_t: PhantomData<(T, U)>, _t: PhantomData<(T, U)>,
} }
impl<S, T, U, E> Clone for FramedService<S, T, U, E> impl<S, T, U> Clone for FramedService<S, T, U>
where where
S: Clone, S: Clone,
E: Clone,
{ {
fn clone(&self) -> Self { fn clone(&self) -> Self {
Self { Self {
factory: self.factory.clone(), factory: self.factory.clone(),
error_service: self.error_service.clone(),
_t: PhantomData, _t: PhantomData,
} }
} }
} }
impl<S, T, U, E> Service for FramedService<S, T, U, E> impl<S, T, U> Service for FramedService<S, T, U>
where where
T: AsyncRead + AsyncWrite, T: AsyncRead + AsyncWrite,
U: Decoder + Encoder, U: Decoder + Encoder,
S: NewService<Request = Request<U>, Response = Option<Response<U>>>, S: NewService<Request = Request<U>, Response = Option<Response<U>>>,
E: NewService<Request = TransportError<S::Service, U>, InitError = S::InitError>,
<<S as NewService>::Service as Service>::Future: 'static, <<S as NewService>::Service as Service>::Future: 'static,
<<S as NewService>::Service as Service>::Error: fmt::Debug + 'static, <<S as NewService>::Service as Service>::Error: 'static,
<U as Encoder>::Item: 'static, <U as Encoder>::Item: 'static,
<U as Decoder>::Error: fmt::Debug + 'static, <U as Encoder>::Error: 'static,
<U as Encoder>::Error: fmt::Debug + 'static,
{ {
type Request = Framed<T, U>; type Request = Framed<T, U>;
type Response = FramedTransport<S::Service, T, U, E::Service>; type Response = FramedTransport<S::Service, T, U>;
type Error = S::InitError; type Error = S::InitError;
type Future = FramedServiceResponseFuture<S, T, U, E>; type Future = FramedServiceResponseFuture<S, T, U>;
fn poll_ready(&mut self) -> Poll<(), Self::Error> { fn poll_ready(&mut self) -> Poll<(), Self::Error> {
Ok(Async::Ready(())) Ok(Async::Ready(()))
@ -125,129 +112,69 @@ where
fn call(&mut self, req: Self::Request) -> Self::Future { fn call(&mut self, req: Self::Request) -> Self::Future {
FramedServiceResponseFuture { FramedServiceResponseFuture {
fut: self fut: self.factory.new_service(),
.factory
.new_service()
.join(self.error_service.new_service()),
framed: Some(req), framed: Some(req),
} }
} }
} }
#[doc(hidden)] #[doc(hidden)]
pub struct FramedServiceResponseFuture<S, T, U, E> pub struct FramedServiceResponseFuture<S, T, U>
where where
T: AsyncRead + AsyncWrite, T: AsyncRead + AsyncWrite,
U: Decoder + Encoder, U: Decoder + Encoder,
S: NewService<Request = Request<U>, Response = Option<Response<U>>>, S: NewService<Request = Request<U>, Response = Option<Response<U>>>,
E: NewService<Request = TransportError<S::Service, U>, InitError = S::InitError>,
<<S as NewService>::Service as Service>::Future: 'static, <<S as NewService>::Service as Service>::Future: 'static,
<<S as NewService>::Service as Service>::Error: fmt::Debug + 'static, <<S as NewService>::Service as Service>::Error: 'static,
<U as Encoder>::Item: 'static, <U as Encoder>::Item: 'static,
<U as Decoder>::Error: fmt::Debug + 'static, <U as Encoder>::Error: 'static,
<U as Encoder>::Error: fmt::Debug + 'static,
{ {
fut: Join<S::Future, E::Future>, fut: S::Future,
framed: Option<Framed<T, U>>, framed: Option<Framed<T, U>>,
} }
impl<S, T, U, E> Future for FramedServiceResponseFuture<S, T, U, E> impl<S, T, U> Future for FramedServiceResponseFuture<S, T, U>
where where
T: AsyncRead + AsyncWrite, T: AsyncRead + AsyncWrite,
U: Decoder + Encoder, U: Decoder + Encoder,
S: NewService<Request = Request<U>, Response = Option<Response<U>>>, S: NewService<Request = Request<U>, Response = Option<Response<U>>>,
E: NewService<Request = TransportError<S::Service, U>, InitError = S::InitError>,
<<S as NewService>::Service as Service>::Future: 'static, <<S as NewService>::Service as Service>::Future: 'static,
<<S as NewService>::Service as Service>::Error: fmt::Debug + 'static, <<S as NewService>::Service as Service>::Error: 'static,
<U as Encoder>::Item: 'static, <U as Encoder>::Item: 'static,
<U as Decoder>::Error: fmt::Debug + 'static, <U as Encoder>::Error: 'static,
<U as Encoder>::Error: fmt::Debug + 'static,
{ {
type Item = FramedTransport<S::Service, T, U, E::Service>; type Item = FramedTransport<S::Service, T, U>;
type Error = S::InitError; type Error = S::InitError;
fn poll(&mut self) -> Poll<Self::Item, Self::Error> { fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
match self.fut.poll()? { match self.fut.poll()? {
Async::NotReady => Ok(Async::NotReady), Async::NotReady => Ok(Async::NotReady),
Async::Ready((service, error_service)) => { Async::Ready(service) => Ok(Async::Ready(FramedTransport::new(
Ok(Async::Ready(FramedTransport::with_error_service( self.framed.take().unwrap(),
self.framed.take().unwrap(), service,
service, ))),
error_service,
)))
}
} }
} }
} }
pub enum TransportError<S: Service, U: Encoder + Decoder> { /// Framed transport errors
pub enum FramedTransportError<S: Service, U: Encoder + Decoder> {
Decoder(<U as Decoder>::Error), Decoder(<U as Decoder>::Error),
Encoder(<U as Encoder>::Error), Encoder(<U as Encoder>::Error),
Service(S::Error), Service(S::Error),
} }
/// Default error handling service
pub struct DefaultErrorHandler<S, U, E>(PhantomData<(S, U, E)>);
impl<S, U, E> Service for DefaultErrorHandler<S, U, E>
where
S: Service,
U: Encoder + Decoder,
S::Error: fmt::Debug,
<U as Decoder>::Error: fmt::Debug,
<U as Encoder>::Error: fmt::Debug,
{
type Request = TransportError<S, U>;
type Response = ();
type Error = ();
type Future = FutureResult<Self::Response, Self::Error>;
fn poll_ready(&mut self) -> Poll<(), Self::Error> {
Ok(Async::Ready(()))
}
fn call(&mut self, req: Self::Request) -> Self::Future {
match req {
TransportError::Service(err) => debug!("Service error: {:?}", err),
TransportError::Decoder(err) => trace!("Service decoder error: {:?}", err),
TransportError::Encoder(err) => trace!("Service encoder error: {:?}", err),
}
ok(())
}
}
impl<S, U, E> NewService for DefaultErrorHandler<S, U, E>
where
S: Service,
U: Encoder + Decoder,
S::Error: fmt::Debug,
<U as Decoder>::Error: fmt::Debug,
<U as Encoder>::Error: fmt::Debug,
{
type Request = TransportError<S, U>;
type Response = ();
type Error = ();
type InitError = E;
type Service = DefaultErrorHandler<S, U, ()>;
type Future = FutureResult<Self::Service, Self::InitError>;
fn new_service(&self) -> Self::Future {
ok(DefaultErrorHandler(PhantomData))
}
}
/// FramedTransport - is a future that reads frames from Framed object /// FramedTransport - is a future that reads frames from Framed object
/// and pass then to the service. /// and pass then to the service.
pub struct FramedTransport<S, T, U, E> pub struct FramedTransport<S, T, U>
where where
S: Service, S: Service,
T: AsyncRead + AsyncWrite, T: AsyncRead + AsyncWrite,
U: Encoder + Decoder, U: Encoder + Decoder,
E: Service,
{ {
service: S, service: S,
error_service: E, state: TransportState<S, U>,
state: TransportState<E>,
framed: Framed<T, U>, framed: Framed<T, U>,
request: Option<Request<U>>, request: Option<Request<U>>,
response: Option<Response<U>>, response: Option<Response<U>>,
@ -256,23 +183,21 @@ where
flushed: bool, flushed: bool,
} }
enum TransportState<E: Service> { enum TransportState<S: Service, U: Encoder + Decoder> {
Processing, Processing,
Error(E::Future), Error(FramedTransportError<S, U>),
EncoderError(E::Future), EncoderError(FramedTransportError<S, U>),
SinkFlushing,
Stopping, Stopping,
} }
impl<S, T, U> FramedTransport<S, T, U, DefaultErrorHandler<S, U, ()>> impl<S, T, U> FramedTransport<S, T, U>
where where
T: AsyncRead + AsyncWrite, T: AsyncRead + AsyncWrite,
U: Decoder + Encoder, U: Decoder + Encoder,
S: Service<Request = Request<U>, Response = Option<Response<U>>>, S: Service<Request = Request<U>, Response = Option<Response<U>>>,
S::Future: 'static, S::Future: 'static,
S::Error: fmt::Debug + 'static, S::Error: 'static,
<U as Encoder>::Error: fmt::Debug + 'static, <U as Encoder>::Error: 'static,
<U as Decoder>::Error: fmt::Debug + 'static,
{ {
pub fn new<F: IntoService<S>>(framed: Framed<T, U>, service: F) -> Self { pub fn new<F: IntoService<S>>(framed: Framed<T, U>, service: F) -> Self {
let (write_tx, write_rx) = mpsc::channel(16); let (write_tx, write_rx) = mpsc::channel(16);
@ -281,62 +206,24 @@ where
write_rx, write_rx,
write_tx, write_tx,
service: service.into_service(), service: service.into_service(),
error_service: DefaultErrorHandler(PhantomData),
state: TransportState::Processing, state: TransportState::Processing,
request: None, request: None,
response: None, response: None,
flushed: true, flushed: true,
} }
} }
/// Set error handler service
pub fn error_handler<E>(self, handler: E) -> FramedTransport<S, T, U, E>
where
E: Service<Request = TransportError<S, U>>,
{
FramedTransport {
framed: self.framed,
request: self.request,
service: self.service,
write_rx: self.write_rx,
write_tx: self.write_tx,
response: self.response,
flushed: self.flushed,
state: TransportState::Processing,
error_service: handler,
}
}
} }
impl<S, T, U, E> FramedTransport<S, T, U, E> impl<S, T, U> FramedTransport<S, T, U>
where where
T: AsyncRead + AsyncWrite, T: AsyncRead + AsyncWrite,
U: Decoder + Encoder, U: Decoder + Encoder,
S: Service<Request = Request<U>, Response = Option<Response<U>>>, S: Service<Request = Request<U>, Response = Option<Response<U>>>,
E: Service<Request = TransportError<S, U>>,
S::Future: 'static, S::Future: 'static,
S::Error: fmt::Debug + 'static, S::Error: 'static,
<U as Encoder>::Item: 'static, <U as Encoder>::Item: 'static,
<U as Encoder>::Error: fmt::Debug + 'static, <U as Encoder>::Error: 'static,
<U as Decoder>::Error: fmt::Debug + 'static,
{ {
pub fn with_error_service<F: IntoService<S>>(
framed: Framed<T, U>, service: F, error_service: E,
) -> Self {
let (write_tx, write_rx) = mpsc::channel(16);
FramedTransport {
framed,
write_rx,
write_tx,
error_service,
service: service.into_service(),
state: TransportState::Processing,
request: None,
response: None,
flushed: true,
}
}
fn poll_service(&mut self) -> bool { fn poll_service(&mut self) -> bool {
match self.service.poll_ready() { match self.service.poll_ready() {
Ok(Async::Ready(_)) => { Ok(Async::Ready(_)) => {
@ -365,9 +252,8 @@ where
return false; return false;
} }
Err(err) => { Err(err) => {
self.state = TransportState::Error( self.state =
self.error_service.call(TransportError::Service(err)), TransportState::Error(FramedTransportError::Service(err));
);
return true; return true;
} }
} }
@ -375,9 +261,8 @@ where
match self.framed.poll() { match self.framed.poll() {
Ok(Async::Ready(Some(el))) => item = Some(el), Ok(Async::Ready(Some(el))) => item = Some(el),
Err(err) => { Err(err) => {
self.state = TransportState::Error( self.state =
self.error_service.call(TransportError::Decoder(err)), TransportState::Error(FramedTransportError::Decoder(err));
);
return true; return true;
} }
Ok(Async::NotReady) => return false, Ok(Async::NotReady) => return false,
@ -390,9 +275,7 @@ where
} }
Ok(Async::NotReady) => return false, Ok(Async::NotReady) => return false,
Err(err) => { Err(err) => {
self.state = TransportState::Error( self.state = TransportState::Error(FramedTransportError::Service(err));
self.error_service.call(TransportError::Service(err)),
);
return true; return true;
} }
} }
@ -408,10 +291,8 @@ where
Ok(AsyncSink::Ready) => None, Ok(AsyncSink::Ready) => None,
Ok(AsyncSink::NotReady(item)) => Some(item), Ok(AsyncSink::NotReady(item)) => Some(item),
Err(err) => { Err(err) => {
trace!("Connection error: {:?}", err); self.state =
self.state = TransportState::EncoderError( TransportState::EncoderError(FramedTransportError::Encoder(err));
self.error_service.call(TransportError::Encoder(err)),
);
return true; return true;
} }
} }
@ -427,10 +308,8 @@ where
} }
Ok(Async::NotReady) => break, Ok(Async::NotReady) => break,
Err(err) => { Err(err) => {
trace!("Connection flush error: {:?}", err); self.state =
self.state = TransportState::EncoderError( TransportState::EncoderError(FramedTransportError::Encoder(err));
self.error_service.call(TransportError::Encoder(err)),
);
return true; return true;
} }
} }
@ -443,9 +322,8 @@ where
Ok(Async::Ready(Some(msg))) => match msg { Ok(Async::Ready(Some(msg))) => match msg {
Ok(msg) => item = Some(msg), Ok(msg) => item = Some(msg),
Err(err) => { Err(err) => {
self.state = TransportState::Error( self.state =
self.error_service.call(TransportError::Service(err)), TransportState::Error(FramedTransportError::Service(err));
);
return true; return true;
} }
}, },
@ -466,23 +344,21 @@ where
} }
} }
impl<S, T, U, E> Future for FramedTransport<S, T, U, E> impl<S, T, U> Future for FramedTransport<S, T, U>
where where
T: AsyncRead + AsyncWrite, T: AsyncRead + AsyncWrite,
U: Decoder + Encoder, U: Decoder + Encoder,
S: Service<Request = Request<U>, Response = Option<Response<U>>>, S: Service<Request = Request<U>, Response = Option<Response<U>>>,
S::Future: 'static, S::Future: 'static,
S::Error: fmt::Debug + 'static, S::Error: 'static,
E: Service<Request = TransportError<S, U>>,
<U as Encoder>::Item: 'static, <U as Encoder>::Item: 'static,
<U as Encoder>::Error: fmt::Debug + 'static, <U as Encoder>::Error: 'static,
<U as Decoder>::Error: fmt::Debug + 'static,
{ {
type Item = (); type Item = ();
type Error = S::Error; type Error = FramedTransportError<S, U>;
fn poll(&mut self) -> Poll<Self::Item, Self::Error> { fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
let state = match self.state { match mem::replace(&mut self.state, TransportState::Processing) {
TransportState::Processing => { TransportState::Processing => {
if self.poll_service() { if self.poll_service() {
return self.poll(); return self.poll();
@ -492,27 +368,18 @@ where
} }
return Ok(Async::NotReady); return Ok(Async::NotReady);
} }
TransportState::Error(ref mut fut) => match fut.poll() { TransportState::Error(err) => {
Err(_) | Ok(Async::Ready(_)) => TransportState::SinkFlushing,
_ => return Ok(Async::NotReady),
},
TransportState::EncoderError(ref mut fut) => match fut.poll() {
Err(_) | Ok(Async::Ready(_)) => return Ok(Async::Ready(())),
_ => return Ok(Async::NotReady),
},
TransportState::SinkFlushing => {
if self.poll_response() { if self.poll_response() {
return self.poll(); return Err(err);
} }
if self.flushed { if self.flushed {
return Ok(Async::Ready(())); return Err(err);
} }
self.state = TransportState::Error(err);
return Ok(Async::NotReady); return Ok(Async::NotReady);
} }
TransportState::EncoderError(err) => return Err(err),
TransportState::Stopping => return Ok(Async::Ready(())), TransportState::Stopping => return Ok(Async::Ready(())),
}; }
self.state = state;
self.poll()
} }
} }