inline inner websocket dispatcher

actix-http/Cargo.toml

@@ -65,11 +65,13 @@ http = "0.2.2"
 httparse = "1.3"
 itoa = "0.4"
 language-tags = "0.2"
+local-channel = "0.1"
 once_cell = "1.5"
 log = "0.4"
 mime = "0.3"
 percent-encoding = "2.1"
 pin-project = "1.0.0"
+pin-project-lite = "0.2"
 rand = "0.8"
 regex = "1.3"
 serde = "1.0"

actix-http/src/ws/dispatcher.rs

@@ -4,7 +4,6 @@ use std::task::{Context, Poll};
 
 use actix_codec::{AsyncRead, AsyncWrite, Framed};
 use actix_service::{IntoService, Service};
-use actix_utils::dispatcher::{Dispatcher as InnerDispatcher, DispatcherError};
 
 use super::{Codec, Frame, Message};
 
@@ -15,7 +14,7 @@ where
     T: AsyncRead + AsyncWrite,
 {
     #[pin]
-    inner: InnerDispatcher<S, T, Codec, Message>,
+    inner: inner::Dispatcher<S, T, Codec, Message>,
 }
 
 impl<S, T> Dispatcher<S, T>
@@ -27,13 +26,13 @@ where
 {
     pub fn new<F: IntoService<S, Frame>>(io: T, service: F) -> Self {
         Dispatcher {
-            inner: InnerDispatcher::new(Framed::new(io, Codec::new()), service),
+            inner: inner::Dispatcher::new(Framed::new(io, Codec::new()), service),
         }
     }
 
     pub fn with<F: IntoService<S, Frame>>(framed: Framed<T, Codec>, service: F) -> Self {
         Dispatcher {
-            inner: InnerDispatcher::new(framed, service),
+            inner: inner::Dispatcher::new(framed, service),
         }
     }
 }
@@ -45,9 +44,374 @@ where
     S::Future: 'static,
     S::Error: 'static,
 {
-    type Output = Result<(), DispatcherError<S::Error, Codec, Message>>;
+    type Output = Result<(), inner::DispatcherError<S::Error, Codec, Message>>;
 
     fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
         self.project().inner.poll(cx)
     }
 }
