move codec to separate crate

actix-codec/CHANGES.md

@@ -0,0 +1,5 @@
+# Changes
+
+## [0.1.0] - 2018-12-09
+
+* Move codec to separate crate

actix-codec/Cargo.toml

@@ -0,0 +1,25 @@
+[package]
+name = "actix-codec"
+version = "0.1.0"
+authors = ["Nikolay Kim <fafhrd91@gmail.com>"]
+description = "Utilities for encoding and decoding frames"
+keywords = ["network", "framework", "async", "futures"]
+homepage = "https://actix.rs"
+repository = "https://github.com/actix/actix-net.git"
+documentation = "https://docs.rs/actix-codec/"
+categories = ["network-programming", "asynchronous"]
+license = "MIT/Apache-2.0"
+exclude = [".gitignore", ".travis.yml", ".cargo/config", "appveyor.yml"]
+edition = "2018"
+workspace = "../"
+
+[lib]
+name = "actix_codec"
+path = "src/lib.rs"
+
+[dependencies]
+bytes = "0.4"
+futures = "0.1.24"
+tokio-io = "0.1"
+tokio-codec = "0.1"
+log = "0.4"

actix-codec/src/bcodec.rs

@@ -0,0 +1,33 @@
+use std::io;
+
+use bytes::{Bytes, BytesMut};
+use tokio_codec::{Decoder, Encoder};
+
+/// Bytes codec.
+///
+/// Reads/Writes chunks of bytes from a stream.
+#[derive(Debug, Copy, Clone)]
+pub struct BytesCodec;
+
+impl Encoder for BytesCodec {
+    type Item = Bytes;
+    type Error = io::Error;
+
+    fn encode(&mut self, item: Bytes, dst: &mut BytesMut) -> Result<(), Self::Error> {
+        dst.extend_from_slice(&item[..]);
+        Ok(())
+    }
+}
+
+impl Decoder for BytesCodec {
+    type Item = BytesMut;
+    type Error = io::Error;
+
+    fn decode(&mut self, src: &mut BytesMut) -> Result<Option<Self::Item>, Self::Error> {
+        if src.is_empty() {
+            Ok(None)
+        } else {
+            Ok(Some(src.take()))
+        }
+    }
+}

actix-codec/src/framed.rs

