diff --git a/azure-pipelines.yml b/azure-pipelines.yml index be5d1e9b5..6e948b034 100644 --- a/azure-pipelines.yml +++ b/azure-pipelines.yml @@ -11,13 +11,14 @@ jobs: # name: rustfmt # Test top level crate -# - template: ci/azure-test-stable.yml -# parameters: -# name: test_tokio -# displayName: Test tokio -# cross: true -# crates: -# - tokio +- template: ci/azure-test-stable.yml + parameters: + name: test_tokio + rust: $(nightly) + displayName: Test tokio + cross: true + crates: + - tokio # Test crates that are platform specific - template: ci/azure-test-stable.yml diff --git a/tokio/Cargo.toml b/tokio/Cargo.toml index 742e70805..79d8dca8c 100644 --- a/tokio/Cargo.toml +++ b/tokio/Cargo.toml @@ -85,6 +85,8 @@ tokio-tcp = { version = "0.2.0", optional = true, path = "../tokio-tcp" } tokio-uds = { version = "0.2.1", optional = true } [dev-dependencies] +tokio-test = { path = "../tokio-test" } +pin-utils = "0.1.0-alpha.4" env_logger = { version = "0.5", default-features = false } flate2 = { version = "1", features = ["tokio"] } futures-cpupool = "0.1" diff --git a/tokio/examples/README.md b/tokio/examples_old/README.md similarity index 100% rename from tokio/examples/README.md rename to tokio/examples_old/README.md diff --git a/tokio/examples/blocking.rs b/tokio/examples_old/blocking.rs similarity index 100% rename from tokio/examples/blocking.rs rename to tokio/examples_old/blocking.rs diff --git a/tokio/examples/chat-combinator-current-thread.rs b/tokio/examples_old/chat-combinator-current-thread.rs similarity index 100% rename from tokio/examples/chat-combinator-current-thread.rs rename to tokio/examples_old/chat-combinator-current-thread.rs diff --git a/tokio/examples/chat-combinator.rs b/tokio/examples_old/chat-combinator.rs similarity index 100% rename from tokio/examples/chat-combinator.rs rename to tokio/examples_old/chat-combinator.rs diff --git a/tokio/examples/chat.rs b/tokio/examples_old/chat.rs similarity index 100% rename from tokio/examples/chat.rs rename to tokio/examples_old/chat.rs diff --git a/tokio/examples/connect.rs b/tokio/examples_old/connect.rs similarity index 100% rename from tokio/examples/connect.rs rename to tokio/examples_old/connect.rs diff --git a/tokio/examples/echo-udp.rs b/tokio/examples_old/echo-udp.rs similarity index 100% rename from tokio/examples/echo-udp.rs rename to tokio/examples_old/echo-udp.rs diff --git a/tokio/examples/echo.rs b/tokio/examples_old/echo.rs similarity index 100% rename from tokio/examples/echo.rs rename to tokio/examples_old/echo.rs diff --git a/tokio/examples/hello_world.rs b/tokio/examples_old/hello_world.rs similarity index 100% rename from tokio/examples/hello_world.rs rename to tokio/examples_old/hello_world.rs diff --git a/tokio/examples/manual-runtime.rs b/tokio/examples_old/manual-runtime.rs similarity index 100% rename from tokio/examples/manual-runtime.rs rename to tokio/examples_old/manual-runtime.rs diff --git a/tokio/examples/print_each_packet.rs b/tokio/examples_old/print_each_packet.rs similarity index 100% rename from tokio/examples/print_each_packet.rs rename to tokio/examples_old/print_each_packet.rs diff --git a/tokio/examples/proxy.rs b/tokio/examples_old/proxy.rs similarity index 100% rename from tokio/examples/proxy.rs rename to tokio/examples_old/proxy.rs diff --git a/tokio/examples/tinydb.rs b/tokio/examples_old/tinydb.rs similarity index 100% rename from tokio/examples/tinydb.rs rename to tokio/examples_old/tinydb.rs diff --git a/tokio/examples/tinyhttp.rs b/tokio/examples_old/tinyhttp.rs similarity index 100% rename from tokio/examples/tinyhttp.rs rename to tokio/examples_old/tinyhttp.rs diff --git a/tokio/examples/udp-client.rs b/tokio/examples_old/udp-client.rs similarity index 100% rename from tokio/examples/udp-client.rs rename to tokio/examples_old/udp-client.rs diff --git a/tokio/examples/udp-codec.rs b/tokio/examples_old/udp-codec.rs similarity index 100% rename from tokio/examples/udp-codec.rs rename to tokio/examples_old/udp-codec.rs diff --git a/tokio/src/executor.rs b/tokio/src/executor.rs index 266da320c..347531316 100644 --- a/tokio/src/executor.rs +++ b/tokio/src/executor.rs @@ -40,7 +40,7 @@ //! [`spawn`]: fn.spawn.html use std::future::Future; -pub use tokio_executor::{Executor, TypedExecutor, DefaultExecutor, SpawnError}; +pub use tokio_executor::{DefaultExecutor, Executor, SpawnError, TypedExecutor}; /// Return value from the `spawn` function. /// @@ -68,7 +68,7 @@ pub struct Spawn(()); /// In this example, a server is started and `spawn` is used to start a new task /// that processes each received connection. /// -/// ```rust +/// ```rust,ignore /// # use futures::{Future, Stream}; /// use tokio::net::TcpListener; /// @@ -100,7 +100,8 @@ pub struct Spawn(()); /// /// [`DefaultExecutor`]: struct.DefaultExecutor.html pub fn spawn(f: F) -> Spawn -where F: Future + 'static + Send +where + F: Future + 'static + Send, { ::tokio_executor::spawn(f); Spawn(()) diff --git a/tokio/src/io/copy.rs b/tokio/src/io/copy.rs new file mode 100644 index 000000000..b407c193b --- /dev/null +++ b/tokio/src/io/copy.rs @@ -0,0 +1,107 @@ +use std::future::Future; +use std::io; +use std::pin::Pin; +use std::task::{Context, Poll}; +use tokio_io::{AsyncRead, AsyncWrite}; + +macro_rules! ready { + ($e:expr) => { + match $e { + ::std::task::Poll::Ready(t) => t, + ::std::task::Poll::Pending => return ::std::task::Poll::Pending, + } + }; +} + +/// 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<'a, R, W> { + reader: &'a mut R, + read_done: bool, + writer: &'a mut W, + 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<'a, R, W>(reader: &'a mut R, writer: &'a mut W) -> Copy<'a, R, W> +where + R: AsyncRead + Unpin, + W: AsyncWrite + Unpin, +{ + Copy { + reader, + read_done: false, + writer, + amt: 0, + pos: 0, + cap: 0, + buf: Box::new([0; 2048]), + } +} + +impl<'a, R, W> Future for Copy<'a, R, W> +where + R: AsyncRead + Unpin, + W: AsyncWrite + Unpin, +{ + type Output = io::Result; + + fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + loop { + // If our buffer is empty, then we need to read some data to + // continue. + if self.pos == self.cap && !self.read_done { + let me = &mut *self; + let n = ready!(Pin::new(&mut *me.reader).poll_read(cx, &mut me.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 me = &mut *self; + let i = ready!(Pin::new(&mut *me.writer).poll_write(cx, &me.buf[me.pos..me.cap]))?; + if i == 0 { + return Poll::Ready(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 { + let me = &mut *self; + ready!(Pin::new(&mut *me.writer).poll_flush(cx))?; + return Poll::Ready(Ok(self.amt)); + } + } + } +} diff --git a/tokio/src/io.rs b/tokio/src/io/mod.rs similarity index 92% rename from tokio/src/io.rs rename to tokio/src/io/mod.rs index 4d2b5e35c..0772a098a 100644 --- a/tokio/src/io.rs +++ b/tokio/src/io/mod.rs @@ -45,3 +45,11 @@ pub use tokio_fs::{stderr, stdin, stdout, Stderr, Stdin, Stdout}; // Re-export io::Error so that users don't have to deal // with conflicts when `use`ing `futures::io` and `std::io`. pub use std::io::{Error, ErrorKind, Result}; + +mod copy; +mod read; +mod write; + +pub use self::copy::{copy, Copy}; +pub use self::read::{read, Read}; +pub use self::write::{write, Write}; diff --git a/tokio/src/io/read.rs b/tokio/src/io/read.rs new file mode 100644 index 000000000..571375715 --- /dev/null +++ b/tokio/src/io/read.rs @@ -0,0 +1,43 @@ +use std::future::Future; +use std::io; +use std::marker::Unpin; +use std::pin::Pin; +use std::task::{Context, Poll}; +use tokio_io::AsyncRead; + +/// 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<'a, R>(reader: &'a mut R, buf: &'a mut [u8]) -> Read<'a, R> +where + R: AsyncRead + Unpin + ?Sized, +{ + Read { reader, 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<'a, R: ?Sized> { + reader: &'a mut R, + buf: &'a mut [u8], +} + +// forward Unpin +impl<'a, R: Unpin + ?Sized> Unpin for Read<'_, R> {} + +impl Future for Read<'_, R> +where + R: AsyncRead + Unpin + ?Sized, +{ + type Output = io::Result; + + fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + let me = &mut *self; + Pin::new(&mut *me.reader).poll_read(cx, me.buf) + } +} diff --git a/tokio/src/io/write.rs b/tokio/src/io/write.rs new file mode 100644 index 000000000..424f478de --- /dev/null +++ b/tokio/src/io/write.rs @@ -0,0 +1,37 @@ +use std::future::Future; +use std::io; +use std::marker::Unpin; +use std::pin::Pin; +use std::task::{Context, Poll}; +use tokio_io::AsyncWrite; + +/// A future to write some of the buffer to an `AsyncWrite`. +#[derive(Debug)] +pub struct Write<'a, W: ?Sized> { + writer: &'a mut W, + buf: &'a [u8], +} + +/// Tries to write some bytes from the given `buf` to the writer in an +/// asynchronous manner, returning a future. +pub fn write<'a, W>(writer: &'a mut W, buf: &'a [u8]) -> Write<'a, W> +where + W: AsyncWrite + Unpin + ?Sized, +{ + Write { writer, buf } +} + +// forward Unpin +impl<'a, W: Unpin + ?Sized> Unpin for Write<'_, W> {} + +impl Future for Write<'_, W> +where + W: AsyncWrite + Unpin + ?Sized, +{ + type Output = io::Result; + + fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + let me = &mut *self; + Pin::new(&mut *me.writer).poll_write(cx, me.buf) + } +} diff --git a/tokio/src/lib.rs b/tokio/src/lib.rs index 080e04639..7436a2409 100644 --- a/tokio/src/lib.rs +++ b/tokio/src/lib.rs @@ -28,7 +28,7 @@ //! //! A simple TCP echo server: //! -//! ```no_run +//! ```no_run,ignore //! use tokio::prelude::*; //! use tokio::io::copy; //! use tokio::net::TcpListener; diff --git a/tokio/src/reactor.rs b/tokio/src/reactor.rs index e343bcf4e..2edec1fd1 100644 --- a/tokio/src/reactor.rs +++ b/tokio/src/reactor.rs @@ -20,7 +20,7 @@ //! //! Let's start with a basic example, establishing a TCP connection. //! -//! ```rust +//! ```rust,ignore //! # fn dox() { //! use tokio::prelude::*; //! use tokio::net::TcpStream; @@ -134,6 +134,4 @@ //! [`std::io::Read`]: https://doc.rust-lang.org/std/io/trait.Read.html //! [`std::io::Write`]: https://doc.rust-lang.org/std/io/trait.Write.html -pub use tokio_reactor::{ - Handle, PollEvented, Reactor, Registration, Turn, -}; +pub use tokio_reactor::{Handle, PollEvented, Reactor, Registration, Turn}; diff --git a/tokio/src/runtime/current_thread/builder.rs b/tokio/src/runtime/current_thread/builder.rs index d7e5323c1..7b0f0c012 100644 --- a/tokio/src/runtime/current_thread/builder.rs +++ b/tokio/src/runtime/current_thread/builder.rs @@ -20,7 +20,7 @@ use std::io; /// /// # Examples /// -/// ``` +/// ```ignore /// use tokio::runtime::current_thread::Builder; /// use tokio_timer::clock::Clock; /// @@ -36,7 +36,7 @@ use std::io; #[derive(Debug)] pub struct Builder { // /// The clock to use - //clock: Clock, +//clock: Clock, } impl Builder { @@ -78,7 +78,8 @@ impl Builder { reactor_handle, //timer_handle, //self.clock.clone(), - executor); + executor, + ); Ok(runtime) } diff --git a/tokio/src/runtime/current_thread/mod.rs b/tokio/src/runtime/current_thread/mod.rs index 5ebeea875..971922a72 100644 --- a/tokio/src/runtime/current_thread/mod.rs +++ b/tokio/src/runtime/current_thread/mod.rs @@ -23,7 +23,7 @@ //! //! For example: //! -//! ``` +//! ```ignore //! use tokio::runtime::current_thread::Runtime; //! use tokio::prelude::*; //! use std::thread; @@ -68,7 +68,7 @@ mod builder; mod runtime; pub use self::builder::Builder; -pub use self::runtime::{Runtime, Handle}; +pub use self::runtime::{Handle, Runtime}; pub use tokio_current_thread::spawn; pub use tokio_current_thread::TaskExecutor; @@ -98,7 +98,6 @@ pub fn run(future: F) where F: Future + 'static, { - let mut r = Runtime::new().expect("failed to start runtime on current thread"); r.spawn(future); r.run().expect("failed to resolve remaining futures"); diff --git a/tokio/src/runtime/current_thread/runtime.rs b/tokio/src/runtime/current_thread/runtime.rs index 85aff37a0..0ef5988ec 100644 --- a/tokio/src/runtime/current_thread/runtime.rs +++ b/tokio/src/runtime/current_thread/runtime.rs @@ -1,13 +1,13 @@ use crate::runtime::current_thread::Builder; -use tokio_current_thread::{self as current_thread, CurrentThread}; use tokio_current_thread::Handle as ExecutorHandle; +use tokio_current_thread::{self as current_thread, CurrentThread}; use tokio_executor; use tokio_reactor::{self, Reactor}; //use tokio_timer::clock::{self, Clock}; //use tokio_timer::timer::{self, Timer}; +use std::error::Error; use std::fmt; use std::future::Future; -use std::error::Error; use std::io; /// Single-threaded runtime provides a way to start reactor @@ -39,7 +39,9 @@ impl Handle { /// This function panics if the spawn fails. Failure occurs if the `CurrentThread` /// instance of the `Handle` does not exist anymore. pub fn spawn(&self, future: F) -> Result<(), tokio_executor::SpawnError> - where F: Future + Send + 'static { + where + F: Future + Send + 'static, + { self.0.spawn(future) } @@ -101,8 +103,8 @@ impl Runtime { reactor_handle: tokio_reactor::Handle, //timer_handle: timer::Handle, //clock: Clock, - executor: CurrentThread) -> Runtime - { + executor: CurrentThread, + ) -> Runtime { Runtime { reactor_handle, //timer_handle, @@ -127,7 +129,7 @@ impl Runtime { /// /// # Examples /// - /// ```rust + /// ```rust,ignore /// # use futures::{future, Future, Stream}; /// use tokio::runtime::current_thread::Runtime; /// @@ -149,7 +151,8 @@ impl Runtime { /// This function panics if the spawn fails. Failure occurs if the executor /// is currently at capacity and is unable to spawn a new future. pub fn spawn(&mut self, future: F) -> &mut Self - where F: Future + 'static, + where + F: Future + 'static, { self.executor.spawn(future); self @@ -172,7 +175,8 @@ impl Runtime { /// The caller is responsible for ensuring that other spawned futures /// complete execution by calling `block_on` or `run`. pub fn block_on(&mut self, f: F) -> F::Output - where F: Future + where + F: Future, { self.enter(|executor| { // Run the provided future @@ -184,13 +188,12 @@ impl Runtime { /// spawned futures have completed. pub fn run(&mut self) -> Result<(), RunError> { self.enter(|executor| executor.run()) - .map_err(|e| RunError { - inner: e, - }) + .map_err(|e| RunError { inner: e }) } fn enter(&mut self, f: F) -> R - where F: FnOnce(&mut current_thread::Entered<'_, Parker>) -> R + where + F: FnOnce(&mut current_thread::Entered<'_, Parker>) -> R, { let Runtime { ref reactor_handle, @@ -208,16 +211,16 @@ impl Runtime { tokio_reactor::with_default(&reactor_handle, &mut enter, |enter| { //clock::with_default(clock, enter, |enter| { // timer::with_default(&timer_handle, enter, |enter| { - // The TaskExecutor is a fake executor that looks into the - // current single-threaded executor when used. This is a trick, - // because we need two mutable references to the executor (one - // to run the provided future, another to install as the default - // one). We use the fake one here as the default one. - let mut default_executor = current_thread::TaskExecutor::current(); - tokio_executor::with_default(&mut default_executor, enter, |enter| { - let mut executor = executor.enter(enter); - f(&mut executor) - }) + // The TaskExecutor is a fake executor that looks into the + // current single-threaded executor when used. This is a trick, + // because we need two mutable references to the executor (one + // to run the provided future, another to install as the default + // one). We use the fake one here as the default one. + let mut default_executor = current_thread::TaskExecutor::current(); + tokio_executor::with_default(&mut default_executor, enter, |enter| { + let mut executor = executor.enter(enter); + f(&mut executor) + }) // }) //}) }) diff --git a/tokio/src/runtime/mod.rs b/tokio/src/runtime/mod.rs index 84a7f390b..34df32230 100644 --- a/tokio/src/runtime/mod.rs +++ b/tokio/src/runtime/mod.rs @@ -35,7 +35,7 @@ //! "seed" the application, blocking the thread until the runtime becomes //! [idle]. //! -//! ```rust +//! ```rust,ignore //! # use futures::{Future, Stream}; //! use tokio::net::TcpListener; //! @@ -66,7 +66,7 @@ //! //! A [`Runtime`] instance can also be used directly. //! -//! ```rust +//! ```rust,ignore //! # use futures::{Future, Stream}; //! use tokio::runtime::Runtime; //! use tokio::net::TcpListener; @@ -111,11 +111,7 @@ pub mod current_thread; //mod threadpool; -pub use self::current_thread::{ - Builder, - Runtime, - run, -}; +pub use self::current_thread::{run, Builder, Runtime}; /* pub use self::threadpool::{ Builder, diff --git a/tokio/tests/buffered.rs b/tokio/tests/buffered.rs index 1cb5b09f8..11f86f794 100644 --- a/tokio/tests/buffered.rs +++ b/tokio/tests/buffered.rs @@ -1,3 +1,4 @@ +#![cfg(feature = "broken")] #![deny(warnings, rust_2018_idioms)] use env_logger; diff --git a/tokio/tests/clock.rs b/tokio/tests/clock.rs index 65ec81098..2fa22c9d3 100644 --- a/tokio/tests/clock.rs +++ b/tokio/tests/clock.rs @@ -1,3 +1,4 @@ +#![cfg(feature = "broken")] #![deny(warnings, rust_2018_idioms)] use env_logger; diff --git a/tokio/tests/drop-core.rs b/tokio/tests/drop-core.rs index d735c3437..98c136e82 100644 --- a/tokio/tests/drop-core.rs +++ b/tokio/tests/drop-core.rs @@ -1,3 +1,4 @@ +#![cfg(feature = "broken")] #![deny(warnings, rust_2018_idioms)] use futures::future; diff --git a/tokio/tests/enumerate.rs b/tokio/tests/enumerate.rs index 76367a322..eaf2766b8 100644 --- a/tokio/tests/enumerate.rs +++ b/tokio/tests/enumerate.rs @@ -1,3 +1,4 @@ +#![cfg(feature = "broken")] #![deny(warnings, rust_2018_idioms)] use futures::sync::mpsc; diff --git a/tokio/tests/global.rs b/tokio/tests/global.rs index da6af2e77..52111a13a 100644 --- a/tokio/tests/global.rs +++ b/tokio/tests/global.rs @@ -1,3 +1,4 @@ +#![cfg(feature = "broken")] #![deny(warnings, rust_2018_idioms)] use env_logger; diff --git a/tokio/tests/io.rs b/tokio/tests/io.rs new file mode 100644 index 000000000..f708ce605 --- /dev/null +++ b/tokio/tests/io.rs @@ -0,0 +1,134 @@ +use bytes::BytesMut; +use pin_utils::pin_mut; +use std::future::Future; +use std::io; +use std::pin::Pin; +use std::task::{Context, Poll}; +use tokio::io::{AsyncRead, AsyncWrite}; +use tokio_test::assert_ready_ok; +use tokio_test::task::MockTask; + +#[test] +fn write() { + struct Wr(BytesMut); + + impl AsyncWrite for Wr { + fn poll_write( + mut self: Pin<&mut Self>, + _cx: &mut Context<'_>, + buf: &[u8], + ) -> Poll> { + self.0.extend(buf); + Ok(buf.len()).into() + } + + fn poll_flush(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll> { + Ok(()).into() + } + + fn poll_shutdown(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll> { + Ok(()).into() + } + } + + let mut task = MockTask::new(); + + task.enter(|cx| { + let mut wr = Wr(BytesMut::with_capacity(64)); + + let write = tokio::io::write(&mut wr, "hello world".as_bytes()); + pin_mut!(write); + + let n = assert_ready_ok!(write.poll(cx)); + assert_eq!(n, 11); + }); +} + +#[test] +fn read() { + struct Rd; + + impl AsyncRead for Rd { + fn poll_read( + self: Pin<&mut Self>, + _cx: &mut Context<'_>, + buf: &mut [u8], + ) -> Poll> { + buf[0..11].copy_from_slice(b"hello world"); + Poll::Ready(Ok(11)) + } + } + + let mut buf = Box::new([0; 11]); + let mut task = MockTask::new(); + + task.enter(|cx| { + let mut rd = Rd; + + let read = tokio::io::read(&mut rd, &mut buf[..]); + pin_mut!(read); + + let n = assert_ready_ok!(read.poll(cx)); + assert_eq!(n, 11); + assert_eq!(buf[..], b"hello world"[..]); + }); +} + +#[test] +fn copy() { + struct Rd(bool); + + impl AsyncRead for Rd { + fn poll_read( + mut self: Pin<&mut Self>, + _cx: &mut Context<'_>, + buf: &mut [u8], + ) -> Poll> { + if self.0 { + buf[0..11].copy_from_slice(b"hello world"); + self.0 = false; + Poll::Ready(Ok(11)) + } else { + Poll::Ready(Ok(0)) + } + } + } + + struct Wr(BytesMut); + + impl Unpin for Wr {} + impl AsyncWrite for Wr { + fn poll_write( + mut self: Pin<&mut Self>, + _cx: &mut Context<'_>, + buf: &[u8], + ) -> Poll> { + self.0.extend(buf); + Ok(buf.len()).into() + } + + fn poll_flush(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll> { + Ok(()).into() + } + + fn poll_shutdown(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll> { + Ok(()).into() + } + } + + let buf = BytesMut::with_capacity(64); + let mut task = MockTask::new(); + + task.enter(|cx| { + let mut rd = Rd(true); + let mut wr = Wr(buf); + + let copy = tokio::io::copy(&mut rd, &mut wr); + pin_mut!(copy); + + let n = assert_ready_ok!(copy.poll(cx)); + + assert_eq!(n, 11); + assert_eq!(wr.0[..], b"hello world"[..]); + }); +} diff --git a/tokio/tests/length_delimited.rs b/tokio/tests/length_delimited.rs index 65ab5564a..2b2fe563c 100644 --- a/tokio/tests/length_delimited.rs +++ b/tokio/tests/length_delimited.rs @@ -1,3 +1,4 @@ +#![cfg(feature = "broken")] #![deny(warnings, rust_2018_idioms)] use bytes::{BufMut, Bytes, BytesMut}; diff --git a/tokio/tests/line-frames.rs b/tokio/tests/line-frames.rs index 35629683d..42e884af8 100644 --- a/tokio/tests/line-frames.rs +++ b/tokio/tests/line-frames.rs @@ -1,3 +1,4 @@ +#![cfg(feature = "broken")] #![deny(warnings, rust_2018_idioms)] use bytes::{BufMut, BytesMut}; diff --git a/tokio/tests/pipe-hup.rs b/tokio/tests/pipe-hup.rs index bb1992873..4d78ace4c 100644 --- a/tokio/tests/pipe-hup.rs +++ b/tokio/tests/pipe-hup.rs @@ -1,3 +1,4 @@ +#![cfg(feature = "broken")] #![cfg(unix)] #![deny(warnings, rust_2018_idioms)] diff --git a/tokio/tests/reactor.rs b/tokio/tests/reactor.rs index 1f8b9902b..5c45cac86 100644 --- a/tokio/tests/reactor.rs +++ b/tokio/tests/reactor.rs @@ -1,3 +1,4 @@ +#![cfg(feature = "broken")] #![deny(warnings, rust_2018_idioms)] use futures::executor::{spawn, Notify, Spawn}; diff --git a/tokio/tests/runtime.rs b/tokio/tests/runtime.rs index 4ec292d95..780c3c115 100644 --- a/tokio/tests/runtime.rs +++ b/tokio/tests/runtime.rs @@ -1,3 +1,4 @@ +#![cfg(feature = "broken")] #![deny(warnings, rust_2018_idioms)] use env_logger; diff --git a/tokio/tests/timer.rs b/tokio/tests/timer.rs index 8cf8ab844..8790f5bad 100644 --- a/tokio/tests/timer.rs +++ b/tokio/tests/timer.rs @@ -1,3 +1,4 @@ +#![cfg(feature = "broken")] #![deny(warnings, rust_2018_idioms)] use env_logger;