actix/actix-net
1
0
Fork 0
mirror of https://github.com/fafhrd91/actix-net synced 2025-06-26 19:47:43 +02:00
Code Issues Releases Wiki Activity

cleanup Unpin constraint; simplify Framed impl

Browse Source
This commit is contained in:
Nikolay Kim
2019-11-19 14:51:40 +06:00
parent 617e40a7e9
commit 3bf83c1d98
30 changed files with 493 additions and 1252 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
actix-codec/Cargo.toml
View File

@ -19,8 +19,8 @@ path = "src/lib.rs"
[dependencies]
bytes = "0.4.12"
pin-utils = "0.1.0-alpha.4"
futures = "0.3.1"
pin-project = "0.4.5"
tokio-io = "0.2.0-alpha.6"
tokio-codec = "0.2.0-alpha.6"
log = "0.4"

378
actix-codec/src/framed.rs
View File

@ -1,35 +1,40 @@
#![allow(deprecated)]
use std::fmt;
use std::io::{self, Read, Write};
use std::io::{self};
use std::pin::Pin;
use std::task::{Context, Poll};
use bytes::BytesMut;
use futures::{Sink, Stream};
use futures::{ready, Sink, Stream};
use pin_project::pin_project;
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;
const INITIAL_CAPACITY: 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.
#[pin_project]
pub struct Framed<T, U> {
inner: FramedRead2<FramedWrite2<Fuse<T, U>>>,
io: T,
codec: U,
eof: bool,
is_readable: bool,
read_buf: BytesMut,
write_buf: BytesMut,
write_lw: usize,
write_hw: usize,
}
pub struct Fuse<T, U>(pub T, pub U);
impl<T, U> Framed<T, U>
where
T: AsyncRead + AsyncWrite + Unpin,
U: Decoder + Encoder + Unpin,
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.
@ -48,17 +53,31 @@ where
/// 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> {
pub fn new(io: T, codec: U) -> Framed<T, U> {
Framed {
inner: framed_read2(framed_write2(Fuse(inner, codec), LW, HW)),
io,
codec,
eof: false,
is_readable: false,
read_buf: BytesMut::with_capacity(INITIAL_CAPACITY),
write_buf: BytesMut::with_capacity(HW),
write_lw: LW,
write_hw: 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> {
pub fn new_with_caps(io: 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)),
io,
codec,
eof: false,
is_readable: false,
read_buf: BytesMut::with_capacity(INITIAL_CAPACITY),
write_buf: BytesMut::with_capacity(hw),
write_lw: lw,
write_hw: hw,
}
}
}
@ -87,26 +106,25 @@ impl<T, U> Framed<T, U> {
/// 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,
),
io: parts.io,
codec: parts.codec,
eof: false,
is_readable: false,
write_buf: parts.write_buf,
write_lw: parts.write_buf_lw,
write_hw: parts.write_buf_hw,
read_buf: parts.read_buf,
}
}
/// Returns a reference to the underlying codec.
pub fn get_codec(&self) -> &U {
&self.inner.get_ref().get_ref().1
&self.codec
}
/// 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
&mut self.codec
}
/// Returns a reference to the underlying I/O stream wrapped by
@ -116,7 +134,7 @@ impl<T, U> Framed<T, U> {
/// 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
&self.io
}
/// Returns a mutable reference to the underlying I/O stream wrapped by
@ -126,17 +144,17 @@ impl<T, U> Framed<T, U> {
/// 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
&mut self.io
}
/// Check if write buffer is empty.
pub fn is_write_buf_empty(&self) -> bool {
self.inner.get_ref().is_empty()
self.write_buf.is_empty()
}
/// Check if write buffer is full.
pub fn is_write_buf_full(&self) -> bool {
self.inner.get_ref().is_full()
self.write_buf.len() >= self.write_hw
}
/// Consumes the `Frame`, returning its underlying I/O stream.
@ -145,19 +163,20 @@ impl<T, U> Framed<T, U> {
/// 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
self.io
}
/// 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,
),
io: self.io,
codec,
eof: self.eof,
is_readable: self.is_readable,
read_buf: self.read_buf,
write_buf: self.write_buf,
write_lw: self.write_lw,
write_hw: self.write_hw,
}
}
@ -166,14 +185,15 @@ impl<T, U> Framed<T, U> {
where
F: Fn(T) -> T2,
{
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(f(inner.0), inner.1), write_buf, lw, hw),
read_buf,
),
io: f(self.io),
codec: self.codec,
eof: self.eof,
is_readable: self.is_readable,
read_buf: self.read_buf,
write_buf: self.write_buf,
write_lw: self.write_lw,
write_hw: self.write_hw,
}
}
@ -182,14 +202,15 @@ impl<T, U> Framed<T, U> {
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,
),
io: self.io,
codec: f(self.codec),
eof: self.eof,
is_readable: self.is_readable,
read_buf: self.read_buf,
write_buf: self.write_buf,
write_lw: self.write_lw,
write_hw: self.write_hw,
}
}
@ -200,16 +221,13 @@ impl<T, U> Framed<T, U> {
/// 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,
io: self.io,
codec: self.codec,
read_buf: self.read_buf,
write_buf: self.write_buf,
write_buf_lw: self.write_lw,
write_buf_hw: self.write_hw,
_priv: (),
}
}
@ -219,92 +237,168 @@ impl<T, U> Framed<T, U> {
/// Serialize item and Write to the inner buffer
pub fn write(&mut self, item: <U as Encoder>::Item) -> Result<(), <U as Encoder>::Error>
where
T: AsyncWrite + Unpin,
U: Encoder + Unpin,
T: AsyncWrite,
U: Encoder,
{
self.inner.get_mut().force_send(item)
let len = self.write_buf.len();
if len < self.write_lw {
self.write_buf.reserve(self.write_hw - len)
}
self.codec.encode(item, &mut self.write_buf)?;
Ok(())
}
pub fn is_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), U::Error>>
where
T: AsyncWrite + Unpin,
U: Encoder + Unpin,
U::Error: From<io::Error>,
{
Pin::new(&mut self.inner.get_mut()).poll_ready(cx)
pub fn is_ready(&self) -> bool {
let len = self.write_buf.len();
len < self.write_hw
}
pub fn next_item(&mut self, cx: &mut Context<'_>) -> Poll<Option<Result<U::Item, U::Error>>>
pub fn next_item(&mut self, cx: &mut Context) -> Poll<Option<Result<U::Item, U::Error>>>
where
T: AsyncRead + Unpin,
U: Decoder + Unpin,
T: AsyncRead,
U: Decoder,
{
Pin::new(&mut self.inner).poll_next(cx)
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 {
match self.codec.decode_eof(&mut self.read_buf) {
Ok(Some(frame)) => return Poll::Ready(Some(Ok(frame))),
Ok(None) => return Poll::Ready(None),
Err(e) => return Poll::Ready(Some(Err(e))),
}
}
log::trace!("attempting to decode a frame");
match self.codec.decode(&mut self.read_buf) {
Ok(Some(frame)) => {
log::trace!("frame decoded from buffer");
return Poll::Ready(Some(Ok(frame)));
}
Err(e) => return Poll::Ready(Some(Err(e))),
_ => {
// Need more data
}
}
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.read_buf.reserve(1);
let cnt = unsafe {
match Pin::new_unchecked(&mut self.io).poll_read_buf(cx, &mut self.read_buf) {
Poll::Pending => return Poll::Pending,
Poll::Ready(Err(e)) => return Poll::Ready(Some(Err(e.into()))),
Poll::Ready(Ok(cnt)) => cnt,
}
};
if cnt == 0 {
self.eof = true;
}
self.is_readable = true;
}
}
pub fn flush(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), U::Error>>
pub fn flush(&mut self, cx: &mut Context) -> Poll<Result<(), U::Error>>
where
T: AsyncWrite + Unpin,
U: Encoder + Unpin,
T: AsyncWrite,
U: Encoder,
{
Pin::new(self.inner.get_mut()).poll_flush(cx)
log::trace!("flushing framed transport");
while !self.write_buf.is_empty() {
log::trace!("writing; remaining={}", self.write_buf.len());
let n = ready!(
unsafe { Pin::new_unchecked(&mut self.io) }.poll_write(cx, &self.write_buf)
)?;
if n == 0 {
return Poll::Ready(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.write_buf.split_to(n);
}
// Try flushing the underlying IO
ready!(unsafe { Pin::new_unchecked(&mut self.io) }.poll_flush(cx))?;
log::trace!("framed transport flushed");
Poll::Ready(Ok(()))
}
pub fn close(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), U::Error>>
pub fn close(&mut self, cx: &mut Context) -> Poll<Result<(), U::Error>>
where
T: AsyncWrite + Unpin,
U: Encoder + Unpin,
T: AsyncWrite,
U: Encoder,
{
Pin::new(&mut self.inner.get_mut()).poll_close(cx)
ready!(unsafe { Pin::new_unchecked(&mut self.io) }.poll_flush(cx))?;
ready!(unsafe { Pin::new_unchecked(&mut self.io) }.poll_shutdown(cx))?;
Poll::Ready(Ok(()))
}
}
impl<T, U> Stream for Framed<T, U>
where
T: AsyncRead + Unpin,
U: Decoder + Unpin,
T: AsyncRead,
U: Decoder,
{
type Item = Result<U::Item, U::Error>;
fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
Pin::new(&mut self.as_mut().inner).poll_next(cx)
fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll<Option<Self::Item>> {
self.next_item(cx)
}
}
impl<T, U> Sink<U::Item> for Framed<T, U>
where
T: AsyncWrite + Unpin,
U: Encoder + Unpin,
T: AsyncWrite,
U: Encoder,
U::Error: From<io::Error>,
{
type Error = U::Error;
fn poll_ready(
mut self: Pin<&mut Self>,
cx: &mut Context<'_>,
) -> Poll<Result<(), Self::Error>> {
Pin::new(&mut self.as_mut().inner.get_mut()).poll_ready(cx)
fn poll_ready(self: Pin<&mut Self>, _: &mut Context) -> Poll<Result<(), Self::Error>> {
if self.is_ready() {
Poll::Ready(Ok(()))
} else {
Poll::Pending
}
}
fn start_send(
mut self: Pin<&mut Self>,
item: <U as Encoder>::Item,
) -> Result<(), Self::Error> {
Pin::new(&mut self.as_mut().inner.get_mut()).start_send(item)
self.write(item)
}
fn poll_flush(
mut self: Pin<&mut Self>,
cx: &mut Context<'_>,
) -> Poll<Result<(), Self::Error>> {
Pin::new(&mut self.as_mut().inner.get_mut()).poll_flush(cx)
fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll<Result<(), Self::Error>> {
self.flush(cx)
}
fn poll_close(
mut self: Pin<&mut Self>,
cx: &mut Context<'_>,
) -> Poll<Result<(), Self::Error>> {
Pin::new(&mut self.as_mut().inner.get_mut()).poll_close(cx)
fn poll_close(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll<Result<(), Self::Error>> {
self.close(cx)
}
}
@ -315,84 +409,12 @@ where
{
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)
.field("io", &self.io)
.field("codec", &self.codec)
.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)
}
fn poll_read(
self: Pin<&mut Self>,
cx: &mut Context<'_>,
buf: &mut [u8],
) -> Poll<io::Result<usize>> {
unsafe { self.map_unchecked_mut(|s| &mut s.0).poll_read(cx, 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 poll_write(
self: Pin<&mut Self>,
cx: &mut Context<'_>,
buf: &[u8],
) -> Poll<io::Result<usize>> {
unsafe { self.map_unchecked_mut(|s| &mut s.0).poll_write(cx, buf) }
}
fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
unsafe { self.map_unchecked_mut(|s| &mut s.0).poll_flush(cx) }
}
fn poll_shutdown(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
unsafe { self.map_unchecked_mut(|s| &mut s.0).poll_shutdown(cx) }
}
}
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.

248
actix-codec/src/framed_read.rs
View File

@ -1,248 +0,0 @@
use std::fmt;
use std::pin::Pin;
use std::task::{Context, Poll};
use bytes::BytesMut;
use futures::{Sink, 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 = Result<D::Item, D::Error>;
fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
unsafe { self.map_unchecked_mut(|s| &mut s.inner).poll_next(cx) }
}
}
impl<I, T, D> Sink<I> for FramedRead<T, D>
where
T: Sink<I>,
{
type Error = T::Error;
fn poll_ready(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
unsafe {
self.map_unchecked_mut(|s| &mut s.inner.inner.0)
.poll_ready(cx)
}
}
fn start_send(self: Pin<&mut Self>, item: I) -> Result<(), Self::Error> {
unsafe {
self.map_unchecked_mut(|s| &mut s.inner.inner.0)
.start_send(item)
}
}
fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
unsafe {
self.map_unchecked_mut(|s| &mut s.inner.inner.0)
.poll_flush(cx)
}
}
fn poll_close(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
unsafe {
self.map_unchecked_mut(|s| &mut s.inner.inner.0)
.poll_close(cx)
}
}
}
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: tokio_io::AsyncRead + Decoder,
{
type Item = Result<T::Item, T::Error>;
fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
let mut this = unsafe { self.get_unchecked_mut() };
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 this.is_readable {
if this.eof {
match this.inner.decode_eof(&mut this.buffer) {
Ok(Some(frame)) => return Poll::Ready(Some(Ok(frame))),
Ok(None) => return Poll::Ready(None),
Err(e) => return Poll::Ready(Some(Err(e))),
}
}
trace!("attempting to decode a frame");
match this.inner.decode(&mut this.buffer) {
Ok(Some(frame)) => {
trace!("frame decoded from buffer");
return Poll::Ready(Some(Ok(frame)));
}
Err(e) => return Poll::Ready(Some(Err(e))),
_ => {
// Need more data
}
}
this.is_readable = false;
}
assert!(!this.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
this.buffer.reserve(1);
let cnt = unsafe {
match Pin::new_unchecked(&mut this.inner).poll_read_buf(cx, &mut this.buffer) {
Poll::Pending => return Poll::Pending,
Poll::Ready(Err(e)) => return Poll::Ready(Some(Err(e.into()))),
Poll::Ready(Ok(cnt)) => cnt,
}
};
if cnt == 0 {
this.eof = true;
}
this.is_readable = true;
}
}
}

333
actix-codec/src/framed_write.rs
View File

@ -1,333 +0,0 @@
use std::fmt;
use std::io::{self, Read};
use std::pin::Pin;
use std::task::{Context, Poll};
use bytes::BytesMut;
use futures::{ready, Sink, 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<E::Item> for FramedWrite<T, E>
where
T: AsyncWrite,
E: Encoder,
{
type Error = E::Error;
fn poll_ready(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
unsafe { self.map_unchecked_mut(|s| &mut s.inner).poll_ready(cx) }
}
fn start_send(self: Pin<&mut Self>, item: <E as Encoder>::Item) -> Result<(), Self::Error> {
unsafe { self.map_unchecked_mut(|s| &mut s.inner).start_send(item) }
}
fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
unsafe { self.map_unchecked_mut(|s| &mut s.inner).poll_flush(cx) }
}
fn poll_close(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
unsafe { self.map_unchecked_mut(|s| &mut s.inner).poll_close(cx) }
}
}
impl<T, D> Stream for FramedWrite<T, D>
where
T: Stream,
{
type Item = T::Item;
fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
unsafe {
self.map_unchecked_mut(|s| &mut s.inner.inner.0)
.poll_next(cx)
}
}
}
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<T::Item> for FramedWrite2<T>
where
T: AsyncWrite + Encoder,
{
type Error = T::Error;
fn poll_ready(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
let len = self.buffer.len();
if len >= self.high_watermark {
Poll::Pending
} else {
Poll::Ready(Ok(()))
}
}
fn start_send(self: Pin<&mut Self>, item: <T as Encoder>::Item) -> Result<(), Self::Error> {
let this = unsafe { self.get_unchecked_mut() };
// Check the buffer capacity
let len = this.buffer.len();
if len < this.low_watermark {
this.buffer.reserve(this.high_watermark - len)
}
this.inner.encode(item, &mut this.buffer)?;
Ok(())
}
fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
let this = unsafe { self.get_unchecked_mut() };
trace!("flushing framed transport");
while !this.buffer.is_empty() {
trace!("writing; remaining={}", this.buffer.len());
let n = ready!(
unsafe { Pin::new_unchecked(&mut this.inner) }.poll_write(cx, &this.buffer)
)?;
if n == 0 {
return Poll::Ready(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 _ = this.buffer.split_to(n);
}
// Try flushing the underlying IO
ready!(unsafe { Pin::new_unchecked(&mut this.inner) }.poll_flush(cx))?;
trace!("framed transport flushed");
Poll::Ready(Ok(()))
}
fn poll_close(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
let mut this = unsafe { self.get_unchecked_mut() };
ready!(
unsafe { Pin::new_unchecked(&mut this).map_unchecked_mut(|s| *s) }.poll_flush(cx)
)?;
ready!(unsafe { Pin::new_unchecked(&mut this.inner) }.poll_shutdown(cx))?;
Poll::Ready(Ok(()))
}
}
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)
}
fn poll_read(
self: Pin<&mut Self>,
cx: &mut Context<'_>,
buf: &mut [u8],
) -> Poll<io::Result<usize>> {
unsafe { self.map_unchecked_mut(|s| &mut s.inner).poll_read(cx, buf) }
}
}

4
actix-codec/src/lib.rs
View File

@ -12,13 +12,9 @@
mod bcodec;
mod framed;
mod framed_read;
mod framed_write;
pub use self::bcodec::BytesCodec;
pub use self::framed::{Framed, FramedParts};
pub use self::framed_read::FramedRead;
pub use self::framed_write::FramedWrite;
pub use tokio_codec::{Decoder, Encoder};
pub use tokio_io::{AsyncRead, AsyncWrite};