From f723d100871e025e4bdd2f47397c9b089e666ce0 Mon Sep 17 00:00:00 2001 From: Bryan Burgers Date: Mon, 4 Jun 2018 22:36:06 -0500 Subject: [PATCH] Create tokio-codec (#360) Create a new tokio-codec crate with many of the contents of `tokio_io::codec`. --- Cargo.toml | 2 + examples/connect.rs | 6 +- examples/print_each_packet.rs | 7 +- examples/tinyhttp.rs | 7 +- examples/udp-codec.rs | 3 +- tests/line-frames.rs | 6 +- tokio-codec/CHANGELOG.md | 3 + tokio-codec/Cargo.toml | 22 ++ tokio-codec/LICENSE | 25 ++ tokio-codec/README.md | 35 +++ tokio-codec/src/bytes_codec.rs | 37 +++ tokio-codec/src/lib.rs | 32 +++ tokio-codec/src/lines_codec.rs | 89 +++++++ {tokio-io => tokio-codec}/tests/codecs.rs | 4 +- {tokio-io => tokio-codec}/tests/framed.rs | 3 +- .../tests/framed_read.rs | 3 +- .../tests/framed_write.rs | 3 +- tokio-fs/CHANGELOG.md | 4 + tokio-fs/Cargo.toml | 1 + tokio-fs/examples/std-echo.rs | 5 +- tokio-io/CHANGELOG.md | 4 + tokio-io/src/_tokio_codec/decoder.rs | 3 + tokio-io/src/_tokio_codec/encoder.rs | 3 + tokio-io/src/_tokio_codec/framed.rs | 248 ++++++++++++++++++ tokio-io/src/_tokio_codec/framed_read.rs | 214 +++++++++++++++ tokio-io/src/_tokio_codec/framed_write.rs | 237 +++++++++++++++++ tokio-io/src/_tokio_codec/mod.rs | 36 +++ tokio-io/src/async_read.rs | 3 + tokio-io/src/codec/bytes_codec.rs | 3 + tokio-io/src/codec/decoder.rs | 31 +++ tokio-io/src/codec/encoder.rs | 3 + tokio-io/src/codec/lines_codec.rs | 3 + tokio-io/src/codec/mod.rs | 8 + tokio-io/src/framed.rs | 6 + tokio-io/src/framed_read.rs | 6 + tokio-io/src/framed_write.rs | 6 + tokio-io/src/length_delimited.rs | 2 + tokio-io/src/lib.rs | 1 + tokio-udp/Cargo.toml | 1 + tokio-udp/src/frame.rs | 2 +- tokio-udp/src/lib.rs | 1 + tokio-udp/tests/udp.rs | 5 +- 42 files changed, 1100 insertions(+), 23 deletions(-) create mode 100644 tokio-codec/CHANGELOG.md create mode 100644 tokio-codec/Cargo.toml create mode 100644 tokio-codec/LICENSE create mode 100644 tokio-codec/README.md create mode 100644 tokio-codec/src/bytes_codec.rs create mode 100644 tokio-codec/src/lib.rs create mode 100644 tokio-codec/src/lines_codec.rs rename {tokio-io => tokio-codec}/tests/codecs.rs (96%) rename {tokio-io => tokio-codec}/tests/framed.rs (96%) rename {tokio-io => tokio-codec}/tests/framed_read.rs (98%) rename {tokio-io => tokio-codec}/tests/framed_write.rs (97%) create mode 100644 tokio-io/src/_tokio_codec/decoder.rs create mode 100644 tokio-io/src/_tokio_codec/encoder.rs create mode 100644 tokio-io/src/_tokio_codec/framed.rs create mode 100644 tokio-io/src/_tokio_codec/framed_read.rs create mode 100644 tokio-io/src/_tokio_codec/framed_write.rs create mode 100644 tokio-io/src/_tokio_codec/mod.rs diff --git a/Cargo.toml b/Cargo.toml index 1569d4eb6..70a215f17 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -23,6 +23,7 @@ keywords = ["io", "async", "non-blocking", "futures"] members = [ "./", + "tokio-codec", "tokio-executor", "tokio-fs", "tokio-io", @@ -39,6 +40,7 @@ travis-ci = { repository = "tokio-rs/tokio" } appveyor = { repository = "carllerche/tokio", id = "s83yxhy9qeb58va7" } [dependencies] +tokio-codec = { version = "0.1.0", path = "tokio-codec" } tokio-io = { version = "0.1.6", path = "tokio-io" } tokio-executor = { version = "0.1.2", path = "tokio-executor" } tokio-reactor = { version = "0.1.1", path = "tokio-reactor" } diff --git a/examples/connect.rs b/examples/connect.rs index f44a5c1ed..5a6b51514 100644 --- a/examples/connect.rs +++ b/examples/connect.rs @@ -17,6 +17,7 @@ #![deny(warnings)] extern crate tokio; +extern crate tokio_codec; extern crate tokio_io; extern crate futures; extern crate bytes; @@ -82,7 +83,7 @@ fn main() { mod codec { use std::io; use bytes::{BufMut, BytesMut}; - use tokio_io::codec::{Encoder, Decoder}; + use tokio_codec::{Encoder, Decoder}; /// A simple `Codec` implementation that just ships bytes around. /// @@ -120,6 +121,7 @@ mod codec { mod tcp { use tokio; + use tokio_codec::Decoder; use tokio::net::TcpStream; use tokio::prelude::*; @@ -151,7 +153,7 @@ mod tcp { // to the TCP stream. This is done to ensure that happens concurrently // with us reading data from the stream. Box::new(tcp.map(move |stream| { - let (sink, stream) = stream.framed(Bytes).split(); + let (sink, stream) = Bytes.framed(stream).split(); tokio::spawn(stdin.forward(sink).then(|result| { if let Err(e) = result { diff --git a/examples/print_each_packet.rs b/examples/print_each_packet.rs index bb054948d..5dc533240 100644 --- a/examples/print_each_packet.rs +++ b/examples/print_each_packet.rs @@ -55,9 +55,10 @@ #![deny(warnings)] extern crate tokio; +extern crate tokio_codec; extern crate tokio_io; -use tokio_io::codec::BytesCodec; +use tokio_codec::{Decoder, BytesCodec}; use tokio::net::TcpListener; use tokio::prelude::*; @@ -99,8 +100,8 @@ fn main() { // We're parsing each socket with the `BytesCodec` included in `tokio_io`, // and then we `split` each codec into the reader/writer halves. // - // See https://docs.rs/tokio-io/0.1/src/tokio_io/codec/bytes_codec.rs.html - let framed = socket.framed(BytesCodec::new()); + // See https://docs.rs/tokio-codec/0.1/src/tokio_codec/bytes_codec.rs.html + let framed = BytesCodec::new().framed(socket); let (_writer, reader) = framed.split(); let processor = reader diff --git a/examples/tinyhttp.rs b/examples/tinyhttp.rs index 9bf3f27a5..d56ff96fd 100644 --- a/examples/tinyhttp.rs +++ b/examples/tinyhttp.rs @@ -21,6 +21,7 @@ extern crate serde_derive; extern crate serde_json; extern crate time; extern crate tokio; +extern crate tokio_codec; extern crate tokio_io; use std::{env, fmt, io}; @@ -29,7 +30,7 @@ use std::net::SocketAddr; use tokio::net::{TcpStream, TcpListener}; use tokio::prelude::*; -use tokio_io::codec::{Encoder, Decoder}; +use tokio_codec::{Encoder, Decoder}; use bytes::BytesMut; use http::header::HeaderValue; @@ -55,10 +56,10 @@ fn main() { } fn process(socket: TcpStream) { - let (tx, rx) = socket + let (tx, rx) = // Frame the socket using the `Http` protocol. This maps the TCP socket // to a Stream + Sink of HTTP frames. - .framed(Http) + Http.framed(socket) // This splits a single `Stream + Sink` value into two separate handles // that can be used independently (even on different tasks or threads). .split(); diff --git a/examples/udp-codec.rs b/examples/udp-codec.rs index 063083e5b..b273a3606 100644 --- a/examples/udp-codec.rs +++ b/examples/udp-codec.rs @@ -9,6 +9,7 @@ #![deny(warnings)] extern crate tokio; +extern crate tokio_codec; extern crate tokio_io; extern crate env_logger; @@ -16,7 +17,7 @@ use std::net::SocketAddr; use tokio::prelude::*; use tokio::net::{UdpSocket, UdpFramed}; -use tokio_io::codec::BytesCodec; +use tokio_codec::BytesCodec; fn main() { let _ = env_logger::init(); diff --git a/tests/line-frames.rs b/tests/line-frames.rs index c6437d84d..4d42f5080 100644 --- a/tests/line-frames.rs +++ b/tests/line-frames.rs @@ -1,6 +1,7 @@ extern crate env_logger; extern crate futures; extern crate tokio; +extern crate tokio_codec; extern crate tokio_io; extern crate tokio_threadpool; extern crate bytes; @@ -11,9 +12,8 @@ use std::net::Shutdown; use bytes::{BytesMut, BufMut}; use futures::{Future, Stream, Sink}; use tokio::net::{TcpListener, TcpStream}; -use tokio_io::codec::{Encoder, Decoder}; +use tokio_codec::{Encoder, Decoder}; use tokio_io::io::{write_all, read}; -use tokio_io::AsyncRead; use tokio_threadpool::Builder; pub struct LineCodec; @@ -61,7 +61,7 @@ fn echo() { let addr = listener.local_addr().unwrap(); let sender = pool.sender().clone(); let srv = listener.incoming().for_each(move |socket| { - let (sink, stream) = socket.framed(LineCodec).split(); + let (sink, stream) = LineCodec.framed(socket).split(); sender.spawn(sink.send_all(stream).map(|_| ()).map_err(|_| ())).unwrap(); Ok(()) }); diff --git a/tokio-codec/CHANGELOG.md b/tokio-codec/CHANGELOG.md new file mode 100644 index 000000000..f96d5068a --- /dev/null +++ b/tokio-codec/CHANGELOG.md @@ -0,0 +1,3 @@ +# Unreleased + +* Initial release (#353) diff --git a/tokio-codec/Cargo.toml b/tokio-codec/Cargo.toml new file mode 100644 index 000000000..1d335b834 --- /dev/null +++ b/tokio-codec/Cargo.toml @@ -0,0 +1,22 @@ +[package] +name = "tokio-codec" + +# When releasing to crates.io: +# - Update html_root_url. +# - Update CHANGELOG.md. +# - Create "v0.1.x" git tag. +version = "0.1.0" +authors = ["Carl Lerche ", "Bryan Burgers "] +license = "MIT" +repository = "https://github.com/tokio-rs/tokio" +homepage = "https://tokio.rs" +documentation = "https://docs.rs/tokio-codec/0.1" +description = """ +Utilities for encoding and decoding frames. +""" +categories = ["asynchronous"] + +[dependencies] +tokio-io = { version = "0.1.6", path = "../tokio-io" } +bytes = "0.4.7" +futures = "0.1.18" diff --git a/tokio-codec/LICENSE b/tokio-codec/LICENSE new file mode 100644 index 000000000..38c1e27b8 --- /dev/null +++ b/tokio-codec/LICENSE @@ -0,0 +1,25 @@ +Copyright (c) 2018 Tokio Contributors + +Permission is hereby granted, free of charge, to any +person obtaining a copy of this software and associated +documentation files (the "Software"), to deal in the +Software without restriction, including without +limitation the rights to use, copy, modify, merge, +publish, distribute, sublicense, and/or sell copies of +the Software, and to permit persons to whom the Software +is furnished to do so, subject to the following +conditions: + +The above copyright notice and this permission notice +shall be included in all copies or substantial portions +of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF +ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED +TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A +PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT +SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY +CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION +OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR +IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER +DEALINGS IN THE SOFTWARE. diff --git a/tokio-codec/README.md b/tokio-codec/README.md new file mode 100644 index 000000000..e0c1a3858 --- /dev/null +++ b/tokio-codec/README.md @@ -0,0 +1,35 @@ +# tokio-codec + +Utilities for encoding and decoding frames. + +[Documentation](https://docs.rs/tokio-codec) + +## Usage + +First, add this to your `Cargo.toml`: + +```toml +[dependencies] +tokio-codec = "0.1" +``` + +Next, add this to your crate: + +```rust +extern crate tokio_codec; +``` + +You can find extensive documentation and examples about how to use this crate +online at [https://tokio.rs](https://tokio.rs). The [API +documentation](https://docs.rs/tokio-codec) is also a great place to get started +for the nitty-gritty. + +## License + +This project is licensed under the [MIT license](LICENSE). + +### Contribution + +Unless you explicitly state otherwise, any contribution intentionally submitted +for inclusion in Tokio by you, shall be licensed as MIT, without any additional +terms or conditions. diff --git a/tokio-codec/src/bytes_codec.rs b/tokio-codec/src/bytes_codec.rs new file mode 100644 index 000000000..d535aef68 --- /dev/null +++ b/tokio-codec/src/bytes_codec.rs @@ -0,0 +1,37 @@ +use bytes::{Bytes, BufMut, BytesMut}; +use tokio_io::_tokio_codec::{Encoder, Decoder}; +use std::io; + +/// A simple `Codec` implementation that just ships bytes around. +#[derive(Copy, Clone, Debug, Eq, PartialEq, Ord, PartialOrd, Hash)] +pub struct BytesCodec(()); + +impl BytesCodec { + /// Creates a new `BytesCodec` for shipping around raw bytes. + pub fn new() -> BytesCodec { BytesCodec(()) } +} + +impl Decoder for BytesCodec { + type Item = BytesMut; + type Error = io::Error; + + fn decode(&mut self, buf: &mut BytesMut) -> Result, io::Error> { + if buf.len() > 0 { + let len = buf.len(); + Ok(Some(buf.split_to(len))) + } else { + Ok(None) + } + } +} + +impl Encoder for BytesCodec { + type Item = Bytes; + type Error = io::Error; + + fn encode(&mut self, data: Bytes, buf: &mut BytesMut) -> Result<(), io::Error> { + buf.reserve(data.len()); + buf.put(data); + Ok(()) + } +} diff --git a/tokio-codec/src/lib.rs b/tokio-codec/src/lib.rs new file mode 100644 index 000000000..2b26b542b --- /dev/null +++ b/tokio-codec/src/lib.rs @@ -0,0 +1,32 @@ +//! 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)] +#![doc(html_root_url = "https://docs.rs/tokio-codec/0.1.0")] + +extern crate bytes; +extern crate tokio_io; + +mod bytes_codec; +mod lines_codec; + +pub use tokio_io::_tokio_codec::{ + Decoder, + Encoder, + Framed, + FramedParts, + FramedRead, + FramedWrite, +}; + +pub use bytes_codec::BytesCodec; +pub use lines_codec::LinesCodec; diff --git a/tokio-codec/src/lines_codec.rs b/tokio-codec/src/lines_codec.rs new file mode 100644 index 000000000..bf4135b8e --- /dev/null +++ b/tokio-codec/src/lines_codec.rs @@ -0,0 +1,89 @@ +use bytes::{BufMut, BytesMut}; +use tokio_io::_tokio_codec::{Encoder, Decoder}; +use std::{io, str}; + +/// A simple `Codec` implementation that splits up data into lines. +#[derive(Clone, Debug, Eq, PartialEq, Ord, PartialOrd, Hash)] +pub struct LinesCodec { + // Stored index of the next index to examine for a `\n` character. + // This is used to optimize searching. + // For example, if `decode` was called with `abc`, it would hold `3`, + // because that is the next index to examine. + // The next time `decode` is called with `abcde\n`, the method will + // only look at `de\n` before returning. + next_index: usize, +} + +impl LinesCodec { + /// Returns a `LinesCodec` for splitting up data into lines. + pub fn new() -> LinesCodec { + LinesCodec { next_index: 0 } + } +} + +fn utf8(buf: &[u8]) -> Result<&str, io::Error> { + str::from_utf8(buf).map_err(|_| + io::Error::new( + io::ErrorKind::InvalidData, + "Unable to decode input as UTF8")) +} + +fn without_carriage_return(s: &[u8]) -> &[u8] { + if let Some(&b'\r') = s.last() { + &s[..s.len() - 1] + } else { + s + } +} + +impl Decoder for LinesCodec { + type Item = String; + type Error = io::Error; + + fn decode(&mut self, buf: &mut BytesMut) -> Result, io::Error> { + if let Some(newline_offset) = + buf[self.next_index..].iter().position(|b| *b == b'\n') + { + let newline_index = newline_offset + self.next_index; + let line = buf.split_to(newline_index + 1); + let line = &line[..line.len()-1]; + let line = without_carriage_return(line); + let line = utf8(line)?; + self.next_index = 0; + Ok(Some(line.to_string())) + } else { + self.next_index = buf.len(); + Ok(None) + } + } + + fn decode_eof(&mut self, buf: &mut BytesMut) -> Result, io::Error> { + Ok(match self.decode(buf)? { + Some(frame) => Some(frame), + None => { + // No terminating newline - return remaining data, if any + if buf.is_empty() || buf == &b"\r"[..] { + None + } else { + let line = buf.take(); + let line = without_carriage_return(&line); + let line = utf8(line)?; + self.next_index = 0; + Some(line.to_string()) + } + } + }) + } +} + +impl Encoder for LinesCodec { + type Item = String; + type Error = io::Error; + + fn encode(&mut self, line: String, buf: &mut BytesMut) -> Result<(), io::Error> { + buf.reserve(line.len() + 1); + buf.put(line); + buf.put_u8(b'\n'); + Ok(()) + } +} diff --git a/tokio-io/tests/codecs.rs b/tokio-codec/tests/codecs.rs similarity index 96% rename from tokio-io/tests/codecs.rs rename to tokio-codec/tests/codecs.rs index 5d63242fd..6359e7c72 100644 --- a/tokio-io/tests/codecs.rs +++ b/tokio-codec/tests/codecs.rs @@ -1,8 +1,8 @@ -extern crate tokio_io; +extern crate tokio_codec; extern crate bytes; use bytes::{BytesMut, Bytes, BufMut}; -use tokio_io::codec::{BytesCodec, LinesCodec, Decoder, Encoder}; +use tokio_codec::{BytesCodec, LinesCodec, Decoder, Encoder}; #[test] fn bytes_decoder() { diff --git a/tokio-io/tests/framed.rs b/tokio-codec/tests/framed.rs similarity index 96% rename from tokio-io/tests/framed.rs rename to tokio-codec/tests/framed.rs index 660cc5d0e..fdedd560e 100644 --- a/tokio-io/tests/framed.rs +++ b/tokio-codec/tests/framed.rs @@ -1,10 +1,11 @@ +extern crate tokio_codec; extern crate tokio_io; extern crate bytes; extern crate futures; use futures::{Stream, Future}; use std::io::{self, Read}; -use tokio_io::codec::{Framed, FramedParts, Decoder, Encoder}; +use tokio_codec::{Framed, FramedParts, Decoder, Encoder}; use tokio_io::AsyncRead; use bytes::{BytesMut, Buf, BufMut, IntoBuf, BigEndian}; diff --git a/tokio-io/tests/framed_read.rs b/tokio-codec/tests/framed_read.rs similarity index 98% rename from tokio-io/tests/framed_read.rs rename to tokio-codec/tests/framed_read.rs index 0dd327372..80dfa5e50 100644 --- a/tokio-io/tests/framed_read.rs +++ b/tokio-codec/tests/framed_read.rs @@ -1,9 +1,10 @@ +extern crate tokio_codec; extern crate tokio_io; extern crate bytes; extern crate futures; use tokio_io::AsyncRead; -use tokio_io::codec::{FramedRead, Decoder}; +use tokio_codec::{FramedRead, Decoder}; use bytes::{BytesMut, Buf, IntoBuf, BigEndian}; use futures::Stream; diff --git a/tokio-io/tests/framed_write.rs b/tokio-codec/tests/framed_write.rs similarity index 97% rename from tokio-io/tests/framed_write.rs rename to tokio-codec/tests/framed_write.rs index 2db21b48d..137fb5be1 100644 --- a/tokio-io/tests/framed_write.rs +++ b/tokio-codec/tests/framed_write.rs @@ -1,9 +1,10 @@ +extern crate tokio_codec; extern crate tokio_io; extern crate bytes; extern crate futures; use tokio_io::AsyncWrite; -use tokio_io::codec::{Encoder, FramedWrite}; +use tokio_codec::{Encoder, FramedWrite}; use futures::{Sink, Poll}; use bytes::{BytesMut, BufMut, BigEndian}; diff --git a/tokio-fs/CHANGELOG.md b/tokio-fs/CHANGELOG.md index 91f5c842c..fb3e7dab5 100644 --- a/tokio-fs/CHANGELOG.md +++ b/tokio-fs/CHANGELOG.md @@ -1,3 +1,7 @@ +# Unreleased + +* Use `tokio-codec` in examples + # 0.1.0 (May 2, 2018) * Initial release diff --git a/tokio-fs/Cargo.toml b/tokio-fs/Cargo.toml index 922eca164..a241e0288 100644 --- a/tokio-fs/Cargo.toml +++ b/tokio-fs/Cargo.toml @@ -27,3 +27,4 @@ tokio-io = { version = "0.1.6", path = "../tokio-io" } rand = "0.4.2" tempdir = "0.3.7" tokio-io = { version = "0.1.6", path = "../tokio-io" } +tokio-codec = { version = "0.1.0", path = "../tokio-codec" } diff --git a/tokio-fs/examples/std-echo.rs b/tokio-fs/examples/std-echo.rs index cdfdf0080..83efa66e5 100644 --- a/tokio-fs/examples/std-echo.rs +++ b/tokio-fs/examples/std-echo.rs @@ -1,12 +1,13 @@ //! Echo everything received on STDIN to STDOUT. +#![deny(deprecated, warnings)] extern crate futures; extern crate tokio_fs; -extern crate tokio_io; +extern crate tokio_codec; extern crate tokio_threadpool; use tokio_fs::{stdin, stdout, stderr}; -use tokio_io::codec::{FramedRead, FramedWrite, LinesCodec}; +use tokio_codec::{FramedRead, FramedWrite, LinesCodec}; use tokio_threadpool::Builder; use futures::{Future, Stream, Sink}; diff --git a/tokio-io/CHANGELOG.md b/tokio-io/CHANGELOG.md index c96c4d1dc..428bdf005 100644 --- a/tokio-io/CHANGELOG.md +++ b/tokio-io/CHANGELOG.md @@ -1,3 +1,7 @@ +# Unreleased + +* Move `codec::{Encode, Decode, Framed*}` into `tokio-codec` (#353) + # 0.1.6 (March 09, 2018) * Add native endian builder fn to length_delimited (#144) diff --git a/tokio-io/src/_tokio_codec/decoder.rs b/tokio-io/src/_tokio_codec/decoder.rs new file mode 100644 index 000000000..9c9fbacb4 --- /dev/null +++ b/tokio-io/src/_tokio_codec/decoder.rs @@ -0,0 +1,3 @@ +// For now, we need to keep the implmentation of Encoder in tokio_io. + +pub use codec::Decoder; diff --git a/tokio-io/src/_tokio_codec/encoder.rs b/tokio-io/src/_tokio_codec/encoder.rs new file mode 100644 index 000000000..9cbe054d7 --- /dev/null +++ b/tokio-io/src/_tokio_codec/encoder.rs @@ -0,0 +1,3 @@ +// For now, we need to keep the implmentation of Encoder in tokio_io. + +pub use codec::Encoder; diff --git a/tokio-io/src/_tokio_codec/framed.rs b/tokio-io/src/_tokio_codec/framed.rs new file mode 100644 index 000000000..de360dba9 --- /dev/null +++ b/tokio-io/src/_tokio_codec/framed.rs @@ -0,0 +1,248 @@ +#![allow(deprecated)] + +use std::io::{self, Read, Write}; +use std::fmt; + +use {AsyncRead, AsyncWrite}; +use codec::{Decoder, Encoder}; +use super::framed_read::{framed_read2, framed_read2_with_buffer, FramedRead2}; +use super::framed_write::{framed_write2, framed_write2_with_buffer, FramedWrite2}; + +use futures::{Stream, Sink, StartSend, Poll}; +use bytes::{BytesMut}; + +/// 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 { + inner: FramedRead2>>, +} + +pub struct Fuse(pub T, pub 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. +/// +/// 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 framed(inner: T, codec: U) -> Framed + where T: AsyncRead + AsyncWrite, + U: Decoder + Encoder, +{ + Framed { + inner: framed_read2(framed_write2(Fuse(inner, codec))), + } +} + +impl Framed { + /// 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, codec: U) -> Framed + { + Framed { + inner: framed_read2_with_buffer(framed_write2_with_buffer(Fuse(parts.inner, codec), parts.writebuf), parts.readbuf), + } + } + + /// 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 + } + + /// 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 + } + + /// Consumes the `Frame`, returning its underlying I/O stream and the buffer + /// with unprocessed data. + /// + /// 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 { + let (inner, readbuf) = self.inner.into_parts(); + let (inner, writebuf) = inner.into_parts(); + FramedParts { inner: inner.0, readbuf: readbuf, writebuf: writebuf } + } + + /// Consumes the `Frame`, returning its underlying I/O stream and the buffer + /// with unprocessed data, and also the current codec state. + /// + /// 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. + /// + /// Note that this function will be removed once the codec has been + /// integrated into `FramedParts` in a new version (see + /// [#53](https://github.com/tokio-rs/tokio-io/pull/53)). + pub fn into_parts_and_codec(self) -> (FramedParts, U) { + let (inner, readbuf) = self.inner.into_parts(); + let (inner, writebuf) = inner.into_parts(); + (FramedParts { inner: inner.0, readbuf: readbuf, writebuf: writebuf }, inner.1) + } +} + +impl Stream for Framed + where T: AsyncRead, + U: Decoder, +{ + type Item = U::Item; + type Error = U::Error; + + fn poll(&mut self) -> Poll, Self::Error> { + self.inner.poll() + } +} + +impl Sink for Framed + where T: AsyncWrite, + U: Encoder, + U::Error: From, +{ + type SinkItem = U::Item; + type SinkError = U::Error; + + fn start_send(&mut self, + item: Self::SinkItem) + -> StartSend + { + 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 fmt::Debug for Framed + 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 Read for Fuse { + fn read(&mut self, dst: &mut [u8]) -> io::Result { + self.0.read(dst) + } +} + +impl AsyncRead for Fuse { + unsafe fn prepare_uninitialized_buffer(&self, buf: &mut [u8]) -> bool { + self.0.prepare_uninitialized_buffer(buf) + } +} + +impl Write for Fuse { + fn write(&mut self, src: &[u8]) -> io::Result { + self.0.write(src) + } + + fn flush(&mut self) -> io::Result<()> { + self.0.flush() + } +} + +impl AsyncWrite for Fuse { + fn shutdown(&mut self) -> Poll<(), io::Error> { + self.0.shutdown() + } +} + +impl Decoder for Fuse { + type Item = U::Item; + type Error = U::Error; + + fn decode(&mut self, buffer: &mut BytesMut) -> Result, Self::Error> { + self.1.decode(buffer) + } + + fn decode_eof(&mut self, buffer: &mut BytesMut) -> Result, Self::Error> { + self.1.decode_eof(buffer) + } +} + +impl Encoder for Fuse { + 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 +{ + /// The inner transport used to read bytes to and write bytes to + pub inner: T, + /// The buffer with read but unprocessed data. + pub readbuf: BytesMut, + /// A buffer with unprocessed data which are not written yet. + pub writebuf: BytesMut +} diff --git a/tokio-io/src/_tokio_codec/framed_read.rs b/tokio-io/src/_tokio_codec/framed_read.rs new file mode 100644 index 000000000..279b1a3bc --- /dev/null +++ b/tokio-io/src/_tokio_codec/framed_read.rs @@ -0,0 +1,214 @@ +#![allow(deprecated)] + +use std::fmt; + +use AsyncRead; +use codec::Decoder; +use super::framed::Fuse; + +use futures::{Async, Poll, Stream, Sink, StartSend}; +use bytes::BytesMut; + +/// A `Stream` of messages decoded from an `AsyncRead`. +pub struct FramedRead { + inner: FramedRead2>, +} + +pub struct FramedRead2 { + inner: T, + eof: bool, + is_readable: bool, + buffer: BytesMut, +} + +const INITIAL_CAPACITY: usize = 8 * 1024; + +// ===== impl FramedRead ===== + +impl FramedRead + where T: AsyncRead, + D: Decoder, +{ + /// Creates a new `FramedRead` with the given `decoder`. + pub fn new(inner: T, decoder: D) -> FramedRead { + FramedRead { + inner: framed_read2(Fuse(inner, decoder)), + } + } +} + +impl FramedRead { + /// 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 Stream for FramedRead + where T: AsyncRead, + D: Decoder, +{ + type Item = D::Item; + type Error = D::Error; + + fn poll(&mut self) -> Poll, Self::Error> { + self.inner.poll() + } +} + +impl Sink for FramedRead + where T: Sink, +{ + type SinkItem = T::SinkItem; + type SinkError = T::SinkError; + + fn start_send(&mut self, + item: Self::SinkItem) + -> StartSend + { + 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 fmt::Debug for FramedRead + 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(inner: T) -> FramedRead2 { + FramedRead2 { + inner: inner, + eof: false, + is_readable: false, + buffer: BytesMut::with_capacity(INITIAL_CAPACITY), + } +} + +pub fn framed_read2_with_buffer(inner: T, mut buf: BytesMut) -> FramedRead2 { + if buf.capacity() < INITIAL_CAPACITY { + let bytes_to_reserve = INITIAL_CAPACITY - buf.capacity(); + buf.reserve(bytes_to_reserve); + } + FramedRead2 { + inner: inner, + eof: false, + is_readable: buf.len() > 0, + buffer: buf, + } +} + +impl FramedRead2 { + 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 Stream for FramedRead2 + where T: AsyncRead + Decoder, +{ + type Item = T::Item; + type Error = T::Error; + + fn poll(&mut self) -> Poll, 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 = try!(self.inner.decode_eof(&mut self.buffer)); + return Ok(Async::Ready(frame)); + } + + trace!("attempting to decode a frame"); + + if let Some(frame) = try!(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; + } + } +} diff --git a/tokio-io/src/_tokio_codec/framed_write.rs b/tokio-io/src/_tokio_codec/framed_write.rs new file mode 100644 index 000000000..9a01e9139 --- /dev/null +++ b/tokio-io/src/_tokio_codec/framed_write.rs @@ -0,0 +1,237 @@ +#![allow(deprecated)] + +use std::io::{self, Read}; +use std::fmt; + +use {AsyncRead, AsyncWrite}; +use codec::{Decoder, Encoder}; +use super::framed::Fuse; + +use futures::{Async, AsyncSink, Poll, Stream, Sink, StartSend}; +use bytes::BytesMut; + +/// A `Sink` of frames encoded to an `AsyncWrite`. +pub struct FramedWrite { + inner: FramedWrite2>, +} + +pub struct FramedWrite2 { + inner: T, + buffer: BytesMut, +} + +const INITIAL_CAPACITY: usize = 8 * 1024; +const BACKPRESSURE_BOUNDARY: usize = INITIAL_CAPACITY; + +impl FramedWrite + where T: AsyncWrite, + E: Encoder, +{ + /// Creates a new `FramedWrite` with the given `encoder`. + pub fn new(inner: T, encoder: E) -> FramedWrite { + FramedWrite { + inner: framed_write2(Fuse(inner, encoder)), + } + } +} + +impl FramedWrite { + /// 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 + } +} + +impl Sink for FramedWrite + where T: AsyncWrite, + E: Encoder, +{ + type SinkItem = E::Item; + type SinkError = E::Error; + + fn start_send(&mut self, item: E::Item) -> StartSend { + 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(try!(self.inner.close())) + } +} + +impl Stream for FramedWrite + where T: Stream, +{ + type Item = T::Item; + type Error = T::Error; + + fn poll(&mut self) -> Poll, Self::Error> { + self.inner.inner.0.poll() + } +} + +impl fmt::Debug for FramedWrite + 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(inner: T) -> FramedWrite2 { + FramedWrite2 { + inner: inner, + buffer: BytesMut::with_capacity(INITIAL_CAPACITY), + } +} + +pub fn framed_write2_with_buffer(inner: T, mut buf: BytesMut) -> FramedWrite2 { + if buf.capacity() < INITIAL_CAPACITY { + let bytes_to_reserve = INITIAL_CAPACITY - buf.capacity(); + buf.reserve(bytes_to_reserve); + } + FramedWrite2 { + inner: inner, + buffer: buf, + } +} + +impl FramedWrite2 { + 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 Sink for FramedWrite2 + where T: AsyncWrite + Encoder, +{ + type SinkItem = T::Item; + type SinkError = T::Error; + + fn start_send(&mut self, item: T::Item) -> StartSend { + // If the buffer is already over 8KiB, then attempt to flush it. If after flushing it's + // *still* over 8KiB, then apply backpressure (reject the send). + if self.buffer.len() >= BACKPRESSURE_BOUNDARY { + try!(self.poll_complete()); + + if self.buffer.len() >= BACKPRESSURE_BOUNDARY { + return Ok(AsyncSink::NotReady(item)); + } + } + + try!(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"); + return Ok(Async::Ready(())); + } + + fn close(&mut self) -> Poll<(), Self::SinkError> { + try_ready!(self.poll_complete()); + Ok(try!(self.inner.shutdown())) + } +} + +impl Decoder for FramedWrite2 { + type Item = T::Item; + type Error = T::Error; + + fn decode(&mut self, src: &mut BytesMut) -> Result, T::Error> { + self.inner.decode(src) + } + + fn decode_eof(&mut self, src: &mut BytesMut) -> Result, T::Error> { + self.inner.decode_eof(src) + } +} + +impl Read for FramedWrite2 { + fn read(&mut self, dst: &mut [u8]) -> io::Result { + self.inner.read(dst) + } +} + +impl AsyncRead for FramedWrite2 { + unsafe fn prepare_uninitialized_buffer(&self, buf: &mut [u8]) -> bool { + self.inner.prepare_uninitialized_buffer(buf) + } +} diff --git a/tokio-io/src/_tokio_codec/mod.rs b/tokio-io/src/_tokio_codec/mod.rs new file mode 100644 index 000000000..7bee1f17c --- /dev/null +++ b/tokio-io/src/_tokio_codec/mod.rs @@ -0,0 +1,36 @@ +//! 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)] +#![doc(hidden, html_root_url = "https://docs.rs/tokio-codec/0.1.0")] + +// _tokio_codec are the items that belong in the `tokio_codec` crate. However, because we need to +// maintain backward compatibility until the next major breaking change, they are defined here. +// When the next breaking change comes, they should be moved to the `tokio_codec` crate and become +// independent. +// +// The primary reason we can't move these to `tokio-codec` now is because, again for backward +// compatibility reasons, we need to keep `Decoder` and `Encoder` in tokio_io::codec. And `Decoder` +// and `Encoder` needs to reference `Framed`. So they all still need to still be in the same +// module. + +mod decoder; +mod encoder; +mod framed; +mod framed_read; +mod framed_write; + +pub use self::decoder::Decoder; +pub use self::encoder::Encoder; +pub use self::framed::{framed, Framed, FramedParts}; +pub use self::framed_read::FramedRead; +pub use self::framed_write::FramedWrite; diff --git a/tokio-io/src/async_read.rs b/tokio-io/src/async_read.rs index 804aca7d0..3c2c3bab0 100644 --- a/tokio-io/src/async_read.rs +++ b/tokio-io/src/async_read.rs @@ -3,6 +3,7 @@ use bytes::BufMut; use futures::{Async, Poll}; use {framed, split, AsyncWrite}; +#[allow(deprecated)] use codec::{Decoder, Encoder, Framed}; use split::{ReadHalf, WriteHalf}; @@ -129,6 +130,8 @@ pub trait AsyncRead: std_io::Read { /// 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. + #[deprecated(since = "0.1.7", note = "Use tokio_codec::Decoder::framed instead")] + #[allow(deprecated)] fn framed(self, codec: T) -> Framed where Self: AsyncWrite + Sized, { diff --git a/tokio-io/src/codec/bytes_codec.rs b/tokio-io/src/codec/bytes_codec.rs index ce26b9e6c..c77f5ca53 100644 --- a/tokio-io/src/codec/bytes_codec.rs +++ b/tokio-io/src/codec/bytes_codec.rs @@ -1,9 +1,12 @@ +#![allow(deprecated)] + use bytes::{Bytes, BufMut, BytesMut}; use codec::{Encoder, Decoder}; use std::io; /// A simple `Codec` implementation that just ships bytes around. #[derive(Copy, Clone, Debug, Eq, PartialEq, Ord, PartialOrd, Hash)] +#[deprecated(since = "0.1.7", note = "Moved to tokio-codec")] pub struct BytesCodec(()); impl BytesCodec { diff --git a/tokio-io/src/codec/decoder.rs b/tokio-io/src/codec/decoder.rs index e4c3d79ee..a3c0b8c1c 100644 --- a/tokio-io/src/codec/decoder.rs +++ b/tokio-io/src/codec/decoder.rs @@ -1,6 +1,11 @@ use std::io; use bytes::BytesMut; +use {AsyncWrite, AsyncRead}; +use super::encoder::Encoder; + +use ::_tokio_codec::{framed, Framed}; + /// Decoding of frames via buffers. /// /// This trait is used when constructing an instance of `Framed` or @@ -11,6 +16,9 @@ use bytes::BytesMut; /// Implementations are able to track state on `self`, which enables /// implementing stateful streaming parsers. In many cases, though, this type /// will simply be a unit struct (e.g. `struct HttpDecoder`). + +// Note: We can't deprecate this trait, because the deprecation carries through to tokio-codec, and +// there doesn't seem to be a way to un-deprecate the re-export. pub trait Decoder { /// The type of decoded frames. type Item; @@ -83,4 +91,27 @@ pub trait Decoder { } } } + + /// 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. + fn framed(self, io: T) -> Framed + where Self: Encoder + Sized, + { + framed(io, self) + } } diff --git a/tokio-io/src/codec/encoder.rs b/tokio-io/src/codec/encoder.rs index 55056c1ce..222990d2e 100644 --- a/tokio-io/src/codec/encoder.rs +++ b/tokio-io/src/codec/encoder.rs @@ -3,6 +3,9 @@ use bytes::BytesMut; /// Trait of helper objects to write out messages as bytes, for use with /// `FramedWrite`. + +// Note: We can't deprecate this trait, because the deprecation carries through to tokio-codec, and +// there doesn't seem to be a way to un-deprecate the re-export. pub trait Encoder { /// The type of items consumed by the `Encoder` type Item; diff --git a/tokio-io/src/codec/lines_codec.rs b/tokio-io/src/codec/lines_codec.rs index 2e1bea4aa..7056d5c8c 100644 --- a/tokio-io/src/codec/lines_codec.rs +++ b/tokio-io/src/codec/lines_codec.rs @@ -1,9 +1,12 @@ +#![allow(deprecated)] + use bytes::{BufMut, BytesMut}; use codec::{Encoder, Decoder}; use std::{io, str}; /// A simple `Codec` implementation that splits up data into lines. #[derive(Clone, Debug, Eq, PartialEq, Ord, PartialOrd, Hash)] +#[deprecated(since = "0.1.7", note = "Moved to tokio-codec")] pub struct LinesCodec { // Stored index of the next index to examine for a `\n` character. // This is used to optimize searching. diff --git a/tokio-io/src/codec/mod.rs b/tokio-io/src/codec/mod.rs index 3a80d6a30..c4dab6564 100644 --- a/tokio-io/src/codec/mod.rs +++ b/tokio-io/src/codec/mod.rs @@ -10,6 +10,14 @@ //! [`Stream`]: # //! [transports]: # +// tokio_io::codec originally held all codec-related helpers. This is now intended to be in +// tokio_codec instead. However, for backward compatibility, this remains here. When the next major +// breaking change comes, `Encoder` and `Decoder` need to be moved to `tokio_codec`, and the rest +// of this module should be removed. + +#![doc(hidden)] +#![allow(deprecated)] + mod decoder; mod encoder; mod bytes_codec; diff --git a/tokio-io/src/framed.rs b/tokio-io/src/framed.rs index 5a3c54a80..7235b1c6d 100644 --- a/tokio-io/src/framed.rs +++ b/tokio-io/src/framed.rs @@ -1,3 +1,5 @@ +#![allow(deprecated)] + use std::io::{self, Read, Write}; use std::fmt; @@ -13,10 +15,14 @@ use bytes::{BytesMut}; /// the `Encoder` and `Decoder` traits to encode and decode frames. /// /// You can create a `Framed` instance by using the `AsyncRead::framed` adapter. +#[deprecated(since = "0.1.7", note = "Moved to tokio-codec")] +#[doc(hidden)] pub struct Framed { inner: FramedRead2>>, } +#[deprecated(since = "0.1.7", note = "Moved to tokio-codec")] +#[doc(hidden)] pub struct Fuse(pub T, pub U); pub fn framed(inner: T, codec: U) -> Framed diff --git a/tokio-io/src/framed_read.rs b/tokio-io/src/framed_read.rs index e04e3acfc..69b7f4631 100644 --- a/tokio-io/src/framed_read.rs +++ b/tokio-io/src/framed_read.rs @@ -1,3 +1,5 @@ +#![allow(deprecated)] + use std::fmt; use AsyncRead; @@ -8,10 +10,14 @@ use futures::{Async, Poll, Stream, Sink, StartSend}; use bytes::BytesMut; /// A `Stream` of messages decoded from an `AsyncRead`. +#[deprecated(since = "0.1.7", note = "Moved to tokio-codec")] +#[doc(hidden)] pub struct FramedRead { inner: FramedRead2>, } +#[deprecated(since = "0.1.7", note = "Moved to tokio-codec")] +#[doc(hidden)] pub struct FramedRead2 { inner: T, eof: bool, diff --git a/tokio-io/src/framed_write.rs b/tokio-io/src/framed_write.rs index 8c60426c3..392300ea9 100644 --- a/tokio-io/src/framed_write.rs +++ b/tokio-io/src/framed_write.rs @@ -1,3 +1,5 @@ +#![allow(deprecated)] + use std::io::{self, Read}; use std::fmt; @@ -9,10 +11,14 @@ use futures::{Async, AsyncSink, Poll, Stream, Sink, StartSend}; use bytes::BytesMut; /// A `Sink` of frames encoded to an `AsyncWrite`. +#[deprecated(since = "0.1.7", note = "Moved to tokio-codec")] +#[doc(hidden)] pub struct FramedWrite { inner: FramedWrite2>, } +#[deprecated(since = "0.1.7", note = "Moved to tokio-codec")] +#[doc(hidden)] pub struct FramedWrite2 { inner: T, buffer: BytesMut, diff --git a/tokio-io/src/length_delimited.rs b/tokio-io/src/length_delimited.rs index b9f847622..6b552f2b5 100644 --- a/tokio-io/src/length_delimited.rs +++ b/tokio-io/src/length_delimited.rs @@ -1,3 +1,5 @@ +#![allow(deprecated)] + use {codec, AsyncRead, AsyncWrite}; use bytes::{Buf, BufMut, BytesMut, IntoBuf}; diff --git a/tokio-io/src/lib.rs b/tokio-io/src/lib.rs index f73785a43..ef9e5d16a 100644 --- a/tokio-io/src/lib.rs +++ b/tokio-io/src/lib.rs @@ -56,6 +56,7 @@ mod length_delimited; mod lines; mod split; mod window; +pub mod _tokio_codec; pub use self::async_read::AsyncRead; pub use self::async_write::AsyncWrite; diff --git a/tokio-udp/Cargo.toml b/tokio-udp/Cargo.toml index 5a74c6a6d..f3cdab730 100644 --- a/tokio-udp/Cargo.toml +++ b/tokio-udp/Cargo.toml @@ -17,6 +17,7 @@ UDP bindings for tokio. categories = ["asynchronous"] [dependencies] +tokio-codec = { version = "0.1.0", path = "../tokio-codec" } tokio-io = { version = "0.1.6", path = "../tokio-io" } tokio-reactor = { version = "0.1.1", path = "../tokio-reactor" } bytes = "0.4" diff --git a/tokio-udp/src/frame.rs b/tokio-udp/src/frame.rs index 83e204d8f..37097ca37 100644 --- a/tokio-udp/src/frame.rs +++ b/tokio-udp/src/frame.rs @@ -5,7 +5,7 @@ use futures::{Async, Poll, Stream, Sink, StartSend, AsyncSink}; use super::UdpSocket; -use tokio_io::codec::{Decoder, Encoder}; +use tokio_codec::{Decoder, Encoder}; use bytes::{BytesMut, BufMut}; /// A unified `Stream` and `Sink` interface to an underlying `UdpSocket`, using diff --git a/tokio-udp/src/lib.rs b/tokio-udp/src/lib.rs index b5f4abcf5..8e9977ec7 100644 --- a/tokio-udp/src/lib.rs +++ b/tokio-udp/src/lib.rs @@ -25,6 +25,7 @@ extern crate futures; extern crate mio; #[macro_use] extern crate log; +extern crate tokio_codec; extern crate tokio_io; extern crate tokio_reactor; diff --git a/tokio-udp/tests/udp.rs b/tokio-udp/tests/udp.rs index 0eb2f1809..ef70ed23f 100644 --- a/tokio-udp/tests/udp.rs +++ b/tokio-udp/tests/udp.rs @@ -1,7 +1,6 @@ -#![allow(deprecated)] - extern crate futures; extern crate tokio_udp; +extern crate tokio_codec; #[macro_use] extern crate tokio_io; extern crate bytes; @@ -13,7 +12,7 @@ use std::net::SocketAddr; use futures::{Future, Poll, Stream, Sink}; use tokio_udp::{UdpSocket, UdpFramed}; -use tokio_io::codec::{Encoder, Decoder}; +use tokio_codec::{Encoder, Decoder}; use bytes::{BytesMut, BufMut}; macro_rules! t {