1
0
mirror of https://github.com/fafhrd91/actix-net synced 2024-12-03 00:22:23 +01:00
actix-net/actix-utils/src/framed.rs

451 lines
13 KiB
Rust
Raw Normal View History

2018-09-13 03:47:39 +02:00
//! Framed dispatcher service and related utilities
2019-01-27 06:41:28 +01:00
use std::collections::VecDeque;
2018-09-13 03:47:39 +02:00
use std::marker::PhantomData;
2018-09-25 05:40:31 +02:00
use std::mem;
2018-09-13 03:47:39 +02:00
use actix_codec::{AsyncRead, AsyncWrite, Decoder, Encoder, Framed};
2018-12-09 19:15:49 +01:00
use actix_service::{IntoNewService, IntoService, NewService, Service};
2018-09-25 06:03:05 +02:00
use futures::future::{ok, FutureResult};
2019-01-27 06:41:28 +01:00
use futures::task::AtomicTask;
2018-12-17 01:26:24 +01:00
use futures::{Async, Future, Poll, Sink, Stream};
use log::debug;
2018-09-13 03:47:39 +02:00
2019-01-27 06:41:28 +01:00
use crate::cell::Cell;
type Request<U> = <U as Decoder>::Item;
type Response<U> = <U as Encoder>::Item;
2018-09-13 03:47:39 +02:00
2018-09-25 05:40:31 +02:00
pub struct FramedNewService<S, T, U> {
2018-09-13 03:47:39 +02:00
factory: S,
_t: PhantomData<(T, U)>,
}
2018-09-25 05:40:31 +02:00
impl<S, T, U> FramedNewService<S, T, U>
2018-09-13 03:47:39 +02:00
where
2018-11-30 03:56:15 +01:00
S: NewService<Request<U>, Response = Response<U>>,
2019-01-27 07:07:27 +01:00
S::Error: 'static,
<S::Service as Service<Request<U>>>::Future: 'static,
T: AsyncRead + AsyncWrite,
U: Decoder + Encoder,
<U as Encoder>::Item: 'static,
2019-01-27 06:41:28 +01:00
<U as Encoder>::Error: std::fmt::Debug,
2018-09-13 03:47:39 +02:00
{
2018-11-30 03:56:15 +01:00
pub fn new<F1: IntoNewService<S, Request<U>>>(factory: F1) -> Self {
2018-09-13 03:47:39 +02:00
Self {
factory: factory.into_new_service(),
_t: PhantomData,
}
}
}
2018-09-25 05:40:31 +02:00
impl<S, T, U> Clone for FramedNewService<S, T, U>
2018-09-13 03:47:39 +02:00
where
S: Clone,
{
fn clone(&self) -> Self {
Self {
factory: self.factory.clone(),
_t: PhantomData,
}
}
}
2018-11-30 03:56:15 +01:00
impl<S, T, U> NewService<Framed<T, U>> for FramedNewService<S, T, U>
2018-09-13 03:47:39 +02:00
where
2018-11-30 03:56:15 +01:00
S: NewService<Request<U>, Response = Response<U>> + Clone,
2019-01-27 07:07:27 +01:00
S::Error: 'static,
<S::Service as Service<Request<U>>>::Future: 'static,
T: AsyncRead + AsyncWrite,
U: Decoder + Encoder,
<U as Encoder>::Item: 'static,
2019-01-27 06:41:28 +01:00
<U as Encoder>::Error: std::fmt::Debug,
2018-09-13 03:47:39 +02:00
{
2018-09-25 05:40:31 +02:00
type Response = FramedTransport<S::Service, T, U>;
2018-09-13 03:47:39 +02:00
type Error = S::InitError;
type InitError = S::InitError;
2018-09-25 05:40:31 +02:00
type Service = FramedService<S, T, U>;
2018-09-13 03:47:39 +02:00
type Future = FutureResult<Self::Service, Self::InitError>;
fn new_service(&self) -> Self::Future {
ok(FramedService {
factory: self.factory.clone(),
_t: PhantomData,
})
}
}
2018-09-25 05:40:31 +02:00
pub struct FramedService<S, T, U> {
2018-09-13 03:47:39 +02:00
factory: S,
_t: PhantomData<(T, U)>,
}
2018-09-25 05:40:31 +02:00
impl<S, T, U> Clone for FramedService<S, T, U>
2018-09-13 03:47:39 +02:00
where
S: Clone,
{
fn clone(&self) -> Self {
Self {
factory: self.factory.clone(),
_t: PhantomData,
}
}
}
2018-11-30 03:56:15 +01:00
impl<S, T, U> Service<Framed<T, U>> for FramedService<S, T, U>
2018-09-13 03:47:39 +02:00
where
2018-11-30 03:56:15 +01:00
S: NewService<Request<U>, Response = Response<U>>,
2019-01-27 07:07:27 +01:00
S::Error: 'static,
<S::Service as Service<Request<U>>>::Future: 'static,
T: AsyncRead + AsyncWrite,
U: Decoder + Encoder,
<U as Encoder>::Item: 'static,
2019-01-27 06:41:28 +01:00
<U as Encoder>::Error: std::fmt::Debug,
2018-09-13 03:47:39 +02:00
{
2018-09-25 05:40:31 +02:00
type Response = FramedTransport<S::Service, T, U>;
2018-09-13 03:47:39 +02:00
type Error = S::InitError;
2018-09-25 05:40:31 +02:00
type Future = FramedServiceResponseFuture<S, T, U>;
2018-09-13 03:47:39 +02:00
fn poll_ready(&mut self) -> Poll<(), Self::Error> {
Ok(Async::Ready(()))
}
2018-11-30 03:56:15 +01:00
fn call(&mut self, req: Framed<T, U>) -> Self::Future {
2018-09-13 03:47:39 +02:00
FramedServiceResponseFuture {
2018-09-25 05:40:31 +02:00
fut: self.factory.new_service(),
2018-09-13 03:47:39 +02:00
framed: Some(req),
}
}
}
#[doc(hidden)]
2018-09-25 05:40:31 +02:00
pub struct FramedServiceResponseFuture<S, T, U>
2018-09-13 03:47:39 +02:00
where
2018-11-30 03:56:15 +01:00
S: NewService<Request<U>, Response = Response<U>>,
2019-01-27 07:07:27 +01:00
S::Error: 'static,
<S::Service as Service<Request<U>>>::Future: 'static,
T: AsyncRead + AsyncWrite,
U: Decoder + Encoder,
<U as Encoder>::Item: 'static,
2019-01-27 06:41:28 +01:00
<U as Encoder>::Error: std::fmt::Debug,
2018-09-13 03:47:39 +02:00
{
2018-09-25 05:40:31 +02:00
fut: S::Future,
2018-09-13 03:47:39 +02:00
framed: Option<Framed<T, U>>,
}
2018-09-25 05:40:31 +02:00
impl<S, T, U> Future for FramedServiceResponseFuture<S, T, U>
2018-09-13 03:47:39 +02:00
where
2018-11-30 03:56:15 +01:00
S: NewService<Request<U>, Response = Response<U>>,
2019-01-27 07:07:27 +01:00
S::Error: 'static,
<S::Service as Service<Request<U>>>::Future: 'static,
T: AsyncRead + AsyncWrite,
U: Decoder + Encoder,
<U as Encoder>::Item: 'static,
2019-01-27 06:41:28 +01:00
<U as Encoder>::Error: std::fmt::Debug,
2018-09-13 03:47:39 +02:00
{
2018-09-25 05:40:31 +02:00
type Item = FramedTransport<S::Service, T, U>;
2018-09-13 03:47:39 +02:00
type Error = S::InitError;
fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
match self.fut.poll()? {
Async::NotReady => Ok(Async::NotReady),
2018-09-25 05:40:31 +02:00
Async::Ready(service) => Ok(Async::Ready(FramedTransport::new(
self.framed.take().unwrap(),
service,
))),
}
}
}
2018-09-25 05:40:31 +02:00
/// Framed transport errors
2018-09-25 07:31:05 +02:00
pub enum FramedTransportError<E, U: Encoder + Decoder> {
Service(E),
Encoder(<U as Encoder>::Error),
Decoder(<U as Decoder>::Error),
}
impl<E, U: Encoder + Decoder> From<E> for FramedTransportError<E, U> {
fn from(err: E) -> Self {
FramedTransportError::Service(err)
}
}
/// FramedTransport - is a future that reads frames from Framed object
2018-09-13 03:47:39 +02:00
/// and pass then to the service.
2018-09-25 05:40:31 +02:00
pub struct FramedTransport<S, T, U>
2018-09-13 03:47:39 +02:00
where
2019-01-27 07:07:27 +01:00
S: Service<Request<U>, Response = Response<U>>,
S::Error: 'static,
S::Future: 'static,
T: AsyncRead + AsyncWrite,
U: Encoder + Decoder,
<U as Encoder>::Item: 'static,
2019-01-27 06:41:28 +01:00
<U as Encoder>::Error: std::fmt::Debug,
2018-09-13 03:47:39 +02:00
{
2019-01-27 07:07:27 +01:00
service: S,
state: TransportState<S, U>,
framed: Framed<T, U>,
inner: Cell<FramedTransportInner<<U as Encoder>::Item, S::Error>>,
2018-09-13 03:47:39 +02:00
}
2018-11-30 03:56:15 +01:00
enum TransportState<S: Service<Request<U>>, U: Encoder + Decoder> {
Processing,
2018-09-25 07:31:05 +02:00
Error(FramedTransportError<S::Error, U>),
2018-12-21 22:00:26 +01:00
FramedError(FramedTransportError<S::Error, U>),
Stopping,
}
2019-01-27 07:07:27 +01:00
struct FramedTransportInner<I, E> {
buf: VecDeque<Result<I, E>>,
2019-01-27 06:41:28 +01:00
task: AtomicTask,
2018-09-13 03:47:39 +02:00
}
2019-01-27 07:07:27 +01:00
impl<S, T, U> FramedTransport<S, T, U>
2018-09-13 03:47:39 +02:00
where
2019-01-27 07:07:27 +01:00
S: Service<Request<U>, Response = Response<U>>,
S::Error: 'static,
S::Future: 'static,
T: AsyncRead + AsyncWrite,
U: Decoder + Encoder,
<U as Encoder>::Item: 'static,
2019-01-27 06:41:28 +01:00
<U as Encoder>::Error: std::fmt::Debug,
2018-09-13 03:47:39 +02:00
{
2019-01-27 07:07:27 +01:00
fn poll_service(&mut self) -> bool {
2019-01-27 06:41:28 +01:00
loop {
match self.service.poll_ready() {
Ok(Async::Ready(_)) => loop {
2018-09-27 05:40:45 +02:00
let item = match self.framed.poll() {
Ok(Async::Ready(Some(el))) => el,
Err(err) => {
2018-09-25 05:40:31 +02:00
self.state =
2018-12-21 22:00:26 +01:00
TransportState::FramedError(FramedTransportError::Decoder(err));
return true;
2018-09-13 03:47:39 +02:00
}
Ok(Async::NotReady) => return false,
Ok(Async::Ready(None)) => {
self.state = TransportState::Stopping;
return true;
2018-09-13 03:47:39 +02:00
}
2018-09-27 05:40:45 +02:00
};
2019-01-27 07:07:27 +01:00
let mut cell = self.inner.clone();
cell.get_mut().task.register();
2019-01-27 06:41:28 +01:00
tokio_current_thread::spawn(self.service.call(item).then(move |item| {
let inner = cell.get_mut();
inner.buf.push_back(item);
inner.task.notify();
Ok(())
}));
},
Ok(Async::NotReady) => return false,
Err(err) => {
self.state = TransportState::Error(FramedTransportError::Service(err));
return true;
2018-09-13 03:47:39 +02:00
}
}
2018-09-13 03:47:39 +02:00
}
}
2018-09-13 03:47:39 +02:00
/// write to sink
fn poll_response(&mut self) -> bool {
2019-01-27 07:07:27 +01:00
let inner = self.inner.get_mut();
2018-09-13 03:47:39 +02:00
loop {
2018-12-17 01:26:24 +01:00
while !self.framed.is_write_buf_full() {
2019-01-27 07:07:27 +01:00
if let Some(msg) = inner.buf.pop_front() {
2019-01-27 06:41:28 +01:00
match msg {
2018-12-17 01:26:24 +01:00
Ok(msg) => {
if let Err(err) = self.framed.force_send(msg) {
2018-12-21 22:00:26 +01:00
self.state = TransportState::FramedError(
2018-12-17 01:26:24 +01:00
FramedTransportError::Encoder(err),
);
return true;
}
}
Err(err) => {
self.state =
TransportState::Error(FramedTransportError::Service(err));
return true;
}
2019-01-27 06:41:28 +01:00
}
} else {
break;
2018-09-13 03:47:39 +02:00
}
2018-12-17 01:26:24 +01:00
}
2018-09-13 03:47:39 +02:00
2018-12-17 01:26:24 +01:00
if !self.framed.is_write_buf_empty() {
2018-09-13 03:47:39 +02:00
match self.framed.poll_complete() {
Ok(Async::NotReady) => break,
Err(err) => {
2018-12-17 01:26:24 +01:00
debug!("Error sending data: {:?}", err);
2018-09-25 05:40:31 +02:00
self.state =
2018-12-21 22:00:26 +01:00
TransportState::FramedError(FramedTransportError::Encoder(err));
return true;
2018-09-13 03:47:39 +02:00
}
2018-12-17 01:26:24 +01:00
Ok(Async::Ready(_)) => (),
2018-09-13 03:47:39 +02:00
}
} else {
break;
}
}
false
}
}
2019-01-27 06:41:28 +01:00
impl<S, T, U> FramedTransport<S, T, U>
where
2019-01-27 07:07:27 +01:00
S: Service<Request<U>, Response = Response<U>>,
S::Error: 'static,
S::Future: 'static,
T: AsyncRead + AsyncWrite,
U: Decoder + Encoder,
<U as Encoder>::Item: 'static,
2019-01-27 06:41:28 +01:00
<U as Encoder>::Error: std::fmt::Debug,
{
pub fn new<F: IntoService<S, Request<U>>>(framed: Framed<T, U>, service: F) -> Self {
2019-01-27 07:07:27 +01:00
FramedTransport {
2019-01-27 06:41:28 +01:00
framed,
service: service.into_service(),
state: TransportState::Processing,
2019-01-27 07:07:27 +01:00
inner: Cell::new(FramedTransportInner {
buf: VecDeque::new(),
task: AtomicTask::new(),
}),
2019-01-27 06:41:28 +01:00
}
}
/// Get reference to a service wrapped by `FramedTransport` instance.
pub fn get_ref(&self) -> &S {
2019-01-27 07:07:27 +01:00
&self.service
2019-01-27 06:41:28 +01:00
}
/// Get mutable reference to a service wrapped by `FramedTransport`
/// instance.
pub fn get_mut(&mut self) -> &mut S {
2019-01-27 07:07:27 +01:00
&mut self.service
2019-01-27 06:41:28 +01:00
}
/// Get reference to a framed instance wrapped by `FramedTransport`
/// instance.
pub fn get_framed(&self) -> &Framed<T, U> {
2019-01-27 07:07:27 +01:00
&self.framed
2019-01-27 06:41:28 +01:00
}
/// Get mutable reference to a framed instance wrapped by `FramedTransport`
/// instance.
pub fn get_framed_mut(&mut self) -> &mut Framed<T, U> {
2019-01-27 07:07:27 +01:00
&mut self.framed
2019-01-27 06:41:28 +01:00
}
}
2018-09-25 05:40:31 +02:00
impl<S, T, U> Future for FramedTransport<S, T, U>
where
2019-01-27 07:07:27 +01:00
S: Service<Request<U>, Response = Response<U>>,
S::Error: 'static,
S::Future: 'static,
T: AsyncRead + AsyncWrite,
U: Decoder + Encoder,
<U as Encoder>::Item: 'static,
2019-01-27 06:41:28 +01:00
<U as Encoder>::Error: std::fmt::Debug,
{
type Item = ();
2018-09-25 07:31:05 +02:00
type Error = FramedTransportError<S::Error, U>;
fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
2019-01-27 07:07:27 +01:00
match mem::replace(&mut self.state, TransportState::Processing) {
TransportState::Processing => {
2019-01-27 07:07:27 +01:00
if self.poll_service() || self.poll_response() {
2018-12-06 23:04:42 +01:00
self.poll()
} else {
Ok(Async::NotReady)
}
}
2018-09-25 05:40:31 +02:00
TransportState::Error(err) => {
2019-01-27 07:07:27 +01:00
if self.framed.is_write_buf_empty()
|| (self.poll_response() || self.framed.is_write_buf_empty())
2018-12-26 20:50:07 +01:00
{
2018-12-06 23:04:42 +01:00
Err(err)
} else {
2019-01-27 07:07:27 +01:00
self.state = TransportState::Error(err);
2018-12-06 23:04:42 +01:00
Ok(Async::NotReady)
}
}
2018-12-21 22:00:26 +01:00
TransportState::FramedError(err) => Err(err),
2018-12-06 23:04:42 +01:00
TransportState::Stopping => Ok(Async::Ready(())),
2018-09-25 05:40:31 +02:00
}
2018-09-13 03:47:39 +02:00
}
}
2018-10-06 00:37:15 +02:00
pub struct IntoFramed<T, U, F>
where
T: AsyncRead + AsyncWrite,
F: Fn() -> U + Send + Clone + 'static,
U: Encoder + Decoder,
{
factory: F,
_t: PhantomData<(T,)>,
}
impl<T, U, F> IntoFramed<T, U, F>
where
T: AsyncRead + AsyncWrite,
F: Fn() -> U + Send + Clone + 'static,
U: Encoder + Decoder,
{
pub fn new(factory: F) -> Self {
IntoFramed {
factory,
_t: PhantomData,
}
}
}
2018-11-30 03:56:15 +01:00
impl<T, U, F> NewService<T> for IntoFramed<T, U, F>
2018-10-06 00:37:15 +02:00
where
T: AsyncRead + AsyncWrite,
F: Fn() -> U + Send + Clone + 'static,
U: Encoder + Decoder,
{
type Response = Framed<T, U>;
type Error = ();
type InitError = ();
type Service = IntoFramedService<T, U, F>;
type Future = FutureResult<Self::Service, Self::InitError>;
fn new_service(&self) -> Self::Future {
ok(IntoFramedService {
factory: self.factory.clone(),
_t: PhantomData,
})
}
}
pub struct IntoFramedService<T, U, F>
where
T: AsyncRead + AsyncWrite,
F: Fn() -> U + Send + Clone + 'static,
U: Encoder + Decoder,
{
factory: F,
_t: PhantomData<(T,)>,
}
2018-11-30 03:56:15 +01:00
impl<T, U, F> Service<T> for IntoFramedService<T, U, F>
2018-10-06 00:37:15 +02:00
where
T: AsyncRead + AsyncWrite,
F: Fn() -> U + Send + Clone + 'static,
U: Encoder + Decoder,
{
type Response = Framed<T, U>;
type Error = ();
type Future = FutureResult<Self::Response, Self::Error>;
fn poll_ready(&mut self) -> Poll<(), Self::Error> {
Ok(Async::Ready(()))
}
2018-11-30 03:56:15 +01:00
fn call(&mut self, req: T) -> Self::Future {
2018-10-06 00:37:15 +02:00
ok(Framed::new(req, (self.factory)()))
}
}