From c7a8743bf97dd397e8d1924b84a1f7f422a74432 Mon Sep 17 00:00:00 2001
From: Nikolay Kim <fafhrd91@gmail.com>
Date: Wed, 11 Dec 2019 16:44:09 +0600
Subject: [PATCH] remove E param

---
 actix-ioframe/CHANGES.md           |   4 +
 actix-ioframe/Cargo.toml           |   2 +-
 actix-ioframe/src/connect.rs       |  25 ++--
 actix-ioframe/src/dispatcher.rs    |  24 ++--
 actix-ioframe/src/item.rs          |  24 ++--
 actix-ioframe/src/service.rs       | 190 ++++++++++++-----------------
 actix-ioframe/src/sink.rs          |  19 +--
 actix-ioframe/tests/test_server.rs |   4 +-
 8 files changed, 130 insertions(+), 162 deletions(-)

diff --git a/actix-ioframe/CHANGES.md b/actix-ioframe/CHANGES.md
index 28fb886d..03d710b7 100644
--- a/actix-ioframe/CHANGES.md
+++ b/actix-ioframe/CHANGES.md
@@ -1,5 +1,9 @@
 # Changes
 
+## [0.4.0] - 2019-12-1
+
+* Remove `E` param
+
 ## [0.3.0-alpha.3] - 2019-12-07
 
 * Migrate to tokio 0.2
diff --git a/actix-ioframe/Cargo.toml b/actix-ioframe/Cargo.toml
index 1df73c98..1fb4438d 100644
--- a/actix-ioframe/Cargo.toml
+++ b/actix-ioframe/Cargo.toml
@@ -1,6 +1,6 @@
 [package]
 name = "actix-ioframe"
-version = "0.3.0"
+version = "0.4.0"
 authors = ["Nikolay Kim <fafhrd91@gmail.com>"]
 description = "Actix framed service"
 keywords = ["network", "framework", "async", "futures"]
diff --git a/actix-ioframe/src/connect.rs b/actix-ioframe/src/connect.rs
index f7534413..29fcb1e2 100644
--- a/actix-ioframe/src/connect.rs
+++ b/actix-ioframe/src/connect.rs
@@ -7,21 +7,21 @@ use futures::Stream;
 
 use crate::sink::Sink;
 
-pub struct Connect<Io, Codec, Err, St = ()>
+pub struct Connect<Io, Codec, St = ()>
 where
     Codec: Encoder + Decoder,
 {
     io: Io,
-    sink: Sink<<Codec as Encoder>::Item, Err>,
+    sink: Sink<<Codec as Encoder>::Item>,
     _t: PhantomData<(St, Codec)>,
 }
 
-impl<Io, Codec, Err> Connect<Io, Codec, Err>
+impl<Io, Codec> Connect<Io, Codec>
 where
     Io: AsyncRead + AsyncWrite,
     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 {
             io,
             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 {
             state: (),
             sink: self.sink,
@@ -39,15 +39,15 @@ where
 }
 
 #[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) 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]
-    pub fn sink(&self) -> &Sink<<Codec as Encoder>::Item, Err> {
+    pub fn sink(&self) -> &Sink<<Codec as Encoder>::Item> {
         &self.sink
     }
 
@@ -62,7 +62,7 @@ impl<Io, St, Codec: Encoder + Decoder, Err> ConnectResult<Io, St, Codec, Err> {
     }
 
     #[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 {
             state,
             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
     Io: AsyncRead + AsyncWrite,
     Codec: Encoder + Decoder,
@@ -83,8 +83,7 @@ where
     }
 }
 
-impl<Io, St, Codec, Err> futures::Sink<<Codec as Encoder>::Item>
-    for ConnectResult<Io, St, Codec, Err>
+impl<Io, St, Codec> futures::Sink<<Codec as Encoder>::Item> for ConnectResult<Io, St, Codec>
 where
     Io: AsyncRead + AsyncWrite,
     Codec: Encoder + Decoder,
