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

refactor service and state manahement

Browse Source
This commit is contained in:
Nikolay Kim 2019-12-29 13:42:42 +06:00
parent 1918c8d4f8
commit 5779da0f49
9 changed files with 345 additions and 527 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
actix-ioframe/CHANGES.md
View File

@ -1,5 +1,13 @@
# Changes
## [0.5.0] - 2019-12-29
* Simplify state management
* Allow to set custom output stream
* Removed disconnect callback
## [0.4.1] - 2019-12-11
* Disconnect callback accepts owned state

11
actix-ioframe/Cargo.toml
View File

@ -1,6 +1,6 @@
[package]
name = "actix-ioframe"
version = "0.4.1"
version = "0.5.0"
authors = ["Nikolay Kim <fafhrd91@gmail.com>"]
description = "Actix framed service"
keywords = ["network", "framework", "async", "futures"]
@ -10,19 +10,18 @@ documentation = "https://docs.rs/actix-ioframe/"
categories = ["network-programming", "asynchronous"]
license = "MIT/Apache-2.0"
edition = "2018"
workspace = ".."
[lib]
name = "actix_ioframe"
path = "src/lib.rs"
[dependencies]
actix-service = "1.0.0"
actix-service = "1.0.1"
actix-codec = "0.2.0"
actix-utils = "1.0.1"
actix-utils = "1.0.4"
actix-rt = "1.0.0"
bytes = "0.5"
either = "1.5.2"
bytes = "0.5.3"
either = "1.5.3"
futures = "0.3.1"
pin-project = "0.4.6"
log = "0.4"

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

