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

remove E param

Browse Source
This commit is contained in:
Nikolay Kim 2019-12-11 16:44:09 +06:00
parent f26fcc703b
commit c7a8743bf9
8 changed files with 130 additions and 162 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
actix-ioframe/CHANGES.md
View File

@ -1,5 +1,9 @@
# Changes # Changes
## [0.4.0] - 2019-12-1
* Remove `E` param
## [0.3.0-alpha.3] - 2019-12-07 ## [0.3.0-alpha.3] - 2019-12-07
* Migrate to tokio 0.2 * Migrate to tokio 0.2

2
actix-ioframe/Cargo.toml
View File

@ -1,6 +1,6 @@
[package] [package]
name = "actix-ioframe" name = "actix-ioframe"
version = "0.3.0" version = "0.4.0"
authors = ["Nikolay Kim <fafhrd91@gmail.com>"] authors = ["Nikolay Kim <fafhrd91@gmail.com>"]
description = "Actix framed service" description = "Actix framed service"
keywords = ["network", "framework", "async", "futures"] keywords = ["network", "framework", "async", "futures"]

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

@ -7,21 +7,21 @@ use futures::Stream;
use crate::sink::Sink; use crate::sink::Sink;
pub struct Connect<Io, Codec, Err, St = ()> pub struct Connect<Io, Codec, St = ()>
where where
Codec: Encoder + Decoder, Codec: Encoder + Decoder,
{ {
io: Io, io: Io,
sink: Sink<<Codec as Encoder>::Item, Err>, sink: Sink<<Codec as Encoder>::Item>,
_t: PhantomData<(St, Codec)>, _t: PhantomData<(St, Codec)>,
} }
impl<Io, Codec, Err> Connect<Io, Codec, Err> impl<Io, Codec> Connect<Io, Codec>
where where
Io: AsyncRead + AsyncWrite, Io: AsyncRead + AsyncWrite,
Codec: Encoder + Decoder, Codec: Encoder + Decoder,
{ {
pub(crate) fn new(io: Io, sink: Sink<<Codec as Encoder>::Item, Err>) -> Self { pub(crate) fn new(io: Io, sink: Sink<<Codec as Encoder>::Item>) -> Self {
Self { Self {
io, io,
sink, sink,
@ -29,7 +29,7 @@ where
} }
} }
pub fn codec(self, codec: Codec) -> ConnectResult<Io, (), Codec, Err> { pub fn codec(self, codec: Codec) -> ConnectResult<Io, (), Codec> {
ConnectResult { ConnectResult {
state: (), state: (),
sink: self.sink, sink: self.sink,
@ -39,15 +39,15 @@ where
} }
#[pin_project::pin_project] #[pin_project::pin_project]
pub struct ConnectResult<Io, St, Codec: Encoder + Decoder, Err> { pub struct ConnectResult<Io, St, Codec: Encoder + Decoder> {
pub(crate) state: St, pub(crate) state: St,
pub(crate) framed: Framed<Io, Codec>, pub(crate) framed: Framed<Io, Codec>,
pub(crate) sink: Sink<<Codec as Encoder>::Item, Err>, pub(crate) sink: Sink<<Codec as Encoder>::Item>,
} }
impl<Io, St, Codec: Encoder + Decoder, Err> ConnectResult<Io, St, Codec, Err> { impl<Io, St, Codec: Encoder + Decoder> ConnectResult<Io, St, Codec> {
#[inline] #[inline]
pub fn sink(&self) -> &Sink<<Codec as Encoder>::Item, Err> { pub fn sink(&self) -> &Sink<<Codec as Encoder>::Item> {
&self.sink &self.sink
} }
@ -62,7 +62,7 @@ impl<Io, St, Codec: Encoder + Decoder, Err> ConnectResult<Io, St, Codec, Err> {
} }
#[inline] #[inline]
pub fn state<S>(self, state: S) -> ConnectResult<Io, S, Codec, Err> { pub fn state<S>(self, state: S) -> ConnectResult<Io, S, Codec> {
ConnectResult { ConnectResult {
state, state,
framed: self.framed, framed: self.framed,
@ -71,7 +71,7 @@ impl<Io, St, Codec: Encoder + Decoder, Err> ConnectResult<Io, St, Codec, Err> {
} }
} }
impl<Io, St, Codec, Err> Stream for ConnectResult<Io, St, Codec, Err> impl<Io, St, Codec> Stream for ConnectResult<Io, St, Codec>
where where
Io: AsyncRead + AsyncWrite, Io: AsyncRead + AsyncWrite,
Codec: Encoder + Decoder, Codec: Encoder + Decoder,
@ -83,8 +83,7 @@ where
} }
} }
impl<Io, St, Codec, Err> futures::Sink<<Codec as Encoder>::Item> impl<Io, St, Codec> futures::Sink<<Codec as Encoder>::Item> for ConnectResult<Io, St, Codec>
for ConnectResult<Io, St, Codec, Err>
where where
Io: AsyncRead + AsyncWrite, Io: AsyncRead + AsyncWrite,
Codec: Encoder + Decoder, Codec: Encoder + Decoder,

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

@ -13,7 +13,7 @@ use crate::error::ServiceError;
use crate::item::Item; use crate::item::Item;
use crate::sink::Sink; use crate::sink::Sink;
type Request<S, U, E> = Item<S, U, E>; type Request<S, U> = Item<S, U>;
type Response<U> = <U as Encoder>::Item; type Response<U> = <U as Encoder>::Item;
pub(crate) enum Message<T> { pub(crate) enum Message<T> {
@ -25,10 +25,10 @@ pub(crate) enum Message<T> {
/// 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.
#[pin_project::pin_project] #[pin_project::pin_project]
pub(crate) struct Dispatcher<St, S, T, U, E> pub(crate) struct Dispatcher<St, S, T, U>
where where
St: Clone, St: Clone,
S: Service<Request = Request<St, U, E>, Response = Option<Response<U>>, Error = E>, S: Service<Request = Request<St, U>, Response = Option<Response<U>>>,
S::Error: 'static, S::Error: 'static,
S::Future: 'static, S::Future: 'static,
T: AsyncRead + AsyncWrite, T: AsyncRead + AsyncWrite,
@ -37,19 +37,19 @@ where
<U as Encoder>::Error: std::fmt::Debug, <U as Encoder>::Error: std::fmt::Debug,
{ {
service: S, service: S,
sink: Sink<<U as Encoder>::Item, E>, sink: Sink<<U as Encoder>::Item>,
state: St, state: St,
dispatch_state: FramedState<S, U>, dispatch_state: FramedState<S, U>,
framed: Framed<T, U>, framed: Framed<T, U>,
rx: mpsc::Receiver<Result<Message<<U as Encoder>::Item>, E>>, rx: mpsc::Receiver<Result<Message<<U as Encoder>::Item>, S::Error>>,
tx: mpsc::Sender<Result<Message<<U as Encoder>::Item>, E>>, tx: mpsc::Sender<Result<Message<<U as Encoder>::Item>, S::Error>>,
disconnect: Option<Rc<dyn Fn(&mut St, bool)>>, disconnect: Option<Rc<dyn Fn(&mut St, bool)>>,
} }
impl<St, S, T, U, E> Dispatcher<St, S, T, U, E> impl<St, S, T, U> Dispatcher<St, S, T, U>
where where
St: Clone, St: Clone,
S: Service<Request = Request<St, U, E>, Response = Option<Response<U>>, Error = E>, S: Service<Request = Request<St, U>, Response = Option<Response<U>>>,
S::Error: 'static, S::Error: 'static,
S::Future: 'static, S::Future: 'static,
T: AsyncRead + AsyncWrite, T: AsyncRead + AsyncWrite,
@ -61,8 +61,8 @@ where
framed: Framed<T, U>, framed: Framed<T, U>,
state: St, state: St,
service: F, service: F,
sink: Sink<<U as Encoder>::Item, E>, sink: Sink<<U as Encoder>::Item>,
rx: mpsc::Receiver<Result<Message<<U as Encoder>::Item>, E>>, rx: mpsc::Receiver<Result<Message<<U as Encoder>::Item>, S::Error>>,
disconnect: Option<Rc<dyn Fn(&mut St, bool)>>, disconnect: Option<Rc<dyn Fn(&mut St, bool)>>,
) -> Self { ) -> Self {
let tx = rx.sender(); let tx = rx.sender();
@ -126,10 +126,10 @@ impl<S: Service, U: Encoder + Decoder> FramedState<S, U> {
} }
} }
impl<St, S, T, U, E> Dispatcher<St, S, T, U, E> impl<St, S, T, U> Dispatcher<St, S, T, U>
where where
St: Clone, St: Clone,
S: Service<Request = Request<St, U, E>, Response = Option<Response<U>>, Error = E>, S: Service<Request = Request<St, U>, Response = Option<Response<U>>>,
S::Error: 'static, S::Error: 'static,
S::Future: 'static, S::Future: 'static,
T: AsyncRead + AsyncWrite, T: AsyncRead + AsyncWrite,

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

@ -5,19 +5,19 @@ use actix_codec::{Decoder, Encoder};
use crate::sink::Sink; use crate::sink::Sink;
pub struct Item<St, Codec: Encoder + Decoder, E> { pub struct Item<St, Codec: Encoder + Decoder> {
state: St, state: St,
sink: Sink<<Codec as Encoder>::Item, E>, sink: Sink<<Codec as Encoder>::Item>,
item: <Codec as Decoder>::Item, item: <Codec as Decoder>::Item,
} }
impl<St, Codec, E> Item<St, Codec, E> impl<St, Codec> Item<St, Codec>
where where
Codec: Encoder + Decoder, Codec: Encoder + Decoder,
{ {
pub(crate) fn new( pub(crate) fn new(
state: St, state: St,
sink: Sink<<Codec as Encoder>::Item, E>, sink: Sink<<Codec as Encoder>::Item>,
item: <Codec as Decoder>::Item, item: <Codec as Decoder>::Item,
) -> Self { ) -> Self {
Item { state, sink, item } Item { state, sink, item }
@ -34,7 +34,7 @@ where
} }
#[inline] #[inline]
pub fn sink(&self) -> &Sink<<Codec as Encoder>::Item, E> { pub fn sink(&self) -> &Sink<<Codec as Encoder>::Item> {
&self.sink &self.sink
} }
@ -44,18 +44,12 @@ where
} }
#[inline] #[inline]
pub fn into_parts( pub fn into_parts(self) -> (St, Sink<<Codec as Encoder>::Item>, <Codec as Decoder>::Item) {
self,
) -> (
St,
Sink<<Codec as Encoder>::Item, E>,
<Codec as Decoder>::Item,
) {
(self.state, self.sink, self.item) (self.state, self.sink, self.item)
} }
} }
impl<St, Codec, E> Deref for Item<St, Codec, E> impl<St, Codec> Deref for Item<St, Codec>
where where
Codec: Encoder + Decoder, Codec: Encoder + Decoder,
{ {
@ -67,7 +61,7 @@ where
} }
} }
impl<St, Codec, E> DerefMut for Item<St, Codec, E> impl<St, Codec> DerefMut for Item<St, Codec>
where where
Codec: Encoder + Decoder, Codec: Encoder + Decoder,
{ {
@ -77,7 +71,7 @@ where
} }
} }
impl<St, Codec, E> fmt::Debug for Item<St, Codec, E> impl<St, Codec> fmt::Debug for Item<St, Codec>
where where
Codec: Encoder + Decoder, Codec: Encoder + Decoder,
<Codec as Decoder>::Item: fmt::Debug, <Codec as Decoder>::Item: fmt::Debug,

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

