#![allow(deprecated)]
use {codec, AsyncRead, AsyncWrite};
use bytes::{Buf, BufMut, BytesMut, IntoBuf};
use bytes::buf::Chain;
use futures::{Async, AsyncSink, Stream, Sink, StartSend, Poll};
use std::{cmp, fmt};
use std::error::Error as StdError;
use std::io::{self, Cursor};
#[deprecated(since = "0.1.8", note = "Moved to tokio-codec")]
#[doc(hidden)]
#[derive(Debug, Clone, Copy)]
pub struct Builder {
max_frame_len: usize,
length_field_len: usize,
length_field_offset: usize,
length_adjustment: isize,
num_skip: Option<usize>,
length_field_is_big_endian: bool,
}
#[deprecated(since = "0.1.8", note = "Moved to tokio-codec")]
#[doc(hidden)]
pub struct Framed<T, B: IntoBuf = BytesMut> {
inner: FramedRead<FramedWrite<T, B>>,
}
#[deprecated(since = "0.1.8", note = "Moved to tokio-codec")]
#[doc(hidden)]
#[derive(Debug)]
pub struct FramedRead<T> {
inner: codec::FramedRead<T, Decoder>,
}
#[deprecated(since = "0.1.8", note = "Moved to tokio-codec")]
#[doc(hidden)]
pub struct FrameTooBig {
_priv: (),
}
#[derive(Debug)]
struct Decoder {
builder: Builder,
state: DecodeState,
}
#[derive(Debug, Clone, Copy)]
enum DecodeState {
Head,
Data(usize),
}
#[deprecated(since = "0.1.8", note = "Moved to tokio-codec")]
#[doc(hidden)]
pub struct FramedWrite<T, B: IntoBuf = BytesMut> {
inner: T,
builder: Builder,
frame: Option<Chain<Cursor<BytesMut>, B::Buf>>,
}
impl<T: AsyncRead + AsyncWrite, B: IntoBuf> Framed<T, B> {
pub fn new(inner: T) -> Framed<T, B> {
Builder::new().new_framed(inner)
}
}
impl<T, B: IntoBuf> Framed<T, B> {
pub fn get_ref(&self) -> &T {
self.inner.get_ref().get_ref()
}
pub fn get_mut(&mut self) -> &mut T {
self.inner.get_mut().get_mut()
}
pub fn into_inner(self) -> T {
self.inner.into_inner().into_inner()
}
}
impl<T: AsyncRead, B: IntoBuf> Stream for Framed<T, B> {
type Item = BytesMut;
type Error = io::Error;
fn poll(&mut self) -> Poll<Option<BytesMut>, io::Error> {
self.inner.poll()
}
}
impl<T: AsyncWrite, B: IntoBuf> Sink for Framed<T, B> {
type SinkItem = B;
type SinkError = io::Error;
fn start_send(&mut self, item: B) -> StartSend<B, io::Error> {
self.inner.start_send(item)
}
fn poll_complete(&mut self) -> Poll<(), io::Error> {
self.inner.poll_complete()
}
fn close(&mut self) -> Poll<(), io::Error> {
self.inner.close()
}
}
impl<T, B: IntoBuf> fmt::Debug for Framed<T, B>
where T: fmt::Debug,
B::Buf: fmt::Debug,
{
fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
f.debug_struct("Framed")
.field("inner", &self.inner)
.finish()
}
}
impl<T: AsyncRead> FramedRead<T> {
pub fn new(inner: T) -> FramedRead<T> {
Builder::new().new_read(inner)
}
}
impl<T> FramedRead<T> {
pub fn max_frame_length(&self) -> usize {
self.inner.decoder().builder.max_frame_len
}
pub fn set_max_frame_length(&mut self, val: usize) {
self.inner.decoder_mut().builder.max_frame_length(val);
}
pub fn get_ref(&self) -> &T {
self.inner.get_ref()
}
pub fn get_mut(&mut self) -> &mut T {
self.inner.get_mut()
}
pub fn into_inner(self) -> T {
self.inner.into_inner()
}
}
impl<T: AsyncRead> Stream for FramedRead<T> {
type Item = BytesMut;
type Error = io::Error;
fn poll(&mut self) -> Poll<Option<BytesMut>, io::Error> {
self.inner.poll()
}
}
impl<T: Sink> Sink for FramedRead<T> {
type SinkItem = T::SinkItem;
type SinkError = T::SinkError;
fn start_send(&mut self, item: T::SinkItem) -> StartSend<T::SinkItem, T::SinkError> {
self.inner.start_send(item)
}
fn poll_complete(&mut self) -> Poll<(), T::SinkError> {
self.inner.poll_complete()
}
fn close(&mut self) -> Poll<(), T::SinkError> {
self.inner.close()
}
}
impl<T: io::Write> io::Write for FramedRead<T> {
fn write(&mut self, src: &[u8]) -> io::Result<usize> {
self.inner.get_mut().write(src)
}
fn flush(&mut self) -> io::Result<()> {
self.inner.get_mut().flush()
}
}
impl<T: AsyncWrite> AsyncWrite for FramedRead<T> {
fn shutdown(&mut self) -> Poll<(), io::Error> {
self.inner.get_mut().shutdown()
}
fn write_buf<B: Buf>(&mut self, buf: &mut B) -> Poll<usize, io::Error> {
self.inner.get_mut().write_buf(buf)
}
}
impl Decoder {
fn decode_head(&mut self, src: &mut BytesMut) -> io::Result<Option<usize>> {
let head_len = self.builder.num_head_bytes();
let field_len = self.builder.length_field_len;
if src.len() < head_len {
return Ok(None);
}
let n = {
let mut src = Cursor::new(&mut *src);
src.advance(self.builder.length_field_offset);
let n = if self.builder.length_field_is_big_endian {
src.get_uint_be(field_len)
} else {
src.get_uint_le(field_len)
};
if n > self.builder.max_frame_len as u64 {
return Err(io::Error::new(io::ErrorKind::InvalidData, FrameTooBig {
_priv: (),
}));
}
let n = n as usize;
let n = if self.builder.length_adjustment < 0 {
n.checked_sub(-self.builder.length_adjustment as usize)
} else {
n.checked_add(self.builder.length_adjustment as usize)
};
match n {
Some(n) => n,
None => return Err(io::Error::new(io::ErrorKind::InvalidInput, "provided length would overflow after adjustment")),
}
};
let num_skip = self.builder.get_num_skip();
if num_skip > 0 {
let _ = src.split_to(num_skip);
}
src.reserve(n);
return Ok(Some(n));
}
fn decode_data(&self, n: usize, src: &mut BytesMut) -> io::Result<Option<BytesMut>> {
if src.len() < n {
return Ok(None);
}
Ok(Some(src.split_to(n)))
}
}
impl codec::Decoder for Decoder {
type Item = BytesMut;
type Error = io::Error;
fn decode(&mut self, src: &mut BytesMut) -> io::Result<Option<BytesMut>> {
let n = match self.state {
DecodeState::Head => {
match try!(self.decode_head(src)) {
Some(n) => {
self.state = DecodeState::Data(n);
n
}
None => return Ok(None),
}
}
DecodeState::Data(n) => n,
};
match try!(self.decode_data(n, src)) {
Some(data) => {
self.state = DecodeState::Head;
src.reserve(self.builder.num_head_bytes());
Ok(Some(data))
}
None => Ok(None),
}
}
}
impl<T: AsyncWrite, B: IntoBuf> FramedWrite<T, B> {
pub fn new(inner: T) -> FramedWrite<T, B> {
Builder::new().new_write(inner)
}
}
impl<T, B: IntoBuf> FramedWrite<T, B> {
pub fn max_frame_length(&self) -> usize {
self.builder.max_frame_len
}
pub fn set_max_frame_length(&mut self, val: usize) {
self.builder.max_frame_length(val);
}
pub fn get_ref(&self) -> &T {
&self.inner
}
pub fn get_mut(&mut self) -> &mut T {
&mut self.inner
}
pub fn into_inner(self) -> T {
self.inner
}
}
impl<T: AsyncWrite, B: IntoBuf> FramedWrite<T, B> {
fn do_write(&mut self) -> Poll<(), io::Error> {
if self.frame.is_none() {
return Ok(Async::Ready(()));
}
loop {
let frame = self.frame.as_mut().unwrap();
if try_ready!(self.inner.write_buf(frame)) == 0 {
return Err(io::Error::new(
io::ErrorKind::WriteZero,
"failed to write frame to transport",
));
}
if !frame.has_remaining() {
break;
}
}
self.frame = None;
Ok(Async::Ready(()))
}
fn set_frame(&mut self, buf: B::Buf) -> io::Result<()> {
let mut head = BytesMut::with_capacity(8);
let n = buf.remaining();
if n > self.builder.max_frame_len {
return Err(io::Error::new(io::ErrorKind::InvalidInput, FrameTooBig {
_priv: (),
}));
}
let n = if self.builder.length_adjustment < 0 {
n.checked_add(-self.builder.length_adjustment as usize)
} else {
n.checked_sub(self.builder.length_adjustment as usize)
};
let n = match n {
Some(n) => n,
None => return Err(io::Error::new(io::ErrorKind::InvalidInput, "provided length would overflow after adjustment")),
};
if self.builder.length_field_is_big_endian {
head.put_uint_be(n as u64, self.builder.length_field_len);
} else {
head.put_uint_le(n as u64, self.builder.length_field_len);
}
debug_assert!(self.frame.is_none());
self.frame = Some(head.into_buf().chain(buf));
Ok(())
}
}
impl<T: AsyncWrite, B: IntoBuf> Sink for FramedWrite<T, B> {
type SinkItem = B;
type SinkError = io::Error;
fn start_send(&mut self, item: B) -> StartSend<B, io::Error> {
if !try!(self.do_write()).is_ready() {
return Ok(AsyncSink::NotReady(item));
}
try!(self.set_frame(item.into_buf()));
Ok(AsyncSink::Ready)
}
fn poll_complete(&mut self) -> Poll<(), io::Error> {
try_ready!(self.do_write());
try_ready!(self.inner.poll_flush());
return Ok(Async::Ready(()));
}
fn close(&mut self) -> Poll<(), io::Error> {
try_ready!(self.poll_complete());
self.inner.shutdown()
}
}
impl<T: Stream, B: IntoBuf> Stream for FramedWrite<T, B> {
type Item = T::Item;
type Error = T::Error;
fn poll(&mut self) -> Poll<Option<T::Item>, T::Error> {
self.inner.poll()
}
}
impl<T: io::Read, B: IntoBuf> io::Read for FramedWrite<T, B> {
fn read(&mut self, dst: &mut [u8]) -> io::Result<usize> {
self.get_mut().read(dst)
}
}
impl<T: AsyncRead, U: IntoBuf> AsyncRead for FramedWrite<T, U> {
fn read_buf<B: BufMut>(&mut self, buf: &mut B) -> Poll<usize, io::Error> {
self.get_mut().read_buf(buf)
}
unsafe fn prepare_uninitialized_buffer(&self, buf: &mut [u8]) -> bool {
self.get_ref().prepare_uninitialized_buffer(buf)
}
}
impl<T, B: IntoBuf> fmt::Debug for FramedWrite<T, B>
where T: fmt::Debug,
B::Buf: fmt::Debug,
{
fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
f.debug_struct("FramedWrite")
.field("inner", &self.inner)
.field("builder", &self.builder)
.field("frame", &self.frame)
.finish()
}
}
impl Builder {
pub fn new() -> Builder {
Builder {
max_frame_len: 8 * 1_024 * 1_024,
length_field_len: 4,
length_field_offset: 0,
length_adjustment: 0,
num_skip: None,
length_field_is_big_endian: true,
}
}
pub fn big_endian(&mut self) -> &mut Self {
self.length_field_is_big_endian = true;
self
}
pub fn little_endian(&mut self) -> &mut Self {
self.length_field_is_big_endian = false;
self
}
pub fn native_endian(&mut self) -> &mut Self {
if cfg!(target_endian = "big") {
self.big_endian()
} else {
self.little_endian()
}
}
pub fn max_frame_length(&mut self, val: usize) -> &mut Self {
self.max_frame_len = val;
self
}
pub fn length_field_length(&mut self, val: usize) -> &mut Self {
assert!(val > 0 && val <= 8, "invalid length field length");
self.length_field_len = val;
self
}
pub fn length_field_offset(&mut self, val: usize) -> &mut Self {
self.length_field_offset = val;
self
}
pub fn length_adjustment(&mut self, val: isize) -> &mut Self {
self.length_adjustment = val;
self
}
pub fn num_skip(&mut self, val: usize) -> &mut Self {
self.num_skip = Some(val);
self
}
pub fn new_read<T>(&self, upstream: T) -> FramedRead<T>
where T: AsyncRead,
{
FramedRead {
inner: codec::FramedRead::new(upstream, Decoder {
builder: *self,
state: DecodeState::Head,
}),
}
}
pub fn new_write<T, B>(&self, inner: T) -> FramedWrite<T, B>
where T: AsyncWrite,
B: IntoBuf,
{
FramedWrite {
inner: inner,
builder: *self,
frame: None,
}
}
pub fn new_framed<T, B>(&self, inner: T) -> Framed<T, B>
where T: AsyncRead + AsyncWrite,
B: IntoBuf
{
let inner = self.new_read(self.new_write(inner));
Framed { inner: inner }
}
fn num_head_bytes(&self) -> usize {
let num = self.length_field_offset + self.length_field_len;
cmp::max(num, self.num_skip.unwrap_or(0))
}
fn get_num_skip(&self) -> usize {
self.num_skip.unwrap_or(self.length_field_offset + self.length_field_len)
}
}
impl fmt::Debug for FrameTooBig {
fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
f.debug_struct("FrameTooBig")
.finish()
}
}
impl fmt::Display for FrameTooBig {
fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
f.write_str(self.description())
}
}
impl StdError for FrameTooBig {
fn description(&self) -> &str {
"frame size too big"
}
}