@@ -0,0 +1,352 @@
+#![allow(deprecated)]
+
+use std::fmt;
+use std::io::{self, Read, Write};
+
+use bytes::BytesMut;
+use futures::{Poll, Sink, StartSend, Stream};
+use tokio_codec::{Decoder, Encoder};
+use tokio_io::{AsyncRead, AsyncWrite};
+
+use super::framed_read::{framed_read2, framed_read2_with_buffer, FramedRead2};
+use super::framed_write::{framed_write2, framed_write2_with_buffer, FramedWrite2};
+
+const LW: usize = 1024;
+const HW: usize = 8 * 1024;
+
+/// A unified `Stream` and `Sink` interface to an underlying I/O object, using
+/// the `Encoder` and `Decoder` traits to encode and decode frames.
+///
+/// You can create a `Framed` instance by using the `AsyncRead::framed` adapter.
+pub struct Framed<T, U> {
+    inner: FramedRead2<FramedWrite2<Fuse<T, U>>>,
+}
+
+pub struct Fuse<T, U>(pub T, pub U);
+
+impl<T, U> Framed<T, U>
+where
+    T: AsyncRead + AsyncWrite,
+    U: Decoder + Encoder,
+{
+    /// Provides a `Stream` and `Sink` interface for reading and writing to this
+    /// `Io` object, using `Decode` and `Encode` to read and write the raw data.
+    ///
+    /// Raw I/O objects work with byte sequences, but higher-level code usually
+    /// wants to batch these into meaningful chunks, called "frames". This
+    /// method layers framing on top of an I/O object, by using the `Codec`
+    /// traits to handle encoding and decoding of messages frames. Note that
+    /// the incoming and outgoing frame types may be distinct.
+    ///
+    /// This function returns a *single* object that is both `Stream` and
+    /// `Sink`; grouping this into a single object is often useful for layering
+    /// things like gzip or TLS, which require both read and write access to the
+    /// underlying object.
+    ///
+    /// If you want to work more directly with the streams and sink, consider
+    /// calling `split` on the `Framed` returned by this method, which will
+    /// break them into separate objects, allowing them to interact more easily.
+    pub fn new(inner: T, codec: U) -> Framed<T, U> {
+        Framed {
+            inner: framed_read2(framed_write2(Fuse(inner, codec), LW, HW)),
+        }
+    }
+
+    /// Same as `Framed::new()` with ability to specify write buffer low/high capacity watermarks.
+    pub fn new_with_caps(inner: T, codec: U, lw: usize, hw: usize) -> Framed<T, U> {
+        debug_assert!((lw < hw) && hw != 0);
+        Framed {
+            inner: framed_read2(framed_write2(Fuse(inner, codec), lw, hw)),
+        }
+    }
+
+    /// Force send item
+    pub fn force_send(
+        &mut self,
+        item: <U as Encoder>::Item,
+    ) -> Result<(), <U as Encoder>::Error> {
+        self.inner.get_mut().force_send(item)
+    }
+}
+
+impl<T, U> Framed<T, U> {
+    /// Provides a `Stream` and `Sink` interface for reading and writing to this
+    /// `Io` object, using `Decode` and `Encode` to read and write the raw data.
+    ///
+    /// Raw I/O objects work with byte sequences, but higher-level code usually
+    /// wants to batch these into meaningful chunks, called "frames". This
+    /// method layers framing on top of an I/O object, by using the `Codec`
+    /// traits to handle encoding and decoding of messages frames. Note that
+    /// the incoming and outgoing frame types may be distinct.
+    ///
+    /// This function returns a *single* object that is both `Stream` and
+    /// `Sink`; grouping this into a single object is often useful for layering
+    /// things like gzip or TLS, which require both read and write access to the
+    /// underlying object.
+    ///
+    /// This objects takes a stream and a readbuffer and a writebuffer. These
+    /// field can be obtained from an existing `Framed` with the
+    /// `into_parts` method.
+    ///
+    /// If you want to work more directly with the streams and sink, consider
+    /// calling `split` on the `Framed` returned by this method, which will
+    /// break them into separate objects, allowing them to interact more easily.
+    pub fn from_parts(parts: FramedParts<T, U>) -> Framed<T, U> {
+        Framed {
+            inner: framed_read2_with_buffer(
+                framed_write2_with_buffer(
+                    Fuse(parts.io, parts.codec),
+                    parts.write_buf,
+                    parts.write_buf_lw,
+                    parts.write_buf_hw,
+                ),
+                parts.read_buf,
+            ),
+        }
+    }
+
+    /// Returns a reference to the underlying codec.
+    pub fn get_codec(&self) -> &U {
+        &self.inner.get_ref().get_ref().1
+    }
+
+    /// Returns a mutable reference to the underlying codec.
+    pub fn get_codec_mut(&mut self) -> &mut U {
+        &mut self.inner.get_mut().get_mut().1
+    }
+
+    /// Returns a reference to the underlying I/O stream wrapped by
+    /// `Frame`.
+    ///
+    /// Note that care should be taken to not tamper with the underlying stream
+    /// of data coming in as it may corrupt the stream of frames otherwise
+    /// being worked with.
+    pub fn get_ref(&self) -> &T {
+        &self.inner.get_ref().get_ref().0
+    }
+
+    /// Returns a mutable reference to the underlying I/O stream wrapped by
+    /// `Frame`.
+    ///
+    /// Note that care should be taken to not tamper with the underlying stream
+    /// of data coming in as it may corrupt the stream of frames otherwise
+    /// being worked with.
+    pub fn get_mut(&mut self) -> &mut T {
+        &mut self.inner.get_mut().get_mut().0
+    }
+
+    /// Check if write buffer is empty.
+    pub fn is_write_buf_empty(&self) -> bool {
+        self.inner.get_ref().is_empty()
+    }
+
+    /// Check if write buffer is full.
+    pub fn is_write_buf_full(&self) -> bool {
+        self.inner.get_ref().is_full()
+    }
+
+    /// Consumes the `Frame`, returning its underlying I/O stream.
+    ///
+    /// Note that care should be taken to not tamper with the underlying stream
+    /// of data coming in as it may corrupt the stream of frames otherwise
+    /// being worked with.
+    pub fn into_inner(self) -> T {
+        self.inner.into_inner().into_inner().0
+    }
+
+    /// Consume the `Frame`, returning `Frame` with different codec.
+    pub fn into_framed<U2>(self, codec: U2) -> Framed<T, U2> {
+        let (inner, read_buf) = self.inner.into_parts();
+        let (inner, write_buf, lw, hw) = inner.into_parts();
+
+        Framed {
+            inner: framed_read2_with_buffer(
+                framed_write2_with_buffer(Fuse(inner.0, codec), write_buf, lw, hw),
+                read_buf,
+            ),
+        }
+    }
+
+    /// Consume the `Frame`, returning `Frame` with different codec.
+    pub fn map_codec<F, U2>(self, f: F) -> Framed<T, U2>
+    where
+        F: Fn(U) -> U2,
+    {
+        let (inner, read_buf) = self.inner.into_parts();
+        let (inner, write_buf, lw, hw) = inner.into_parts();
+
+        Framed {
+            inner: framed_read2_with_buffer(
+                framed_write2_with_buffer(Fuse(inner.0, f(inner.1)), write_buf, lw, hw),
+                read_buf,
+            ),
+        }
+    }
+
+    /// Consumes the `Frame`, returning its underlying I/O stream, the buffer
+    /// with unprocessed data, and the codec.
+    ///
+    /// Note that care should be taken to not tamper with the underlying stream
+    /// of data coming in as it may corrupt the stream of frames otherwise
+    /// being worked with.
+    pub fn into_parts(self) -> FramedParts<T, U> {
+        let (inner, read_buf) = self.inner.into_parts();
+        let (inner, write_buf, write_buf_lw, write_buf_hw) = inner.into_parts();
+
+        FramedParts {
+            io: inner.0,
+            codec: inner.1,
+            read_buf,
+            write_buf,
+            write_buf_lw,
+            write_buf_hw,
+            _priv: (),
+        }
+    }
+}
+
+impl<T, U> Stream for Framed<T, U>
+where
+    T: AsyncRead,
+    U: Decoder,
+{
+    type Item = U::Item;
+    type Error = U::Error;
+
+    fn poll(&mut self) -> Poll<Option<Self::Item>, Self::Error> {
+        self.inner.poll()
+    }
+}
+
+impl<T, U> Sink for Framed<T, U>
+where
+    T: AsyncWrite,
+    U: Encoder,
+    U::Error: From<io::Error>,
+{
+    type SinkItem = U::Item;
+    type SinkError = U::Error;
+
+    fn start_send(
+        &mut self,
+        item: Self::SinkItem,
+    ) -> StartSend<Self::SinkItem, Self::SinkError> {
+        self.inner.get_mut().start_send(item)
+    }
+
+    fn poll_complete(&mut self) -> Poll<(), Self::SinkError> {
+        self.inner.get_mut().poll_complete()
+    }
+
+    fn close(&mut self) -> Poll<(), Self::SinkError> {
+        self.inner.get_mut().close()
+    }
+}
+
+impl<T, U> fmt::Debug for Framed<T, U>
+where
+    T: fmt::Debug,
+    U: fmt::Debug,
+{
+    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+        f.debug_struct("Framed")
+            .field("io", &self.inner.get_ref().get_ref().0)
+            .field("codec", &self.inner.get_ref().get_ref().1)
+            .finish()
+    }
+}
+
+// ===== impl Fuse =====
+
+impl<T: Read, U> Read for Fuse<T, U> {
+    fn read(&mut self, dst: &mut [u8]) -> io::Result<usize> {
+        self.0.read(dst)
+    }
+}
+
+impl<T: AsyncRead, U> AsyncRead for Fuse<T, U> {
+    unsafe fn prepare_uninitialized_buffer(&self, buf: &mut [u8]) -> bool {
+        self.0.prepare_uninitialized_buffer(buf)
+    }
+}
+
+impl<T: Write, U> Write for Fuse<T, U> {
+    fn write(&mut self, src: &[u8]) -> io::Result<usize> {
+        self.0.write(src)
+    }
+
+    fn flush(&mut self) -> io::Result<()> {
+        self.0.flush()
+    }
+}
+
+impl<T: AsyncWrite, U> AsyncWrite for Fuse<T, U> {
+    fn shutdown(&mut self) -> Poll<(), io::Error> {
+        self.0.shutdown()
+    }
+}
+
+impl<T, U: Decoder> Decoder for Fuse<T, U> {
+    type Item = U::Item;
+    type Error = U::Error;
+
+    fn decode(&mut self, buffer: &mut BytesMut) -> Result<Option<Self::Item>, Self::Error> {
+        self.1.decode(buffer)
+    }
+
+    fn decode_eof(&mut self, buffer: &mut BytesMut) -> Result<Option<Self::Item>, Self::Error> {
+        self.1.decode_eof(buffer)
+    }
+}
+
+impl<T, U: Encoder> Encoder for Fuse<T, U> {
+    type Item = U::Item;
+    type Error = U::Error;
+
+    fn encode(&mut self, item: Self::Item, dst: &mut BytesMut) -> Result<(), Self::Error> {
+        self.1.encode(item, dst)
+    }
+}
+
+/// `FramedParts` contains an export of the data of a Framed transport.
+/// It can be used to construct a new `Framed` with a different codec.
+/// It contains all current buffers and the inner transport.
+#[derive(Debug)]
+pub struct FramedParts<T, U> {
+    /// The inner transport used to read bytes to and write bytes to
+    pub io: T,
+
+    /// The codec
+    pub codec: U,
+
+    /// The buffer with read but unprocessed data.
+    pub read_buf: BytesMut,
+
+    /// A buffer with unprocessed data which are not written yet.
+    pub write_buf: BytesMut,
+
+    /// A buffer low watermark capacity
+    pub write_buf_lw: usize,
+
+    /// A buffer high watermark capacity
+    pub write_buf_hw: usize,
+
+    /// This private field allows us to add additional fields in the future in a
+    /// backwards compatible way.
+    _priv: (),
+}
+
+impl<T, U> FramedParts<T, U> {
+    /// Create a new, default, `FramedParts`
+    pub fn new(io: T, codec: U) -> FramedParts<T, U> {
+        FramedParts {
+            io,
+            codec,
+            read_buf: BytesMut::new(),
+            write_buf: BytesMut::new(),
+            write_buf_lw: LW,
+            write_buf_hw: HW,
+            _priv: (),
+        }
+    }
+}