@ -3,17 +3,15 @@ use std::pin::Pin;
use std::task::{Context, Poll};
use actix_codec::{AsyncRead, AsyncWrite, Decoder, Encoder, Framed};
use actix_utils::mpsc::Receiver;
use futures::Stream;
use crate::sink::Sink;
pub struct Connect<Io, Codec, St = ()>
pub struct Connect<Io, Codec>
where
Codec: Encoder + Decoder,
{
io: Io,
sink: Sink<<Codec as Encoder>::Item>,
_t: PhantomData<(St, Codec)>,
_t: PhantomData<Codec>,
}
impl<Io, Codec> Connect<Io, Codec>
@ -21,36 +19,33 @@ where
Io: AsyncRead + AsyncWrite,
Codec: Encoder + Decoder,
{
pub(crate) fn new(io: Io, sink: Sink<<Codec as Encoder>::Item>) -> Self {
pub(crate) fn new(io: Io) -> Self {
Self {
io,
sink,
_t: PhantomData,
}
}
pub fn codec(self, codec: Codec) -> ConnectResult<Io, (), Codec> {
pub fn codec(
self,
codec: Codec,
) -> ConnectResult<Io, (), Codec, Receiver<<Codec as Encoder>::Item>> {
ConnectResult {
state: (),
sink: self.sink,
out: None,
framed: Framed::new(self.io, codec),
}
}
}
#[pin_project::pin_project]
pub struct ConnectResult<Io, St, Codec: Encoder + Decoder> {
pub struct ConnectResult<Io, St, Codec: Encoder + Decoder, Out> {
pub(crate) state: St,
pub(crate) out: Option<Out>,
pub(crate) framed: Framed<Io, Codec>,
pub(crate) sink: Sink<<Codec as Encoder>::Item>,
}
impl<Io, St, Codec: Encoder + Decoder> ConnectResult<Io, St, Codec> {
#[inline]
pub fn sink(&self) -> &Sink<<Codec as Encoder>::Item> {
&self.sink
}
impl<Io, St, Codec: Encoder + Decoder, Out: Unpin> ConnectResult<Io, St, Codec, Out> {
#[inline]
pub fn get_ref(&self) -> &Io {
self.framed.get_ref()
@ -61,17 +56,28 @@ impl<Io, St, Codec: Encoder + Decoder> ConnectResult<Io, St, Codec> {
self.framed.get_mut()
}
pub fn out<U>(self, out: U) -> ConnectResult<Io, St, Codec, U>
where
U: Stream<Item = <Codec as Encoder>::Item> + Unpin,
{
ConnectResult {
state: self.state,
framed: self.framed,
out: Some(out),
}
}
#[inline]
pub fn state<S>(self, state: S) -> ConnectResult<Io, S, Codec> {
pub fn state<S>(self, state: S) -> ConnectResult<Io, S, Codec, Out> {
ConnectResult {
state,
framed: self.framed,
sink: self.sink,
out: self.out,
}
}
}
impl<Io, St, Codec> Stream for ConnectResult<Io, St, Codec>
impl<Io, St, Codec, Out> Stream for ConnectResult<Io, St, Codec, Out>
where
Io: AsyncRead + AsyncWrite,
Codec: Encoder + Decoder,
@ -83,7 +89,8 @@ where
}
}
impl<Io, St, Codec> futures::Sink<<Codec as Encoder>::Item> for ConnectResult<Io, St, Codec>
impl<Io, St, Codec, Out> futures::Sink<<Codec as Encoder>::Item>
for ConnectResult<Io, St, Codec, Out>
where
Io: AsyncRead + AsyncWrite,
Codec: Encoder + Decoder,

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

@ -1,81 +1,56 @@
//! Framed dispatcher service and related utilities
use std::pin::Pin;
use std::rc::Rc;
use std::task::{Context, Poll};
use actix_codec::{AsyncRead, AsyncWrite, Decoder, Encoder, Framed};
use actix_service::{IntoService, Service};
use actix_utils::{mpsc, oneshot};
use futures::{FutureExt, Stream};
use actix_service::Service;
use actix_utils::mpsc;
use futures::Stream;
use log::debug;
use crate::error::ServiceError;
use crate::item::Item;
use crate::sink::Sink;
type Request<S, U> = Item<S, U>;
type Request<U> = <U as Decoder>::Item;
type Response<U> = <U as Encoder>::Item;
pub(crate) enum Message<T> {
Item(T),
WaitClose(oneshot::Sender<()>),
Close,
}
/// FramedTransport - is a future that reads frames from Framed object
/// and pass then to the service.
#[pin_project::pin_project]
pub(crate) struct Dispatcher<St, S, T, U>
pub(crate) struct Dispatcher<S, T, U, Out>
where
St: Clone,
S: Service<Request = Request<St, U>, Response = Option<Response<U>>>,
S: Service<Request = Request<U>, Response = Option<Response<U>>>,
S::Error: 'static,
S::Future: 'static,
T: AsyncRead + AsyncWrite,
U: Encoder + Decoder,
<U as Encoder>::Item: 'static,
<U as Encoder>::Error: std::fmt::Debug,
Out: Stream<Item = <U as Encoder>::Item> + Unpin,
{
service: S,
sink: Sink<<U as Encoder>::Item>,
state: St,
dispatch_state: FramedState<S, U>,
sink: Option<Out>,
state: FramedState<S, U>,
framed: Framed<T, U>,
rx: mpsc::Receiver<Result<Message<<U as Encoder>::Item>, S::Error>>,
tx: mpsc::Sender<Result<Message<<U as Encoder>::Item>, S::Error>>,
disconnect: Option<Rc<dyn Fn(St, bool)>>,
rx: mpsc::Receiver<Result<<U as Encoder>::Item, S::Error>>,
}
impl<St, S, T, U> Dispatcher<St, S, T, U>
impl<S, T, U, Out> Dispatcher<S, T, U, Out>
where
St: Clone,
S: Service<Request = Request<St, U>, Response = Option<Response<U>>>,
S: Service<Request = Request<U>, Response = Option<Response<U>>>,
S::Error: 'static,
S::Future: 'static,
T: AsyncRead + AsyncWrite,
U: Decoder + Encoder,
<U as Encoder>::Item: 'static,
<U as Encoder>::Error: std::fmt::Debug,
Out: Stream<Item = <U as Encoder>::Item> + Unpin,
{
pub(crate) fn new<F: IntoService<S>>(
framed: Framed<T, U>,
state: St,
service: F,
sink: Sink<<U as Encoder>::Item>,
rx: mpsc::Receiver<Result<Message<<U as Encoder>::Item>, S::Error>>,
disconnect: Option<Rc<dyn Fn(St, bool)>>,
) -> Self {
let tx = rx.sender();
pub(crate) fn new(framed: Framed<T, U>, service: S, sink: Option<Out>) -> Self {
Dispatcher {
framed,
state,
sink,
disconnect,
rx,
tx,
service: service.into_service(),
dispatch_state: FramedState::Processing,
service,
framed,
rx: mpsc::channel().1,
state: FramedState::Processing,
}
}
}
@ -84,33 +59,11 @@ enum FramedState<S: Service, U: Encoder + Decoder> {
Processing,
Error(ServiceError<S::Error, U>),
FramedError(ServiceError<S::Error, U>),
FlushAndStop(Vec<oneshot::Sender<()>>),
FlushAndStop,
Stopping,
}
impl<S: Service, U: Encoder + Decoder> FramedState<S, U> {
fn stop(&mut self, tx: Option<oneshot::Sender<()>>) {
match self {
FramedState::FlushAndStop(ref mut vec) => {
if let Some(tx) = tx {
vec.push(tx)
}
}
FramedState::Processing => {
*self = FramedState::FlushAndStop(if let Some(tx) = tx {
vec![tx]
} else {
Vec::new()
})
}
FramedState::Error(_) | FramedState::FramedError(_) | FramedState::Stopping => {
if let Some(tx) = tx {
let _ = tx.send(());
}
}
}
}
fn take_error(&mut self) -> ServiceError<S::Error, U> {
match std::mem::replace(self, FramedState::Processing) {
FramedState::Error(err) => err,
@ -126,16 +79,16 @@ impl<S: Service, U: Encoder + Decoder> FramedState<S, U> {
}
}
impl<St, S, T, U> Dispatcher<St, S, T, U>
impl<S, T, U, Out> Dispatcher<S, T, U, Out>
where
St: Clone,
S: Service<Request = Request<St, U>, Response = Option<Response<U>>>,
S: Service<Request = Request<U>, Response = Option<Response<U>>>,
S::Error: 'static,
S::Future: 'static,
T: AsyncRead + AsyncWrite,
U: Decoder + Encoder,
<U as Encoder>::Item: 'static,
<U as Encoder>::Error: std::fmt::Debug,
Out: Stream<Item = <U as Encoder>::Item> + Unpin,
{
fn poll_read(&mut self, cx: &mut Context<'_>) -> bool {
loop {
@ -144,35 +97,32 @@ where
let item = match self.framed.next_item(cx) {
Poll::Ready(Some(Ok(el))) => el,
Poll::Ready(Some(Err(err))) => {
self.dispatch_state =
FramedState::FramedError(ServiceError::Decoder(err));
self.state = FramedState::FramedError(ServiceError::Decoder(err));
return true;
}
Poll::Pending => return false,
Poll::Ready(None) => {
log::trace!("Client disconnected");
self.dispatch_state = FramedState::Stopping;
self.state = FramedState::Stopping;
return true;
}
};
let tx = self.tx.clone();
actix_rt::spawn(
self.service
.call(Item::new(self.state.clone(), self.sink.clone(), item))
.map(move |item| {
let tx = self.rx.sender();
let fut = self.service.call(item);
actix_rt::spawn(async move {
let item = fut.await;
let item = match item {
Ok(Some(item)) => Ok(Message::Item(item)),
Ok(Some(item)) => Ok(item),
Ok(None) => return,
Err(err) => Err(err),
};
let _ = tx.send(item);
}),
);
});
}
Poll::Pending => return false,
Poll::Ready(Err(err)) => {
self.dispatch_state = FramedState::Error(ServiceError::Service(err));
self.state = FramedState::Error(ServiceError::Service(err));
return true;
}
}
@ -184,27 +134,39 @@ where
loop {
while !self.framed.is_write_buf_full() {
match Pin::new(&mut self.rx).poll_next(cx) {
Poll::Ready(Some(Ok(Message::Item(msg)))) => {
Poll::Ready(Some(Ok(msg))) => {
if let Err(err) = self.framed.write(msg) {
self.dispatch_state =
self.state = FramedState::FramedError(ServiceError::Encoder(err));
return true;
}
continue;
}
Poll::Ready(Some(Err(err))) => {
self.state = FramedState::Error(ServiceError::Service(err));
return true;
}
Poll::Ready(None) | Poll::Pending => (),
}
if self.sink.is_some() {
match Pin::new(self.sink.as_mut().unwrap()).poll_next(cx) {
Poll::Ready(Some(msg)) => {
if let Err(err) = self.framed.write(msg) {
self.state =
FramedState::FramedError(ServiceError::Encoder(err));
return true;
}
continue;
}
Poll::Ready(Some(Ok(Message::Close))) => {
self.dispatch_state.stop(None);
Poll::Ready(None) => {
let _ = self.sink.take();
self.state = FramedState::FlushAndStop;
return true;
}
Poll::Ready(Some(Ok(Message::WaitClose(tx)))) => {
self.dispatch_state.stop(Some(tx));
return true;
Poll::Pending => (),
}
Poll::Ready(Some(Err(err))) => {
self.dispatch_state = FramedState::Error(ServiceError::Service(err));
return true;
}
Poll::Ready(None) | Poll::Pending => break,
}
break;
}
if !self.framed.is_write_buf_empty() {
@ -213,8 +175,7 @@ where
Poll::Ready(Ok(_)) => (),
Poll::Ready(Err(err)) => {
debug!("Error sending data: {:?}", err);
self.dispatch_state =
FramedState::FramedError(ServiceError::Encoder(err));
self.state = FramedState::FramedError(ServiceError::Encoder(err));
return true;
}
}
@ -229,14 +190,16 @@ where
&mut self,
cx: &mut Context<'_>,
) -> Poll<Result<(), ServiceError<S::Error, U>>> {
match self.dispatch_state {
FramedState::Processing => {
if self.poll_read(cx) || self.poll_write(cx) {
self.poll(cx)
match self.state {
FramedState::Processing => loop {
let read = self.poll_read(cx);
let write = self.poll_write(cx);
if read || write {
continue;
} else {
Poll::Pending
}
return Poll::Pending;
}
},
FramedState::Error(_) => {
// flush write buffer
if !self.framed.is_write_buf_empty() {
@ -244,12 +207,21 @@ where
return Poll::Pending;
}
}
if let Some(ref disconnect) = self.disconnect {
(&*disconnect)(self.state.clone(), true);
Poll::Ready(Err(self.state.take_error()))
}
Poll::Ready(Err(self.dispatch_state.take_error()))
FramedState::FlushAndStop => {
// drain service responses
match Pin::new(&mut self.rx).poll_next(cx) {
Poll::Ready(Some(Ok(msg))) => {
if let Err(_) = self.framed.write(msg) {
return Poll::Ready(Ok(()));
}
FramedState::FlushAndStop(ref mut vec) => {
}
Poll::Ready(Some(Err(_))) => return Poll::Ready(Ok(())),
Poll::Ready(None) | Poll::Pending => (),
}
// flush io
if !self.framed.is_write_buf_empty() {
match self.framed.flush(cx) {
Poll::Ready(Err(err)) => {
@ -261,26 +233,10 @@ where
Poll::Ready(_) => (),
}
};
for tx in vec.drain(..) {
let _ = tx.send(());
}
if let Some(ref disconnect) = self.disconnect {
(&*disconnect)(self.state.clone(), false);
}
Poll::Ready(Ok(()))
}
FramedState::FramedError(_) => {
if let Some(ref disconnect) = self.disconnect {
(&*disconnect)(self.state.clone(), true);
}
Poll::Ready(Err(self.dispatch_state.take_framed_error()))
}
FramedState::Stopping => {
if let Some(ref disconnect) = self.disconnect {
(&*disconnect)(self.state.clone(), false);
}
Poll::Ready(Ok(()))
}
FramedState::FramedError(_) => Poll::Ready(Err(self.state.take_framed_error())),
FramedState::Stopping => Poll::Ready(Ok(())),
}
}
}

82
actix-ioframe/src/item.rs
View File

@ -1,82 +0,0 @@
use std::fmt;
use std::ops::{Deref, DerefMut};
use actix_codec::{Decoder, Encoder};
use crate::sink::Sink;
pub struct Item<St, Codec: Encoder + Decoder> {
state: St,
sink: Sink<<Codec as Encoder>::Item>,
item: <Codec as Decoder>::Item,
}
impl<St, Codec> Item<St, Codec>
where
Codec: Encoder + Decoder,
{
pub(crate) fn new(
state: St,
sink: Sink<<Codec as Encoder>::Item>,
item: <Codec as Decoder>::Item,
) -> Self {
Item { state, sink, item }
}
#[inline]
pub fn state(&self) -> &St {
&self.state
}
#[inline]
pub fn state_mut(&mut self) -> &mut St {
&mut self.state
}
#[inline]
pub fn sink(&self) -> &Sink<<Codec as Encoder>::Item> {
&self.sink
}
#[inline]
pub fn into_inner(self) -> <Codec as Decoder>::Item {
self.item
}
#[inline]
pub fn into_parts(self) -> (St, Sink<<Codec as Encoder>::Item>, <Codec as Decoder>::Item) {
(self.state, self.sink, self.item)
}
}
impl<St, Codec> Deref for Item<St, Codec>
where
Codec: Encoder + Decoder,
{
type Target = <Codec as Decoder>::Item;
#[inline]
fn deref(&self) -> &<Codec as Decoder>::Item {
&self.item
}
}
impl<St, Codec> DerefMut for Item<St, Codec>
where
Codec: Encoder + Decoder,
{
#[inline]
fn deref_mut(&mut self) -> &mut <Codec as Decoder>::Item {
&mut self.item
}
}
impl<St, Codec> fmt::Debug for Item<St, Codec>
where
Codec: Encoder + Decoder,
<Codec as Decoder>::Item: fmt::Debug,
{
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.debug_tuple("Item").field(&self.item).finish()
}
}

8
actix-ioframe/src/lib.rs
View File

@ -1,15 +1,11 @@
#![deny(rust_2018_idioms, warnings)]
// #![deny(rust_2018_idioms, warnings)]
#![allow(clippy::type_complexity, clippy::too_many_arguments)]
mod connect;
mod dispatcher;
mod error;
mod item;
mod service;
mod sink;
pub use self::connect::{Connect, ConnectResult};
pub use self::error::ServiceError;
pub use self::item::Item;
pub use self::service::{Builder, NewServiceBuilder, ServiceBuilder};
pub use self::sink::Sink;
pub use self::service::{Builder, FactoryBuilder};

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

@ -4,108 +4,57 @@ use std::pin::Pin;
use std::rc::Rc;
use std::task::{Context, Poll};
use actix_codec::{AsyncRead, AsyncWrite, Decoder, Encoder};
use actix_codec::{AsyncRead, AsyncWrite, Decoder, Encoder, Framed};
use actix_service::{IntoService, IntoServiceFactory, Service, ServiceFactory};
use actix_utils::mpsc;
use either::Either;
use futures::future::{FutureExt, LocalBoxFuture};
use futures::{ready, Stream};
use pin_project::project;
use crate::connect::{Connect, ConnectResult};
use crate::dispatcher::{Dispatcher, Message};
use crate::dispatcher::Dispatcher;
use crate::error::ServiceError;
use crate::item::Item;
use crate::sink::Sink;
type RequestItem<S, U> = Item<S, U>;
type RequestItem<U> = <U as Decoder>::Item;
type ResponseItem<U> = Option<<U as Encoder>::Item>;
type ServiceResult<U, E> = Result<Message<<U as Encoder>::Item>, E>;
/// Service builder - structure that follows the builder pattern
/// for building instances for framed services.
pub struct Builder<St, Codec>(PhantomData<(St, Codec)>);
impl<St: Clone, Codec> Default for Builder<St, Codec> {
fn default() -> Builder<St, Codec> {
Builder::new()
}
}
impl<St: Clone, Codec> Builder<St, Codec> {
pub fn new() -> Builder<St, Codec> {
Builder(PhantomData)
}
/// Construct framed handler service with specified connect service
pub fn service<Io, C, F>(self, connect: F) -> ServiceBuilder<St, C, Io, Codec>
where
F: IntoService<C>,
Io: AsyncRead + AsyncWrite,
C: Service<Request = Connect<Io, Codec>, Response = ConnectResult<Io, St, Codec>>,
Codec: Decoder + Encoder,
{
ServiceBuilder {
connect: connect.into_service(),
disconnect: None,
_t: PhantomData,
}
}
/// Construct framed handler new service with specified connect service
pub fn factory<Io, C, F>(self, connect: F) -> NewServiceBuilder<St, C, Io, Codec>
where
F: IntoServiceFactory<C>,
Io: AsyncRead + AsyncWrite,
C: ServiceFactory<
Config = (),
Request = Connect<Io, Codec>,
Response = ConnectResult<Io, St, Codec>,
>,
C::Error: 'static,
C::Future: 'static,
Codec: Decoder + Encoder,
{
NewServiceBuilder {
connect: connect.into_factory(),
disconnect: None,
_t: PhantomData,
}
}
}
pub struct ServiceBuilder<St, C, Io, Codec> {
pub struct Builder<St, C, Io, Codec, Out> {
connect: C,
disconnect: Option<Rc<dyn Fn(St, bool)>>,
_t: PhantomData<(St, Io, Codec)>,
_t: PhantomData<(St, Io, Codec, Out)>,
}
impl<St, C, Io, Codec> ServiceBuilder<St, C, Io, Codec>
impl<St, C, Io, Codec, Out> Builder<St, C, Io, Codec, Out>
where
St: Clone,
C: Service<Request = Connect<Io, Codec>, Response = ConnectResult<Io, St, Codec>>,
C: Service<Request = Connect<Io, Codec>, Response = ConnectResult<Io, St, Codec, Out>>,
Io: AsyncRead + AsyncWrite,
Codec: Decoder + Encoder,
<Codec as Encoder>::Item: 'static,
<Codec as Encoder>::Error: std::fmt::Debug,
Out: Stream<Item = <Codec as Encoder>::Item> + Unpin,
{
/// Callback to execute on disconnect
///
/// Second parameter indicates error occured during disconnect.
pub fn disconnect<F, Out>(mut self, disconnect: F) -> Self
/// Construct framed handler service with specified connect service
pub fn new<F>(connect: F) -> Builder<St, C, Io, Codec, Out>
where
F: Fn(St, bool) + 'static,
F: IntoService<C>,
Io: AsyncRead + AsyncWrite,
C: Service<Request = Connect<Io, Codec>, Response = ConnectResult<Io, St, Codec, Out>>,
Codec: Decoder + Encoder,
Out: Stream<Item = <Codec as Encoder>::Item>,
{
self.disconnect = Some(Rc::new(disconnect));
self
Builder {
connect: connect.into_service(),
_t: PhantomData,
}
}
/// Provide stream items handler service and construct service factory.
pub fn finish<F, T>(self, service: F) -> FramedServiceImpl<St, C, T, Io, Codec>
pub fn build<F, T>(self, service: F) -> FramedServiceImpl<St, C, T, Io, Codec, Out>
where
F: IntoServiceFactory<T>,
T: ServiceFactory<
Config = St,
Request = RequestItem<St, Codec>,
Request = RequestItem<Codec>,
Response = ResponseItem<Codec>,
Error = C::Error,
InitError = C::Error,
@ -114,211 +63,261 @@ where
FramedServiceImpl {
connect: self.connect,
handler: Rc::new(service.into_factory()),
disconnect: self.disconnect.clone(),
_t: PhantomData,
}
}
}
pub struct NewServiceBuilder<St, C, Io, Codec> {
/// Service builder - structure that follows the builder pattern
/// for building instances for framed services.
pub struct FactoryBuilder<St, C, Io, Codec, Out> {
connect: C,
disconnect: Option<Rc<dyn Fn(St, bool)>>,
_t: PhantomData<(St, Io, Codec)>,
_t: PhantomData<(St, Io, Codec, Out)>,
}
impl<St, C, Io, Codec> NewServiceBuilder<St, C, Io, Codec>
impl<St, C, Io, Codec, Out> FactoryBuilder<St, C, Io, Codec, Out>
where
St: Clone,
Io: AsyncRead + AsyncWrite,
C: ServiceFactory<
Config = (),
Request = Connect<Io, Codec>,
Response = ConnectResult<Io, St, Codec>,
Response = ConnectResult<Io, St, Codec, Out>,
>,
C::Error: 'static,
C::Future: 'static,
Codec: Decoder + Encoder,
<Codec as Encoder>::Item: 'static,
<Codec as Encoder>::Error: std::fmt::Debug,
Out: Stream<Item = <Codec as Encoder>::Item> + Unpin,
{
/// Callback to execute on disconnect
///
/// Second parameter indicates error occured during disconnect.
pub fn disconnect<F>(mut self, disconnect: F) -> Self
/// Construct framed handler new service with specified connect service
pub fn new<F>(connect: F) -> FactoryBuilder<St, C, Io, Codec, Out>
where
F: Fn(St, bool) + 'static,
F: IntoServiceFactory<C>,
Io: AsyncRead + AsyncWrite,
C: ServiceFactory<
Config = (),
Request = Connect<Io, Codec>,
Response = ConnectResult<Io, St, Codec, Out>,
>,
Codec: Decoder + Encoder,
Out: Stream<Item = <Codec as Encoder>::Item> + Unpin,
{
self.disconnect = Some(Rc::new(disconnect));
self
FactoryBuilder {
connect: connect.into_factory(),
_t: PhantomData,
}
}
pub fn finish<F, T, Cfg>(self, service: F) -> FramedService<St, C, T, Io, Codec, Cfg>
pub fn build<F, T, Cfg>(self, service: F) -> FramedService<St, C, T, Io, Codec, Out, Cfg>
where
F: IntoServiceFactory<T>,
T: ServiceFactory<
Config = St,
Request = RequestItem<St, Codec>,
Request = RequestItem<Codec>,
Response = ResponseItem<Codec>,
Error = C::Error,
InitError = C::Error,
> + 'static,
>,
{
FramedService {
connect: self.connect,
handler: Rc::new(service.into_factory()),
disconnect: self.disconnect,
_t: PhantomData,
}
}
}
pub struct FramedService<St, C, T, Io, Codec, Cfg> {
pub struct FramedService<St, C, T, Io, Codec, Out, Cfg> {
connect: C,
handler: Rc<T>,
disconnect: Option<Rc<dyn Fn(St, bool)>>,
_t: PhantomData<(St, Io, Codec, Cfg)>,
_t: PhantomData<(St, Io, Codec, Out, Cfg)>,
}
impl<St, C, T, Io, Codec, Cfg> ServiceFactory for FramedService<St, C, T, Io, Codec, Cfg>
impl<St, C, T, Io, Codec, Out, Cfg> ServiceFactory
for FramedService<St, C, T, Io, Codec, Out, Cfg>
where
St: Clone + 'static,
Io: AsyncRead + AsyncWrite,
C: ServiceFactory<
Config = (),
Request = Connect<Io, Codec>,
Response = ConnectResult<Io, St, Codec>,
Response = ConnectResult<Io, St, Codec, Out>,
>,
C::Error: 'static,
C::Future: 'static,
T: ServiceFactory<
Config = St,
Request = RequestItem<St, Codec>,
Request = RequestItem<Codec>,
Response = ResponseItem<Codec>,
Error = C::Error,
InitError = C::Error,
> + 'static,
>,
<T::Service as Service>::Error: 'static,
<T::Service as Service>::Future: 'static,
Codec: Decoder + Encoder,
<Codec as Encoder>::Item: 'static,
<Codec as Encoder>::Error: std::fmt::Debug,
Out: Stream<Item = <Codec as Encoder>::Item> + Unpin,
{
type Config = Cfg;
type Request = Io;
type Response = ();
type Error = ServiceError<C::Error, Codec>;
type InitError = C::InitError;
type Service = FramedServiceImpl<St, C::Service, T, Io, Codec>;
type Future = LocalBoxFuture<'static, Result<Self::Service, Self::InitError>>;
type Service = FramedServiceImpl<St, C::Service, T, Io, Codec, Out>;
type Future = FramedServiceResponse<St, C, T, Io, Codec, Out>;
fn new_service(&self, _: Cfg) -> Self::Future {
let handler = self.handler.clone();
let disconnect = self.disconnect.clone();
// create connect service and then create service impl
self.connect
.new_service(())
.map(move |result| {
result.map(move |connect| FramedServiceImpl {
connect,
handler,
disconnect,
_t: PhantomData,
})
})
.boxed_local()
FramedServiceResponse {
fut: self.connect.new_service(()),
handler: self.handler.clone(),
}
}
}
pub struct FramedServiceImpl<St, C, T, Io, Codec> {
connect: C,
handler: Rc<T>,
disconnect: Option<Rc<dyn Fn(St, bool)>>,
_t: PhantomData<(St, Io, Codec)>,
}
impl<St, C, T, Io, Codec> Service for FramedServiceImpl<St, C, T, Io, Codec>
#[pin_project::pin_project]
pub struct FramedServiceResponse<St, C, T, Io, Codec, Out>
where
St: Clone,
Io: AsyncRead + AsyncWrite,
C: Service<Request = Connect<Io, Codec>, Response = ConnectResult<Io, St, Codec>>,
C::Error: 'static,
C: ServiceFactory<
Config = (),
Request = Connect<Io, Codec>,
Response = ConnectResult<Io, St, Codec, Out>,
>,
T: ServiceFactory<
Config = St,
Request = RequestItem<St, Codec>,
Request = RequestItem<Codec>,
Response = ResponseItem<Codec>,
Error = C::Error,
InitError = C::Error,
>,
<T::Service as Service>::Error: 'static,
<T::Service as Service>::Future: 'static,
Codec: Decoder + Encoder,
<Codec as Encoder>::Item: 'static,
<Codec as Encoder>::Error: std::fmt::Debug,
Out: Stream<Item = <Codec as Encoder>::Item> + Unpin,
{
#[pin]
fut: C::Future,
handler: Rc<T>,
}
impl<St, C, T, Io, Codec, Out> Future for FramedServiceResponse<St, C, T, Io, Codec, Out>
where
Io: AsyncRead + AsyncWrite,
C: ServiceFactory<
Config = (),
Request = Connect<Io, Codec>,
Response = ConnectResult<Io, St, Codec, Out>,
>,
T: ServiceFactory<
Config = St,
Request = RequestItem<Codec>,
Response = ResponseItem<Codec>,
Error = C::Error,
InitError = C::Error,
>,
<T::Service as Service>::Error: 'static,
<T::Service as Service>::Future: 'static,
Codec: Decoder + Encoder,
<Codec as Encoder>::Item: 'static,
<Codec as Encoder>::Error: std::fmt::Debug,
Out: Stream<Item = <Codec as Encoder>::Item> + Unpin,
{
type Output = Result<FramedServiceImpl<St, C::Service, T, Io, Codec, Out>, C::InitError>;
fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
let this = self.project();
let connect = ready!(this.fut.poll(cx))?;
Poll::Ready(Ok(FramedServiceImpl {
connect,
handler: this.handler.clone(),
_t: PhantomData,
}))
}
}
pub struct FramedServiceImpl<St, C, T, Io, Codec, Out> {
connect: C,
handler: Rc<T>,
_t: PhantomData<(St, Io, Codec, Out)>,
}
impl<St, C, T, Io, Codec, Out> Service for FramedServiceImpl<St, C, T, Io, Codec, Out>
where
Io: AsyncRead + AsyncWrite,
C: Service<Request = Connect<Io, Codec>, Response = ConnectResult<Io, St, Codec, Out>>,
T: ServiceFactory<
Config = St,
Request = RequestItem<Codec>,
Response = ResponseItem<Codec>,
Error = C::Error,
InitError = C::Error,
>,
<T::Service as Service>::Error: 'static,
<T::Service as Service>::Future: 'static,
Codec: Decoder + Encoder,
<Codec as Encoder>::Item: 'static,
<Codec as Encoder>::Error: std::fmt::Debug,
Out: Stream<Item = <Codec as Encoder>::Item> + Unpin,
{
type Request = Io;
type Response = ();
type Error = ServiceError<C::Error, Codec>;
type Future = FramedServiceImplResponse<St, Io, Codec, C, T>;
type Future = FramedServiceImplResponse<St, Io, Codec, Out, C, T>;
fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
self.connect.poll_ready(cx).map_err(|e| e.into())
}
fn call(&mut self, req: Io) -> Self::Future {
let (tx, rx) = mpsc::channel();
let sink = Sink::new(Rc::new(move |msg| {
let _ = tx.send(Ok(msg));
}));
FramedServiceImplResponse {
inner: FramedServiceImplResponseInner::Connect(
self.connect.call(Connect::new(req, sink.clone())),
self.connect.call(Connect::new(req)),
self.handler.clone(),
self.disconnect.clone(),
Some(rx),
),
}
}
}
#[pin_project::pin_project]
pub struct FramedServiceImplResponse<St, Io, Codec, C, T>
pub struct FramedServiceImplResponse<St, Io, Codec, Out, C, T>
where
St: Clone,
C: Service<Request = Connect<Io, Codec>, Response = ConnectResult<Io, St, Codec>>,
C::Error: 'static,
C: Service<Request = Connect<Io, Codec>, Response = ConnectResult<Io, St, Codec, Out>>,
T: ServiceFactory<
Config = St,
Request = RequestItem<St, Codec>,
Request = RequestItem<Codec>,
Response = ResponseItem<Codec>,
Error = C::Error,
InitError = C::Error,
>,
<T::Service as Service>::Error: 'static,
<T::Service as Service>::Future: 'static,
Io: AsyncRead + AsyncWrite,
Codec: Encoder + Decoder,
<Codec as Encoder>::Item: 'static,
<Codec as Encoder>::Error: std::fmt::Debug,
Out: Stream<Item = <Codec as Encoder>::Item> + Unpin,
{
#[pin]
inner: FramedServiceImplResponseInner<St, Io, Codec, C, T>,
inner: FramedServiceImplResponseInner<St, Io, Codec, Out, C, T>,
}
impl<St, Io, Codec, C, T> Future for FramedServiceImplResponse<St, Io, Codec, C, T>
impl<St, Io, Codec, Out, C, T> Future for FramedServiceImplResponse<St, Io, Codec, Out, C, T>
where
St: Clone,
C: Service<Request = Connect<Io, Codec>, Response = ConnectResult<Io, St, Codec>>,
C::Error: 'static,
C: Service<Request = Connect<Io, Codec>, Response = ConnectResult<Io, St, Codec, Out>>,
T: ServiceFactory<
Config = St,
Request = RequestItem<St, Codec>,
Request = RequestItem<Codec>,
Response = ResponseItem<Codec>,
Error = C::Error,
InitError = C::Error,
>,
<T::Service as Service>::Error: 'static,
<T::Service as Service>::Future: 'static,
Io: AsyncRead + AsyncWrite,
Codec: Encoder + Decoder,
<Codec as Encoder>::Item: 'static,
<Codec as Encoder>::Error: std::fmt::Debug,
Out: Stream<Item = <Codec as Encoder>::Item> + Unpin,
{
type Output = Result<(), ServiceError<C::Error, Codec>>;
@ -338,94 +337,71 @@ where
}
#[pin_project::pin_project]
enum FramedServiceImplResponseInner<St, Io, Codec, C, T>
enum FramedServiceImplResponseInner<St, Io, Codec, Out, C, T>
where
St: Clone,
C: Service<Request = Connect<Io, Codec>, Response = ConnectResult<Io, St, Codec>>,
C::Error: 'static,
C: Service<Request = Connect<Io, Codec>, Response = ConnectResult<Io, St, Codec, Out>>,
T: ServiceFactory<
Config = St,
Request = RequestItem<St, Codec>,
Request = RequestItem<Codec>,
Response = ResponseItem<Codec>,
Error = C::Error,
InitError = C::Error,
>,
<T::Service as Service>::Error: 'static,
<T::Service as Service>::Future: 'static,
Io: AsyncRead + AsyncWrite,
Codec: Encoder + Decoder,
<Codec as Encoder>::Item: 'static,
<Codec as Encoder>::Error: std::fmt::Debug,
Out: Stream<Item = <Codec as Encoder>::Item> + Unpin,
{
Connect(
#[pin] C::Future,
Rc<T>,
Option<Rc<dyn Fn(St, bool)>>,
Option<mpsc::Receiver<ServiceResult<Codec, C::Error>>>,
),
Handler(
#[pin] T::Future,
Option<ConnectResult<Io, St, Codec>>,
Option<Rc<dyn Fn(St, bool)>>,
Option<mpsc::Receiver<ServiceResult<Codec, C::Error>>>,
),
Dispatcher(Dispatcher<St, T::Service, Io, Codec>),
Connect(#[pin] C::Future, Rc<T>),
Handler(#[pin] T::Future, Option<Framed<Io, Codec>>, Option<Out>),
Dispatcher(Dispatcher<T::Service, Io, Codec, Out>),
}
impl<St, Io, Codec, C, T> FramedServiceImplResponseInner<St, Io, Codec, C, T>
impl<St, Io, Codec, Out, C, T> FramedServiceImplResponseInner<St, Io, Codec, Out, C, T>
where
St: Clone,
C: Service<Request = Connect<Io, Codec>, Response = ConnectResult<Io, St, Codec>>,
C::Error: 'static,
C: Service<Request = Connect<Io, Codec>, Response = ConnectResult<Io, St, Codec, Out>>,
T: ServiceFactory<
Config = St,
Request = RequestItem<St, Codec>,
Request = RequestItem<Codec>,
Response = ResponseItem<Codec>,
Error = C::Error,
InitError = C::Error,
>,
<T::Service as Service>::Error: 'static,
<T::Service as Service>::Future: 'static,
Io: AsyncRead + AsyncWrite,
Codec: Encoder + Decoder,
<Codec as Encoder>::Item: 'static,
<Codec as Encoder>::Error: std::fmt::Debug,
Out: Stream<Item = <Codec as Encoder>::Item> + Unpin,
{
#[project]
fn poll(
self: Pin<&mut Self>,
cx: &mut Context<'_>,
) -> Either<
FramedServiceImplResponseInner<St, Io, Codec, C, T>,
FramedServiceImplResponseInner<St, Io, Codec, Out, C, T>,
Poll<Result<(), ServiceError<C::Error, Codec>>>,
> {
#[project]
match self.project() {
FramedServiceImplResponseInner::Connect(fut, handler, disconnect, rx) => {
match fut.poll(cx) {
Poll::Ready(Ok(res)) => {
Either::Left(FramedServiceImplResponseInner::Handler(
handler.new_service(res.state.clone()),
Some(res),
disconnect.take(),
rx.take(),
))
}
FramedServiceImplResponseInner::Connect(fut, handler) => match fut.poll(cx) {
Poll::Ready(Ok(res)) => Either::Left(FramedServiceImplResponseInner::Handler(
handler.new_service(res.state),
Some(res.framed),
res.out,
)),
Poll::Pending => Either::Right(Poll::Pending),
Poll::Ready(Err(e)) => Either::Right(Poll::Ready(Err(e.into()))),
}
}
FramedServiceImplResponseInner::Handler(fut, res, disconnect, rx) => {
},
FramedServiceImplResponseInner::Handler(fut, framed, out) => {
match fut.poll(cx) {
Poll::Ready(Ok(handler)) => {
let res = res.take().unwrap();
Either::Left(FramedServiceImplResponseInner::Dispatcher(
Dispatcher::new(
res.framed,
res.state,
handler,
res.sink,
rx.take().unwrap(),
disconnect.take(),
),
Dispatcher::new(framed.take().unwrap(), handler, out.take()),
))
}
Poll::Pending => Either::Right(Poll::Pending),

45
actix-ioframe/src/sink.rs
View File

@ -1,45 +0,0 @@
use std::fmt;
use std::rc::Rc;
use actix_utils::oneshot;
use futures::future::{Future, FutureExt};
use crate::dispatcher::Message;
pub struct Sink<T>(Rc<dyn Fn(Message<T>)>);
impl<T> Clone for Sink<T> {
fn clone(&self) -> Self {
Sink(self.0.clone())
}
}
impl<T> Sink<T> {
pub(crate) fn new(tx: Rc<dyn Fn(Message<T>)>) -> Self {
Sink(tx)
}
/// Close connection
pub fn close(&self) {
(self.0)(Message::Close);
}
/// Close connection
pub fn wait_close(&self) -> impl Future<Output = ()> {
let (tx, rx) = oneshot::channel();
(self.0)(Message::WaitClose(tx));
rx.map(|_| ())
}
/// Send item
pub fn send(&self, item: T) {
(self.0)(Message::Item(item));
}
}
impl<T> fmt::Debug for Sink<T> {
fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result {
fmt.debug_struct("Sink").finish()
}
}

61
actix-ioframe/tests/test_server.rs
View File

@ -1,43 +1,49 @@
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;
use std::time::Duration;
use std::cell::Cell;
use std::rc::Rc;
use actix_codec::BytesCodec;
use actix_rt::time::delay_for;
use actix_service::{fn_service, Service};
use actix_service::{fn_factory_with_config, fn_service, IntoService, Service};
use actix_testing::TestServer;
use actix_utils::mpsc;
use bytes::{Bytes, BytesMut};
use futures::future::ok;
use actix_ioframe::{Builder, Connect};
use actix_ioframe::{Builder, Connect, FactoryBuilder};
#[derive(Clone)]
struct State;
struct State(Option<mpsc::Sender<Bytes>>);
#[actix_rt::test]
async fn test_disconnect() -> std::io::Result<()> {
let disconnect = Arc::new(AtomicBool::new(false));
let disconnect1 = disconnect.clone();
async fn test_basic() {
let client_item = Rc::new(Cell::new(false));
let srv = TestServer::with(move || {
let disconnect1 = disconnect1.clone();
Builder::new()
.factory(fn_service(|conn: Connect<_, _>| {
ok(conn.codec(BytesCodec).state(State))
FactoryBuilder::new(fn_service(|conn: Connect<_, _>| {
ok(conn.codec(BytesCodec).state(State(None)))
}))
.disconnect(move |_, _| {
disconnect1.store(true, Ordering::Relaxed);
})
.finish(fn_service(|_t| ok(None)))
// echo
.build(fn_service(|t: BytesMut| ok(Some(t.freeze()))))
});
let mut client = Builder::new()
.service(|conn: Connect<_, _>| {
let conn = conn.codec(BytesCodec).state(State);
conn.sink().close();
ok(conn)
let item = client_item.clone();
let mut client = Builder::new(fn_service(move |conn: Connect<_, _>| {
async move {
let (tx, rx) = mpsc::channel();
let _ = tx.send(Bytes::from_static(b"Hello"));
Ok(conn.codec(BytesCodec).out(rx).state(State(Some(tx))))
}
}))
.build(fn_factory_with_config(move |mut cfg: State| {
let item = item.clone();
ok((move |t: BytesMut| {
assert_eq!(t.freeze(), Bytes::from_static(b"Hello"));
item.set(true);
// drop Sender, which will close connection
cfg.0.take();
ok::<_, ()>(None)
})
.finish(fn_service(|_t| ok(None)));
.into_service())
}));
let conn = actix_connect::default_connector()
.call(actix_connect::Connect::with(String::new(), srv.addr()))
@ -45,8 +51,5 @@ async fn test_disconnect() -> std::io::Result<()> {
.unwrap();
client.call(conn.into_parts().0).await.unwrap();
let _ = delay_for(Duration::from_millis(100)).await;
assert!(disconnect.load(Ordering::Relaxed));
Ok(())
assert!(client_item.get());
}