diff --git a/actix-ioframe/src/dispatcher.rs b/actix-ioframe/src/dispatcher.rs
index 8db9a436..965215f7 100644
--- a/actix-ioframe/src/dispatcher.rs
+++ b/actix-ioframe/src/dispatcher.rs
@@ -13,7 +13,7 @@ use crate::error::ServiceError;
 use crate::item::Item;
 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;
 
 pub(crate) enum Message<T> {
@@ -25,10 +25,10 @@ pub(crate) enum Message<T> {
 /// 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, E>
+pub(crate) struct Dispatcher<St, S, T, U>
 where
     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::Future: 'static,
     T: AsyncRead + AsyncWrite,
@@ -37,19 +37,19 @@ where
     <U as Encoder>::Error: std::fmt::Debug,
 {
     service: S,
-    sink: Sink<<U as Encoder>::Item, E>,
+    sink: Sink<<U as Encoder>::Item>,
     state: St,
     dispatch_state: FramedState<S, U>,
     framed: Framed<T, U>,
-    rx: mpsc::Receiver<Result<Message<<U as Encoder>::Item>, E>>,
-    tx: mpsc::Sender<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>, S::Error>>,
     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
     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::Future: 'static,
     T: AsyncRead + AsyncWrite,
@@ -61,8 +61,8 @@ where
         framed: Framed<T, U>,
         state: St,
         service: F,
-        sink: Sink<<U as Encoder>::Item, E>,
-        rx: mpsc::Receiver<Result<Message<<U as Encoder>::Item>, E>>,
+        sink: Sink<<U as Encoder>::Item>,
+        rx: mpsc::Receiver<Result<Message<<U as Encoder>::Item>, S::Error>>,
         disconnect: Option<Rc<dyn Fn(&mut St, bool)>>,
     ) -> Self {
         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
     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::Future: 'static,
     T: AsyncRead + AsyncWrite,
diff --git a/actix-ioframe/src/item.rs b/actix-ioframe/src/item.rs
index 03db8131..42008d9e 100644
--- a/actix-ioframe/src/item.rs
+++ b/actix-ioframe/src/item.rs
@@ -5,19 +5,19 @@ use actix_codec::{Decoder, Encoder};
 
 use crate::sink::Sink;
 
-pub struct Item<St, Codec: Encoder + Decoder, E> {
+pub struct Item<St, Codec: Encoder + Decoder> {
     state: St,
-    sink: Sink<<Codec as Encoder>::Item, E>,
+    sink: Sink<<Codec as Encoder>::Item>,
     item: <Codec as Decoder>::Item,
 }
 
-impl<St, Codec, E> Item<St, Codec, E>
+impl<St, Codec> Item<St, Codec>
 where
     Codec: Encoder + Decoder,
 {
     pub(crate) fn new(
         state: St,
-        sink: Sink<<Codec as Encoder>::Item, E>,
+        sink: Sink<<Codec as Encoder>::Item>,
         item: <Codec as Decoder>::Item,
     ) -> Self {
         Item { state, sink, item }
@@ -34,7 +34,7 @@ where
     }
 
     #[inline]
-    pub fn sink(&self) -> &Sink<<Codec as Encoder>::Item, E> {
+    pub fn sink(&self) -> &Sink<<Codec as Encoder>::Item> {
         &self.sink
     }
 
@@ -44,18 +44,12 @@ where
     }
 
     #[inline]
-    pub fn into_parts(
-        self,
-    ) -> (
-        St,
-        Sink<<Codec as Encoder>::Item, E>,
-        <Codec as Decoder>::Item,
-    ) {
+    pub fn into_parts(self) -> (St, Sink<<Codec as Encoder>::Item>, <Codec as Decoder>::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
     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
     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
     Codec: Encoder + Decoder,
     <Codec as Decoder>::Item: fmt::Debug,
diff --git a/actix-ioframe/src/service.rs b/actix-ioframe/src/service.rs
index 82576f04..ee781970 100644
--- a/actix-ioframe/src/service.rs
+++ b/actix-ioframe/src/service.rs
@@ -17,7 +17,7 @@ use crate::error::ServiceError;
 use crate::item::Item;
 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 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
-    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
         F: IntoService<C>,
         Io: AsyncRead + AsyncWrite,
-        C: Service<
-            Request = Connect<Io, Codec, E>,
-            Response = ConnectResult<Io, St, Codec, E>,
-            Error = E,
-        >,
+        C: Service<Request = Connect<Io, Codec>, Response = ConnectResult<Io, St, Codec>>,
         Codec: Decoder + Encoder,
     {
         ServiceBuilder {
@@ -56,17 +52,16 @@ impl<St: Clone, Codec> Builder<St, Codec> {
     }
 
     /// 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
         F: IntoServiceFactory<C>,
-        E: 'static,
         Io: AsyncRead + AsyncWrite,
         C: ServiceFactory<
             Config = (),
-            Request = Connect<Io, Codec, E>,
-            Response = ConnectResult<Io, St, Codec, E>,
-            Error = E,
+            Request = Connect<Io, Codec>,
+            Response = ConnectResult<Io, St, Codec>,
         >,
+        C::Error: 'static,
         C::Future: 'static,
         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,
     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
     St: Clone,
-    C: Service<
-        Request = Connect<Io, Codec, Err>,
-        Response = ConnectResult<Io, St, Codec, Err>,
-        Error = Err,
-    >,
+    C: Service<Request = Connect<Io, Codec>, Response = ConnectResult<Io, St, Codec>>,
     Io: AsyncRead + AsyncWrite,
     Codec: Decoder + Encoder,
     <Codec as Encoder>::Item: 'static,
@@ -109,15 +100,15 @@ where
     }
 
     /// 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
         F: IntoServiceFactory<T>,
         T: ServiceFactory<
             Config = St,
-            Request = RequestItem<St, Codec, Err>,
+            Request = RequestItem<St, Codec>,
             Response = ResponseItem<Codec>,
-            Error = Err,
-            InitError = Err,
+            Error = C::Error,
+            InitError = C::Error,
         >,
     {
         FramedServiceImpl {
@@ -129,23 +120,22 @@ where
     }
 }
 
-pub struct NewServiceBuilder<St, C, Io, Codec, Err> {
+pub struct NewServiceBuilder<St, C, Io, Codec> {
     connect: C,
     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
     St: Clone,
     Io: AsyncRead + AsyncWrite,
-    Err: 'static,
     C: ServiceFactory<
         Config = (),
-        Request = Connect<Io, Codec, Err>,
-        Response = ConnectResult<Io, St, Codec, Err>,
-        Error = Err,
+        Request = Connect<Io, Codec>,
+        Response = ConnectResult<Io, St, Codec>,
     >,
+    C::Error: 'static,
     C::Future: 'static,
     Codec: Decoder + Encoder,
     <Codec as Encoder>::Item: 'static,
@@ -162,15 +152,15 @@ where
         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
         F: IntoServiceFactory<T>,
         T: ServiceFactory<
                 Config = St,
-                Request = RequestItem<St, Codec, Err>,
+                Request = RequestItem<St, Codec>,
                 Response = ResponseItem<Codec>,
-                Error = Err,
-                InitError = Err,
+                Error = C::Error,
+                InitError = C::Error,
             > + 'static,
     {
         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,
     handler: Rc<T>,
     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
-    for FramedService<St, C, T, Io, Codec, Err, Cfg>
+impl<St, C, T, Io, Codec, Cfg> ServiceFactory for FramedService<St, C, T, Io, Codec, Cfg>
 where
     St: Clone + 'static,
     Io: AsyncRead + AsyncWrite,
     C: ServiceFactory<
         Config = (),
-        Request = Connect<Io, Codec, Err>,
-        Response = ConnectResult<Io, St, Codec, Err>,
-        Error = Err,
+        Request = Connect<Io, Codec>,
+        Response = ConnectResult<Io, St, Codec>,
     >,
+    C::Error: 'static,
     C::Future: 'static,
     T: ServiceFactory<
             Config = St,
-            Request = RequestItem<St, Codec, Err>,
+            Request = RequestItem<St, Codec>,
             Response = ResponseItem<Codec>,
-            Error = Err,
-            InitError = Err,
+            Error = C::Error,
+            InitError = C::Error,
         > + 'static,
     <T::Service as Service>::Future: 'static,
-    Err: 'static,
     Codec: Decoder + Encoder,
     <Codec as Encoder>::Item: 'static,
     <Codec as Encoder>::Error: std::fmt::Debug,
@@ -219,7 +207,7 @@ where
     type Response = ();
     type Error = ServiceError<C::Error, Codec>;
     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>>;
 
     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,
     handler: Rc<T>,
     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
     St: Clone,
     Io: AsyncRead + AsyncWrite,
-    C: Service<
-        Request = Connect<Io, Codec, Err>,
-        Response = ConnectResult<Io, St, Codec, Err>,
-        Error = Err,
-    >,
-    Err: 'static,
+    C: Service<Request = Connect<Io, Codec>, Response = ConnectResult<Io, St, Codec>>,
+    C::Error: 'static,
     T: ServiceFactory<
         Config = St,
-        Request = RequestItem<St, Codec, Err>,
+        Request = RequestItem<St, Codec>,
         Response = ResponseItem<Codec>,
-        Error = Err,
-        InitError = Err,
+        Error = C::Error,
+        InitError = C::Error,
     >,
     <T::Service as Service>::Future: 'static,
     Codec: Decoder + Encoder,
@@ -272,8 +256,8 @@ where
 {
     type Request = Io;
     type Response = ();
-    type Error = ServiceError<Err, Codec>;
-    type Future = FramedServiceImplResponse<St, Io, Codec, Err, C, T>;
+    type Error = ServiceError<C::Error, Codec>;
+    type Future = FramedServiceImplResponse<St, Io, Codec, C, T>;
 
     fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
         self.connect.poll_ready(cx).map_err(|e| e.into())
@@ -281,7 +265,9 @@ where
 
     fn call(&mut self, req: Io) -> Self::Future {
         let (tx, rx) = mpsc::channel();
-        let sink = Sink::new(tx);
+        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())),
@@ -294,21 +280,17 @@ where
 }
 
 #[pin_project::pin_project]
-pub struct FramedServiceImplResponse<St, Io, Codec, Err, C, T>
+pub struct FramedServiceImplResponse<St, Io, Codec, C, T>
 where
     St: Clone,
-    C: Service<
-        Request = Connect<Io, Codec, Err>,
-        Response = ConnectResult<Io, St, Codec, Err>,
-        Error = Err,
-    >,
-    Err: 'static,
+    C: Service<Request = Connect<Io, Codec>, Response = ConnectResult<Io, St, Codec>>,
+    C::Error: 'static,
     T: ServiceFactory<
         Config = St,
-        Request = RequestItem<St, Codec, Err>,
+        Request = RequestItem<St, Codec>,
         Response = ResponseItem<Codec>,
-        Error = Err,
-        InitError = Err,
+        Error = C::Error,
+        InitError = C::Error,
     >,
     <T::Service as Service>::Future: 'static,
     Io: AsyncRead + AsyncWrite,
@@ -317,24 +299,20 @@ where
     <Codec as Encoder>::Error: std::fmt::Debug,
 {
     #[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
     St: Clone,
-    C: Service<
-        Request = Connect<Io, Codec, Err>,
-        Response = ConnectResult<Io, St, Codec, Err>,
-        Error = Err,
-    >,
-    Err: 'static,
+    C: Service<Request = Connect<Io, Codec>, Response = ConnectResult<Io, St, Codec>>,
+    C::Error: 'static,
     T: ServiceFactory<
         Config = St,
-        Request = RequestItem<St, Codec, Err>,
+        Request = RequestItem<St, Codec>,
         Response = ResponseItem<Codec>,
-        Error = Err,
-        InitError = Err,
+        Error = C::Error,
+        InitError = C::Error,
     >,
     <T::Service as Service>::Future: 'static,
     Io: AsyncRead + AsyncWrite,
@@ -342,7 +320,7 @@ where
     <Codec as Encoder>::Item: 'static,
     <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> {
         let mut this = self.as_mut().project();
@@ -360,21 +338,17 @@ where
 }
 
 #[pin_project::pin_project]
-enum FramedServiceImplResponseInner<St, Io, Codec, Err, C, T>
+enum FramedServiceImplResponseInner<St, Io, Codec, C, T>
 where
     St: Clone,
-    C: Service<
-        Request = Connect<Io, Codec, Err>,
-        Response = ConnectResult<Io, St, Codec, Err>,
-        Error = Err,
-    >,
-    Err: 'static,
+    C: Service<Request = Connect<Io, Codec>, Response = ConnectResult<Io, St, Codec>>,
+    C::Error: 'static,
     T: ServiceFactory<
         Config = St,
-        Request = RequestItem<St, Codec, Err>,
+        Request = RequestItem<St, Codec>,
         Response = ResponseItem<Codec>,
-        Error = Err,
-        InitError = Err,
+        Error = C::Error,
+        InitError = C::Error,
     >,
     <T::Service as Service>::Future: 'static,
     Io: AsyncRead + AsyncWrite,
@@ -386,32 +360,28 @@ where
         #[pin] C::Future,
         Rc<T>,
         Option<Rc<dyn Fn(&mut St, bool)>>,
-        Option<mpsc::Receiver<ServiceResult<Codec, Err>>>,
+        Option<mpsc::Receiver<ServiceResult<Codec, C::Error>>>,
     ),
     Handler(
         #[pin] T::Future,
-        Option<ConnectResult<Io, St, Codec, Err>>,
+        Option<ConnectResult<Io, St, Codec>>,
         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
     St: Clone,
-    C: Service<
-        Request = Connect<Io, Codec, Err>,
-        Response = ConnectResult<Io, St, Codec, Err>,
-        Error = Err,
-    >,
-    Err: 'static,
+    C: Service<Request = Connect<Io, Codec>, Response = ConnectResult<Io, St, Codec>>,
+    C::Error: 'static,
     T: ServiceFactory<
         Config = St,
-        Request = RequestItem<St, Codec, Err>,
+        Request = RequestItem<St, Codec>,
         Response = ResponseItem<Codec>,
-        Error = Err,
-        InitError = Err,
+        Error = C::Error,
+        InitError = C::Error,
     >,
     <T::Service as Service>::Future: 'static,
     Io: AsyncRead + AsyncWrite,
@@ -424,8 +394,8 @@ where
         self: Pin<&mut Self>,
         cx: &mut Context<'_>,
     ) -> Either<
-        FramedServiceImplResponseInner<St, Io, Codec, Err, C, T>,
-        Poll<Result<(), ServiceError<Err, Codec>>>,
+        FramedServiceImplResponseInner<St, Io, Codec, C, T>,
+        Poll<Result<(), ServiceError<C::Error, Codec>>>,
     > {
         #[project]
         match self.project() {
diff --git a/actix-ioframe/src/sink.rs b/actix-ioframe/src/sink.rs
index 4e0fe025..eb4ff703 100644
--- a/actix-ioframe/src/sink.rs
+++ b/actix-ioframe/src/sink.rs
@@ -1,43 +1,44 @@
 use std::fmt;
+use std::rc::Rc;
 
-use actix_utils::{mpsc, oneshot};
+use actix_utils::oneshot;
 use futures::future::{Future, FutureExt};
 
 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 {
         Sink(self.0.clone())
     }
 }
 
-impl<T, E> Sink<T, E> {
-    pub(crate) fn new(tx: mpsc::Sender<Result<Message<T>, E>>) -> Self {
+impl<T> Sink<T> {
+    pub(crate) fn new(tx: Rc<dyn Fn(Message<T>)>) -> Self {
         Sink(tx)
     }
 
     /// Close connection
     pub fn close(&self) {
-        let _ = self.0.send(Ok(Message::Close));
+        (self.0)(Message::Close);
     }
 
     /// Close connection
     pub fn wait_close(&self) -> impl Future<Output = ()> {
         let (tx, rx) = oneshot::channel();
-        let _ = self.0.send(Ok(Message::WaitClose(tx)));
+        (self.0)(Message::WaitClose(tx));
 
         rx.map(|_| ())
     }
 
     /// Send item
     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 {
         fmt.debug_struct("Sink").finish()
     }
diff --git a/actix-ioframe/tests/test_server.rs b/actix-ioframe/tests/test_server.rs
index 35bdb983..0a6746f8 100644
--- a/actix-ioframe/tests/test_server.rs
+++ b/actix-ioframe/tests/test_server.rs
@@ -22,7 +22,7 @@ async fn test_disconnect() -> std::io::Result<()> {
         let disconnect1 = disconnect1.clone();
 
         Builder::new()
-            .factory(fn_service(|conn: Connect<_, _, _>| {
+            .factory(fn_service(|conn: Connect<_, _>| {
                 ok(conn.codec(BytesCodec).state(State))
             }))
             .disconnect(move |_, _| {
@@ -32,7 +32,7 @@ async fn test_disconnect() -> std::io::Result<()> {
     });
 
     let mut client = Builder::new()
-        .service(|conn: Connect<_, _, _>| {
+        .service(|conn: Connect<_, _>| {
             let conn = conn.codec(BytesCodec).state(State);
             conn.sink().close();
             ok(conn)