actix-codec/src/framed2.rs

@@ -0,0 +1,313 @@
+#![allow(deprecated)]
+
+use std::fmt;
+use std::io::{self, Read, Write};
+
+use bytes::BytesMut;
+use futures::{Poll, Sink, StartSend, Stream};
+use tokio_codec::{Decoder, Encoder};
+use tokio_io::{AsyncRead, AsyncWrite};
+
+use super::framed_read::{framed_read2, framed_read2_with_buffer, FramedRead2};
+use super::framed_write::{framed_write2, framed_write2_with_buffer, FramedWrite2};
+
+/// A unified `Stream` and `Sink` interface to an underlying I/O object, using
+/// the `Encoder` and `Decoder` traits to encode and decode frames.
+///
+/// You can create a `Framed` instance by using the `AsyncRead::framed` adapter.
+pub struct Framed2<T, D, E> {
+    inner: FramedRead2<FramedWrite2<Fuse2<T, D, E>>>,
+}
+
+pub struct Fuse2<T, D, E>(pub T, pub D, pub E);
+
+impl<T, D, E> Framed2<T, D, E>
+where
+    T: AsyncRead + AsyncWrite,
+    D: Decoder,
+    E: Encoder,
+{
+    /// Provides a `Stream` and `Sink` interface for reading and writing to this
+    /// `Io` object, using `Decode` and `Encode` to read and write the raw data.
+    ///
+    /// Raw I/O objects work with byte sequences, but higher-level code usually
+    /// wants to batch these into meaningful chunks, called "frames". This
+    /// method layers framing on top of an I/O object, by using the `Codec`
+    /// traits to handle encoding and decoding of messages frames. Note that
+    /// the incoming and outgoing frame types may be distinct.
+    ///
+    /// This function returns a *single* object that is both `Stream` and
+    /// `Sink`; grouping this into a single object is often useful for layering
+    /// things like gzip or TLS, which require both read and write access to the
+    /// underlying object.
+    ///
+    /// If you want to work more directly with the streams and sink, consider
+    /// calling `split` on the `Framed` returned by this method, which will
+    /// break them into separate objects, allowing them to interact more easily.
+    pub fn new(inner: T, decoder: D, encoder: E) -> Framed2<T, D, E> {
+        Framed2 {
+            inner: framed_read2(framed_write2(Fuse2(inner, decoder, encoder))),
+        }
+    }
+}
+
+impl<T, D, E> Framed2<T, D, E> {
+    /// Provides a `Stream` and `Sink` interface for reading and writing to this
+    /// `Io` object, using `Decode` and `Encode` to read and write the raw data.
+    ///
+    /// Raw I/O objects work with byte sequences, but higher-level code usually
+    /// wants to batch these into meaningful chunks, called "frames". This
+    /// method layers framing on top of an I/O object, by using the `Codec`
+    /// traits to handle encoding and decoding of messages frames. Note that
+    /// the incoming and outgoing frame types may be distinct.
+    ///
+    /// This function returns a *single* object that is both `Stream` and
+    /// `Sink`; grouping this into a single object is often useful for layering
+    /// things like gzip or TLS, which require both read and write access to the
+    /// underlying object.
+    ///
+    /// This objects takes a stream and a readbuffer and a writebuffer. These
+    /// field can be obtained from an existing `Framed` with the
+    /// `into_parts` method.
+    ///
+    /// If you want to work more directly with the streams and sink, consider
+    /// calling `split` on the `Framed` returned by this method, which will
+    /// break them into separate objects, allowing them to interact more easily.
+    pub fn from_parts(parts: FramedParts2<T, D, E>) -> Framed2<T, D, E> {
+        Framed2 {
+            inner: framed_read2_with_buffer(
+                framed_write2_with_buffer(
+                    Fuse2(parts.io, parts.decoder, parts.encoder),
+                    parts.write_buf,
+                ),
+                parts.read_buf,
+            ),
+        }
+    }
+
+    /// Returns a reference to the underlying I/O stream wrapped by
+    /// `Frame`.
+    ///
+    /// Note that care should be taken to not tamper with the underlying stream
+    /// of data coming in as it may corrupt the stream of frames otherwise
+    /// being worked with.
+    pub fn get_ref(&self) -> &T {
+        &self.inner.get_ref().get_ref().0
+    }
+
+    /// Returns a mutable reference to the underlying I/O stream wrapped by
+    /// `Frame`.
+    ///
+    /// Note that care should be taken to not tamper with the underlying stream
+    /// of data coming in as it may corrupt the stream of frames otherwise
+    /// being worked with.
+    pub fn get_mut(&mut self) -> &mut T {
+        &mut self.inner.get_mut().get_mut().0
+    }
+
+    /// Returns a reference to the underlying decoder.
+    pub fn decocer(&self) -> &D {
+        &self.inner.get_ref().get_ref().1
+    }
+
+    /// Returns a mutable reference to the underlying decoder.
+    pub fn decoder_mut(&mut self) -> &mut D {
+        &mut self.inner.get_mut().get_mut().1
+    }
+
+    /// Returns a reference to the underlying encoder.
+    pub fn encoder(&self) -> &E {
+        &self.inner.get_ref().get_ref().2
+    }
+
+    /// Returns a mutable reference to the underlying codec.
+    pub fn encoder_mut(&mut self) -> &mut E {
+        &mut self.inner.get_mut().get_mut().2
+    }
+
+    /// Consumes the `Frame`, returning its underlying I/O stream.
+    ///
+    /// Note that care should be taken to not tamper with the underlying stream
+    /// of data coming in as it may corrupt the stream of frames otherwise
+    /// being worked with.
+    pub fn into_inner(self) -> T {
+        self.inner.into_inner().into_inner().0
+    }
+
+    /// Consume the `Frame`, returning `Frame` with different codec.
+    pub fn switch_encoder<E2>(self, encoder: E2) -> Framed2<T, D, E2> {
+        let (inner, read_buf) = self.inner.into_parts();
+        let (inner, write_buf) = inner.into_parts();
+
+        Framed2 {
+            inner: framed_read2_with_buffer(
+                framed_write2_with_buffer(Fuse2(inner.0, inner.1, encoder), write_buf),
+                read_buf,
+            ),
+        }
+    }
+
+    /// Consumes the `Frame`, returning its underlying I/O stream, the buffer
+    /// with unprocessed data, and the codec.
+    ///
+    /// Note that care should be taken to not tamper with the underlying stream
+    /// of data coming in as it may corrupt the stream of frames otherwise
+    /// being worked with.
+    pub fn into_parts(self) -> FramedParts2<T, D, E> {
+        let (inner, read_buf) = self.inner.into_parts();
+        let (inner, write_buf) = inner.into_parts();
+
+        FramedParts2 {
+            io: inner.0,
+            decoder: inner.1,
+            encoder: inner.2,
+            read_buf: read_buf,
+            write_buf: write_buf,
+            _priv: (),
+        }
+    }
+}
+
+impl<T, D, E> Stream for Framed2<T, D, E>
+where
+    T: AsyncRead,
+    D: Decoder,
+{
+    type Item = D::Item;
+    type Error = D::Error;
+
+    fn poll(&mut self) -> Poll<Option<Self::Item>, Self::Error> {
+        self.inner.poll()
+    }
+}
+
+impl<T, D, E> Sink for Framed2<T, D, E>
+where
+    T: AsyncWrite,
+    E: Encoder,
+    E::Error: From<io::Error>,
+{
+    type SinkItem = E::Item;
+    type SinkError = E::Error;
+
+    fn start_send(
+        &mut self,
+        item: Self::SinkItem,
+    ) -> StartSend<Self::SinkItem, Self::SinkError> {
+        self.inner.get_mut().start_send(item)
+    }
+
+    fn poll_complete(&mut self) -> Poll<(), Self::SinkError> {
+        self.inner.get_mut().poll_complete()
+    }
+
+    fn close(&mut self) -> Poll<(), Self::SinkError> {
+        self.inner.get_mut().close()
+    }
+}
+
+impl<T, D, E> fmt::Debug for Framed2<T, D, E>
+where
+    T: fmt::Debug,
+    D: fmt::Debug,
+    E: fmt::Debug,
+{
+    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+        f.debug_struct("Framed2")
+            .field("io", &self.inner.get_ref().get_ref().0)
+            .field("decoder", &self.inner.get_ref().get_ref().1)
+            .field("encoder", &self.inner.get_ref().get_ref().2)
+            .finish()
+    }
+}
+
+// ===== impl Fuse2 =====
+
+impl<T: Read, D, E> Read for Fuse2<T, D, E> {
+    fn read(&mut self, dst: &mut [u8]) -> io::Result<usize> {
+        self.0.read(dst)
+    }
+}
+
+impl<T: AsyncRead, D, E> AsyncRead for Fuse2<T, D, E> {
+    unsafe fn prepare_uninitialized_buffer(&self, buf: &mut [u8]) -> bool {
+        self.0.prepare_uninitialized_buffer(buf)
+    }
+}
+
+impl<T: Write, D, E> Write for Fuse2<T, D, E> {
+    fn write(&mut self, src: &[u8]) -> io::Result<usize> {
+        self.0.write(src)
+    }
+
+    fn flush(&mut self) -> io::Result<()> {
+        self.0.flush()
+    }
+}
+
+impl<T: AsyncWrite, D, E> AsyncWrite for Fuse2<T, D, E> {
+    fn shutdown(&mut self) -> Poll<(), io::Error> {
+        self.0.shutdown()
+    }
+}
+
+impl<T, D: Decoder, E> Decoder for Fuse2<T, D, E> {
+    type Item = D::Item;
+    type Error = D::Error;
+
+    fn decode(&mut self, buffer: &mut BytesMut) -> Result<Option<Self::Item>, Self::Error> {
+        self.1.decode(buffer)
+    }
+
+    fn decode_eof(&mut self, buffer: &mut BytesMut) -> Result<Option<Self::Item>, Self::Error> {
+        self.1.decode_eof(buffer)
+    }
+}
+
+impl<T, D, E: Encoder> Encoder for Fuse2<T, D, E> {
+    type Item = E::Item;
+    type Error = E::Error;
+
+    fn encode(&mut self, item: Self::Item, dst: &mut BytesMut) -> Result<(), Self::Error> {
+        self.2.encode(item, dst)
+    }
+}
+
+/// `FramedParts` contains an export of the data of a Framed transport.
+/// It can be used to construct a new `Framed` with a different codec.
+/// It contains all current buffers and the inner transport.
+#[derive(Debug)]
+pub struct FramedParts2<T, D, E> {
+    /// The inner transport used to read bytes to and write bytes to
+    pub io: T,
+
+    /// The decoder
+    pub decoder: D,
+
+    /// The encoder
+    pub encoder: E,
+
+    /// The buffer with read but unprocessed data.
+    pub read_buf: BytesMut,
+
+    /// A buffer with unprocessed data which are not written yet.
+    pub write_buf: BytesMut,
+
+    /// This private field allows us to add additional fields in the future in a
+    /// backwards compatible way.
+    _priv: (),
+}
+
+impl<T, D, E> FramedParts2<T, D, E> {
+    /// Create a new, default, `FramedParts`
+    pub fn new(io: T, decoder: D, encoder: E) -> FramedParts2<T, D, E> {
+        FramedParts2 {
+            io,
+            decoder,
+            encoder,
+            read_buf: BytesMut::new(),
+            write_buf: BytesMut::new(),
+            _priv: (),
+        }
+    }
+}