+
+/// Framed dispatcher service and related utilities.
+mod inner {
+    // allow dead code since this mod was ripped from actix-utils
+    #![allow(dead_code)]
+
+    use core::{
+        fmt,
+        future::Future,
+        mem,
+        pin::Pin,
+        task::{Context, Poll},
+    };
+
+    use actix_service::{IntoService, Service};
+    use futures_core::stream::Stream;
+    use local_channel::mpsc;
+    use log::debug;
+    use pin_project_lite::pin_project;
+
+    use actix_codec::{AsyncRead, AsyncWrite, Decoder, Encoder, Framed};
+
+    /// Framed transport errors
+    pub enum DispatcherError<E, U: Encoder<I> + Decoder, I> {
+        /// Inner service error.
+        Service(E),
+
+        /// Frame encoding error.
+        Encoder(<U as Encoder<I>>::Error),
+
+        /// Frame decoding error.
+        Decoder(<U as Decoder>::Error),
+    }
+
+    impl<E, U: Encoder<I> + Decoder, I> From<E> for DispatcherError<E, U, I> {
+        fn from(err: E) -> Self {
+            DispatcherError::Service(err)
+        }
+    }
+
+    impl<E, U: Encoder<I> + Decoder, I> fmt::Debug for DispatcherError<E, U, I>
+    where
+        E: fmt::Debug,
+        <U as Encoder<I>>::Error: fmt::Debug,
+        <U as Decoder>::Error: fmt::Debug,
+    {
+        fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result {
+            match *self {
+                DispatcherError::Service(ref e) => {
+                    write!(fmt, "DispatcherError::Service({:?})", e)
+                }
+                DispatcherError::Encoder(ref e) => {
+                    write!(fmt, "DispatcherError::Encoder({:?})", e)
+                }
+                DispatcherError::Decoder(ref e) => {
+                    write!(fmt, "DispatcherError::Decoder({:?})", e)
+                }
+            }
+        }
+    }
+
+    impl<E, U: Encoder<I> + Decoder, I> fmt::Display for DispatcherError<E, U, I>
+    where
+        E: fmt::Display,
+        <U as Encoder<I>>::Error: fmt::Debug,
+        <U as Decoder>::Error: fmt::Debug,
+    {
+        fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result {
+            match *self {
+                DispatcherError::Service(ref e) => write!(fmt, "{}", e),
+                DispatcherError::Encoder(ref e) => write!(fmt, "{:?}", e),
+                DispatcherError::Decoder(ref e) => write!(fmt, "{:?}", e),
+            }
+        }
+    }
+
+    /// Message type wrapper for signalling end of message stream.
+    pub enum Message<T> {
+        /// Message item.
+        Item(T),
+
+        /// Signal from service to flush all messages and stop processing.
+        Close,
+    }
+
+    pin_project! {
+        /// A future that reads frames from a [`Framed`] object and passes them to a [`Service`].
+        pub struct Dispatcher<S, T, U, I>
+        where
+            S: Service<<U as Decoder>::Item, Response = I>,
+            S::Error: 'static,
+            S::Future: 'static,
+            T: AsyncRead,
+            T: AsyncWrite,
+            U: Encoder<I>,
+            U: Decoder,
+            I: 'static,
+            <U as Encoder<I>>::Error: fmt::Debug,
+        {
+            service: S,
+            state: State<S, U, I>,
+            #[pin]
+            framed: Framed<T, U>,
+            rx: mpsc::Receiver<Result<Message<I>, S::Error>>,
+            tx: mpsc::Sender<Result<Message<I>, S::Error>>,
+        }
+    }
+
+    enum State<S, U, I>
+    where
+        S: Service<<U as Decoder>::Item>,
+        U: Encoder<I> + Decoder,
+    {
+        Processing,
+        Error(DispatcherError<S::Error, U, I>),
+        FramedError(DispatcherError<S::Error, U, I>),
+        FlushAndStop,
+        Stopping,
+    }
+
+    impl<S, U, I> State<S, U, I>
+    where
+        S: Service<<U as Decoder>::Item>,
+        U: Encoder<I> + Decoder,
+    {
+        fn take_error(&mut self) -> DispatcherError<S::Error, U, I> {
+            match mem::replace(self, State::Processing) {
+                State::Error(err) => err,
+                _ => panic!(),
+            }
+        }
+
+        fn take_framed_error(&mut self) -> DispatcherError<S::Error, U, I> {
+            match mem::replace(self, State::Processing) {
+                State::FramedError(err) => err,
+                _ => panic!(),
+            }
+        }
+    }
+
+    impl<S, T, U, I> Dispatcher<S, T, U, I>
+    where
+        S: Service<<U as Decoder>::Item, Response = I>,
+        S::Error: 'static,
+        S::Future: 'static,
+        T: AsyncRead + AsyncWrite,
+        U: Decoder + Encoder<I>,
+        I: 'static,
+        <U as Decoder>::Error: fmt::Debug,
+        <U as Encoder<I>>::Error: fmt::Debug,
+    {
+        /// Create new `Dispatcher`.
+        pub fn new<F>(framed: Framed<T, U>, service: F) -> Self
+        where
+            F: IntoService<S, <U as Decoder>::Item>,
+        {
+            let (tx, rx) = mpsc::channel();
+            Dispatcher {
+                framed,
+                rx,
+                tx,
+                service: service.into_service(),
+                state: State::Processing,
+            }
+        }
+
+        /// Construct new `Dispatcher` instance with customer `mpsc::Receiver`
+        pub fn with_rx<F>(
+            framed: Framed<T, U>,
+            service: F,
+            rx: mpsc::Receiver<Result<Message<I>, S::Error>>,
+        ) -> Self
+        where
+            F: IntoService<S, <U as Decoder>::Item>,
+        {
+            let tx = rx.sender();
+            Dispatcher {
+                framed,
+                rx,
+                tx,
+                service: service.into_service(),
+                state: State::Processing,
+            }
+        }
+
+        /// Get sender handle.
+        pub fn tx(&self) -> mpsc::Sender<Result<Message<I>, S::Error>> {
+            self.tx.clone()
+        }
+
+        /// Get reference to a service wrapped by `Dispatcher` instance.
+        pub fn service(&self) -> &S {
+            &self.service
+        }
+
+        /// Get mutable reference to a service wrapped by `Dispatcher` instance.
+        pub fn service_mut(&mut self) -> &mut S {
+            &mut self.service
+        }
+
+        /// Get reference to a framed instance wrapped by `Dispatcher` instance.
+        pub fn framed(&self) -> &Framed<T, U> {
+            &self.framed
+        }
+
+        /// Get mutable reference to a framed instance wrapped by `Dispatcher` instance.
+        pub fn framed_mut(&mut self) -> &mut Framed<T, U> {
+            &mut self.framed
+        }
+
+        /// Read from framed object.
+        fn poll_read(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> bool
+        where
+            S: Service<<U as Decoder>::Item, Response = I>,
+            S::Error: 'static,
+            S::Future: 'static,
+            T: AsyncRead + AsyncWrite,
+            U: Decoder + Encoder<I>,
+            I: 'static,
+            <U as Encoder<I>>::Error: fmt::Debug,
+        {
+            loop {
+                let this = self.as_mut().project();
+                match this.service.poll_ready(cx) {
+                    Poll::Ready(Ok(_)) => {
+                        let item = match this.framed.next_item(cx) {
+                            Poll::Ready(Some(Ok(el))) => el,
+                            Poll::Ready(Some(Err(err))) => {
+                                *this.state =
+                                    State::FramedError(DispatcherError::Decoder(err));
+                                return true;
+                            }
+                            Poll::Pending => return false,
+                            Poll::Ready(None) => {
+                                *this.state = State::Stopping;
+                                return true;
+                            }
+                        };
+
+                        let tx = this.tx.clone();
+                        let fut = this.service.call(item);
+                        actix_rt::spawn(async move {
+                            let item = fut.await;
+                            let _ = tx.send(item.map(Message::Item));
+                        });
+                    }
+                    Poll::Pending => return false,
+                    Poll::Ready(Err(err)) => {
+                        *this.state = State::Error(DispatcherError::Service(err));
+                        return true;
+                    }
+                }
+            }
+        }
+
+        /// Write to framed object.
+        fn poll_write(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> bool
+        where
+            S: Service<<U as Decoder>::Item, Response = I>,
+            S::Error: 'static,
+            S::Future: 'static,
+            T: AsyncRead + AsyncWrite,
+            U: Decoder + Encoder<I>,
+            I: 'static,
+            <U as Encoder<I>>::Error: fmt::Debug,
+        {
+            loop {
+                let mut this = self.as_mut().project();
+                while !this.framed.is_write_buf_full() {
+                    match Pin::new(&mut this.rx).poll_next(cx) {
+                        Poll::Ready(Some(Ok(Message::Item(msg)))) => {
+                            if let Err(err) = this.framed.as_mut().write(msg) {
+                                *this.state =
+                                    State::FramedError(DispatcherError::Encoder(err));
+                                return true;
+                            }
+                        }
+                        Poll::Ready(Some(Ok(Message::Close))) => {
+                            *this.state = State::FlushAndStop;
+                            return true;
+                        }
+                        Poll::Ready(Some(Err(err))) => {
+                            *this.state = State::Error(DispatcherError::Service(err));
+                            return true;
+                        }
+                        Poll::Ready(None) | Poll::Pending => break,
+                    }
+                }
+
+                if !this.framed.is_write_buf_empty() {
+                    match this.framed.flush(cx) {
+                        Poll::Pending => break,
+                        Poll::Ready(Ok(_)) => {}
+                        Poll::Ready(Err(err)) => {
+                            debug!("Error sending data: {:?}", err);
+                            *this.state =
+                                State::FramedError(DispatcherError::Encoder(err));
+                            return true;
+                        }
+                    }
+                } else {
+                    break;
+                }
+            }
+
+            false
+        }
+    }
+
+    impl<S, T, U, I> Future for Dispatcher<S, T, U, I>
+    where
+        S: Service<<U as Decoder>::Item, Response = I>,
+        S::Error: 'static,
+        S::Future: 'static,
+        T: AsyncRead + AsyncWrite,
+        U: Decoder + Encoder<I>,
+        I: 'static,
+        <U as Encoder<I>>::Error: fmt::Debug,
+        <U as Decoder>::Error: fmt::Debug,
+    {
+        type Output = Result<(), DispatcherError<S::Error, U, I>>;
+
+        fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+            loop {
+                let this = self.as_mut().project();
+
+                return match this.state {
+                    State::Processing => {
+                        if self.as_mut().poll_read(cx) || self.as_mut().poll_write(cx) {
+                            continue;
+                        } else {
+                            Poll::Pending
+                        }
+                    }
+                    State::Error(_) => {
+                        // flush write buffer
+                        if !this.framed.is_write_buf_empty()
+                            && this.framed.flush(cx).is_pending()
+                        {
+                            return Poll::Pending;
+                        }
+                        Poll::Ready(Err(this.state.take_error()))
+                    }
+                    State::FlushAndStop => {
+                        if !this.framed.is_write_buf_empty() {
+                            this.framed.flush(cx).map(|res| {
+                                if let Err(err) = res {
+                                    debug!("Error sending data: {:?}", err);
+                                }
+
+                                Ok(())
+                            })
+                        } else {
+                            Poll::Ready(Ok(()))
+                        }
+                    }
+                    State::FramedError(_) => {
+                        Poll::Ready(Err(this.state.take_framed_error()))
+                    }
+                    State::Stopping => Poll::Ready(Ok(())),
+                };
+            }
+        }
+    }
+}

docs/graphs/net-only.dot

@@ -1,19 +1,42 @@
 digraph {
+    rankdir=TB
+
     subgraph cluster_net {
-        label="actix/actix-net";
+        label="actix-net"
+
         "actix-codec" "actix-macros" "actix-rt" "actix-server" "actix-service"
         "actix-tls" "actix-tracing" "actix-utils" "actix-router"
-        "local-channel" "local-waker"
     }
     
-    "actix-codec" -> { "actix-rt" "actix-service" "local-channel" "tokio" }
+    subgraph cluster_other {
+        label="other actix owned crates"
+
+        {
+            rank=same
+            "local-channel" "local-waker" "bytestring"
+        }
+    }
+
+    subgraph cluster_tokio {
+        label="tokio"
+
+        "tokio" [fontcolor = darkgreen]
+        "tokio-util" [fontcolor = darkgreen]
+    }
+
+    "actix-codec" -> { "tokio" "tokio-util" }
     "actix-utils" -> { "local-waker" }
     "actix-tracing" -> { "actix-service" }
     "actix-tls" -> { "actix-service" "actix-codec" "actix-utils" "actix-rt" }
-    "actix-server" -> { "actix-service" "actix-rt" "actix-codec" "actix-utils" "tokio" }
+    "actix-server" -> { "actix-service" "actix-rt" "actix-utils" "tokio" }
     "actix-rt" -> { "actix-macros" "tokio" }
+    "actix-router" -> { "bytestring" }
 
     "local-channel" -> { "local-waker" }
 
-    "tokio" [fontcolor = darkgreen]
+    // invisible edges to force nicer layout
+    edge [style=invis]
+    "actix-macros" -> "tokio"
+    "actix-service" -> "bytestring"
+    "actix-macros" -> "bytestring"
 }