diff --git a/.appveyor.yml b/.appveyor.yml index c54fc1d39..31ba12150 100644 --- a/.appveyor.yml +++ b/.appveyor.yml @@ -13,4 +13,4 @@ install: build: false test_script: - - cargo test --target %TARGET% + - cargo test --all --target %TARGET% diff --git a/.travis.yml b/.travis.yml index 4a6a346c4..c79ff7ceb 100644 --- a/.travis.yml +++ b/.travis.yml @@ -1,28 +1,34 @@ +--- language: rust sudo: false matrix: include: - - rust: 1.21.0 - - rust: stable - - os: osx - - rust: beta - - rust: nightly - before_script: - - pip install 'travis-cargo<0.2' --user && export PATH=$HOME/.local/bin:$PATH - script: - - cargo doc --no-deps --all-features - after_success: - - travis-cargo --only nightly doc-upload + - rust: 1.21.0 + - rust: stable + before_deploy: cargo doc --all --no-deps + - os: osx + - rust: beta + - rust: nightly script: - - cargo test +- cargo test --all + +deploy: + provider: pages + skip_cleanup: true + github_token: $GH_TOKEN + target_branch: gh-pages + local_dir: target/doc + on: + branch: master + repo: tokio-rs/tokio + rust: stable env: global: - secure: iwlN1zfUCp/5BAAheqIRSFIqiM9zSwfIGcVDw/V7jHveqXyNzmCs7H58/cd90WLqonqpPX0t5GF66oTjms4v0DFjgXr/k4358qeSZaV082V3baNrVpCDHeCQV0SvKsfiYxDDJGSUL1WIUP+tqqDm4+ksZQP3LnwZojkABjWz5CBNt4kX+Wz5ZbYqtQoxyuZba5UyPY2CXJtubvCVPGMJULuUpklYxXZ4dWM2olzGgVJ8rE8udhSZ4ER4JgxB0KUx3/5TwHHzgyPEsWR4bKN6JzBjIczQofXUcUXXdoZBs23H/VhCpzKcn3/oJ8btVYPzwtdj5FmVB1aVR/gjPo2bSGi/sofq+LwL/1HJXkM+kjl8m2dLLcDBKqNYNERtVA1++LhkMWAFRgGYe8v8Ryxjiue1NF5LgAIA/fjK0uI1DELTzTf/TKrM+AtPDNTvhOft4/YD+hoImjwk6nv6PBb2TiTYnc79Qf4AZ65tv1qtsAUPuw4plLaccHQAO4ldYVXn4u9c+iisJwvovs6jo06bF3U3qtdI5gXsrI9+T25TrXvYb+IREo0MHzYEM0KlPFnscEArzC3eajuSd36ARFP3lDc+gp2RPs89iJjowms0eRyepp7Cu6XO3Cd2pfAX8AqvnmttZf4Nm51ONeiBPXPXItUkJm49MCpMJywU1IZcWZg= - notifications: email: on_success: never diff --git a/Cargo.toml b/Cargo.toml index 0ad15e9ea..b4fbe13e6 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -18,6 +18,13 @@ the rest of the tokio crates. """ categories = ["asynchronous"] +[workspace] + +members = [ + "./", + "tokio-io", +] + [badges] travis-ci = { repository = "tokio-rs/tokio" } appveyor = { repository = "alexcrichton/tokio" } diff --git a/tokio-io/CHANGELOG.md b/tokio-io/CHANGELOG.md new file mode 100644 index 000000000..c56d8a84e --- /dev/null +++ b/tokio-io/CHANGELOG.md @@ -0,0 +1,26 @@ +# 0.1.4 (November 10, 2017) + +* Use `FrameTooBig` as length delimited error type (#70). +* Provide `Bytes` and `Lines` codecs (#78). +* Provide `AllowStdIo` wrapper (#76). + +# 0.1.3 (August 14, 2017) + +* Fix bug involving zero sized writes in copy helper (#57). +* Add get / set accessors for length delimited max frame length setting. (#65). +* Add `Framed::into_parts_and_codec` (#59). + +# 0.1.2 (May 23, 2017) + +* Add `from_parts` and `into_parts` to the framing combinators. +* Support passing an initialized buffer to the framing combinators. +* Add `length_adjustment` support to length delimited encoding (#48). + +# 0.1.1 (March 22, 2017) + +* Add some omitted `Self: Sized` bounds. +* Add missing "inner" fns. + +# 0.1.0 (March 15, 2017) + +* Initial release diff --git a/tokio-io/Cargo.toml b/tokio-io/Cargo.toml new file mode 100644 index 000000000..aed93cb16 --- /dev/null +++ b/tokio-io/Cargo.toml @@ -0,0 +1,17 @@ +[package] +name = "tokio-io" +version = "0.1.4" +authors = ["Alex Crichton ", "Carl Lerche "] +license = "MIT/Apache-2.0" +repository = "https://github.com/tokio-rs/tokio-io" +homepage = "https://tokio.rs" +documentation = "https://docs.rs/tokio-io/0.1" +description = """ +Core I/O primitives for asynchronous I/O in Rust. +""" +categories = ["asynchronous"] + +[dependencies] +bytes = "0.4" +futures = "0.1.11" +log = "0.4" diff --git a/tokio-io/LICENSE-APACHE b/tokio-io/LICENSE-APACHE new file mode 100644 index 000000000..16fe87b06 --- /dev/null +++ b/tokio-io/LICENSE-APACHE @@ -0,0 +1,201 @@ + Apache License + Version 2.0, January 2004 + http://www.apache.org/licenses/ + +TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION + +1. Definitions. + + "License" shall mean the terms and conditions for use, reproduction, + and distribution as defined by Sections 1 through 9 of this document. + + "Licensor" shall mean the copyright owner or entity authorized by + the copyright owner that is granting the License. + + "Legal Entity" shall mean the union of the acting entity and all + other entities that control, are controlled by, or are under common + control with that entity. For the purposes of this definition, + "control" means (i) the power, direct or indirect, to cause the + direction or management of such entity, whether by contract or + otherwise, or (ii) ownership of fifty percent (50%) or more of the + outstanding shares, or (iii) beneficial ownership of such entity. + + "You" (or "Your") shall mean an individual or Legal Entity + exercising permissions granted by this License. + + "Source" form shall mean the preferred form for making modifications, + including but not limited to software source code, documentation + source, and configuration files. + + "Object" form shall mean any form resulting from mechanical + transformation or translation of a Source form, including but + not limited to compiled object code, generated documentation, + and conversions to other media types. + + "Work" shall mean the work of authorship, whether in Source or + Object form, made available under the License, as indicated by a + copyright notice that is included in or attached to the work + (an example is provided in the Appendix below). + + "Derivative Works" shall mean any work, whether in Source or Object + form, that is based on (or derived from) the Work and for which the + editorial revisions, annotations, elaborations, or other modifications + represent, as a whole, an original work of authorship. For the purposes + of this License, Derivative Works shall not include works that remain + separable from, or merely link (or bind by name) to the interfaces of, + the Work and Derivative Works thereof. + + "Contribution" shall mean any work of authorship, including + the original version of the Work and any modifications or additions + to that Work or Derivative Works thereof, that is intentionally + submitted to Licensor for inclusion in the Work by the copyright owner + or by an individual or Legal Entity authorized to submit on behalf of + the copyright owner. For the purposes of this definition, "submitted" + means any form of electronic, verbal, or written communication sent + to the Licensor or its representatives, including but not limited to + communication on electronic mailing lists, source code control systems, + and issue tracking systems that are managed by, or on behalf of, the + Licensor for the purpose of discussing and improving the Work, but + excluding communication that is conspicuously marked or otherwise + designated in writing by the copyright owner as "Not a Contribution." + + "Contributor" shall mean Licensor and any individual or Legal Entity + on behalf of whom a Contribution has been received by Licensor and + subsequently incorporated within the Work. + +2. Grant of Copyright License. Subject to the terms and conditions of + this License, each Contributor hereby grants to You a perpetual, + worldwide, non-exclusive, no-charge, royalty-free, irrevocable + copyright license to reproduce, prepare Derivative Works of, + publicly display, publicly perform, sublicense, and distribute the + Work and such Derivative Works in Source or Object form. + +3. Grant of Patent License. Subject to the terms and conditions of + this License, each Contributor hereby grants to You a perpetual, + worldwide, non-exclusive, no-charge, royalty-free, irrevocable + (except as stated in this section) patent license to make, have made, + use, offer to sell, sell, import, and otherwise transfer the Work, + where such license applies only to those patent claims licensable + by such Contributor that are necessarily infringed by their + Contribution(s) alone or by combination of their Contribution(s) + with the Work to which such Contribution(s) was submitted. If You + institute patent litigation against any entity (including a + cross-claim or counterclaim in a lawsuit) alleging that the Work + or a Contribution incorporated within the Work constitutes direct + or contributory patent infringement, then any patent licenses + granted to You under this License for that Work shall terminate + as of the date such litigation is filed. + +4. Redistribution. You may reproduce and distribute copies of the + Work or Derivative Works thereof in any medium, with or without + modifications, and in Source or Object form, provided that You + meet the following conditions: + + (a) You must give any other recipients of the Work or + Derivative Works a copy of this License; and + + (b) You must cause any modified files to carry prominent notices + stating that You changed the files; and + + (c) You must retain, in the Source form of any Derivative Works + that You distribute, all copyright, patent, trademark, and + attribution notices from the Source form of the Work, + excluding those notices that do not pertain to any part of + the Derivative Works; and + + (d) If the Work includes a "NOTICE" text file as part of its + distribution, then any Derivative Works that You distribute must + include a readable copy of the attribution notices contained + within such NOTICE file, excluding those notices that do not + pertain to any part of the Derivative Works, in at least one + of the following places: within a NOTICE text file distributed + as part of the Derivative Works; within the Source form or + documentation, if provided along with the Derivative Works; or, + within a display generated by the Derivative Works, if and + wherever such third-party notices normally appear. The contents + of the NOTICE file are for informational purposes only and + do not modify the License. You may add Your own attribution + notices within Derivative Works that You distribute, alongside + or as an addendum to the NOTICE text from the Work, provided + that such additional attribution notices cannot be construed + as modifying the License. + + You may add Your own copyright statement to Your modifications and + may provide additional or different license terms and conditions + for use, reproduction, or distribution of Your modifications, or + for any such Derivative Works as a whole, provided Your use, + reproduction, and distribution of the Work otherwise complies with + the conditions stated in this License. + +5. Submission of Contributions. Unless You explicitly state otherwise, + any Contribution intentionally submitted for inclusion in the Work + by You to the Licensor shall be under the terms and conditions of + this License, without any additional terms or conditions. + Notwithstanding the above, nothing herein shall supersede or modify + the terms of any separate license agreement you may have executed + with Licensor regarding such Contributions. + +6. Trademarks. This License does not grant permission to use the trade + names, trademarks, service marks, or product names of the Licensor, + except as required for reasonable and customary use in describing the + origin of the Work and reproducing the content of the NOTICE file. + +7. Disclaimer of Warranty. Unless required by applicable law or + agreed to in writing, Licensor provides the Work (and each + Contributor provides its Contributions) on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or + implied, including, without limitation, any warranties or conditions + of TITLE, NON-INFRINGEMENT, MERCHANTABILITY, or FITNESS FOR A + PARTICULAR PURPOSE. You are solely responsible for determining the + appropriateness of using or redistributing the Work and assume any + risks associated with Your exercise of permissions under this License. + +8. Limitation of Liability. In no event and under no legal theory, + whether in tort (including negligence), contract, or otherwise, + unless required by applicable law (such as deliberate and grossly + negligent acts) or agreed to in writing, shall any Contributor be + liable to You for damages, including any direct, indirect, special, + incidental, or consequential damages of any character arising as a + result of this License or out of the use or inability to use the + Work (including but not limited to damages for loss of goodwill, + work stoppage, computer failure or malfunction, or any and all + other commercial damages or losses), even if such Contributor + has been advised of the possibility of such damages. + +9. Accepting Warranty or Additional Liability. While redistributing + the Work or Derivative Works thereof, You may choose to offer, + and charge a fee for, acceptance of support, warranty, indemnity, + or other liability obligations and/or rights consistent with this + License. However, in accepting such obligations, You may act only + on Your own behalf and on Your sole responsibility, not on behalf + of any other Contributor, and only if You agree to indemnify, + defend, and hold each Contributor harmless for any liability + incurred by, or claims asserted against, such Contributor by reason + of your accepting any such warranty or additional liability. + +END OF TERMS AND CONDITIONS + +APPENDIX: How to apply the Apache License to your work. + + To apply the Apache License to your work, attach the following + boilerplate notice, with the fields enclosed by brackets "[]" + replaced with your own identifying information. (Don't include + the brackets!) The text should be enclosed in the appropriate + comment syntax for the file format. We also recommend that a + file or class name and description of purpose be included on the + same "printed page" as the copyright notice for easier + identification within third-party archives. + +Copyright [yyyy] [name of copyright owner] + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. diff --git a/tokio-io/LICENSE-MIT b/tokio-io/LICENSE-MIT new file mode 100644 index 000000000..2a7bdc066 --- /dev/null +++ b/tokio-io/LICENSE-MIT @@ -0,0 +1,25 @@ +Copyright (c) 2017 Tokio Authors + +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-io/README.md b/tokio-io/README.md new file mode 100644 index 000000000..68e05406d --- /dev/null +++ b/tokio-io/README.md @@ -0,0 +1,45 @@ +# tokio-io + +Core I/O abstractions for the Tokio stack. + +[![Build Status](https://travis-ci.org/tokio-rs/tokio-io.svg?branch=master)](https://travis-ci.org/tokio-rs/tokio-io) + +[Documentation](https://docs.rs/tokio-io) + +## Usage + +First, add this to your `Cargo.toml`: + +```toml +[dependencies] +tokio-io = "0.1" +``` + +Next, add this to your crate: + +```rust +extern crate tokio_io; +``` + +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-io) is also a great place to get started +for the nitty-gritty. + + +# License + +This project is licensed under either of + + * Apache License, Version 2.0, ([LICENSE-APACHE](LICENSE-APACHE) or + http://www.apache.org/licenses/LICENSE-2.0) + * MIT license ([LICENSE-MIT](LICENSE-MIT) or + http://opensource.org/licenses/MIT) + +at your option. + +### Contribution + +Unless you explicitly state otherwise, any contribution intentionally submitted +for inclusion in Tokio by you, as defined in the Apache-2.0 license, shall be +dual licensed as above, without any additional terms or conditions. diff --git a/tokio-io/src/allow_std.rs b/tokio-io/src/allow_std.rs new file mode 100644 index 000000000..d57eaa89b --- /dev/null +++ b/tokio-io/src/allow_std.rs @@ -0,0 +1,81 @@ +use {AsyncRead, AsyncWrite}; +use futures::{Async, Poll}; +use std::{fmt, io}; + +/// A simple wrapper type which allows types which implement only +/// implement `std::io::Read` or `std::io::Write` +/// to be used in contexts which expect an `AsyncRead` or `AsyncWrite`. +/// +/// If these types issue an error with the kind `io::ErrorKind::WouldBlock`, +/// it is expected that they will notify the current task on readiness. +/// Synchronous `std` types should not issue errors of this kind and +/// are safe to use in this context. However, using these types with +/// `AllowStdIo` will cause the event loop to block, so they should be used +/// with care. +#[derive(Debug, Copy, Clone, Eq, PartialEq, Ord, PartialOrd, Hash)] +pub struct AllowStdIo(T); + +impl AllowStdIo { + /// Creates a new `AllowStdIo` from an existing IO object. + pub fn new(io: T) -> Self { + AllowStdIo(io) + } + + /// Returns a reference to the contained IO object. + pub fn get_ref(&self) -> &T { + &self.0 + } + + /// Returns a mutable reference to the contained IO object. + pub fn get_mut(&mut self) -> &mut T { + &mut self.0 + } + + /// Consumes self and returns the contained IO object. + pub fn into_inner(self) -> T { + self.0 + } +} + +impl io::Write for AllowStdIo where T: io::Write { + fn write(&mut self, buf: &[u8]) -> io::Result { + self.0.write(buf) + } + fn flush(&mut self) -> io::Result<()> { + self.0.flush() + } + fn write_all(&mut self, buf: &[u8]) -> io::Result<()> { + self.0.write_all(buf) + } + fn write_fmt(&mut self, fmt: fmt::Arguments) -> io::Result<()> { + self.0.write_fmt(fmt) + } +} + +impl AsyncWrite for AllowStdIo where T: io::Write { + fn shutdown(&mut self) -> Poll<(), io::Error> { + Ok(Async::Ready(())) + } +} + +impl io::Read for AllowStdIo where T: io::Read { + fn read(&mut self, buf: &mut [u8]) -> io::Result { + self.0.read(buf) + } + // TODO: implement the `initializer` fn when it stabilizes. + // See rust-lang/rust #42788 + fn read_to_end(&mut self, buf: &mut Vec) -> io::Result { + self.0.read_to_end(buf) + } + fn read_to_string(&mut self, buf: &mut String) -> io::Result { + self.0.read_to_string(buf) + } + fn read_exact(&mut self, buf: &mut [u8]) -> io::Result<()> { + self.0.read_exact(buf) + } +} + +impl AsyncRead for AllowStdIo where T: io::Read { + // TODO: override prepare_unitialized_buffer once `Read::initializer` is stable. + // See rust-lang/rust #42788 +} diff --git a/tokio-io/src/codec.rs b/tokio-io/src/codec.rs new file mode 100644 index 000000000..aa91f224d --- /dev/null +++ b/tokio-io/src/codec.rs @@ -0,0 +1,359 @@ +//! 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]: # + +pub use codecs::{BytesCodec, LinesCodec}; +pub use framed::{Framed, FramedParts}; +pub use framed_read::{FramedRead, Decoder}; +pub use framed_write::{FramedWrite, Encoder}; + +pub mod length_delimited { + //! Frame a stream of bytes based on a length prefix + //! + //! Many protocols delimit their frames by prefacing frame data with a + //! frame head that specifies the length of the frame. The + //! `length_delimited` module provides utilities for handling the length + //! based framing. This allows the consumer to work with entire frames + //! without having to worry about buffering or other framing logic. + //! + //! # Getting started + //! + //! If implementing a protocol from scratch, using length delimited framing + //! is an easy way to get started. [`Framed::new()`] will adapt a + //! full-duplex byte stream with a length delimited framer using default + //! configuration values. + //! + //! ``` + //! use tokio_io::{AsyncRead, AsyncWrite}; + //! use tokio_io::codec::length_delimited; + //! + //! fn bind_transport(io: T) + //! -> length_delimited::Framed + //! { + //! length_delimited::Framed::new(io) + //! } + //! ``` + //! + //! The returned transport implements `Sink + Stream` for `BytesMut`. It + //! encodes the frame with a big-endian `u32` header denoting the frame + //! payload length: + //! + //! ```text + //! +----------+--------------------------------+ + //! | len: u32 | frame payload | + //! +----------+--------------------------------+ + //! ``` + //! + //! Specifically, given the following: + //! + //! ``` + //! # extern crate tokio_io; + //! # extern crate bytes; + //! # extern crate futures; + //! # + //! use tokio_io::{AsyncRead, AsyncWrite}; + //! use tokio_io::codec::length_delimited; + //! use bytes::BytesMut; + //! use futures::{Sink, Future}; + //! + //! fn write_frame(io: T) { + //! let mut transport = length_delimited::Framed::new(io); + //! let frame = BytesMut::from("hello world"); + //! + //! transport.send(frame).wait().unwrap(); + //! } + //! # + //! # pub fn main() {} + //! ``` + //! + //! The encoded frame will look like this: + //! + //! ```text + //! +---- len: u32 ----+---- data ----+ + //! | \x00\x00\x00\x0b | hello world | + //! +------------------+--------------+ + //! ``` + //! + //! # Decoding + //! + //! [`FramedRead`] adapts an [`AsyncRead`] into a `Stream` of [`BytesMut`], + //! such that each yielded [`BytesMut`] value contains the contents of an + //! entire frame. There are many configuration parameters enabling + //! [`FrameRead`] to handle a wide range of protocols. Here are some + //! examples that will cover the various options at a high level. + //! + //! ## Example 1 + //! + //! The following will parse a `u16` length field at offset 0, including the + //! frame head in the yielded `BytesMut`. + //! + //! ``` + //! # use tokio_io::AsyncRead; + //! # use tokio_io::codec::length_delimited; + //! # fn bind_read(io: T) { + //! length_delimited::Builder::new() + //! .length_field_offset(0) // default value + //! .length_field_length(2) + //! .length_adjustment(0) // default value + //! .num_skip(0) // Do not strip frame header + //! .new_read(io); + //! # } + //! ``` + //! + //! The following frame will be decoded as such: + //! + //! ```text + //! INPUT DECODED + //! +-- len ---+--- Payload ---+ +-- len ---+--- Payload ---+ + //! | \x00\x0B | Hello world | --> | \x00\x0B | Hello world | + //! +----------+---------------+ +----------+---------------+ + //! ``` + //! + //! The value of the length field is 11 (`\x0B`) which represents the length + //! of the payload, `hello world`. By default, [`FramedRead`] assumes that + //! the length field represents the number of bytes that **follows** the + //! length field. Thus, the entire frame has a length of 13: 2 bytes for the + //! frame head + 11 bytes for the payload. + //! + //! ## Example 2 + //! + //! The following will parse a `u16` length field at offset 0, omitting the + //! frame head in the yielded `BytesMut`. + //! + //! ``` + //! # use tokio_io::AsyncRead; + //! # use tokio_io::codec::length_delimited; + //! # fn bind_read(io: T) { + //! length_delimited::Builder::new() + //! .length_field_offset(0) // default value + //! .length_field_length(2) + //! .length_adjustment(0) // default value + //! // `num_skip` is not needed, the default is to skip + //! .new_read(io); + //! # } + //! ``` + //! + //! The following frame will be decoded as such: + //! + //! ```text + //! INPUT DECODED + //! +-- len ---+--- Payload ---+ +--- Payload ---+ + //! | \x00\x0B | Hello world | --> | Hello world | + //! +----------+---------------+ +---------------+ + //! ``` + //! + //! This is similar to the first example, the only difference is that the + //! frame head is **not** included in the yielded `BytesMut` value. + //! + //! ## Example 3 + //! + //! The following will parse a `u16` length field at offset 0, including the + //! frame head in the yielded `BytesMut`. In this case, the length field + //! **includes** the frame head length. + //! + //! ``` + //! # use tokio_io::AsyncRead; + //! # use tokio_io::codec::length_delimited; + //! # fn bind_read(io: T) { + //! length_delimited::Builder::new() + //! .length_field_offset(0) // default value + //! .length_field_length(2) + //! .length_adjustment(-2) // size of head + //! .num_skip(0) + //! .new_read(io); + //! # } + //! ``` + //! + //! The following frame will be decoded as such: + //! + //! ```text + //! INPUT DECODED + //! +-- len ---+--- Payload ---+ +-- len ---+--- Payload ---+ + //! | \x00\x0D | Hello world | --> | \x00\x0D | Hello world | + //! +----------+---------------+ +----------+---------------+ + //! ``` + //! + //! In most cases, the length field represents the length of the payload + //! only, as shown in the previous examples. However, in some protocols the + //! length field represents the length of the whole frame, including the + //! head. In such cases, we specify a negative `length_adjustment` to adjust + //! the value provided in the frame head to represent the payload length. + //! + //! ## Example 4 + //! + //! The following will parse a 3 byte length field at offset 0 in a 5 byte + //! frame head, including the frame head in the yielded `BytesMut`. + //! + //! ``` + //! # use tokio_io::AsyncRead; + //! # use tokio_io::codec::length_delimited; + //! # fn bind_read(io: T) { + //! length_delimited::Builder::new() + //! .length_field_offset(0) // default value + //! .length_field_length(3) + //! .length_adjustment(2) // remaining head + //! .num_skip(0) + //! .new_read(io); + //! # } + //! ``` + //! + //! The following frame will be decoded as such: + //! + //! ```text + //! INPUT + //! +---- len -----+- head -+--- Payload ---+ + //! | \x00\x00\x0B | \xCAFE | Hello world | + //! +--------------+--------+---------------+ + //! + //! DECODED + //! +---- len -----+- head -+--- Payload ---+ + //! | \x00\x00\x0B | \xCAFE | Hello world | + //! +--------------+--------+---------------+ + //! ``` + //! + //! A more advanced example that shows a case where there is extra frame + //! head data between the length field and the payload. In such cases, it is + //! usually desirable to include the frame head as part of the yielded + //! `BytesMut`. This lets consumers of the length delimited framer to + //! process the frame head as needed. + //! + //! The positive `length_adjustment` value lets `FramedRead` factor in the + //! additional head into the frame length calculation. + //! + //! ## Example 5 + //! + //! The following will parse a `u16` length field at offset 1 of a 4 byte + //! frame head. The first byte and the length field will be omitted from the + //! yielded `BytesMut`, but the trailing 2 bytes of the frame head will be + //! included. + //! + //! ``` + //! # use tokio_io::AsyncRead; + //! # use tokio_io::codec::length_delimited; + //! # fn bind_read(io: T) { + //! length_delimited::Builder::new() + //! .length_field_offset(1) // length of hdr1 + //! .length_field_length(2) + //! .length_adjustment(1) // length of hdr2 + //! .num_skip(3) // length of hdr1 + LEN + //! .new_read(io); + //! # } + //! ``` + //! + //! The following frame will be decoded as such: + //! + //! ```text + //! INPUT + //! +- hdr1 -+-- len ---+- hdr2 -+--- Payload ---+ + //! | \xCA | \x00\x0B | \xFE | Hello world | + //! +--------+----------+--------+---------------+ + //! + //! DECODED + //! +- hdr2 -+--- Payload ---+ + //! | \xFE | Hello world | + //! +--------+---------------+ + //! ``` + //! + //! The length field is situated in the middle of the frame head. In this + //! case, the first byte in the frame head could be a version or some other + //! identifier that is not needed for processing. On the other hand, the + //! second half of the head is needed. + //! + //! `length_field_offset` indicates how many bytes to skip before starting + //! to read the length field. `length_adjustment` is the number of bytes to + //! skip starting at the end of the length field. In this case, it is the + //! second half of the head. + //! + //! ## Example 6 + //! + //! The following will parse a `u16` length field at offset 1 of a 4 byte + //! frame head. The first byte and the length field will be omitted from the + //! yielded `BytesMut`, but the trailing 2 bytes of the frame head will be + //! included. In this case, the length field **includes** the frame head + //! length. + //! + //! ``` + //! # use tokio_io::AsyncRead; + //! # use tokio_io::codec::length_delimited; + //! # fn bind_read(io: T) { + //! length_delimited::Builder::new() + //! .length_field_offset(1) // length of hdr1 + //! .length_field_length(2) + //! .length_adjustment(-3) // length of hdr1 + LEN, negative + //! .num_skip(3) + //! .new_read(io); + //! # } + //! ``` + //! + //! The following frame will be decoded as such: + //! + //! ```text + //! INPUT + //! +- hdr1 -+-- len ---+- hdr2 -+--- Payload ---+ + //! | \xCA | \x00\x0F | \xFE | Hello world | + //! +--------+----------+--------+---------------+ + //! + //! DECODED + //! +- hdr2 -+--- Payload ---+ + //! | \xFE | Hello world | + //! +--------+---------------+ + //! ``` + //! + //! Similar to the example above, the difference is that the length field + //! represents the length of the entire frame instead of just the payload. + //! The length of `hdr1` and `len` must be counted in `length_adjustment`. + //! Note that the length of `hdr2` does **not** need to be explicitly set + //! anywhere because it already is factored into the total frame length that + //! is read from the byte stream. + //! + //! # Encoding + //! + //! [`FramedWrite`] adapts an [`AsyncWrite`] into a `Sink` of [`BytesMut`], + //! such that each submitted [`BytesMut`] is prefaced by a length field. + //! There are fewer configuration options than [`FramedRead`]. Given + //! protocols that have more complex frame heads, an encoder should probably + //! be written by hand using [`Encoder`]. + //! + //! Here is a simple example, given a `FramedWrite` with the following + //! configuration: + //! + //! ``` + //! # extern crate tokio_io; + //! # extern crate bytes; + //! # use tokio_io::AsyncWrite; + //! # use tokio_io::codec::length_delimited; + //! # use bytes::BytesMut; + //! # fn write_frame(io: T) { + //! # let _: length_delimited::FramedWrite = + //! length_delimited::Builder::new() + //! .length_field_length(2) + //! .new_write(io); + //! # } + //! # pub fn main() {} + //! ``` + //! + //! A payload of `hello world` will be encoded as: + //! + //! ```text + //! +- len: u16 -+---- data ----+ + //! | \x00\x0b | hello world | + //! +------------+--------------+ + //! ``` + //! + //! [`FramedRead`]: struct.FramedRead.html + //! [`FramedWrite`]: struct.FramedWrite.html + //! [`AsyncRead`]: ../../trait.AsyncRead.html + //! [`AsyncWrite`]: ../../trait.AsyncWrite.html + //! [`Encoder`]: ../trait.Encoder.html + //! [`BytesMut`]: https://docs.rs/bytes/~0.4/bytes/struct.BytesMut.html + + pub use ::length_delimited::*; +} diff --git a/tokio-io/src/codecs.rs b/tokio-io/src/codecs.rs new file mode 100644 index 000000000..f61448d1f --- /dev/null +++ b/tokio-io/src/codecs.rs @@ -0,0 +1,123 @@ +use bytes::{Bytes, BufMut, BytesMut}; +use codec::{Encoder, Decoder}; +use std::{io, str}; + +/// 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(()) + } +} + +/// 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/src/copy.rs b/tokio-io/src/copy.rs new file mode 100644 index 000000000..8b8c0fe4e --- /dev/null +++ b/tokio-io/src/copy.rs @@ -0,0 +1,96 @@ +use std::io; + +use futures::{Future, Poll}; + +use {AsyncRead, AsyncWrite}; + +/// A future which will copy all data from a reader into a writer. +/// +/// Created by the [`copy`] function, this future will resolve to the number of +/// bytes copied or an error if one happens. +/// +/// [`copy`]: fn.copy.html +#[derive(Debug)] +pub struct Copy { + reader: Option, + read_done: bool, + writer: Option, + pos: usize, + cap: usize, + amt: u64, + buf: Box<[u8]>, +} + +/// Creates a future which represents copying all the bytes from one object to +/// another. +/// +/// The returned future will copy all the bytes read from `reader` into the +/// `writer` specified. This future will only complete once the `reader` has hit +/// EOF and all bytes have been written to and flushed from the `writer` +/// provided. +/// +/// On success the number of bytes is returned and the `reader` and `writer` are +/// consumed. On error the error is returned and the I/O objects are consumed as +/// well. +pub fn copy(reader: R, writer: W) -> Copy + where R: AsyncRead, + W: AsyncWrite, +{ + Copy { + reader: Some(reader), + read_done: false, + writer: Some(writer), + amt: 0, + pos: 0, + cap: 0, + buf: Box::new([0; 2048]), + } +} + +impl Future for Copy + where R: AsyncRead, + W: AsyncWrite, +{ + type Item = (u64, R, W); + type Error = io::Error; + + fn poll(&mut self) -> Poll<(u64, R, W), io::Error> { + loop { + // If our buffer is empty, then we need to read some data to + // continue. + if self.pos == self.cap && !self.read_done { + let reader = self.reader.as_mut().unwrap(); + let n = try_nb!(reader.read(&mut self.buf)); + if n == 0 { + self.read_done = true; + } else { + self.pos = 0; + self.cap = n; + } + } + + // If our buffer has some data, let's write it out! + while self.pos < self.cap { + let writer = self.writer.as_mut().unwrap(); + let i = try_nb!(writer.write(&self.buf[self.pos..self.cap])); + if i == 0 { + return Err(io::Error::new(io::ErrorKind::WriteZero, + "write zero byte into writer")); + } else { + self.pos += i; + self.amt += i as u64; + } + } + + // If we've written al the data and we've seen EOF, flush out the + // data and finish the transfer. + // done with the entire transfer. + if self.pos == self.cap && self.read_done { + try_nb!(self.writer.as_mut().unwrap().flush()); + let reader = self.reader.take().unwrap(); + let writer = self.writer.take().unwrap(); + return Ok((self.amt, reader, writer).into()) + } + } + } +} diff --git a/tokio-io/src/flush.rs b/tokio-io/src/flush.rs new file mode 100644 index 000000000..29065f9f9 --- /dev/null +++ b/tokio-io/src/flush.rs @@ -0,0 +1,44 @@ +use std::io; + +use futures::{Poll, Future, Async}; + +use AsyncWrite; + +/// A future used to fully flush an I/O object. +/// +/// Resolves to the underlying I/O object once the flush operation is complete. +/// +/// Created by the [`flush`] function. +/// +/// [`flush`]: fn.flush.html +#[derive(Debug)] +pub struct Flush { + a: Option, +} + +/// Creates a future which will entirely flush an I/O object and then yield the +/// object itself. +/// +/// This function will consume the object provided if an error happens, and +/// otherwise it will repeatedly call `flush` until it sees `Ok(())`, scheduling +/// a retry if `WouldBlock` is seen along the way. +pub fn flush(a: A) -> Flush + where A: AsyncWrite, +{ + Flush { + a: Some(a), + } +} + +impl Future for Flush + where A: AsyncWrite, +{ + type Item = A; + type Error = io::Error; + + fn poll(&mut self) -> Poll { + try_nb!(self.a.as_mut().unwrap().flush()); + Ok(Async::Ready(self.a.take().unwrap())) + } +} + diff --git a/tokio-io/src/framed.rs b/tokio-io/src/framed.rs new file mode 100644 index 000000000..e32554707 --- /dev/null +++ b/tokio-io/src/framed.rs @@ -0,0 +1,228 @@ +use std::io::{self, Read, Write}; +use std::fmt; + +use {AsyncRead, AsyncWrite}; +use framed_read::{framed_read2, framed_read2_with_buffer, FramedRead2, Decoder}; +use framed_write::{framed_write2, framed_write2_with_buffer, FramedWrite2, Encoder}; + +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); + +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 +} \ No newline at end of file diff --git a/tokio-io/src/framed_read.rs b/tokio-io/src/framed_read.rs new file mode 100644 index 000000000..6ccfd1aa9 --- /dev/null +++ b/tokio-io/src/framed_read.rs @@ -0,0 +1,295 @@ +use std::{fmt, io}; + +use AsyncRead; +use framed::Fuse; + +use futures::{Async, Poll, Stream, Sink, StartSend}; +use bytes::BytesMut; + +/// Decoding of frames via buffers. +/// +/// This trait is used when constructing an instance of `Framed` or +/// `FramedRead`. An implementation of `Decoder` takes a byte stream that has +/// already been buffered in `src` and decodes the data into a stream of +/// `Self::Item` frames. +/// +/// 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`). +pub trait Decoder { + /// The type of decoded frames. + type Item; + + /// The type of unrecoverable frame decoding errors. + /// + /// If an individual message is ill-formed but can be ignored without + /// interfering with the processing of future messages, it may be more + /// useful to report the failure as an `Item`. + /// + /// `From` is required in the interest of making `Error` suitable + /// for returning directly from a `FramedRead`, and to enable the default + /// implementation of `decode_eof` to yield an `io::Error` when the decoder + /// fails to consume all available data. + /// + /// Note that implementors of this trait can simply indicate `type Error = + /// io::Error` to use I/O errors as this type. + type Error: From; + + /// Attempts to decode a frame from the provided buffer of bytes. + /// + /// This method is called by `FramedRead` whenever bytes are ready to be + /// parsed. The provided buffer of bytes is what's been read so far, and + /// this instance of `Decode` can determine whether an entire frame is in + /// the buffer and is ready to be returned. + /// + /// If an entire frame is available, then this instance will remove those + /// bytes from the buffer provided and return them as a decoded + /// frame. Note that removing bytes from the provided buffer doesn't always + /// necessarily copy the bytes, so this should be an efficient operation in + /// most circumstances. + /// + /// If the bytes look valid, but a frame isn't fully available yet, then + /// `Ok(None)` is returned. This indicates to the `Framed` instance that + /// it needs to read some more bytes before calling this method again. + /// + /// Note that the bytes provided may be empty. If a previous call to + /// `decode` consumed all the bytes in the buffer then `decode` will be + /// called again until it returns `None`, indicating that more bytes need to + /// be read. + /// + /// Finally, if the bytes in the buffer are malformed then an error is + /// returned indicating why. This informs `Framed` that the stream is now + /// corrupt and should be terminated. + fn decode(&mut self, src: &mut BytesMut) -> Result, Self::Error>; + + /// A default method available to be called when there are no more bytes + /// available to be read from the underlying I/O. + /// + /// This method defaults to calling `decode` and returns an error if + /// `Ok(None)` is returned while there is unconsumed data in `buf`. + /// Typically this doesn't need to be implemented unless the framing + /// protocol differs near the end of the stream. + /// + /// Note that the `buf` argument may be empty. If a previous call to + /// `decode_eof` consumed all the bytes in the buffer, `decode_eof` will be + /// called again until it returns `None`, indicating that there are no more + /// frames to yield. This behavior enables returning finalization frames + /// that may not be based on inbound data. + fn decode_eof(&mut self, buf: &mut BytesMut) -> Result, Self::Error> { + match try!(self.decode(buf)) { + Some(frame) => Ok(Some(frame)), + None => { + if buf.is_empty() { + Ok(None) + } else { + Err(io::Error::new(io::ErrorKind::Other, + "bytes remaining on stream").into()) + } + } + } + } +} + +/// 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/framed_write.rs b/tokio-io/src/framed_write.rs new file mode 100644 index 000000000..f9d0edcb2 --- /dev/null +++ b/tokio-io/src/framed_write.rs @@ -0,0 +1,256 @@ +use std::io::{self, Read}; +use std::fmt; + +use {AsyncRead, AsyncWrite}; +use codec::Decoder; +use framed::Fuse; + +use futures::{Async, AsyncSink, Poll, Stream, Sink, StartSend}; +use bytes::BytesMut; + +/// Trait of helper objects to write out messages as bytes, for use with +/// `FramedWrite`. +pub trait Encoder { + /// The type of items consumed by the `Encoder` + type Item; + + /// The type of encoding errors. + /// + /// `FramedWrite` requires `Encoder`s errors to implement `From` + /// in the interest letting it return `Error`s directly. + type Error: From; + + /// Encodes a frame into the buffer provided. + /// + /// This method will encode `item` into the byte buffer provided by `dst`. + /// The `dst` provided is an internal buffer of the `Framed` instance and + /// will be written out when possible. + fn encode(&mut self, item: Self::Item, dst: &mut BytesMut) + -> Result<(), Self::Error>; +} + +/// 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_nb!(self.inner.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_nb!(self.inner.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/io.rs b/tokio-io/src/io.rs new file mode 100644 index 000000000..0d8b5e98c --- /dev/null +++ b/tokio-io/src/io.rs @@ -0,0 +1,23 @@ +//! I/O conveniences when working with primitives in `tokio-core` +//! +//! Contains various combinators to work with I/O objects and type definitions +//! as well. +//! +//! A description of the high-level I/O combinators can be [found online] in +//! addition to a description of the [low level details]. +//! +//! [found online]: https://tokio.rs/docs/getting-started/core/ +//! [low level details]: https://tokio.rs/docs/going-deeper-tokio/core-low-level/ + +pub use allow_std::AllowStdIo; +pub use copy::{copy, Copy}; +pub use flush::{flush, Flush}; +pub use lines::{lines, Lines}; +pub use read::{read, Read}; +pub use read_exact::{read_exact, ReadExact}; +pub use read_to_end::{read_to_end, ReadToEnd}; +pub use read_until::{read_until, ReadUntil}; +pub use shutdown::{shutdown, Shutdown}; +pub use split::{ReadHalf, WriteHalf}; +pub use window::Window; +pub use write_all::{write_all, WriteAll}; diff --git a/tokio-io/src/length_delimited.rs b/tokio-io/src/length_delimited.rs new file mode 100644 index 000000000..54e305ee1 --- /dev/null +++ b/tokio-io/src/length_delimited.rs @@ -0,0 +1,883 @@ +use {codec, AsyncRead, AsyncWrite}; + +use bytes::{Buf, BufMut, BytesMut, IntoBuf, BigEndian, LittleEndian}; +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}; + +/// Configure length delimited `FramedRead`, `FramedWrite`, and `Framed` values. +/// +/// `Builder` enables constructing configured length delimited framers. Note +/// that not all configuration settings apply to both encoding and decoding. See +/// the documentation for specific methods for more detail. +#[derive(Debug, Clone, Copy)] +pub struct Builder { + // Maximum frame length + max_frame_len: usize, + + // Number of bytes representing the field length + length_field_len: usize, + + // Number of bytes in the header before the length field + length_field_offset: usize, + + // Adjust the length specified in the header field by this amount + length_adjustment: isize, + + // Total number of bytes to skip before reading the payload, if not set, + // `length_field_len + length_field_offset` + num_skip: Option, + + // Length field byte order (little or big endian) + length_field_is_big_endian: bool, +} + +/// Adapts a byte stream into a unified `Stream` and `Sink` that works over +/// entire frame values. +/// +/// See [module level] documentation for more detail. +/// +/// [module level]: index.html +pub struct Framed { + inner: FramedRead>, +} + +/// Adapts a byte stream to a `Stream` yielding entire frame values. +/// +/// See [module level] documentation for more detail. +/// +/// [module level]: index.html +#[derive(Debug)] +pub struct FramedRead { + inner: codec::FramedRead, +} + +/// An error when the number of bytes read is more than max frame length. +pub struct FrameTooBig { + _priv: (), +} + +#[derive(Debug)] +struct Decoder { + // Configuration values + builder: Builder, + + // Read state + state: DecodeState, +} + +#[derive(Debug, Clone, Copy)] +enum DecodeState { + Head, + Data(usize), +} + +/// Adapts a byte stream to a `Sink` accepting entire frame values. +/// +/// See [module level] documentation for more detail. +/// +/// [module level]: index.html +pub struct FramedWrite { + // I/O type + inner: T, + + // Configuration values + builder: Builder, + + // Current frame being written + frame: Option, B::Buf>>, +} + +// ===== impl Framed ===== + +impl Framed { + /// Creates a new `Framed` with default configuration values. + pub fn new(inner: T) -> Framed { + Builder::new().new_framed(inner) + } +} + +impl Framed { + /// Returns a reference to the underlying I/O stream wrapped by `Framed`. + /// + /// 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() + } + + /// Returns a mutable reference to the underlying I/O stream wrapped by + /// `Framed`. + /// + /// 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 { + self.inner.get_mut().get_mut() + } + + /// Consumes the `Framed`, 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() + } +} + +impl Stream for Framed { + type Item = BytesMut; + type Error = io::Error; + + fn poll(&mut self) -> Poll, io::Error> { + self.inner.poll() + } +} + +impl Sink for Framed { + type SinkItem = B; + type SinkError = io::Error; + + fn start_send(&mut self, item: B) -> StartSend { + 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 fmt::Debug for Framed + 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 FramedRead ===== + +impl FramedRead { + /// Creates a new `FramedRead` with default configuration values. + pub fn new(inner: T) -> FramedRead { + Builder::new().new_read(inner) + } +} + +impl FramedRead { + /// Returns the current max frame setting + /// + /// This is the largest size this codec will accept from the wire. Larger + /// frames will be rejected. + pub fn max_frame_length(&self) -> usize { + self.inner.decoder().builder.max_frame_len + } + + /// Updates the max frame setting. + /// + /// The change takes effect the next time a frame is decoded. In other + /// words, if a frame is currently in process of being decoded with a frame + /// size greater than `val` but less than the max frame length in effect + /// before calling this function, then the frame will be allowed. + pub fn set_max_frame_length(&mut self, val: usize) { + self.inner.decoder_mut().builder.max_frame_length(val); + } + + /// 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.get_ref() + } + + /// 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 { + self.inner.get_mut() + } + + /// 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.into_inner() + } +} + +impl Stream for FramedRead { + type Item = BytesMut; + type Error = io::Error; + + fn poll(&mut self) -> Poll, io::Error> { + self.inner.poll() + } +} + +impl Sink for FramedRead { + type SinkItem = T::SinkItem; + type SinkError = T::SinkError; + + fn start_send(&mut self, item: T::SinkItem) -> StartSend { + 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 io::Write for FramedRead { + fn write(&mut self, src: &[u8]) -> io::Result { + self.inner.get_mut().write(src) + } + + fn flush(&mut self) -> io::Result<()> { + self.inner.get_mut().flush() + } +} + +impl AsyncWrite for FramedRead { + fn shutdown(&mut self) -> Poll<(), io::Error> { + self.inner.get_mut().shutdown() + } + + fn write_buf(&mut self, buf: &mut B) -> Poll { + self.inner.get_mut().write_buf(buf) + } +} + +// ===== impl Decoder ====== + +impl Decoder { + fn decode_head(&mut self, src: &mut BytesMut) -> io::Result> { + let head_len = self.builder.num_head_bytes(); + let field_len = self.builder.length_field_len; + + if src.len() < head_len { + // Not enough data + return Ok(None); + } + + let n = { + let mut src = Cursor::new(&mut *src); + + // Skip the required bytes + src.advance(self.builder.length_field_offset); + + // match endianess + let n = if self.builder.length_field_is_big_endian { + src.get_uint::(field_len) + } else { + src.get_uint::(field_len) + }; + + if n > self.builder.max_frame_len as u64 { + return Err(io::Error::new(io::ErrorKind::InvalidData, FrameTooBig { + _priv: (), + })); + } + + // The check above ensures there is no overflow + let n = n as usize; + + // Adjust `n` with bounds checking + 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) + }; + + // Error handling + 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); + } + + // Ensure that the buffer has enough space to read the incoming + // payload + src.reserve(n); + + return Ok(Some(n)); + } + + fn decode_data(&self, n: usize, src: &mut BytesMut) -> io::Result> { + // At this point, the buffer has already had the required capacity + // reserved. All there is to do is read. + 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> { + 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) => { + // Update the decode state + self.state = DecodeState::Head; + + // Make sure the buffer has enough space to read the next head + src.reserve(self.builder.num_head_bytes()); + + Ok(Some(data)) + } + None => Ok(None), + } + } +} + +// ===== impl FramedWrite ===== + +impl FramedWrite { + /// Creates a new `FramedWrite` with default configuration values. + pub fn new(inner: T) -> FramedWrite { + Builder::new().new_write(inner) + } +} + +impl FramedWrite { + /// Returns the current max frame setting + /// + /// This is the largest size this codec will write to the wire. Larger + /// frames will be rejected. + pub fn max_frame_length(&self) -> usize { + self.builder.max_frame_len + } + + /// Updates the max frame setting. + /// + /// The change takes effect the next time a frame is encoded. In other + /// words, if a frame is currently in process of being encoded with a frame + /// size greater than `val` but less than the max frame length in effect + /// before calling this function, then the frame will be allowed. + pub fn set_max_frame_length(&mut self, val: usize) { + self.builder.max_frame_length(val); + } + + /// 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 + } + + /// 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 + } + + /// 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 + } +} + +impl FramedWrite { + // If there is a buffered frame, try to write it to `T` + 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(); + try_ready!(self.inner.write_buf(frame)); + + 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: (), + })); + } + + // Adjust `n` with bounds checking + 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) + }; + + // Error handling + 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::(n as u64, self.builder.length_field_len); + } else { + head.put_uint::(n as u64, self.builder.length_field_len); + } + + debug_assert!(self.frame.is_none()); + + self.frame = Some(head.into_buf().chain(buf)); + + Ok(()) + } +} + +impl Sink for FramedWrite { + type SinkItem = B; + type SinkError = io::Error; + + fn start_send(&mut self, item: B) -> StartSend { + 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> { + // Write any buffered frame to T + try_ready!(self.do_write()); + + // Try flushing the underlying IO + try_nb!(self.inner.flush()); + + return Ok(Async::Ready(())); + } + + fn close(&mut self) -> Poll<(), io::Error> { + try_ready!(self.poll_complete()); + self.inner.shutdown() + } +} + +impl Stream for FramedWrite { + type Item = T::Item; + type Error = T::Error; + + fn poll(&mut self) -> Poll, T::Error> { + self.inner.poll() + } +} + +impl io::Read for FramedWrite { + fn read(&mut self, dst: &mut [u8]) -> io::Result { + self.get_mut().read(dst) + } +} + +impl AsyncRead for FramedWrite { + fn read_buf(&mut self, buf: &mut B) -> Poll { + self.get_mut().read_buf(buf) + } + + unsafe fn prepare_uninitialized_buffer(&self, buf: &mut [u8]) -> bool { + self.get_ref().prepare_uninitialized_buffer(buf) + } +} + +impl fmt::Debug for FramedWrite + 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 ===== + +impl Builder { + /// Creates a new length delimited framer builder with default configuration + /// values. + /// + /// # Examples + /// + /// ``` + /// # use tokio_io::AsyncRead; + /// use tokio_io::codec::length_delimited::Builder; + /// + /// # fn bind_read(io: T) { + /// Builder::new() + /// .length_field_offset(0) + /// .length_field_length(2) + /// .length_adjustment(0) + /// .num_skip(0) + /// .new_read(io); + /// # } + /// ``` + pub fn new() -> Builder { + Builder { + // Default max frame length of 8MB + max_frame_len: 8 * 1_024 * 1_024, + + // Default byte length of 4 + length_field_len: 4, + + // Default to the header field being at the start of the header. + length_field_offset: 0, + + length_adjustment: 0, + + // Total number of bytes to skip before reading the payload, if not set, + // `length_field_len + length_field_offset` + num_skip: None, + + // Default to reading the length field in network (big) endian. + length_field_is_big_endian: true, + } + } + + /// Read the length field as a big endian integer + /// + /// This is the default setting. + /// + /// This configuration option applies to both encoding and decoding. + /// + /// # Examples + /// + /// ``` + /// # use tokio_io::AsyncRead; + /// use tokio_io::codec::length_delimited::Builder; + /// + /// # fn bind_read(io: T) { + /// Builder::new() + /// .big_endian() + /// .new_read(io); + /// # } + /// ``` + pub fn big_endian(&mut self) -> &mut Self { + self.length_field_is_big_endian = true; + self + } + + /// Read the length field as a little endian integer + /// + /// The default setting is big endian. + /// + /// This configuration option applies to both encoding and decoding. + /// + /// # Examples + /// + /// ``` + /// # use tokio_io::AsyncRead; + /// use tokio_io::codec::length_delimited::Builder; + /// + /// # fn bind_read(io: T) { + /// Builder::new() + /// .little_endian() + /// .new_read(io); + /// # } + /// ``` + pub fn little_endian(&mut self) -> &mut Self { + self.length_field_is_big_endian = false; + self + } + + /// Sets the max frame length + /// + /// This configuration option applies to both encoding and decoding. The + /// default value is 8MB. + /// + /// When decoding, the length field read from the byte stream is checked + /// against this setting **before** any adjustments are applied. When + /// encoding, the length of the submitted payload is checked against this + /// setting. + /// + /// When frames exceed the max length, an `io::Error` with the custom value + /// of the `FrameTooBig` type will be returned. + /// + /// # Examples + /// + /// ``` + /// # use tokio_io::AsyncRead; + /// use tokio_io::codec::length_delimited::Builder; + /// + /// # fn bind_read(io: T) { + /// Builder::new() + /// .max_frame_length(8 * 1024) + /// .new_read(io); + /// # } + /// ``` + pub fn max_frame_length(&mut self, val: usize) -> &mut Self { + self.max_frame_len = val; + self + } + + /// Sets the number of bytes used to represent the length field + /// + /// The default value is `4`. The max value is `8`. + /// + /// This configuration option applies to both encoding and decoding. + /// + /// # Examples + /// + /// ``` + /// # use tokio_io::AsyncRead; + /// use tokio_io::codec::length_delimited::Builder; + /// + /// # fn bind_read(io: T) { + /// Builder::new() + /// .length_field_length(4) + /// .new_read(io); + /// # } + /// ``` + 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 + } + + /// Sets the number of bytes in the header before the length field + /// + /// This configuration option only applies to decoding. + /// + /// # Examples + /// + /// ``` + /// # use tokio_io::AsyncRead; + /// use tokio_io::codec::length_delimited::Builder; + /// + /// # fn bind_read(io: T) { + /// Builder::new() + /// .length_field_offset(1) + /// .new_read(io); + /// # } + /// ``` + pub fn length_field_offset(&mut self, val: usize) -> &mut Self { + self.length_field_offset = val; + self + } + + /// Delta between the payload length specified in the header and the real + /// payload length + /// + /// # Examples + /// + /// ``` + /// # use tokio_io::AsyncRead; + /// use tokio_io::codec::length_delimited::Builder; + /// + /// # fn bind_read(io: T) { + /// Builder::new() + /// .length_adjustment(-2) + /// .new_read(io); + /// # } + /// ``` + pub fn length_adjustment(&mut self, val: isize) -> &mut Self { + self.length_adjustment = val; + self + } + + /// Sets the number of bytes to skip before reading the payload + /// + /// Default value is `length_field_len + length_field_offset` + /// + /// This configuration option only applies to decoding + /// + /// # Examples + /// + /// ``` + /// # use tokio_io::AsyncRead; + /// use tokio_io::codec::length_delimited::Builder; + /// + /// # fn bind_read(io: T) { + /// Builder::new() + /// .num_skip(4) + /// .new_read(io); + /// # } + /// ``` + pub fn num_skip(&mut self, val: usize) -> &mut Self { + self.num_skip = Some(val); + self + } + + /// Create a configured length delimited `FramedRead` + /// + /// # Examples + /// + /// ``` + /// # use tokio_io::AsyncRead; + /// use tokio_io::codec::length_delimited::Builder; + /// + /// # fn bind_read(io: T) { + /// Builder::new() + /// .length_field_offset(0) + /// .length_field_length(2) + /// .length_adjustment(0) + /// .num_skip(0) + /// .new_read(io); + /// # } + /// ``` + pub fn new_read(&self, upstream: T) -> FramedRead + where T: AsyncRead, + { + FramedRead { + inner: codec::FramedRead::new(upstream, Decoder { + builder: *self, + state: DecodeState::Head, + }), + } + } + + /// Create a configured length delimited `FramedWrite` + /// + /// # Examples + /// + /// ``` + /// # extern crate tokio_io; + /// # extern crate bytes; + /// # use tokio_io::AsyncWrite; + /// # use tokio_io::codec::length_delimited; + /// # use bytes::BytesMut; + /// # fn write_frame(io: T) { + /// # let _: length_delimited::FramedWrite = + /// length_delimited::Builder::new() + /// .length_field_length(2) + /// .new_write(io); + /// # } + /// # pub fn main() {} + /// ``` + pub fn new_write(&self, inner: T) -> FramedWrite + where T: AsyncWrite, + B: IntoBuf, + { + FramedWrite { + inner: inner, + builder: *self, + frame: None, + } + } + + /// Create a configured length delimited `Framed` + /// + /// # Examples + /// + /// ``` + /// # extern crate tokio_io; + /// # extern crate bytes; + /// # use tokio_io::{AsyncRead, AsyncWrite}; + /// # use tokio_io::codec::length_delimited; + /// # use bytes::BytesMut; + /// # fn write_frame(io: T) { + /// # let _: length_delimited::Framed = + /// length_delimited::Builder::new() + /// .length_field_length(2) + /// .new_framed(io); + /// # } + /// # pub fn main() {} + /// ``` + pub fn new_framed(&self, inner: T) -> Framed + 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 FrameTooBig ===== + +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" + } +} diff --git a/tokio-io/src/lib.rs b/tokio-io/src/lib.rs new file mode 100644 index 000000000..4c6d67ef7 --- /dev/null +++ b/tokio-io/src/lib.rs @@ -0,0 +1,389 @@ +//! Core I/O traits and combinators when working with Tokio. +//! +//! A description of the high-level I/O combinators can be [found online] in +//! addition to a description of the [low level details]. +//! +//! [found online]: https://tokio.rs/docs/getting-started/core/ +//! [low level details]: https://tokio.rs/docs/going-deeper-tokio/core-low-level/ + +#![deny(missing_docs, missing_debug_implementations, warnings)] +#![doc(html_root_url = "https://docs.rs/tokio-io/0.1")] + +#[macro_use] +extern crate log; + +#[macro_use] +extern crate futures; +extern crate bytes; + +use std::io as std_io; +use std::io::Write; + +use futures::{Async, Future, Poll, Stream}; +use bytes::{Buf, BufMut}; + +/// A convenience typedef around a `Future` whose error component is `io::Error` +pub type IoFuture = Box + Send>; + +/// A convenience typedef around a `Stream` whose error component is `io::Error` +pub type IoStream = Box + Send>; + +/// A convenience macro for working with `io::Result` from the `Read` and +/// `Write` traits. +/// +/// This macro takes `io::Result` as input, and returns `T` as the output. If +/// the input type is of the `Err` variant, then `Poll::NotReady` is returned if +/// it indicates `WouldBlock` or otherwise `Err` is returned. +#[macro_export] +macro_rules! try_nb { + ($e:expr) => (match $e { + Ok(t) => t, + Err(ref e) if e.kind() == ::std::io::ErrorKind::WouldBlock => { + return Ok(::futures::Async::NotReady) + } + Err(e) => return Err(e.into()), + }) +} + +pub mod io; +pub mod codec; + +mod allow_std; +mod codecs; +mod copy; +mod flush; +mod framed; +mod framed_read; +mod framed_write; +mod length_delimited; +mod lines; +mod read; +mod read_exact; +mod read_to_end; +mod read_until; +mod shutdown; +mod split; +mod window; +mod write_all; + +use codec::{Decoder, Encoder, Framed}; +use split::{ReadHalf, WriteHalf}; + +/// A trait for readable objects which operated in an asynchronous and +/// futures-aware fashion. +/// +/// This trait inherits from `io::Read` and indicates as a marker that an I/O +/// object is **nonblocking**, meaning that it will return an error instead of +/// blocking when bytes are unavailable, but the stream hasn't reached EOF. +/// Specifically this means that the `read` function for types that implement +/// this trait can have a few return values: +/// +/// * `Ok(n)` means that `n` bytes of data was immediately read and placed into +/// the output buffer, where `n` == 0 implies that EOF has been reached. +/// * `Err(e) if e.kind() == ErrorKind::WouldBlock` means that no data was read +/// into the buffer provided. The I/O object is not currently readable but may +/// become readable in the future. Most importantly, **the current future's +/// task is scheduled to get unparked when the object is readable**. This +/// means that like `Future::poll` you'll receive a notification when the I/O +/// object is readable again. +/// * `Err(e)` for other errors are standard I/O errors coming from the +/// underlying object. +/// +/// This trait importantly means that the `read` method only works in the +/// context of a future's task. The object may panic if used outside of a task. +pub trait AsyncRead: std_io::Read { + /// Prepares an uninitialized buffer to be safe to pass to `read`. Returns + /// `true` if the supplied buffer was zeroed out. + /// + /// While it would be highly unusual, implementations of [`io::Read`] are + /// able to read data from the buffer passed as an argument. Because of + /// this, the buffer passed to [`io::Read`] must be initialized memory. In + /// situations where large numbers of buffers are used, constantly having to + /// zero out buffers can be expensive. + /// + /// This function does any necessary work to prepare an uninitialized buffer + /// to be safe to pass to `read`. If `read` guarantees to never attempt read + /// data out of the supplied buffer, then `prepare_uninitialized_buffer` + /// doesn't need to do any work. + /// + /// If this function returns `true`, then the memory has been zeroed out. + /// This allows implementations of `AsyncRead` which are composed of + /// multiple sub implementations to efficiently implement + /// `prepare_uninitialized_buffer`. + /// + /// This function isn't actually `unsafe` to call but `unsafe` to implement. + /// The implementor must ensure that either the whole `buf` has been zeroed + /// or `read_buf()` overwrites the buffer without reading it and returns + /// correct value. + /// + /// This function is called from [`read_buf`]. + /// + /// [`io::Read`]: https://doc.rust-lang.org/std/io/trait.Read.html + /// [`read_buf`]: #method.read_buf + unsafe fn prepare_uninitialized_buffer(&self, buf: &mut [u8]) -> bool { + for i in 0..buf.len() { + buf[i] = 0; + } + + true + } + + /// Pull some bytes from this source into the specified `Buf`, returning + /// how many bytes were read. + /// + /// The `buf` provided will have bytes read into it and the internal cursor + /// will be advanced if any bytes were read. Note that this method typically + /// will not reallocate the buffer provided. + fn read_buf(&mut self, buf: &mut B) -> Poll + where Self: Sized, + { + if !buf.has_remaining_mut() { + return Ok(Async::Ready(0)); + } + + unsafe { + let n = { + let b = buf.bytes_mut(); + + self.prepare_uninitialized_buffer(b); + + try_nb!(self.read(b)) + }; + + buf.advance_mut(n); + Ok(Async::Ready(n)) + } + } + + /// 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, codec: T) -> Framed + where Self: AsyncWrite + Sized, + { + framed::framed(self, codec) + } + + /// Helper method for splitting this read/write object into two halves. + /// + /// The two halves returned implement the `Read` and `Write` traits, + /// respectively. + fn split(self) -> (ReadHalf, WriteHalf) + where Self: AsyncWrite + Sized, + { + split::split(self) + } +} + +impl AsyncRead for Box { + unsafe fn prepare_uninitialized_buffer(&self, buf: &mut [u8]) -> bool { + (**self).prepare_uninitialized_buffer(buf) + } +} + +impl<'a, T: ?Sized + AsyncRead> AsyncRead for &'a mut T { + unsafe fn prepare_uninitialized_buffer(&self, buf: &mut [u8]) -> bool { + (**self).prepare_uninitialized_buffer(buf) + } +} + +impl<'a> AsyncRead for &'a [u8] { + unsafe fn prepare_uninitialized_buffer(&self, _buf: &mut [u8]) -> bool { + false + } +} + +/// A trait for writable objects which operated in an asynchronous and +/// futures-aware fashion. +/// +/// This trait inherits from `io::Write` and indicates that an I/O object is +/// **nonblocking**, meaning that it will return an error instead of blocking +/// when bytes cannot currently be written, but hasn't closed. Specifically +/// this means that the `write` function for types that implement this trait +/// can have a few return values: +/// +/// * `Ok(n)` means that `n` bytes of data was immediately written . +/// * `Err(e) if e.kind() == ErrorKind::WouldBlock` means that no data was +/// written from the buffer provided. The I/O object is not currently +/// writable but may become writable in the future. Most importantly, **the +/// current future's task is scheduled to get unparked when the object is +/// readable**. This means that like `Future::poll` you'll receive a +/// notification when the I/O object is writable again. +/// * `Err(e)` for other errors are standard I/O errors coming from the +/// underlying object. +/// +/// This trait importantly means that the `write` method only works in the +/// context of a future's task. The object may panic if used outside of a task. +/// +/// Note that this trait also represents that the `Write::flush` method works +/// very similarly to the `write` method, notably that `Ok(())` means that the +/// writer has successfully been flushed, a "would block" error means that the +/// current task is ready to receive a notification when flushing can make more +/// progress, and otherwise normal errors can happen as well. +pub trait AsyncWrite: std_io::Write { + /// Initiates or attempts to shut down this writer, returning success when + /// the I/O connection has completely shut down. + /// + /// This method is intended to be used for asynchronous shutdown of I/O + /// connections. For example this is suitable for implementing shutdown of a + /// TLS connection or calling `TcpStream::shutdown` on a proxied connection. + /// Protocols sometimes need to flush out final pieces of data or otherwise + /// perform a graceful shutdown handshake, reading/writing more data as + /// appropriate. This method is the hook for such protocols to implement the + /// graceful shutdown logic. + /// + /// This `shutdown` method is required by implementors of the + /// `AsyncWrite` trait. Wrappers typically just want to proxy this call + /// through to the wrapped type, and base types will typically implement + /// shutdown logic here or just return `Ok(().into())`. Note that if you're + /// wrapping an underlying `AsyncWrite` a call to `shutdown` implies that + /// transitively the entire stream has been shut down. After your wrapper's + /// shutdown logic has been executed you should shut down the underlying + /// stream. + /// + /// Invocation of a `shutdown` implies an invocation of `flush`. Once this + /// method returns `Ready` it implies that a flush successfully happened + /// before the shutdown happened. That is, callers don't need to call + /// `flush` before calling `shutdown`. They can rely that by calling + /// `shutdown` any pending buffered data will be written out. + /// + /// # Return value + /// + /// This function returns a `Poll<(), io::Error>` classified as such: + /// + /// * `Ok(Async::Ready(()))` - indicates that the connection was + /// successfully shut down and is now safe to deallocate/drop/close + /// resources associated with it. This method means that the current task + /// will no longer receive any notifications due to this method and the + /// I/O object itself is likely no longer usable. + /// + /// * `Ok(Async::NotReady)` - indicates that shutdown is initiated but could + /// not complete just yet. This may mean that more I/O needs to happen to + /// continue this shutdown operation. The current task is scheduled to + /// receive a notification when it's otherwise ready to continue the + /// shutdown operation. When woken up this method should be called again. + /// + /// * `Err(e)` - indicates a fatal error has happened with shutdown, + /// indicating that the shutdown operation did not complete successfully. + /// This typically means that the I/O object is no longer usable. + /// + /// # Errors + /// + /// This function can return normal I/O errors through `Err`, described + /// above. Additionally this method may also render the underlying + /// `Write::write` method no longer usable (e.g. will return errors in the + /// future). It's recommended that once `shutdown` is called the + /// `write` method is no longer called. + /// + /// # Panics + /// + /// This function will panic if not called within the context of a future's + /// task. + fn shutdown(&mut self) -> Poll<(), std_io::Error>; + + /// Write a `Buf` into this value, returning how many bytes were written. + /// + /// Note that this method will advance the `buf` provided automatically by + /// the number of bytes written. + fn write_buf(&mut self, buf: &mut B) -> Poll + where Self: Sized, + { + if !buf.has_remaining() { + return Ok(Async::Ready(0)); + } + + let n = try_nb!(self.write(buf.bytes())); + buf.advance(n); + Ok(Async::Ready(n)) + } +} + +impl AsyncWrite for Box { + fn shutdown(&mut self) -> Poll<(), std_io::Error> { + (**self).shutdown() + } +} +impl<'a, T: ?Sized + AsyncWrite> AsyncWrite for &'a mut T { + fn shutdown(&mut self) -> Poll<(), std_io::Error> { + (**self).shutdown() + } +} + +impl AsyncRead for std_io::Repeat { + unsafe fn prepare_uninitialized_buffer(&self, _: &mut [u8]) -> bool { + false + } +} + +impl AsyncWrite for std_io::Sink { + fn shutdown(&mut self) -> Poll<(), std_io::Error> { + Ok(().into()) + } +} + +// TODO: Implement `prepare_uninitialized_buffer` for `io::Take`. +// This is blocked on rust-lang/rust#27269 +impl AsyncRead for std_io::Take { +} + +// TODO: Implement `prepare_uninitialized_buffer` when upstream exposes inner +// parts +impl AsyncRead for std_io::Chain + where T: AsyncRead, + U: AsyncRead, +{ +} + +impl AsyncWrite for std_io::BufWriter { + fn shutdown(&mut self) -> Poll<(), std_io::Error> { + try_nb!(self.flush()); + self.get_mut().shutdown() + } +} + +impl AsyncRead for std_io::BufReader { + unsafe fn prepare_uninitialized_buffer(&self, buf: &mut [u8]) -> bool { + self.get_ref().prepare_uninitialized_buffer(buf) + } +} + +impl> AsyncRead for std_io::Cursor { +} + +impl<'a> AsyncWrite for std_io::Cursor<&'a mut [u8]> { + fn shutdown(&mut self) -> Poll<(), std_io::Error> { + Ok(().into()) + } +} + +impl AsyncWrite for std_io::Cursor> { + fn shutdown(&mut self) -> Poll<(), std_io::Error> { + Ok(().into()) + } +} + +impl AsyncWrite for std_io::Cursor> { + fn shutdown(&mut self) -> Poll<(), std_io::Error> { + Ok(().into()) + } +} + +fn _assert_objects() { + fn _assert() {} + _assert::>(); + _assert::>(); +} diff --git a/tokio-io/src/lines.rs b/tokio-io/src/lines.rs new file mode 100644 index 000000000..6263a1607 --- /dev/null +++ b/tokio-io/src/lines.rs @@ -0,0 +1,60 @@ +use std::io::{self, BufRead}; +use std::mem; + +use futures::{Poll, Stream}; + +use AsyncRead; + +/// Combinator created by the top-level `lines` method which is a stream over +/// the lines of text on an I/O object. +#[derive(Debug)] +pub struct Lines { + io: A, + line: String, +} + +/// Creates a new stream from the I/O object given representing the lines of +/// input that are found on `A`. +/// +/// This method takes an asynchronous I/O object, `a`, and returns a `Stream` of +/// lines that the object contains. The returned stream will reach its end once +/// `a` reaches EOF. +pub fn lines(a: A) -> Lines + where A: AsyncRead + BufRead, +{ + Lines { + io: a, + line: String::new(), + } +} + +impl Lines { + /// Returns the underlying I/O object. + /// + /// Note that this may lose data already read into internal buffers. It's + /// recommended to only call this once the stream has reached its end. + pub fn into_inner(self) -> A { + self.io + } +} + +impl Stream for Lines + where A: AsyncRead + BufRead, +{ + type Item = String; + type Error = io::Error; + + fn poll(&mut self) -> Poll, io::Error> { + let n = try_nb!(self.io.read_line(&mut self.line)); + if n == 0 && self.line.len() == 0 { + return Ok(None.into()) + } + if self.line.ends_with("\n") { + self.line.pop(); + if self.line.ends_with("\r") { + self.line.pop(); + } + } + Ok(Some(mem::replace(&mut self.line, String::new())).into()) + } +} diff --git a/tokio-io/src/read.rs b/tokio-io/src/read.rs new file mode 100644 index 000000000..abfb459c8 --- /dev/null +++ b/tokio-io/src/read.rs @@ -0,0 +1,56 @@ +use std::io; +use std::mem; + +use futures::{Future, Poll}; + +use AsyncRead; + +#[derive(Debug)] +enum State { + Pending { + rd: R, + buf: T, + }, + Empty, +} + +/// Tries to read some bytes directly into the given `buf` in asynchronous +/// manner, returning a future type. +/// +/// The returned future will resolve to both the I/O stream and the buffer +/// as well as the number of bytes read once the read operation is completed. +pub fn read(rd: R, buf: T) -> Read + where R: AsyncRead, + T: AsMut<[u8]> +{ + Read { state: State::Pending { rd: rd, buf: buf } } +} + +/// A future which can be used to easily read available number of bytes to fill +/// a buffer. +/// +/// Created by the [`read`] function. +#[derive(Debug)] +pub struct Read { + state: State, +} + +impl Future for Read + where R: AsyncRead, + T: AsMut<[u8]> +{ + type Item = (R, T, usize); + type Error = io::Error; + + fn poll(&mut self) -> Poll<(R, T, usize), io::Error> { + let nread = match self.state { + State::Pending { ref mut rd, ref mut buf } => try_nb!(rd.read(&mut buf.as_mut()[..])), + State::Empty => panic!("poll a Read after it's done"), + }; + + match mem::replace(&mut self.state, State::Empty) { + State::Pending { rd, buf } => Ok((rd, buf, nread).into()), + State::Empty => panic!("invalid internal state"), + } + } +} diff --git a/tokio-io/src/read_exact.rs b/tokio-io/src/read_exact.rs new file mode 100644 index 000000000..142512421 --- /dev/null +++ b/tokio-io/src/read_exact.rs @@ -0,0 +1,83 @@ +use std::io; +use std::mem; + +use futures::{Poll, Future}; + +use AsyncRead; + +/// A future which can be used to easily read exactly enough bytes to fill +/// a buffer. +/// +/// Created by the [`read_exact`] function. +/// +/// [`read_exact`]: fn.read_exact.html +#[derive(Debug)] +pub struct ReadExact { + state: State, +} + +#[derive(Debug)] +enum State { + Reading { + a: A, + buf: T, + pos: usize, + }, + Empty, +} + +/// Creates a future which will read exactly enough bytes to fill `buf`, +/// returning an error if EOF is hit sooner. +/// +/// The returned future will resolve to both the I/O stream as well as the +/// buffer once the read operation is completed. +/// +/// In the case of an error the buffer and the object will be discarded, with +/// the error yielded. In the case of success the object will be destroyed and +/// the buffer will be returned, with all data read from the stream appended to +/// the buffer. +pub fn read_exact(a: A, buf: T) -> ReadExact + where A: AsyncRead, + T: AsMut<[u8]>, +{ + ReadExact { + state: State::Reading { + a: a, + buf: buf, + pos: 0, + }, + } +} + +fn eof() -> io::Error { + io::Error::new(io::ErrorKind::UnexpectedEof, "early eof") +} + +impl Future for ReadExact + where A: AsyncRead, + T: AsMut<[u8]>, +{ + type Item = (A, T); + type Error = io::Error; + + fn poll(&mut self) -> Poll<(A, T), io::Error> { + match self.state { + State::Reading { ref mut a, ref mut buf, ref mut pos } => { + let buf = buf.as_mut(); + while *pos < buf.len() { + let n = try_nb!(a.read(&mut buf[*pos..])); + *pos += n; + if n == 0 { + return Err(eof()) + } + } + } + State::Empty => panic!("poll a ReadExact after it's done"), + } + + match mem::replace(&mut self.state, State::Empty) { + State::Reading { a, buf, .. } => Ok((a, buf).into()), + State::Empty => panic!(), + } + } +} diff --git a/tokio-io/src/read_to_end.rs b/tokio-io/src/read_to_end.rs new file mode 100644 index 000000000..519f8a76d --- /dev/null +++ b/tokio-io/src/read_to_end.rs @@ -0,0 +1,68 @@ +use std::io; +use std::mem; + +use futures::{Poll, Future}; + +use AsyncRead; + +/// A future which can be used to easily read the entire contents of a stream +/// into a vector. +/// +/// Created by the [`read_to_end`] function. +/// +/// [`read_to_end`]: fn.read_to_end.html +#[derive(Debug)] +pub struct ReadToEnd { + state: State, +} + +#[derive(Debug)] +enum State { + Reading { + a: A, + buf: Vec, + }, + Empty, +} + +/// Creates a future which will read all the bytes associated with the I/O +/// object `A` into the buffer provided. +/// +/// In the case of an error the buffer and the object will be discarded, with +/// the error yielded. In the case of success the object will be destroyed and +/// the buffer will be returned, with all data read from the stream appended to +/// the buffer. +pub fn read_to_end(a: A, buf: Vec) -> ReadToEnd + where A: AsyncRead, +{ + ReadToEnd { + state: State::Reading { + a: a, + buf: buf, + } + } +} + +impl Future for ReadToEnd + where A: AsyncRead, +{ + type Item = (A, Vec); + type Error = io::Error; + + fn poll(&mut self) -> Poll<(A, Vec), io::Error> { + match self.state { + State::Reading { ref mut a, ref mut buf } => { + // If we get `Ok`, then we know the stream hit EOF and we're done. If we + // hit "would block" then all the read data so far is in our buffer, and + // otherwise we propagate errors + try_nb!(a.read_to_end(buf)); + }, + State::Empty => panic!("poll ReadToEnd after it's done"), + } + + match mem::replace(&mut self.state, State::Empty) { + State::Reading { a, buf } => Ok((a, buf).into()), + State::Empty => unreachable!(), + } + } +} diff --git a/tokio-io/src/read_until.rs b/tokio-io/src/read_until.rs new file mode 100644 index 000000000..73c982924 --- /dev/null +++ b/tokio-io/src/read_until.rs @@ -0,0 +1,74 @@ +use std::io::{self, BufRead}; +use std::mem; + +use futures::{Poll, Future}; + +use AsyncRead; + +/// A future which can be used to easily read the contents of a stream into a +/// vector until the delimiter is reached. +/// +/// Created by the [`read_until`] function. +/// +/// [`read_until`]: fn.read_until.html +#[derive(Debug)] +pub struct ReadUntil { + state: State, +} + +#[derive(Debug)] +enum State { + Reading { + a: A, + byte: u8, + buf: Vec, + }, + Empty, +} + +/// Creates a future which will read all the bytes associated with the I/O +/// object `A` into the buffer provided until the delimiter `byte` is reached. +/// This method is the async equivalent to [`BufRead::read_until`]. +/// +/// In case of an error the buffer and the object will be discarded, with +/// the error yielded. In the case of success the object will be destroyed and +/// the buffer will be returned, with all bytes up to, and including, the delimiter +/// (if found). +/// +/// [`BufRead::read_until`]: https://doc.rust-lang.org/std/io/trait.BufRead.html#method.read_until +pub fn read_until(a: A, byte: u8, buf: Vec) -> ReadUntil + where A: AsyncRead + BufRead, +{ + ReadUntil { + state: State::Reading { + a: a, + byte: byte, + buf: buf, + } + } +} + +impl Future for ReadUntil + where A: AsyncRead + BufRead +{ + type Item = (A, Vec); + type Error = io::Error; + + fn poll(&mut self) -> Poll<(A, Vec), io::Error> { + match self.state { + State::Reading { ref mut a, byte, ref mut buf } => { + // If we get `Ok(n)`, then we know the stream hit EOF or the delimiter. + // and just return it, as we are finished. + // If we hit "would block" then all the read data so far + // is in our buffer, and otherwise we propagate errors. + try_nb!(a.read_until(byte, buf)); + }, + State::Empty => panic!("poll ReadUntil after it's done"), + } + + match mem::replace(&mut self.state, State::Empty) { + State::Reading { a, byte: _, buf } => Ok((a, buf).into()), + State::Empty => unreachable!(), + } + } +} diff --git a/tokio-io/src/shutdown.rs b/tokio-io/src/shutdown.rs new file mode 100644 index 000000000..18040eadc --- /dev/null +++ b/tokio-io/src/shutdown.rs @@ -0,0 +1,44 @@ +use std::io; + +use futures::{Poll, Future, Async}; + +use AsyncWrite; + +/// A future used to fully shutdown an I/O object. +/// +/// Resolves to the underlying I/O object once the shutdown operation is +/// complete. +/// +/// Created by the [`shutdown`] function. +/// +/// [`shutdown`]: fn.shutdown.html +#[derive(Debug)] +pub struct Shutdown { + a: Option, +} + +/// Creates a future which will entirely shutdown an I/O object and then yield +/// the object itself. +/// +/// This function will consume the object provided if an error happens, and +/// otherwise it will repeatedly call `shutdown` until it sees `Ok(())`, +/// scheduling a retry if `WouldBlock` is seen along the way. +pub fn shutdown(a: A) -> Shutdown + where A: AsyncWrite, +{ + Shutdown { + a: Some(a), + } +} + +impl Future for Shutdown + where A: AsyncWrite, +{ + type Item = A; + type Error = io::Error; + + fn poll(&mut self) -> Poll { + try_nb!(self.a.as_mut().unwrap().shutdown()); + Ok(Async::Ready(self.a.take().unwrap())) + } +} diff --git a/tokio-io/src/split.rs b/tokio-io/src/split.rs new file mode 100644 index 000000000..8993c4f77 --- /dev/null +++ b/tokio-io/src/split.rs @@ -0,0 +1,80 @@ +use std::io::{self, Read, Write}; + +use futures::{Async, Poll}; +use futures::sync::BiLock; +use bytes::{Buf, BufMut}; + +use {AsyncRead, AsyncWrite}; + +/// The readable half of an object returned from `AsyncRead::split`. +#[derive(Debug)] +pub struct ReadHalf { + handle: BiLock, +} + +/// The writable half of an object returned from `AsyncRead::split`. +#[derive(Debug)] +pub struct WriteHalf { + handle: BiLock, +} + +pub fn split(t: T) -> (ReadHalf, WriteHalf) { + let (a, b) = BiLock::new(t); + (ReadHalf { handle: a }, WriteHalf { handle: b }) +} + +fn would_block() -> io::Error { + io::Error::new(io::ErrorKind::WouldBlock, "would block") +} + +impl Read for ReadHalf { + fn read(&mut self, buf: &mut [u8]) -> io::Result { + match self.handle.poll_lock() { + Async::Ready(mut l) => l.read(buf), + Async::NotReady => Err(would_block()), + } + } +} + +impl AsyncRead for ReadHalf { + fn read_buf(&mut self, buf: &mut B) -> Poll { + match self.handle.poll_lock() { + Async::Ready(mut l) => l.read_buf(buf), + Async::NotReady => Err(would_block()), + } + } +} + +impl Write for WriteHalf { + fn write(&mut self, buf: &[u8]) -> io::Result { + match self.handle.poll_lock() { + Async::Ready(mut l) => l.write(buf), + Async::NotReady => Err(would_block()), + } + } + + fn flush(&mut self) -> io::Result<()> { + match self.handle.poll_lock() { + Async::Ready(mut l) => l.flush(), + Async::NotReady => Err(would_block()), + } + } +} + +impl AsyncWrite for WriteHalf { + fn shutdown(&mut self) -> Poll<(), io::Error> { + match self.handle.poll_lock() { + Async::Ready(mut l) => l.shutdown(), + Async::NotReady => Err(would_block()), + } + } + + fn write_buf(&mut self, buf: &mut B) -> Poll + where Self: Sized, + { + match self.handle.poll_lock() { + Async::Ready(mut l) => l.write_buf(buf), + Async::NotReady => Err(would_block()), + } + } +} diff --git a/tokio-io/src/window.rs b/tokio-io/src/window.rs new file mode 100644 index 000000000..4ded9ad40 --- /dev/null +++ b/tokio-io/src/window.rs @@ -0,0 +1,117 @@ +use std::ops; + +/// A owned window around an underlying buffer. +/// +/// Normally slices work great for considering sub-portions of a buffer, but +/// unfortunately a slice is a *borrowed* type in Rust which has an associated +/// lifetime. When working with future and async I/O these lifetimes are not +/// always appropriate, and are sometimes difficult to store in tasks. This +/// type strives to fill this gap by providing an "owned slice" around an +/// underlying buffer of bytes. +/// +/// A `Window` wraps an underlying buffer, `T`, and has configurable +/// start/end indexes to alter the behavior of the `AsRef<[u8]>` implementation +/// that this type carries. +/// +/// This type can be particularly useful when working with the `write_all` +/// combinator in this crate. Data can be sliced via `Window`, consumed by +/// `write_all`, and then earned back once the write operation finishes through +/// the `into_inner` method on this type. +#[derive(Debug)] +pub struct Window { + inner: T, + range: ops::Range, +} + +impl> Window { + /// Creates a new window around the buffer `t` defaulting to the entire + /// slice. + /// + /// Further methods can be called on the returned `Window` to alter the + /// window into the data provided. + pub fn new(t: T) -> Window { + Window { + range: 0..t.as_ref().len(), + inner: t, + } + } + + /// Gets a shared reference to the underlying buffer inside of this + /// `Window`. + pub fn get_ref(&self) -> &T { + &self.inner + } + + /// Gets a mutable reference to the underlying buffer inside of this + /// `Window`. + pub fn get_mut(&mut self) -> &mut T { + &mut self.inner + } + + /// Consumes this `Window`, returning the underlying buffer. + pub fn into_inner(self) -> T { + self.inner + } + + /// Returns the starting index of this window into the underlying buffer + /// `T`. + pub fn start(&self) -> usize { + self.range.start + } + + /// Returns the end index of this window into the underlying buffer + /// `T`. + pub fn end(&self) -> usize { + self.range.end + } + + /// Changes the starting index of this window to the index specified. + /// + /// Returns the windows back to chain multiple calls to this method. + /// + /// # Panics + /// + /// This method will panic if `start` is out of bounds for the underlying + /// slice or if it comes after the `end` configured in this window. + pub fn set_start(&mut self, start: usize) -> &mut Window { + assert!(start <= self.inner.as_ref().len()); + assert!(start <= self.range.end); + self.range.start = start; + self + } + + /// Changes the end index of this window to the index specified. + /// + /// Returns the windows back to chain multiple calls to this method. + /// + /// # Panics + /// + /// This method will panic if `end` is out of bounds for the underlying + /// slice or if it comes before the `start` configured in this window. + pub fn set_end(&mut self, end: usize) -> &mut Window { + assert!(end <= self.inner.as_ref().len()); + assert!(self.range.start <= end); + self.range.end = end; + self + } + + // TODO: how about a generic set() method along the lines of: + // + // buffer.set(..3) + // .set(0..2) + // .set(4..) + // + // etc. +} + +impl> AsRef<[u8]> for Window { + fn as_ref(&self) -> &[u8] { + &self.inner.as_ref()[self.range.start..self.range.end] + } +} + +impl> AsMut<[u8]> for Window { + fn as_mut(&mut self) -> &mut [u8] { + &mut self.inner.as_mut()[self.range.start..self.range.end] + } +} diff --git a/tokio-io/src/write_all.rs b/tokio-io/src/write_all.rs new file mode 100644 index 000000000..d000f1b9c --- /dev/null +++ b/tokio-io/src/write_all.rs @@ -0,0 +1,86 @@ +use std::io; +use std::mem; + +use futures::{Poll, Future}; + +use AsyncWrite; + +/// A future used to write the entire contents of some data to a stream. +/// +/// This is created by the [`write_all`] top-level method. +/// +/// [`write_all`]: fn.write_all.html +#[derive(Debug)] +pub struct WriteAll { + state: State, +} + +#[derive(Debug)] +enum State { + Writing { + a: A, + buf: T, + pos: usize, + }, + Empty, +} + +/// Creates a future that will write the entire contents of the buffer `buf` to +/// the stream `a` provided. +/// +/// The returned future will not return until all the data has been written, and +/// the future will resolve to the stream as well as the buffer (for reuse if +/// needed). +/// +/// Any error which happens during writing will cause both the stream and the +/// buffer to get destroyed. +/// +/// The `buf` parameter here only requires the `AsRef<[u8]>` trait, which should +/// be broadly applicable to accepting data which can be converted to a slice. +/// The `Window` struct is also available in this crate to provide a different +/// window into a slice if necessary. +pub fn write_all(a: A, buf: T) -> WriteAll + where A: AsyncWrite, + T: AsRef<[u8]>, +{ + WriteAll { + state: State::Writing { + a: a, + buf: buf, + pos: 0, + }, + } +} + +fn zero_write() -> io::Error { + io::Error::new(io::ErrorKind::WriteZero, "zero-length write") +} + +impl Future for WriteAll + where A: AsyncWrite, + T: AsRef<[u8]>, +{ + type Item = (A, T); + type Error = io::Error; + + fn poll(&mut self) -> Poll<(A, T), io::Error> { + match self.state { + State::Writing { ref mut a, ref buf, ref mut pos } => { + let buf = buf.as_ref(); + while *pos < buf.len() { + let n = try_nb!(a.write(&buf[*pos..])); + *pos += n; + if n == 0 { + return Err(zero_write()) + } + } + } + State::Empty => panic!("poll a WriteAll after it's done"), + } + + match mem::replace(&mut self.state, State::Empty) { + State::Writing { a, buf, .. } => Ok((a, buf).into()), + State::Empty => panic!(), + } + } +} diff --git a/tokio-io/tests/async_read.rs b/tokio-io/tests/async_read.rs new file mode 100644 index 000000000..7aa9b1d21 --- /dev/null +++ b/tokio-io/tests/async_read.rs @@ -0,0 +1,149 @@ +extern crate tokio_io; +extern crate bytes; +extern crate futures; + +use tokio_io::AsyncRead; +use bytes::{BytesMut, BufMut}; +use futures::Async; + +use std::io::{self, Read}; + +#[test] +fn read_buf_success() { + struct R; + + impl Read for R { + fn read(&mut self, buf: &mut [u8]) -> io::Result { + buf[0..11].copy_from_slice(b"hello world"); + Ok(11) + } + } + + impl AsyncRead for R {} + + let mut buf = BytesMut::with_capacity(65); + + let n = match R.read_buf(&mut buf).unwrap() { + Async::Ready(n) => n, + _ => panic!(), + }; + + assert_eq!(11, n); + assert_eq!(buf[..], b"hello world"[..]); +} + +#[test] +fn read_buf_error() { + struct R; + + impl Read for R { + fn read(&mut self, _: &mut [u8]) -> io::Result { + Err(io::Error::new(io::ErrorKind::Other, "other")) + } + } + + impl AsyncRead for R {} + + let mut buf = BytesMut::with_capacity(65); + + let err = R.read_buf(&mut buf).unwrap_err(); + assert_eq!(err.kind(), io::ErrorKind::Other); +} + +#[test] +fn read_buf_no_capacity() { + struct R; + + impl Read for R { + fn read(&mut self, _: &mut [u8]) -> io::Result { + unimplemented!(); + } + } + + impl AsyncRead for R {} + + // Can't create BytesMut w/ zero capacity, so fill it up + let mut buf = BytesMut::with_capacity(64); + buf.put(&[0; 64][..]); + + let n = match R.read_buf(&mut buf).unwrap() { + Async::Ready(n) => n, + _ => panic!(), + }; + + assert_eq!(0, n); +} + +#[test] +fn read_buf_no_uninitialized() { + struct R; + + impl Read for R { + fn read(&mut self, buf: &mut [u8]) -> io::Result { + for b in buf { + assert_eq!(0, *b); + } + + Ok(0) + } + } + + impl AsyncRead for R {} + + // Can't create BytesMut w/ zero capacity, so fill it up + let mut buf = BytesMut::with_capacity(64); + + let n = match R.read_buf(&mut buf).unwrap() { + Async::Ready(n) => n, + _ => panic!(), + }; + + assert_eq!(0, n); +} + +#[test] +fn read_buf_uninitialized_ok() { + struct R; + + impl Read for R { + fn read(&mut self, buf: &mut [u8]) -> io::Result { + assert_eq!(buf[0..11], b"hello world"[..]); + Ok(0) + } + } + + impl AsyncRead for R { + unsafe fn prepare_uninitialized_buffer(&self, _: &mut [u8]) -> bool { + false + } + } + + // Can't create BytesMut w/ zero capacity, so fill it up + let mut buf = BytesMut::with_capacity(64); + unsafe { + buf.bytes_mut()[0..11].copy_from_slice(b"hello world"); + } + + let n = match R.read_buf(&mut buf).unwrap() { + Async::Ready(n) => n, + _ => panic!(), + }; + + assert_eq!(0, n); +} + +#[test] +fn read_buf_translate_wouldblock_to_not_ready() { + struct R; + + impl Read for R { + fn read(&mut self, _: &mut [u8]) -> io::Result { + Err(io::Error::new(io::ErrorKind::WouldBlock, "")) + } + } + + impl AsyncRead for R {} + + let mut buf = BytesMut::with_capacity(65); + assert!(!R.read_buf(&mut buf).unwrap().is_ready()); +} diff --git a/tokio-io/tests/codecs.rs b/tokio-io/tests/codecs.rs new file mode 100644 index 000000000..5d63242fd --- /dev/null +++ b/tokio-io/tests/codecs.rs @@ -0,0 +1,76 @@ +extern crate tokio_io; +extern crate bytes; + +use bytes::{BytesMut, Bytes, BufMut}; +use tokio_io::codec::{BytesCodec, LinesCodec, Decoder, Encoder}; + +#[test] +fn bytes_decoder() { + let mut codec = BytesCodec::new(); + let buf = &mut BytesMut::new(); + buf.put_slice(b"abc"); + assert_eq!("abc", codec.decode(buf).unwrap().unwrap()); + assert_eq!(None, codec.decode(buf).unwrap()); + assert_eq!(None, codec.decode(buf).unwrap()); + buf.put_slice(b"a"); + assert_eq!("a", codec.decode(buf).unwrap().unwrap()); +} + +#[test] +fn bytes_encoder() { + let mut codec = BytesCodec::new(); + + // Default capacity of BytesMut + #[cfg(target_pointer_width = "64")] + const INLINE_CAP: usize = 4 * 8 - 1; + #[cfg(target_pointer_width = "32")] + const INLINE_CAP: usize = 4 * 4 - 1; + + let mut buf = BytesMut::new(); + codec.encode(Bytes::from_static(&[0; INLINE_CAP + 1]), &mut buf).unwrap(); + + // Default capacity of Framed Read + const INITIAL_CAPACITY: usize = 8 * 1024; + + let mut buf = BytesMut::with_capacity(INITIAL_CAPACITY); + codec.encode(Bytes::from_static(&[0; INITIAL_CAPACITY + 1]), &mut buf).unwrap(); +} + +#[test] +fn lines_decoder() { + let mut codec = LinesCodec::new(); + let buf = &mut BytesMut::new(); + buf.reserve(200); + buf.put("line 1\nline 2\r\nline 3\n\r\n\r"); + assert_eq!("line 1", codec.decode(buf).unwrap().unwrap()); + assert_eq!("line 2", codec.decode(buf).unwrap().unwrap()); + assert_eq!("line 3", codec.decode(buf).unwrap().unwrap()); + assert_eq!("", codec.decode(buf).unwrap().unwrap()); + assert_eq!(None, codec.decode(buf).unwrap()); + assert_eq!(None, codec.decode_eof(buf).unwrap()); + buf.put("k"); + assert_eq!(None, codec.decode(buf).unwrap()); + assert_eq!("\rk", codec.decode_eof(buf).unwrap().unwrap()); + assert_eq!(None, codec.decode(buf).unwrap()); + assert_eq!(None, codec.decode_eof(buf).unwrap()); +} + +#[test] +fn lines_encoder() { + let mut codec = BytesCodec::new(); + + // Default capacity of BytesMut + #[cfg(target_pointer_width = "64")] + const INLINE_CAP: usize = 4 * 8 - 1; + #[cfg(target_pointer_width = "32")] + const INLINE_CAP: usize = 4 * 4 - 1; + + let mut buf = BytesMut::new(); + codec.encode(Bytes::from_static(&[b'a'; INLINE_CAP + 1]), &mut buf).unwrap(); + + // Default capacity of Framed Read + const INITIAL_CAPACITY: usize = 8 * 1024; + + let mut buf = BytesMut::with_capacity(INITIAL_CAPACITY); + codec.encode(Bytes::from_static(&[b'a'; INITIAL_CAPACITY + 1]), &mut buf).unwrap(); +} diff --git a/tokio-io/tests/framed.rs b/tokio-io/tests/framed.rs new file mode 100644 index 000000000..7b2693be1 --- /dev/null +++ b/tokio-io/tests/framed.rs @@ -0,0 +1,97 @@ +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_io::AsyncRead; +use bytes::{BytesMut, Buf, BufMut, IntoBuf, BigEndian}; + +const INITIAL_CAPACITY: usize = 8 * 1024; + +struct U32Codec; + +impl Decoder for U32Codec { + type Item = u32; + type Error = io::Error; + + fn decode(&mut self, buf: &mut BytesMut) -> io::Result> { + if buf.len() < 4 { + return Ok(None); + } + + let n = buf.split_to(4).into_buf().get_u32::(); + Ok(Some(n)) + } +} + +impl Encoder for U32Codec { + type Item = u32; + type Error = io::Error; + + fn encode(&mut self, item: u32, dst: &mut BytesMut) -> io::Result<()> { + // Reserve space + dst.reserve(4); + dst.put_u32::(item); + Ok(()) + } +} + +struct DontReadIntoThis; + +impl Read for DontReadIntoThis { + fn read(&mut self, _: &mut [u8]) -> io::Result { + Err(io::Error::new(io::ErrorKind::Other, + "Read into something you weren't supposed to.")) + } +} + +impl AsyncRead for DontReadIntoThis {} + +#[test] +fn can_read_from_existing_buf() { + let parts = FramedParts { + inner: DontReadIntoThis, + readbuf: vec![0, 0, 0, 42].into(), + writebuf: BytesMut::with_capacity(0), + }; + let framed = Framed::from_parts(parts, U32Codec); + + let num = framed + .into_future() + .map(|(first_num, _)| { + first_num.unwrap() + }) + .wait() + .map_err(|e| e.0) + .unwrap(); + assert_eq!(num, 42); +} + +#[test] +fn external_buf_grows_to_init() { + let parts = FramedParts { + inner: DontReadIntoThis, + readbuf: vec![0, 0, 0, 42].into(), + writebuf: BytesMut::with_capacity(0), + }; + let framed = Framed::from_parts(parts, U32Codec); + let FramedParts { readbuf, .. } = framed.into_parts(); + + assert_eq!(readbuf.capacity(), INITIAL_CAPACITY); +} + +#[test] +fn external_buf_does_not_shrink() { + let parts = FramedParts { + inner: DontReadIntoThis, + readbuf: vec![0; INITIAL_CAPACITY * 2].into(), + writebuf: BytesMut::with_capacity(0), + }; + let framed = Framed::from_parts(parts, U32Codec); + let FramedParts { readbuf, .. } = framed.into_parts(); + + assert_eq!(readbuf.capacity(), INITIAL_CAPACITY * 2); +} + diff --git a/tokio-io/tests/framed_read.rs b/tokio-io/tests/framed_read.rs new file mode 100644 index 000000000..0dd327372 --- /dev/null +++ b/tokio-io/tests/framed_read.rs @@ -0,0 +1,215 @@ +extern crate tokio_io; +extern crate bytes; +extern crate futures; + +use tokio_io::AsyncRead; +use tokio_io::codec::{FramedRead, Decoder}; + +use bytes::{BytesMut, Buf, IntoBuf, BigEndian}; +use futures::Stream; +use futures::Async::{Ready, NotReady}; + +use std::io::{self, Read}; +use std::collections::VecDeque; + +macro_rules! mock { + ($($x:expr,)*) => {{ + let mut v = VecDeque::new(); + v.extend(vec![$($x),*]); + Mock { calls: v } + }}; +} + +struct U32Decoder; + +impl Decoder for U32Decoder { + type Item = u32; + type Error = io::Error; + + fn decode(&mut self, buf: &mut BytesMut) -> io::Result> { + if buf.len() < 4 { + return Ok(None); + } + + let n = buf.split_to(4).into_buf().get_u32::(); + Ok(Some(n)) + } +} + +#[test] +fn read_multi_frame_in_packet() { + let mock = mock! { + Ok(b"\x00\x00\x00\x00\x00\x00\x00\x01\x00\x00\x00\x02".to_vec()), + }; + + let mut framed = FramedRead::new(mock, U32Decoder); + assert_eq!(Ready(Some(0)), framed.poll().unwrap()); + assert_eq!(Ready(Some(1)), framed.poll().unwrap()); + assert_eq!(Ready(Some(2)), framed.poll().unwrap()); + assert_eq!(Ready(None), framed.poll().unwrap()); +} + +#[test] +fn read_multi_frame_across_packets() { + let mock = mock! { + Ok(b"\x00\x00\x00\x00".to_vec()), + Ok(b"\x00\x00\x00\x01".to_vec()), + Ok(b"\x00\x00\x00\x02".to_vec()), + }; + + let mut framed = FramedRead::new(mock, U32Decoder); + assert_eq!(Ready(Some(0)), framed.poll().unwrap()); + assert_eq!(Ready(Some(1)), framed.poll().unwrap()); + assert_eq!(Ready(Some(2)), framed.poll().unwrap()); + assert_eq!(Ready(None), framed.poll().unwrap()); +} + +#[test] +fn read_not_ready() { + let mock = mock! { + Err(io::Error::new(io::ErrorKind::WouldBlock, "")), + Ok(b"\x00\x00\x00\x00".to_vec()), + Ok(b"\x00\x00\x00\x01".to_vec()), + }; + + let mut framed = FramedRead::new(mock, U32Decoder); + assert_eq!(NotReady, framed.poll().unwrap()); + assert_eq!(Ready(Some(0)), framed.poll().unwrap()); + assert_eq!(Ready(Some(1)), framed.poll().unwrap()); + assert_eq!(Ready(None), framed.poll().unwrap()); +} + +#[test] +fn read_partial_then_not_ready() { + let mock = mock! { + Ok(b"\x00\x00".to_vec()), + Err(io::Error::new(io::ErrorKind::WouldBlock, "")), + Ok(b"\x00\x00\x00\x00\x00\x01\x00\x00\x00\x02".to_vec()), + }; + + let mut framed = FramedRead::new(mock, U32Decoder); + assert_eq!(NotReady, framed.poll().unwrap()); + assert_eq!(Ready(Some(0)), framed.poll().unwrap()); + assert_eq!(Ready(Some(1)), framed.poll().unwrap()); + assert_eq!(Ready(Some(2)), framed.poll().unwrap()); + assert_eq!(Ready(None), framed.poll().unwrap()); +} + +#[test] +fn read_err() { + let mock = mock! { + Err(io::Error::new(io::ErrorKind::Other, "")), + }; + + let mut framed = FramedRead::new(mock, U32Decoder); + assert_eq!(io::ErrorKind::Other, framed.poll().unwrap_err().kind()); +} + +#[test] +fn read_partial_then_err() { + let mock = mock! { + Ok(b"\x00\x00".to_vec()), + Err(io::Error::new(io::ErrorKind::Other, "")), + }; + + let mut framed = FramedRead::new(mock, U32Decoder); + assert_eq!(io::ErrorKind::Other, framed.poll().unwrap_err().kind()); +} + +#[test] +fn read_partial_would_block_then_err() { + let mock = mock! { + Ok(b"\x00\x00".to_vec()), + Err(io::Error::new(io::ErrorKind::WouldBlock, "")), + Err(io::Error::new(io::ErrorKind::Other, "")), + }; + + let mut framed = FramedRead::new(mock, U32Decoder); + assert_eq!(NotReady, framed.poll().unwrap()); + assert_eq!(io::ErrorKind::Other, framed.poll().unwrap_err().kind()); +} + +#[test] +fn huge_size() { + let data = [0; 32 * 1024]; + + let mut framed = FramedRead::new(&data[..], BigDecoder); + assert_eq!(Ready(Some(0)), framed.poll().unwrap()); + assert_eq!(Ready(None), framed.poll().unwrap()); + + struct BigDecoder; + + impl Decoder for BigDecoder { + type Item = u32; + type Error = io::Error; + + fn decode(&mut self, buf: &mut BytesMut) -> io::Result> { + if buf.len() < 32 * 1024 { + return Ok(None); + } + buf.split_to(32 * 1024); + Ok(Some(0)) + } + } +} + +#[test] +fn data_remaining_is_error() { + let data = [0; 5]; + + let mut framed = FramedRead::new(&data[..], U32Decoder); + assert_eq!(Ready(Some(0)), framed.poll().unwrap()); + assert!(framed.poll().is_err()); +} + +#[test] +fn multi_frames_on_eof() { + struct MyDecoder(Vec); + + impl Decoder for MyDecoder { + type Item = u32; + type Error = io::Error; + + fn decode(&mut self, _buf: &mut BytesMut) -> io::Result> { + unreachable!(); + } + + fn decode_eof(&mut self, _buf: &mut BytesMut) -> io::Result> { + if self.0.is_empty() { + return Ok(None); + } + + Ok(Some(self.0.remove(0))) + } + } + + let mut framed = FramedRead::new(mock!(), MyDecoder(vec![0, 1, 2, 3])); + assert_eq!(Ready(Some(0)), framed.poll().unwrap()); + assert_eq!(Ready(Some(1)), framed.poll().unwrap()); + assert_eq!(Ready(Some(2)), framed.poll().unwrap()); + assert_eq!(Ready(Some(3)), framed.poll().unwrap()); + assert_eq!(Ready(None), framed.poll().unwrap()); +} + +// ===== Mock ====== + +struct Mock { + calls: VecDeque>>, +} + +impl Read for Mock { + fn read(&mut self, dst: &mut [u8]) -> io::Result { + match self.calls.pop_front() { + Some(Ok(data)) => { + debug_assert!(dst.len() >= data.len()); + dst[..data.len()].copy_from_slice(&data[..]); + Ok(data.len()) + } + Some(Err(e)) => Err(e), + None => Ok(0), + } + } +} + +impl AsyncRead for Mock { +} diff --git a/tokio-io/tests/framed_write.rs b/tokio-io/tests/framed_write.rs new file mode 100644 index 000000000..801602684 --- /dev/null +++ b/tokio-io/tests/framed_write.rs @@ -0,0 +1,133 @@ +extern crate tokio_io; +extern crate bytes; +extern crate futures; + +use tokio_io::AsyncWrite; +use tokio_io::codec::{Encoder, FramedWrite}; + +use futures::{Sink, Poll}; +use bytes::{BytesMut, BufMut, BigEndian}; + +use std::io::{self, Write}; +use std::collections::VecDeque; + +macro_rules! mock { + ($($x:expr,)*) => {{ + let mut v = VecDeque::new(); + v.extend(vec![$($x),*]); + Mock { calls: v } + }}; +} + +struct U32Encoder; + +impl Encoder for U32Encoder { + type Item = u32; + type Error = io::Error; + + fn encode(&mut self, item: u32, dst: &mut BytesMut) -> io::Result<()> { + // Reserve space + dst.reserve(4); + dst.put_u32::(item); + Ok(()) + } +} + +#[test] +fn write_multi_frame_in_packet() { + let mock = mock! { + Ok(b"\x00\x00\x00\x00\x00\x00\x00\x01\x00\x00\x00\x02".to_vec()), + }; + + let mut framed = FramedWrite::new(mock, U32Encoder); + assert!(framed.start_send(0).unwrap().is_ready()); + assert!(framed.start_send(1).unwrap().is_ready()); + assert!(framed.start_send(2).unwrap().is_ready()); + + // Nothing written yet + assert_eq!(1, framed.get_ref().calls.len()); + + // Flush the writes + assert!(framed.poll_complete().unwrap().is_ready()); + + assert_eq!(0, framed.get_ref().calls.len()); +} + +#[test] +fn write_hits_backpressure() { + const ITER: usize = 2 * 1024; + + let mut mock = mock! { + // Block the `ITER`th write + Err(io::Error::new(io::ErrorKind::WouldBlock, "not ready")), + Ok(b"".to_vec()), + }; + + for i in 0..(ITER + 1) { + let mut b = BytesMut::with_capacity(4); + b.put_u32::(i as u32); + + // Append to the end + match mock.calls.back_mut().unwrap() { + &mut Ok(ref mut data) => { + // Write in 2kb chunks + if data.len() < ITER { + data.extend_from_slice(&b[..]); + continue; + } + } + _ => unreachable!(), + } + + // Push a new new chunk + mock.calls.push_back(Ok(b[..].to_vec())); + } + + let mut framed = FramedWrite::new(mock, U32Encoder); + + for i in 0..ITER { + assert!(framed.start_send(i as u32).unwrap().is_ready()); + } + + // This should reject + assert!(!framed.start_send(ITER as u32).unwrap().is_ready()); + + // This should succeed and start flushing the buffer. + assert!(framed.start_send(ITER as u32).unwrap().is_ready()); + + // Flush the rest of the buffer + assert!(framed.poll_complete().unwrap().is_ready()); + + // Ensure the mock is empty + assert_eq!(0, framed.get_ref().calls.len()); +} + +// ===== Mock ====== + +struct Mock { + calls: VecDeque>>, +} + +impl Write for Mock { + fn write(&mut self, src: &[u8]) -> io::Result { + match self.calls.pop_front() { + Some(Ok(data)) => { + assert!(src.len() >= data.len()); + assert_eq!(&data[..], &src[..data.len()]); + Ok(data.len()) + } + Some(Err(e)) => Err(e), + None => panic!("unexpected write; {:?}", src), + } + } + + fn flush(&mut self) -> io::Result<()> { + Ok(()) + } +} + +impl AsyncWrite for Mock { + fn shutdown(&mut self) -> Poll<(), io::Error> { + Ok(().into()) + } +} diff --git a/tokio-io/tests/length_delimited.rs b/tokio-io/tests/length_delimited.rs new file mode 100644 index 000000000..a94ae688d --- /dev/null +++ b/tokio-io/tests/length_delimited.rs @@ -0,0 +1,532 @@ +extern crate tokio_io; +extern crate futures; + +use tokio_io::{AsyncRead, AsyncWrite}; +use tokio_io::codec::length_delimited::*; + +use futures::{Stream, Sink, Poll}; +use futures::Async::*; + +use std::io; +use std::collections::VecDeque; + +macro_rules! mock { + ($($x:expr,)*) => {{ + let mut v = VecDeque::new(); + v.extend(vec![$($x),*]); + Mock { calls: v } + }}; +} + + +#[test] +fn read_empty_io_yields_nothing() { + let mut io = FramedRead::new(mock!()); + + assert_eq!(io.poll().unwrap(), Ready(None)); +} + +#[test] +fn read_single_frame_one_packet() { + let mut io = FramedRead::new(mock! { + Ok(b"\x00\x00\x00\x09abcdefghi"[..].into()), + }); + + assert_eq!(io.poll().unwrap(), Ready(Some(b"abcdefghi"[..].into()))); + assert_eq!(io.poll().unwrap(), Ready(None)); +} + +#[test] +fn read_single_frame_one_packet_little_endian() { + let mut io = Builder::new() + .little_endian() + .new_read(mock! { + Ok(b"\x09\x00\x00\x00abcdefghi"[..].into()), + }); + + assert_eq!(io.poll().unwrap(), Ready(Some(b"abcdefghi"[..].into()))); + assert_eq!(io.poll().unwrap(), Ready(None)); +} + +#[test] +fn read_single_multi_frame_one_packet() { + let mut data: Vec = vec![]; + data.extend_from_slice(b"\x00\x00\x00\x09abcdefghi"); + data.extend_from_slice(b"\x00\x00\x00\x03123"); + data.extend_from_slice(b"\x00\x00\x00\x0bhello world"); + + let mut io = FramedRead::new(mock! { + Ok(data.into()), + }); + + assert_eq!(io.poll().unwrap(), Ready(Some(b"abcdefghi"[..].into()))); + assert_eq!(io.poll().unwrap(), Ready(Some(b"123"[..].into()))); + assert_eq!(io.poll().unwrap(), Ready(Some(b"hello world"[..].into()))); + assert_eq!(io.poll().unwrap(), Ready(None)); +} + +#[test] +fn read_single_frame_multi_packet() { + let mut io = FramedRead::new(mock! { + Ok(b"\x00\x00"[..].into()), + Ok(b"\x00\x09abc"[..].into()), + Ok(b"defghi"[..].into()), + }); + + assert_eq!(io.poll().unwrap(), Ready(Some(b"abcdefghi"[..].into()))); + assert_eq!(io.poll().unwrap(), Ready(None)); +} + +#[test] +fn read_multi_frame_multi_packet() { + let mut io = FramedRead::new(mock! { + Ok(b"\x00\x00"[..].into()), + Ok(b"\x00\x09abc"[..].into()), + Ok(b"defghi"[..].into()), + Ok(b"\x00\x00\x00\x0312"[..].into()), + Ok(b"3\x00\x00\x00\x0bhello world"[..].into()), + }); + + assert_eq!(io.poll().unwrap(), Ready(Some(b"abcdefghi"[..].into()))); + assert_eq!(io.poll().unwrap(), Ready(Some(b"123"[..].into()))); + assert_eq!(io.poll().unwrap(), Ready(Some(b"hello world"[..].into()))); + assert_eq!(io.poll().unwrap(), Ready(None)); +} + +#[test] +fn read_single_frame_multi_packet_wait() { + let mut io = FramedRead::new(mock! { + Ok(b"\x00\x00"[..].into()), + Err(would_block()), + Ok(b"\x00\x09abc"[..].into()), + Err(would_block()), + Ok(b"defghi"[..].into()), + Err(would_block()), + }); + + assert_eq!(io.poll().unwrap(), NotReady); + assert_eq!(io.poll().unwrap(), NotReady); + assert_eq!(io.poll().unwrap(), Ready(Some(b"abcdefghi"[..].into()))); + assert_eq!(io.poll().unwrap(), NotReady); + assert_eq!(io.poll().unwrap(), Ready(None)); +} + +#[test] +fn read_multi_frame_multi_packet_wait() { + let mut io = FramedRead::new(mock! { + Ok(b"\x00\x00"[..].into()), + Err(would_block()), + Ok(b"\x00\x09abc"[..].into()), + Err(would_block()), + Ok(b"defghi"[..].into()), + Err(would_block()), + Ok(b"\x00\x00\x00\x0312"[..].into()), + Err(would_block()), + Ok(b"3\x00\x00\x00\x0bhello world"[..].into()), + Err(would_block()), + }); + + + assert_eq!(io.poll().unwrap(), NotReady); + assert_eq!(io.poll().unwrap(), NotReady); + assert_eq!(io.poll().unwrap(), Ready(Some(b"abcdefghi"[..].into()))); + assert_eq!(io.poll().unwrap(), NotReady); + assert_eq!(io.poll().unwrap(), NotReady); + assert_eq!(io.poll().unwrap(), Ready(Some(b"123"[..].into()))); + assert_eq!(io.poll().unwrap(), Ready(Some(b"hello world"[..].into()))); + assert_eq!(io.poll().unwrap(), NotReady); + assert_eq!(io.poll().unwrap(), Ready(None)); +} + +#[test] +fn read_incomplete_head() { + let mut io = FramedRead::new(mock! { + Ok(b"\x00\x00"[..].into()), + }); + + assert!(io.poll().is_err()); +} + +#[test] +fn read_incomplete_head_multi() { + let mut io = FramedRead::new(mock! { + Err(would_block()), + Ok(b"\x00"[..].into()), + Err(would_block()), + }); + + assert_eq!(io.poll().unwrap(), NotReady); + assert_eq!(io.poll().unwrap(), NotReady); + assert!(io.poll().is_err()); +} + +#[test] +fn read_incomplete_payload() { + let mut io = FramedRead::new(mock! { + Ok(b"\x00\x00\x00\x09ab"[..].into()), + Err(would_block()), + Ok(b"cd"[..].into()), + Err(would_block()), + }); + + assert_eq!(io.poll().unwrap(), NotReady); + assert_eq!(io.poll().unwrap(), NotReady); + assert!(io.poll().is_err()); +} + +#[test] +fn read_max_frame_len() { + let mut io = Builder::new() + .max_frame_length(5) + .new_read(mock! { + Ok(b"\x00\x00\x00\x09abcdefghi"[..].into()), + }); + + assert_eq!(io.poll().unwrap_err().kind(), io::ErrorKind::InvalidData); +} + +#[test] +fn read_update_max_frame_len_at_rest() { + let mut io = Builder::new() + .new_read(mock! { + Ok(b"\x00\x00\x00\x09abcdefghi"[..].into()), + Ok(b"\x00\x00\x00\x09abcdefghi"[..].into()), + }); + + assert_eq!(io.poll().unwrap(), Ready(Some(b"abcdefghi"[..].into()))); + io.set_max_frame_length(5); + assert_eq!(io.poll().unwrap_err().kind(), io::ErrorKind::InvalidData); +} + +#[test] +fn read_update_max_frame_len_in_flight() { + let mut io = Builder::new() + .new_read(mock! { + Ok(b"\x00\x00\x00\x09abcd"[..].into()), + Err(would_block()), + Ok(b"efghi"[..].into()), + Ok(b"\x00\x00\x00\x09abcdefghi"[..].into()), + }); + + assert_eq!(io.poll().unwrap(), NotReady); + io.set_max_frame_length(5); + assert_eq!(io.poll().unwrap(), Ready(Some(b"abcdefghi"[..].into()))); + assert_eq!(io.poll().unwrap_err().kind(), io::ErrorKind::InvalidData); +} + +#[test] +fn read_one_byte_length_field() { + let mut io = Builder::new() + .length_field_length(1) + .new_read(mock! { + Ok(b"\x09abcdefghi"[..].into()), + }); + + assert_eq!(io.poll().unwrap(), Ready(Some(b"abcdefghi"[..].into()))); + assert_eq!(io.poll().unwrap(), Ready(None)); +} + +#[test] +fn read_header_offset() { + let mut io = Builder::new() + .length_field_length(2) + .length_field_offset(4) + .new_read(mock! { + Ok(b"zzzz\x00\x09abcdefghi"[..].into()), + }); + + assert_eq!(io.poll().unwrap(), Ready(Some(b"abcdefghi"[..].into()))); + assert_eq!(io.poll().unwrap(), Ready(None)); +} + +#[test] +fn read_single_multi_frame_one_packet_skip_none_adjusted() { + let mut data: Vec = vec![]; + data.extend_from_slice(b"xx\x00\x09abcdefghi"); + data.extend_from_slice(b"yy\x00\x03123"); + data.extend_from_slice(b"zz\x00\x0bhello world"); + + let mut io = Builder::new() + .length_field_length(2) + .length_field_offset(2) + .num_skip(0) + .length_adjustment(4) + .new_read(mock! { + Ok(data.into()), + }); + + assert_eq!(io.poll().unwrap(), Ready(Some(b"xx\x00\x09abcdefghi"[..].into()))); + assert_eq!(io.poll().unwrap(), Ready(Some(b"yy\x00\x03123"[..].into()))); + assert_eq!(io.poll().unwrap(), Ready(Some(b"zz\x00\x0bhello world"[..].into()))); + assert_eq!(io.poll().unwrap(), Ready(None)); +} + +#[test] +fn read_single_multi_frame_one_packet_length_includes_head() { + let mut data: Vec = vec![]; + data.extend_from_slice(b"\x00\x0babcdefghi"); + data.extend_from_slice(b"\x00\x05123"); + data.extend_from_slice(b"\x00\x0dhello world"); + + let mut io = Builder::new() + .length_field_length(2) + .length_adjustment(-2) + .new_read(mock! { + Ok(data.into()), + }); + + assert_eq!(io.poll().unwrap(), Ready(Some(b"abcdefghi"[..].into()))); + assert_eq!(io.poll().unwrap(), Ready(Some(b"123"[..].into()))); + assert_eq!(io.poll().unwrap(), Ready(Some(b"hello world"[..].into()))); + assert_eq!(io.poll().unwrap(), Ready(None)); +} + +#[test] +fn write_single_frame_length_adjusted() { + let mut io = Builder::new() + .length_adjustment(-2) + .new_write(mock! { + Ok(b"\x00\x00\x00\x0b"[..].into()), + Ok(b"abcdefghi"[..].into()), + Ok(Flush), + }); + assert!(io.start_send("abcdefghi").unwrap().is_ready()); + assert!(io.poll_complete().unwrap().is_ready()); + assert!(io.get_ref().calls.is_empty()); +} + +#[test] +fn write_nothing_yields_nothing() { + let mut io: FramedWrite<_, &'static [u8]> = FramedWrite::new(mock!()); + assert!(io.poll_complete().unwrap().is_ready()); +} + +#[test] +fn write_single_frame_one_packet() { + let mut io = FramedWrite::new(mock! { + Ok(b"\x00\x00\x00\x09"[..].into()), + Ok(b"abcdefghi"[..].into()), + Ok(Flush), + }); + + assert!(io.start_send("abcdefghi").unwrap().is_ready()); + assert!(io.poll_complete().unwrap().is_ready()); + assert!(io.get_ref().calls.is_empty()); +} + +#[test] +fn write_single_multi_frame_one_packet() { + let mut io = FramedWrite::new(mock! { + Ok(b"\x00\x00\x00\x09"[..].into()), + Ok(b"abcdefghi"[..].into()), + Ok(b"\x00\x00\x00\x03"[..].into()), + Ok(b"123"[..].into()), + Ok(b"\x00\x00\x00\x0b"[..].into()), + Ok(b"hello world"[..].into()), + Ok(Flush), + }); + + assert!(io.start_send("abcdefghi").unwrap().is_ready()); + assert!(io.start_send("123").unwrap().is_ready()); + assert!(io.start_send("hello world").unwrap().is_ready()); + assert!(io.poll_complete().unwrap().is_ready()); + assert!(io.get_ref().calls.is_empty()); +} + +#[test] +fn write_single_multi_frame_multi_packet() { + let mut io = FramedWrite::new(mock! { + Ok(b"\x00\x00\x00\x09"[..].into()), + Ok(b"abcdefghi"[..].into()), + Ok(Flush), + Ok(b"\x00\x00\x00\x03"[..].into()), + Ok(b"123"[..].into()), + Ok(Flush), + Ok(b"\x00\x00\x00\x0b"[..].into()), + Ok(b"hello world"[..].into()), + Ok(Flush), + }); + + assert!(io.start_send("abcdefghi").unwrap().is_ready()); + assert!(io.poll_complete().unwrap().is_ready()); + assert!(io.start_send("123").unwrap().is_ready()); + assert!(io.poll_complete().unwrap().is_ready()); + assert!(io.start_send("hello world").unwrap().is_ready()); + assert!(io.poll_complete().unwrap().is_ready()); + assert!(io.get_ref().calls.is_empty()); +} + +#[test] +fn write_single_frame_would_block() { + let mut io = FramedWrite::new(mock! { + Err(would_block()), + Ok(b"\x00\x00"[..].into()), + Err(would_block()), + Ok(b"\x00\x09"[..].into()), + Ok(b"abcdefghi"[..].into()), + Ok(Flush), + }); + + assert!(io.start_send("abcdefghi").unwrap().is_ready()); + assert!(!io.poll_complete().unwrap().is_ready()); + assert!(!io.poll_complete().unwrap().is_ready()); + assert!(io.poll_complete().unwrap().is_ready()); + + assert!(io.get_ref().calls.is_empty()); +} + +#[test] +fn write_single_frame_little_endian() { + let mut io = Builder::new() + .little_endian() + .new_write(mock! { + Ok(b"\x09\x00\x00\x00"[..].into()), + Ok(b"abcdefghi"[..].into()), + Ok(Flush), + }); + + assert!(io.start_send("abcdefghi").unwrap().is_ready()); + assert!(io.poll_complete().unwrap().is_ready()); + assert!(io.get_ref().calls.is_empty()); +} + + +#[test] +fn write_single_frame_with_short_length_field() { + let mut io = Builder::new() + .length_field_length(1) + .new_write(mock! { + Ok(b"\x09"[..].into()), + Ok(b"abcdefghi"[..].into()), + Ok(Flush), + }); + + assert!(io.start_send("abcdefghi").unwrap().is_ready()); + assert!(io.poll_complete().unwrap().is_ready()); + assert!(io.get_ref().calls.is_empty()); +} + +#[test] +fn write_max_frame_len() { + let mut io = Builder::new() + .max_frame_length(5) + .new_write(mock! { }); + + assert_eq!(io.start_send("abcdef").unwrap_err().kind(), io::ErrorKind::InvalidInput); + assert!(io.get_ref().calls.is_empty()); +} + +#[test] +fn write_update_max_frame_len_at_rest() { + let mut io = Builder::new() + .new_write(mock! { + Ok(b"\x00\x00\x00\x06"[..].into()), + Ok(b"abcdef"[..].into()), + Ok(Flush), + }); + + assert!(io.start_send("abcdef").unwrap().is_ready()); + assert!(io.poll_complete().unwrap().is_ready()); + io.set_max_frame_length(5); + assert_eq!(io.start_send("abcdef").unwrap_err().kind(), io::ErrorKind::InvalidInput); + assert!(io.get_ref().calls.is_empty()); +} + +#[test] +fn write_update_max_frame_len_in_flight() { + let mut io = Builder::new() + .new_write(mock! { + Ok(b"\x00\x00\x00\x06"[..].into()), + Ok(b"ab"[..].into()), + Err(would_block()), + Ok(b"cdef"[..].into()), + Ok(Flush), + }); + + assert!(io.start_send("abcdef").unwrap().is_ready()); + assert!(!io.poll_complete().unwrap().is_ready()); + io.set_max_frame_length(5); + assert!(io.poll_complete().unwrap().is_ready()); + assert_eq!(io.start_send("abcdef").unwrap_err().kind(), io::ErrorKind::InvalidInput); + assert!(io.get_ref().calls.is_empty()); +} + +// ===== Test utils ===== + +fn would_block() -> io::Error { + io::Error::new(io::ErrorKind::WouldBlock, "would block") +} + +struct Mock { + calls: VecDeque>, +} + +enum Op { + Data(Vec), + Flush, +} + +use self::Op::*; + +impl io::Read for Mock { + fn read(&mut self, dst: &mut [u8]) -> io::Result { + match self.calls.pop_front() { + Some(Ok(Op::Data(data))) => { + debug_assert!(dst.len() >= data.len()); + dst[..data.len()].copy_from_slice(&data[..]); + Ok(data.len()) + } + Some(Ok(_)) => panic!(), + Some(Err(e)) => Err(e), + None => Ok(0), + } + } +} + +impl AsyncRead for Mock { +} + +impl io::Write for Mock { + fn write(&mut self, src: &[u8]) -> io::Result { + match self.calls.pop_front() { + Some(Ok(Op::Data(data))) => { + let len = data.len(); + assert!(src.len() >= len, "expect={:?}; actual={:?}", data, src); + assert_eq!(&data[..], &src[..len]); + Ok(len) + } + Some(Ok(_)) => panic!(), + Some(Err(e)) => Err(e), + None => Ok(0), + } + } + + fn flush(&mut self) -> io::Result<()> { + match self.calls.pop_front() { + Some(Ok(Op::Flush)) => { + Ok(()) + } + Some(Ok(_)) => panic!(), + Some(Err(e)) => Err(e), + None => Ok(()), + } + } +} + +impl AsyncWrite for Mock { + fn shutdown(&mut self) -> Poll<(), io::Error> { + Ok(Ready(())) + } +} + +impl<'a> From<&'a [u8]> for Op { + fn from(src: &'a [u8]) -> Op { + Op::Data(src.into()) + } +} + +impl From> for Op { + fn from(src: Vec) -> Op { + Op::Data(src) + } +}