actix-codec/src/framed_read.rs

@@ -0,0 +1,218 @@
+use std::fmt;
+
+use bytes::BytesMut;
+use futures::{try_ready, Async, Poll, Sink, StartSend, Stream};
+use log::trace;
+use tokio_codec::Decoder;
+use tokio_io::AsyncRead;
+
+use super::framed::Fuse;
+
+/// A `Stream` of messages decoded from an `AsyncRead`.
+pub struct FramedRead<T, D> {
+    inner: FramedRead2<Fuse<T, D>>,
+}
+
+pub struct FramedRead2<T> {
+    inner: T,
+    eof: bool,
+    is_readable: bool,
+    buffer: BytesMut,
+}
+
+const INITIAL_CAPACITY: usize = 8 * 1024;
+
+// ===== impl FramedRead =====
+
+impl<T, D> FramedRead<T, D>
+where
+    T: AsyncRead,
+    D: Decoder,
+{
+    /// Creates a new `FramedRead` with the given `decoder`.
+    pub fn new(inner: T, decoder: D) -> FramedRead<T, D> {
+        FramedRead {
+            inner: framed_read2(Fuse(inner, decoder)),
+        }
+    }
+}
+
+impl<T, D> FramedRead<T, D> {
+    /// Returns a reference to the underlying I/O stream wrapped by
+    /// `FramedRead`.
+    ///
+    /// Note that care should be taken to not tamper with the underlying stream
+    /// of data coming in as it may corrupt the stream of frames otherwise
+    /// being worked with.
+    pub fn get_ref(&self) -> &T {
+        &self.inner.inner.0
+    }
+
+    /// Returns a mutable reference to the underlying I/O stream wrapped by
+    /// `FramedRead`.
+    ///
+    /// Note that care should be taken to not tamper with the underlying stream
+    /// of data coming in as it may corrupt the stream of frames otherwise
+    /// being worked with.
+    pub fn get_mut(&mut self) -> &mut T {
+        &mut self.inner.inner.0
+    }
+
+    /// Consumes the `FramedRead`, returning its underlying I/O stream.
+    ///
+    /// Note that care should be taken to not tamper with the underlying stream
+    /// of data coming in as it may corrupt the stream of frames otherwise
+    /// being worked with.
+    pub fn into_inner(self) -> T {
+        self.inner.inner.0
+    }
+
+    /// Returns a reference to the underlying decoder.
+    pub fn decoder(&self) -> &D {
+        &self.inner.inner.1
+    }
+
+    /// Returns a mutable reference to the underlying decoder.
+    pub fn decoder_mut(&mut self) -> &mut D {
+        &mut self.inner.inner.1
+    }
+}
+
+impl<T, D> Stream for FramedRead<T, D>
+where
+    T: AsyncRead,
+    D: Decoder,
+{
+    type Item = D::Item;
+    type Error = D::Error;
+
+    fn poll(&mut self) -> Poll<Option<Self::Item>, Self::Error> {
+        self.inner.poll()
+    }
+}
+
+impl<T, D> Sink for FramedRead<T, D>
+where
+    T: Sink,
+{
+    type SinkItem = T::SinkItem;
+    type SinkError = T::SinkError;
+
+    fn start_send(
+        &mut self,
+        item: Self::SinkItem,
+    ) -> StartSend<Self::SinkItem, Self::SinkError> {
+        self.inner.inner.0.start_send(item)
+    }
+
+    fn poll_complete(&mut self) -> Poll<(), Self::SinkError> {
+        self.inner.inner.0.poll_complete()
+    }
+
+    fn close(&mut self) -> Poll<(), Self::SinkError> {
+        self.inner.inner.0.close()
+    }
+}
+
+impl<T, D> fmt::Debug for FramedRead<T, D>
+where
+    T: fmt::Debug,
+    D: fmt::Debug,
+{
+    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+        f.debug_struct("FramedRead")
+            .field("inner", &self.inner.inner.0)
+            .field("decoder", &self.inner.inner.1)
+            .field("eof", &self.inner.eof)
+            .field("is_readable", &self.inner.is_readable)
+            .field("buffer", &self.inner.buffer)
+            .finish()
+    }
+}
+
+// ===== impl FramedRead2 =====
+
+pub fn framed_read2<T>(inner: T) -> FramedRead2<T> {
+    FramedRead2 {
+        inner,
+        eof: false,
+        is_readable: false,
+        buffer: BytesMut::with_capacity(INITIAL_CAPACITY),
+    }
+}
+
+pub fn framed_read2_with_buffer<T>(inner: T, mut buf: BytesMut) -> FramedRead2<T> {
+    if buf.capacity() < INITIAL_CAPACITY {
+        let bytes_to_reserve = INITIAL_CAPACITY - buf.capacity();
+        buf.reserve(bytes_to_reserve);
+    }
+    FramedRead2 {
+        inner,
+        eof: false,
+        is_readable: !buf.is_empty(),
+        buffer: buf,
+    }
+}
+
+impl<T> FramedRead2<T> {
+    pub fn get_ref(&self) -> &T {
+        &self.inner
+    }
+
+    pub fn into_inner(self) -> T {
+        self.inner
+    }
+
+    pub fn into_parts(self) -> (T, BytesMut) {
+        (self.inner, self.buffer)
+    }
+
+    pub fn get_mut(&mut self) -> &mut T {
+        &mut self.inner
+    }
+}
+
+impl<T> Stream for FramedRead2<T>
+where
+    T: AsyncRead + Decoder,
+{
+    type Item = T::Item;
+    type Error = T::Error;
+
+    fn poll(&mut self) -> Poll<Option<Self::Item>, Self::Error> {
+        loop {
+            // Repeatedly call `decode` or `decode_eof` as long as it is
+            // "readable". Readable is defined as not having returned `None`. If
+            // the upstream has returned EOF, and the decoder is no longer
+            // readable, it can be assumed that the decoder will never become
+            // readable again, at which point the stream is terminated.
+            if self.is_readable {
+                if self.eof {
+                    let frame = self.inner.decode_eof(&mut self.buffer)?;
+                    return Ok(Async::Ready(frame));
+                }
+
+                trace!("attempting to decode a frame");
+
+                if let Some(frame) = self.inner.decode(&mut self.buffer)? {
+                    trace!("frame decoded from buffer");
+                    return Ok(Async::Ready(Some(frame)));
+                }
+
+                self.is_readable = false;
+            }
+
+            assert!(!self.eof);
+
+            // Otherwise, try to read more data and try again. Make sure we've
+            // got room for at least one byte to read to ensure that we don't
+            // get a spurious 0 that looks like EOF
+            self.buffer.reserve(1);
+            if 0 == try_ready!(self.inner.read_buf(&mut self.buffer)) {
+                self.eof = true;
+            }
+
+            self.is_readable = true;
+        }
+    }
+}