@ -17,7 +17,7 @@ use crate::error::ServiceError;
use crate::item::Item; use crate::item::Item;
use crate::sink::Sink; use crate::sink::Sink;
type RequestItem<S, U, E> = Item<S, U, E>; type RequestItem<S, U> = Item<S, U>;
type ResponseItem<U> = Option<<U as Encoder>::Item>; type ResponseItem<U> = Option<<U as Encoder>::Item>;
type ServiceResult<U, E> = Result<Message<<U as Encoder>::Item>, E>; type ServiceResult<U, E> = Result<Message<<U as Encoder>::Item>, E>;
@ -37,15 +37,11 @@ impl<St: Clone, Codec> Builder<St, Codec> {
} }
/// Construct framed handler service with specified connect service /// Construct framed handler service with specified connect service
pub fn service<Io, C, F, E>(self, connect: F) -> ServiceBuilder<St, C, Io, Codec, E> pub fn service<Io, C, F>(self, connect: F) -> ServiceBuilder<St, C, Io, Codec>
where where
F: IntoService<C>, F: IntoService<C>,
Io: AsyncRead + AsyncWrite, Io: AsyncRead + AsyncWrite,
C: Service< C: Service<Request = Connect<Io, Codec>, Response = ConnectResult<Io, St, Codec>>,
Request = Connect<Io, Codec, E>,
Response = ConnectResult<Io, St, Codec, E>,
Error = E,
>,
Codec: Decoder + Encoder, Codec: Decoder + Encoder,
{ {
ServiceBuilder { ServiceBuilder {
@ -56,17 +52,16 @@ impl<St: Clone, Codec> Builder<St, Codec> {
} }
/// Construct framed handler new service with specified connect service /// Construct framed handler new service with specified connect service
pub fn factory<Io, C, F, E>(self, connect: F) -> NewServiceBuilder<St, C, Io, Codec, E> pub fn factory<Io, C, F>(self, connect: F) -> NewServiceBuilder<St, C, Io, Codec>
where where
F: IntoServiceFactory<C>, F: IntoServiceFactory<C>,
E: 'static,
Io: AsyncRead + AsyncWrite, Io: AsyncRead + AsyncWrite,
C: ServiceFactory< C: ServiceFactory<
Config = (), Config = (),
Request = Connect<Io, Codec, E>, Request = Connect<Io, Codec>,
Response = ConnectResult<Io, St, Codec, E>, Response = ConnectResult<Io, St, Codec>,
Error = E,
>, >,
C::Error: 'static,
C::Future: 'static, C::Future: 'static,
Codec: Decoder + Encoder, Codec: Decoder + Encoder,
{ {
@ -78,20 +73,16 @@ impl<St: Clone, Codec> Builder<St, Codec> {
} }
} }
pub struct ServiceBuilder<St, C, Io, Codec, Err> { pub struct ServiceBuilder<St, C, Io, Codec> {
connect: C, connect: C,
disconnect: Option<Rc<dyn Fn(&mut St, bool)>>, disconnect: Option<Rc<dyn Fn(&mut St, bool)>>,
_t: PhantomData<(St, Io, Codec, Err)>, _t: PhantomData<(St, Io, Codec)>,
} }
impl<St, C, Io, Codec, Err> ServiceBuilder<St, C, Io, Codec, Err> impl<St, C, Io, Codec> ServiceBuilder<St, C, Io, Codec>
where where
St: Clone, St: Clone,
C: Service< C: Service<Request = Connect<Io, Codec>, Response = ConnectResult<Io, St, Codec>>,
Request = Connect<Io, Codec, Err>,
Response = ConnectResult<Io, St, Codec, Err>,
Error = Err,
>,
Io: AsyncRead + AsyncWrite, Io: AsyncRead + AsyncWrite,
Codec: Decoder + Encoder, Codec: Decoder + Encoder,
<Codec as Encoder>::Item: 'static, <Codec as Encoder>::Item: 'static,
@ -109,15 +100,15 @@ where
} }
/// Provide stream items handler service and construct service factory. /// Provide stream items handler service and construct service factory.
pub fn finish<F, T>(self, service: F) -> FramedServiceImpl<St, C, T, Io, Codec, Err> pub fn finish<F, T>(self, service: F) -> FramedServiceImpl<St, C, T, Io, Codec>
where where
F: IntoServiceFactory<T>, F: IntoServiceFactory<T>,
T: ServiceFactory< T: ServiceFactory<
Config = St, Config = St,
Request = RequestItem<St, Codec, Err>, Request = RequestItem<St, Codec>,
Response = ResponseItem<Codec>, Response = ResponseItem<Codec>,
Error = Err, Error = C::Error,
InitError = Err, InitError = C::Error,
>, >,
{ {
FramedServiceImpl { FramedServiceImpl {
@ -129,23 +120,22 @@ where
} }
} }
pub struct NewServiceBuilder<St, C, Io, Codec, Err> { pub struct NewServiceBuilder<St, C, Io, Codec> {
connect: C, connect: C,
disconnect: Option<Rc<dyn Fn(&mut St, bool)>>, disconnect: Option<Rc<dyn Fn(&mut St, bool)>>,
_t: PhantomData<(St, Io, Codec, Err)>, _t: PhantomData<(St, Io, Codec)>,
} }
impl<St, C, Io, Codec, Err> NewServiceBuilder<St, C, Io, Codec, Err> impl<St, C, Io, Codec> NewServiceBuilder<St, C, Io, Codec>
where where
St: Clone, St: Clone,
Io: AsyncRead + AsyncWrite, Io: AsyncRead + AsyncWrite,
Err: 'static,
C: ServiceFactory< C: ServiceFactory<
Config = (), Config = (),
Request = Connect<Io, Codec, Err>, Request = Connect<Io, Codec>,
Response = ConnectResult<Io, St, Codec, Err>, Response = ConnectResult<Io, St, Codec>,
Error = Err,
>, >,
C::Error: 'static,
C::Future: 'static, C::Future: 'static,
Codec: Decoder + Encoder, Codec: Decoder + Encoder,
<Codec as Encoder>::Item: 'static, <Codec as Encoder>::Item: 'static,
@ -162,15 +152,15 @@ where
self self
} }
pub fn finish<F, T, Cfg>(self, service: F) -> FramedService<St, C, T, Io, Codec, Err, Cfg> pub fn finish<F, T, Cfg>(self, service: F) -> FramedService<St, C, T, Io, Codec, Cfg>
where where
F: IntoServiceFactory<T>, F: IntoServiceFactory<T>,
T: ServiceFactory< T: ServiceFactory<
Config = St, Config = St,
Request = RequestItem<St, Codec, Err>, Request = RequestItem<St, Codec>,
Response = ResponseItem<Codec>, Response = ResponseItem<Codec>,
Error = Err, Error = C::Error,
InitError = Err, InitError = C::Error,
> + 'static, > + 'static,
{ {
FramedService { FramedService {
@ -182,34 +172,32 @@ where
} }
} }
pub struct FramedService<St, C, T, Io, Codec, Err, Cfg> { pub struct FramedService<St, C, T, Io, Codec, Cfg> {
connect: C, connect: C,
handler: Rc<T>, handler: Rc<T>,
disconnect: Option<Rc<dyn Fn(&mut St, bool)>>, disconnect: Option<Rc<dyn Fn(&mut St, bool)>>,
_t: PhantomData<(St, Io, Codec, Err, Cfg)>, _t: PhantomData<(St, Io, Codec, Cfg)>,
} }
impl<St, C, T, Io, Codec, Err, Cfg> ServiceFactory impl<St, C, T, Io, Codec, Cfg> ServiceFactory for FramedService<St, C, T, Io, Codec, Cfg>
for FramedService<St, C, T, Io, Codec, Err, Cfg>
where where
St: Clone + 'static, St: Clone + 'static,
Io: AsyncRead + AsyncWrite, Io: AsyncRead + AsyncWrite,
C: ServiceFactory< C: ServiceFactory<
Config = (), Config = (),
Request = Connect<Io, Codec, Err>, Request = Connect<Io, Codec>,
Response = ConnectResult<Io, St, Codec, Err>, Response = ConnectResult<Io, St, Codec>,
Error = Err,
>, >,
C::Error: 'static,
C::Future: 'static, C::Future: 'static,
T: ServiceFactory< T: ServiceFactory<
Config = St, Config = St,
Request = RequestItem<St, Codec, Err>, Request = RequestItem<St, Codec>,
Response = ResponseItem<Codec>, Response = ResponseItem<Codec>,
Error = Err, Error = C::Error,
InitError = Err, InitError = C::Error,
> + 'static, > + 'static,
<T::Service as Service>::Future: 'static, <T::Service as Service>::Future: 'static,
Err: 'static,
Codec: Decoder + Encoder, Codec: Decoder + Encoder,
<Codec as Encoder>::Item: 'static, <Codec as Encoder>::Item: 'static,
<Codec as Encoder>::Error: std::fmt::Debug, <Codec as Encoder>::Error: std::fmt::Debug,
@ -219,7 +207,7 @@ where
type Response = (); type Response = ();
type Error = ServiceError<C::Error, Codec>; type Error = ServiceError<C::Error, Codec>;
type InitError = C::InitError; type InitError = C::InitError;
type Service = FramedServiceImpl<St, C::Service, T, Io, Codec, Err>; type Service = FramedServiceImpl<St, C::Service, T, Io, Codec>;
type Future = LocalBoxFuture<'static, Result<Self::Service, Self::InitError>>; type Future = LocalBoxFuture<'static, Result<Self::Service, Self::InitError>>;
fn new_service(&self, _: Cfg) -> Self::Future { fn new_service(&self, _: Cfg) -> Self::Future {
@ -241,29 +229,25 @@ where
} }
} }
pub struct FramedServiceImpl<St, C, T, Io, Codec, Err> { pub struct FramedServiceImpl<St, C, T, Io, Codec> {
connect: C, connect: C,
handler: Rc<T>, handler: Rc<T>,
disconnect: Option<Rc<dyn Fn(&mut St, bool)>>, disconnect: Option<Rc<dyn Fn(&mut St, bool)>>,
_t: PhantomData<(St, Io, Codec, Err)>, _t: PhantomData<(St, Io, Codec)>,
} }
impl<St, C, T, Io, Codec, Err> Service for FramedServiceImpl<St, C, T, Io, Codec, Err> impl<St, C, T, Io, Codec> Service for FramedServiceImpl<St, C, T, Io, Codec>
where where
St: Clone, St: Clone,
Io: AsyncRead + AsyncWrite, Io: AsyncRead + AsyncWrite,
C: Service< C: Service<Request = Connect<Io, Codec>, Response = ConnectResult<Io, St, Codec>>,
Request = Connect<Io, Codec, Err>, C::Error: 'static,
Response = ConnectResult<Io, St, Codec, Err>,
Error = Err,
>,
Err: 'static,
T: ServiceFactory< T: ServiceFactory<
Config = St, Config = St,
Request = RequestItem<St, Codec, Err>, Request = RequestItem<St, Codec>,
Response = ResponseItem<Codec>, Response = ResponseItem<Codec>,
Error = Err, Error = C::Error,
InitError = Err, InitError = C::Error,
>, >,
<T::Service as Service>::Future: 'static, <T::Service as Service>::Future: 'static,
Codec: Decoder + Encoder, Codec: Decoder + Encoder,
@ -272,8 +256,8 @@ where
{ {
type Request = Io; type Request = Io;
type Response = (); type Response = ();
type Error = ServiceError<Err, Codec>; type Error = ServiceError<C::Error, Codec>;
type Future = FramedServiceImplResponse<St, Io, Codec, Err, C, T>; type Future = FramedServiceImplResponse<St, Io, Codec, C, T>;
fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> { fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
self.connect.poll_ready(cx).map_err(|e| e.into()) self.connect.poll_ready(cx).map_err(|e| e.into())
@ -281,7 +265,9 @@ where
fn call(&mut self, req: Io) -> Self::Future { fn call(&mut self, req: Io) -> Self::Future {
let (tx, rx) = mpsc::channel(); let (tx, rx) = mpsc::channel();
let sink = Sink::new(tx); let sink = Sink::new(Rc::new(move |msg| {
let _ = tx.send(Ok(msg));
}));
FramedServiceImplResponse { FramedServiceImplResponse {
inner: FramedServiceImplResponseInner::Connect( inner: FramedServiceImplResponseInner::Connect(
self.connect.call(Connect::new(req, sink.clone())), self.connect.call(Connect::new(req, sink.clone())),
@ -294,21 +280,17 @@ where
} }
#[pin_project::pin_project] #[pin_project::pin_project]
pub struct FramedServiceImplResponse<St, Io, Codec, Err, C, T> pub struct FramedServiceImplResponse<St, Io, Codec, C, T>
where where
St: Clone, St: Clone,
C: Service< C: Service<Request = Connect<Io, Codec>, Response = ConnectResult<Io, St, Codec>>,
Request = Connect<Io, Codec, Err>, C::Error: 'static,
Response = ConnectResult<Io, St, Codec, Err>,
Error = Err,
>,
Err: 'static,
T: ServiceFactory< T: ServiceFactory<
Config = St, Config = St,
Request = RequestItem<St, Codec, Err>, Request = RequestItem<St, Codec>,
Response = ResponseItem<Codec>, Response = ResponseItem<Codec>,
Error = Err, Error = C::Error,
InitError = Err, InitError = C::Error,
>, >,
<T::Service as Service>::Future: 'static, <T::Service as Service>::Future: 'static,
Io: AsyncRead + AsyncWrite, Io: AsyncRead + AsyncWrite,
@ -317,24 +299,20 @@ where
<Codec as Encoder>::Error: std::fmt::Debug, <Codec as Encoder>::Error: std::fmt::Debug,
{ {
#[pin] #[pin]
inner: FramedServiceImplResponseInner<St, Io, Codec, Err, C, T>, inner: FramedServiceImplResponseInner<St, Io, Codec, C, T>,
} }
impl<St, Io, Codec, Err, C, T> Future for FramedServiceImplResponse<St, Io, Codec, Err, C, T> impl<St, Io, Codec, C, T> Future for FramedServiceImplResponse<St, Io, Codec, C, T>
where where
St: Clone, St: Clone,
C: Service< C: Service<Request = Connect<Io, Codec>, Response = ConnectResult<Io, St, Codec>>,
Request = Connect<Io, Codec, Err>, C::Error: 'static,
Response = ConnectResult<Io, St, Codec, Err>,
Error = Err,
>,
Err: 'static,
T: ServiceFactory< T: ServiceFactory<
Config = St, Config = St,
Request = RequestItem<St, Codec, Err>, Request = RequestItem<St, Codec>,
Response = ResponseItem<Codec>, Response = ResponseItem<Codec>,
Error = Err, Error = C::Error,
InitError = Err, InitError = C::Error,
>, >,
<T::Service as Service>::Future: 'static, <T::Service as Service>::Future: 'static,
Io: AsyncRead + AsyncWrite, Io: AsyncRead + AsyncWrite,
@ -342,7 +320,7 @@ where
<Codec as Encoder>::Item: 'static, <Codec as Encoder>::Item: 'static,
<Codec as Encoder>::Error: std::fmt::Debug, <Codec as Encoder>::Error: std::fmt::Debug,
{ {
type Output = Result<(), ServiceError<Err, Codec>>; type Output = Result<(), ServiceError<C::Error, Codec>>;
fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> { fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
let mut this = self.as_mut().project(); let mut this = self.as_mut().project();
@ -360,21 +338,17 @@ where
} }
#[pin_project::pin_project] #[pin_project::pin_project]
enum FramedServiceImplResponseInner<St, Io, Codec, Err, C, T> enum FramedServiceImplResponseInner<St, Io, Codec, C, T>
where where
St: Clone, St: Clone,
C: Service< C: Service<Request = Connect<Io, Codec>, Response = ConnectResult<Io, St, Codec>>,
Request = Connect<Io, Codec, Err>, C::Error: 'static,
Response = ConnectResult<Io, St, Codec, Err>,
Error = Err,
>,
Err: 'static,
T: ServiceFactory< T: ServiceFactory<
Config = St, Config = St,
Request = RequestItem<St, Codec, Err>, Request = RequestItem<St, Codec>,
Response = ResponseItem<Codec>, Response = ResponseItem<Codec>,
Error = Err, Error = C::Error,
InitError = Err, InitError = C::Error,
>, >,
<T::Service as Service>::Future: 'static, <T::Service as Service>::Future: 'static,
Io: AsyncRead + AsyncWrite, Io: AsyncRead + AsyncWrite,
@ -386,32 +360,28 @@ where
#[pin] C::Future, #[pin] C::Future,
Rc<T>, Rc<T>,
Option<Rc<dyn Fn(&mut St, bool)>>, Option<Rc<dyn Fn(&mut St, bool)>>,
Option<mpsc::Receiver<ServiceResult<Codec, Err>>>, Option<mpsc::Receiver<ServiceResult<Codec, C::Error>>>,
), ),
Handler( Handler(
#[pin] T::Future, #[pin] T::Future,
Option<ConnectResult<Io, St, Codec, Err>>, Option<ConnectResult<Io, St, Codec>>,
Option<Rc<dyn Fn(&mut St, bool)>>, Option<Rc<dyn Fn(&mut St, bool)>>,
Option<mpsc::Receiver<ServiceResult<Codec, Err>>>, Option<mpsc::Receiver<ServiceResult<Codec, C::Error>>>,
), ),
Dispatcher(Dispatcher<St, T::Service, Io, Codec, Err>), Dispatcher(Dispatcher<St, T::Service, Io, Codec>),
} }
impl<St, Io, Codec, Err, C, T> FramedServiceImplResponseInner<St, Io, Codec, Err, C, T> impl<St, Io, Codec, C, T> FramedServiceImplResponseInner<St, Io, Codec, C, T>
where where
St: Clone, St: Clone,
C: Service< C: Service<Request = Connect<Io, Codec>, Response = ConnectResult<Io, St, Codec>>,
Request = Connect<Io, Codec, Err>, C::Error: 'static,
Response = ConnectResult<Io, St, Codec, Err>,
Error = Err,
>,
Err: 'static,
T: ServiceFactory< T: ServiceFactory<
Config = St, Config = St,
Request = RequestItem<St, Codec, Err>, Request = RequestItem<St, Codec>,
Response = ResponseItem<Codec>, Response = ResponseItem<Codec>,
Error = Err, Error = C::Error,
InitError = Err, InitError = C::Error,
>, >,
<T::Service as Service>::Future: 'static, <T::Service as Service>::Future: 'static,
Io: AsyncRead + AsyncWrite, Io: AsyncRead + AsyncWrite,
@ -424,8 +394,8 @@ where
self: Pin<&mut Self>, self: Pin<&mut Self>,
cx: &mut Context<'_>, cx: &mut Context<'_>,
) -> Either< ) -> Either<
FramedServiceImplResponseInner<St, Io, Codec, Err, C, T>, FramedServiceImplResponseInner<St, Io, Codec, C, T>,
Poll<Result<(), ServiceError<Err, Codec>>>, Poll<Result<(), ServiceError<C::Error, Codec>>>,
> { > {
#[project] #[project]
match self.project() { match self.project() {

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

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

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

@ -22,7 +22,7 @@ async fn test_disconnect() -> std::io::Result<()> {
let disconnect1 = disconnect1.clone(); let disconnect1 = disconnect1.clone();
Builder::new() Builder::new()
.factory(fn_service(|conn: Connect<_, _, _>| { .factory(fn_service(|conn: Connect<_, _>| {
ok(conn.codec(BytesCodec).state(State)) ok(conn.codec(BytesCodec).state(State))
})) }))
.disconnect(move |_, _| { .disconnect(move |_, _| {
@ -32,7 +32,7 @@ async fn test_disconnect() -> std::io::Result<()> {
}); });
let mut client = Builder::new() let mut client = Builder::new()
.service(|conn: Connect<_, _, _>| { .service(|conn: Connect<_, _>| {
let conn = conn.codec(BytesCodec).state(State); let conn = conn.codec(BytesCodec).state(State);
conn.sink().close(); conn.sink().close();
ok(conn) ok(conn)