actix-codec/src/framed_write.rs

@@ -0,0 +1,303 @@
+use std::fmt;
+use std::io::{self, Read};
+
+use bytes::BytesMut;
+use futures::{try_ready, Async, AsyncSink, Poll, Sink, StartSend, Stream};
+use log::trace;
+use tokio_codec::{Decoder, Encoder};
+use tokio_io::{AsyncRead, AsyncWrite};
+
+use super::framed::Fuse;
+
+/// A `Sink` of frames encoded to an `AsyncWrite`.
+pub struct FramedWrite<T, E> {
+    inner: FramedWrite2<Fuse<T, E>>,
+}
+
+pub struct FramedWrite2<T> {
+    inner: T,
+    buffer: BytesMut,
+    low_watermark: usize,
+    high_watermark: usize,
+}
+
+impl<T, E> FramedWrite<T, E>
+where
+    T: AsyncWrite,
+    E: Encoder,
+{
+    /// Creates a new `FramedWrite` with the given `encoder`.
+    pub fn new(inner: T, encoder: E, lw: usize, hw: usize) -> FramedWrite<T, E> {
+        FramedWrite {
+            inner: framed_write2(Fuse(inner, encoder), lw, hw),
+        }
+    }
+}
+
+impl<T, E> FramedWrite<T, E> {
+    /// Returns a reference to the underlying I/O stream wrapped by
+    /// `FramedWrite`.
+    ///
+    /// Note that care should be taken to not tamper with the underlying stream
+    /// of data coming in as it may corrupt the stream of frames otherwise
+    /// being worked with.
+    pub fn get_ref(&self) -> &T {
+        &self.inner.inner.0
+    }
+
+    /// Returns a mutable reference to the underlying I/O stream wrapped by
+    /// `FramedWrite`.
+    ///
+    /// Note that care should be taken to not tamper with the underlying stream
+    /// of data coming in as it may corrupt the stream of frames otherwise
+    /// being worked with.
+    pub fn get_mut(&mut self) -> &mut T {
+        &mut self.inner.inner.0
+    }
+
+    /// Consumes the `FramedWrite`, returning its underlying I/O stream.
+    ///
+    /// Note that care should be taken to not tamper with the underlying stream
+    /// of data coming in as it may corrupt the stream of frames otherwise
+    /// being worked with.
+    pub fn into_inner(self) -> T {
+        self.inner.inner.0
+    }
+
+    /// Returns a reference to the underlying decoder.
+    pub fn encoder(&self) -> &E {
+        &self.inner.inner.1
+    }
+
+    /// Returns a mutable reference to the underlying decoder.
+    pub fn encoder_mut(&mut self) -> &mut E {
+        &mut self.inner.inner.1
+    }
+
+    /// Check if write buffer is full
+    pub fn is_full(&self) -> bool {
+        self.inner.is_full()
+    }
+
+    /// Check if write buffer is empty.
+    pub fn is_empty(&self) -> bool {
+        self.inner.is_empty()
+    }
+}
+
+impl<T, E> FramedWrite<T, E>
+where
+    E: Encoder,
+{
+    /// Force send item
+    pub fn force_send(&mut self, item: E::Item) -> Result<(), E::Error> {
+        self.inner.force_send(item)
+    }
+}
+
+impl<T, E> Sink for FramedWrite<T, E>
+where
+    T: AsyncWrite,
+    E: Encoder,
+{
+    type SinkItem = E::Item;
+    type SinkError = E::Error;
+
+    fn start_send(&mut self, item: E::Item) -> StartSend<E::Item, E::Error> {
+        self.inner.start_send(item)
+    }
+
+    fn poll_complete(&mut self) -> Poll<(), Self::SinkError> {
+        self.inner.poll_complete()
+    }
+
+    fn close(&mut self) -> Poll<(), Self::SinkError> {
+        Ok(self.inner.close()?)
+    }
+}
+
+impl<T, D> Stream for FramedWrite<T, D>
+where
+    T: Stream,
+{
+    type Item = T::Item;
+    type Error = T::Error;
+
+    fn poll(&mut self) -> Poll<Option<Self::Item>, Self::Error> {
+        self.inner.inner.0.poll()
+    }
+}
+
+impl<T, U> fmt::Debug for FramedWrite<T, U>
+where
+    T: fmt::Debug,
+    U: fmt::Debug,
+{
+    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+        f.debug_struct("FramedWrite")
+            .field("inner", &self.inner.get_ref().0)
+            .field("encoder", &self.inner.get_ref().1)
+            .field("buffer", &self.inner.buffer)
+            .finish()
+    }
+}
+
+// ===== impl FramedWrite2 =====
+
+pub fn framed_write2<T>(
+    inner: T,
+    low_watermark: usize,
+    high_watermark: usize,
+) -> FramedWrite2<T> {
+    FramedWrite2 {
+        inner,
+        low_watermark,
+        high_watermark,
+        buffer: BytesMut::with_capacity(high_watermark),
+    }
+}
+
+pub fn framed_write2_with_buffer<T>(
+    inner: T,
+    mut buffer: BytesMut,
+    low_watermark: usize,
+    high_watermark: usize,
+) -> FramedWrite2<T> {
+    if buffer.capacity() < high_watermark {
+        let bytes_to_reserve = high_watermark - buffer.capacity();
+        buffer.reserve(bytes_to_reserve);
+    }
+    FramedWrite2 {
+        inner,
+        buffer,
+        low_watermark,
+        high_watermark,
+    }
+}
+
+impl<T> FramedWrite2<T> {
+    pub fn get_ref(&self) -> &T {
+        &self.inner
+    }
+
+    pub fn into_inner(self) -> T {
+        self.inner
+    }
+
+    pub fn into_parts(self) -> (T, BytesMut, usize, usize) {
+        (
+            self.inner,
+            self.buffer,
+            self.low_watermark,
+            self.high_watermark,
+        )
+    }
+
+    pub fn get_mut(&mut self) -> &mut T {
+        &mut self.inner
+    }
+
+    pub fn is_full(&self) -> bool {
+        self.buffer.len() >= self.high_watermark
+    }
+
+    pub fn is_empty(&self) -> bool {
+        self.buffer.is_empty()
+    }
+}
+
+impl<T> FramedWrite2<T>
+where
+    T: Encoder,
+{
+    pub fn force_send(&mut self, item: T::Item) -> Result<(), T::Error> {
+        let len = self.buffer.len();
+        if len < self.low_watermark {
+            self.buffer.reserve(self.high_watermark - len)
+        }
+        self.inner.encode(item, &mut self.buffer)?;
+        Ok(())
+    }
+}
+
+impl<T> Sink for FramedWrite2<T>
+where
+    T: AsyncWrite + Encoder,
+{
+    type SinkItem = T::Item;
+    type SinkError = T::Error;
+
+    fn start_send(&mut self, item: T::Item) -> StartSend<T::Item, T::Error> {
+        // Check the buffer capacity
+        let len = self.buffer.len();
+        if len >= self.high_watermark {
+            return Ok(AsyncSink::NotReady(item));
+        }
+        if len < self.low_watermark {
+            self.buffer.reserve(self.high_watermark - len)
+        }
+
+        self.inner.encode(item, &mut self.buffer)?;
+
+        Ok(AsyncSink::Ready)
+    }
+
+    fn poll_complete(&mut self) -> Poll<(), Self::SinkError> {
+        trace!("flushing framed transport");
+
+        while !self.buffer.is_empty() {
+            trace!("writing; remaining={}", self.buffer.len());
+
+            let n = try_ready!(self.inner.poll_write(&self.buffer));
+
+            if n == 0 {
+                return Err(io::Error::new(
+                    io::ErrorKind::WriteZero,
+                    "failed to \
+                     write frame to transport",
+                )
+                .into());
+            }
+
+            // TODO: Add a way to `bytes` to do this w/o returning the drained
+            // data.
+            let _ = self.buffer.split_to(n);
+        }
+
+        // Try flushing the underlying IO
+        try_ready!(self.inner.poll_flush());
+
+        trace!("framed transport flushed");
+        Ok(Async::Ready(()))
+    }
+
+    fn close(&mut self) -> Poll<(), Self::SinkError> {
+        try_ready!(self.poll_complete());
+        Ok(self.inner.shutdown()?)
+    }
+}
+
+impl<T: Decoder> Decoder for FramedWrite2<T> {
+    type Item = T::Item;
+    type Error = T::Error;
+
+    fn decode(&mut self, src: &mut BytesMut) -> Result<Option<T::Item>, T::Error> {
+        self.inner.decode(src)
+    }
+
+    fn decode_eof(&mut self, src: &mut BytesMut) -> Result<Option<T::Item>, T::Error> {
+        self.inner.decode_eof(src)
+    }
+}
+
+impl<T: Read> Read for FramedWrite2<T> {
+    fn read(&mut self, dst: &mut [u8]) -> io::Result<usize> {
+        self.inner.read(dst)
+    }
+}
+
+impl<T: AsyncRead> AsyncRead for FramedWrite2<T> {
+    unsafe fn prepare_uninitialized_buffer(&self, buf: &mut [u8]) -> bool {
+        self.inner.prepare_uninitialized_buffer(buf)
+    }
+}

actix-codec/src/lib.rs

@@ -0,0 +1,28 @@
+//! Utilities for encoding and decoding frames.
+//!
+//! Contains adapters to go from streams of bytes, [`AsyncRead`] and
+//! [`AsyncWrite`], to framed streams implementing [`Sink`] and [`Stream`].
+//! Framed streams are also known as [transports].
+//!
+//! [`AsyncRead`]: #
+//! [`AsyncWrite`]: #
+//! [`Sink`]: #
+//! [`Stream`]: #
+//! [transports]: #
+
+// #![deny(missing_docs, missing_debug_implementations, warnings)]
+
+mod bcodec;
+mod framed;
+// mod framed2;
+mod framed_read;
+mod framed_write;
+
+pub use self::bcodec::BytesCodec;
+pub use self::framed::{Framed, FramedParts};
+// pub use self::framed2::{Framed2, FramedParts2};
+pub use self::framed_read::FramedRead;
+pub use self::framed_write::FramedWrite;
+
+pub use tokio_codec::{Decoder, Encoder};
+pub use tokio_io::{AsyncRead, AsyncWrite};