diff --git a/.cirrus.yml b/.cirrus.yml index 43b0b51fd..e78c2d015 100644 --- a/.cirrus.yml +++ b/.cirrus.yml @@ -8,7 +8,7 @@ freebsd_instance: task: name: FreeBSD 12.0 env: - LOOM_MAX_DURATION: 10 + LOOM_MAX_PREEMPTIONS: 2 setup_script: - pkg install -y curl - curl https://sh.rustup.rs -sSf --output rustup.sh @@ -33,7 +33,7 @@ task: echo "~~~~~~~~~~~~~~~~~~~~" test_script: - . $HOME/.cargo/env - - cargo test --all --lib && cargo test --all --tests + - cargo test --all - cargo doc --all --no-deps # TODO: Re-enable # i686_test_script: diff --git a/azure-pipelines.yml b/azure-pipelines.yml index d6777c177..22cd40e13 100644 --- a/azure-pipelines.yml +++ b/azure-pipelines.yml @@ -26,7 +26,17 @@ jobs: cross: true crates: tokio: - - default + - codec + - fs + - io + - reactor + - rt-full + - net + - sync + - tcp + - timer + - udp + - uds # Test crates that are platform specific - template: ci/azure-test-stable.yml @@ -42,7 +52,6 @@ jobs: tokio-signal: [] tokio-tcp: - async-traits -# - tokio-tls tokio-udp: [] tokio-uds: - async-traits @@ -54,7 +63,7 @@ jobs: displayName: Test sub crates - rust: $(nightly) crates: - # - tokio-buf + tokio-buf: [] tokio-codec: [] tokio-current-thread: [] tokio-executor: [] diff --git a/ci/azure-test-stable.yml b/ci/azure-test-stable.yml index 67ff590a3..0e93f6161 100644 --- a/ci/azure-test-stable.yml +++ b/ci/azure-test-stable.yml @@ -20,40 +20,42 @@ jobs: # rust_version: stable rust_version: ${{ parameters.rust }} -# - template: azure-is-release.yml -# -# - ${{ each crate in parameters.crates }}: -# - script: cargo test -# env: -# LOOM_MAX_DURATION: 10 -# CI: 'True' -# displayName: cargo test -p ${{ crate }} -# workingDirectory: $(Build.SourcesDirectory)/${{ crate }} -# condition: and(succeeded(), ne(variables['isRelease'], 'true')) - - - template: azure-patch-crates.yml + - template: azure-is-release.yml - ${{ each crate in parameters.crates }}: # Run with default crate features - - script: cargo test --tests + - script: cargo test env: - LOOM_MAX_DURATION: 10 + LOOM_MAX_PREEMPTIONS: 2 CI: 'True' - displayName: ${{ crate.key }} - cargo test --tests - workingDirectory: $(Build.SourcesDirectory)/${{ crate.key }} - - script: cargo test --examples - displayName: ${{ crate.key }} - cargo test --examples + displayName: ${{ crate.key }} - cargo test workingDirectory: $(Build.SourcesDirectory)/${{ crate.key }} # Run with each specified feature - ${{ each feature in crate.value }}: - script: cargo test --tests --no-default-features --features ${{ feature }} env: - LOOM_MAX_DURATION: 10 + LOOM_MAX_PREEMPTIONS: 2 CI: 'True' - displayName: ${{ crate.key }} - cargo test --tests --features ${{ feature }} + displayName: ${{ crate.key }} - cargo test --features ${{ feature }} workingDirectory: $(Build.SourcesDirectory)/${{ crate.key }} - - script: cargo test --examples --no-default-features --features ${{ feature }} - displayName: ${{ crate.key }} - cargo test --examples --features ${{ feature }} + - template: azure-patch-crates.yml + + - ${{ each crate in parameters.crates }}: + # Run with default crate features + - script: cargo test + env: + LOOM_MAX_PREEMPTIONS: 2 + CI: 'True' + displayName: ${{ crate.key }} - cargo test + workingDirectory: $(Build.SourcesDirectory)/${{ crate.key }} + + # Run with each specified feature + - ${{ each feature in crate.value }}: + - script: cargo test --tests --no-default-features --features ${{ feature }} + env: + LOOM_MAX_PREEMPTIONS: 2 + CI: 'True' + displayName: ${{ crate.key }} - cargo test --features ${{ feature }} workingDirectory: $(Build.SourcesDirectory)/${{ crate.key }} diff --git a/tokio-codec/Cargo.toml b/tokio-codec/Cargo.toml index 86aa1f8e0..8f56d07ca 100644 --- a/tokio-codec/Cargo.toml +++ b/tokio-codec/Cargo.toml @@ -24,11 +24,11 @@ publish = false [dependencies] tokio-io = { version = "0.2.0", path = "../tokio-io" } bytes = "0.4.7" -futures-core-preview = "0.3.0-alpha.17" -futures-sink-preview = "0.3.0-alpha.17" +futures-core-preview = "= 0.3.0-alpha.17" +futures-sink-preview = "= 0.3.0-alpha.17" log = "0.4" [dev-dependencies] -futures-preview = "0.3.0-alpha.17" -tokio-current-thread = { version = "0.2.0", path = "../tokio-current-thread" } +futures-util-preview = "= 0.3.0-alpha.17" +tokio = { version = "0.2.0", path = "../tokio" } tokio-test = { version = "0.2.0", path = "../tokio-test" } diff --git a/tokio-codec/src/length_delimited.rs b/tokio-codec/src/length_delimited.rs index f08f93176..336ca1445 100644 --- a/tokio-codec/src/length_delimited.rs +++ b/tokio-codec/src/length_delimited.rs @@ -41,9 +41,8 @@ //! ``` //! #![feature(async_await)] //! -//! use tokio_io::{AsyncRead, AsyncWrite}; -//! use tokio_codec::{Framed, LengthDelimitedCodec}; -//! use futures::SinkExt; +//! use tokio::codec::{Framed, LengthDelimitedCodec}; +//! use tokio::prelude::*; //! //! use bytes::Bytes; //! diff --git a/tokio-codec/tests/framed.rs b/tokio-codec/tests/framed.rs index ec5bb5a5f..624f8ffb9 100644 --- a/tokio-codec/tests/framed.rs +++ b/tokio-codec/tests/framed.rs @@ -1,17 +1,15 @@ +#![feature(async_await)] #![deny(warnings, rust_2018_idioms)] +use tokio::prelude::*; +use tokio_codec::{Decoder, Encoder, Framed, FramedParts}; +use tokio_test::assert_ok; + +use bytes::{Buf, BufMut, BytesMut, IntoBuf}; use std::io::{self, Read}; use std::pin::Pin; use std::task::{Context, Poll}; -use tokio_codec::{Decoder, Encoder, Framed, FramedParts}; -use tokio_current_thread::block_on_all; -use tokio_io::AsyncRead; - -use bytes::{Buf, BufMut, BytesMut, IntoBuf}; -use futures::future::FutureExt; -use futures::stream::StreamExt; - const INITIAL_CAPACITY: usize = 8 * 1024; /// Encode and decode u32 values. @@ -65,19 +63,13 @@ impl AsyncRead for DontReadIntoThis { } } -#[test] -fn can_read_from_existing_buf() { +#[tokio::test] +async fn can_read_from_existing_buf() { let mut parts = FramedParts::new(DontReadIntoThis, U32Codec); parts.read_buf = vec![0, 0, 0, 42].into(); - let framed = Framed::from_parts(parts); - - let num = block_on_all( - framed - .into_future() - .map(|(first_num, _)| first_num.unwrap()), - ) - .unwrap(); + let mut framed = Framed::from_parts(parts); + let num = assert_ok!(framed.next().await.unwrap()); assert_eq!(num, 42); } diff --git a/tokio-codec/tests/framed_read.rs b/tokio-codec/tests/framed_read.rs index 64d574c08..e4ef53c28 100644 --- a/tokio-codec/tests/framed_read.rs +++ b/tokio-codec/tests/framed_read.rs @@ -1,19 +1,18 @@ +#![feature(async_await)] #![deny(warnings, rust_2018_idioms)] +use tokio::prelude::*; +use tokio_codec::{Decoder, FramedRead}; +use tokio_test::assert_ready; +use tokio_test::task::MockTask; + +use bytes::{Buf, BytesMut, IntoBuf}; use std::collections::VecDeque; -use std::io::{self, Read}; +use std::io; use std::pin::Pin; use std::task::Poll::{Pending, Ready}; use std::task::{Context, Poll}; -use bytes::{Buf, BytesMut, IntoBuf}; -use futures::Stream; - -use tokio_codec::{Decoder, FramedRead}; -use tokio_io::AsyncRead; -use tokio_test::assert_ready; -use tokio_test::task::MockTask; - macro_rules! mock { ($($x:expr,)*) => {{ let mut v = VecDeque::new(); @@ -261,29 +260,23 @@ 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 { fn poll_read( - self: Pin<&mut Self>, + mut self: Pin<&mut Self>, _cx: &mut Context<'_>, buf: &mut [u8], ) -> Poll> { - match Pin::get_mut(self).read(buf) { - Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => Pending, - other => Ready(other), + use io::ErrorKind::WouldBlock; + + match self.calls.pop_front() { + Some(Ok(data)) => { + debug_assert!(buf.len() >= data.len()); + buf[..data.len()].copy_from_slice(&data[..]); + Ready(Ok(data.len())) + } + Some(Err(ref e)) if e.kind() == WouldBlock => Pending, + Some(Err(e)) => Ready(Err(e)), + None => Ready(Ok(0)), } } } @@ -293,10 +286,10 @@ struct Slice<'a>(&'a [u8]); impl<'a> AsyncRead for Slice<'a> { fn poll_read( - self: Pin<&mut Self>, - _cx: &mut Context<'_>, + mut self: Pin<&mut Self>, + cx: &mut Context<'_>, buf: &mut [u8], ) -> Poll> { - Ready(Pin::get_mut(self).0.read(buf)) + Pin::new(&mut self.0).poll_read(cx, buf) } } diff --git a/tokio-codec/tests/length_delimited.rs b/tokio-codec/tests/length_delimited.rs new file mode 100644 index 000000000..8c1eaed23 --- /dev/null +++ b/tokio-codec/tests/length_delimited.rs @@ -0,0 +1,761 @@ +#![deny(warnings, rust_2018_idioms)] + +use tokio::codec::*; +use tokio::io::{AsyncRead, AsyncWrite}; +use tokio::prelude::*; +use tokio_test::task::MockTask; +use tokio_test::{ + assert_err, assert_ok, assert_pending, assert_ready, assert_ready_err, assert_ready_ok, +}; + +use bytes::{BufMut, Bytes, BytesMut}; +use futures_util::pin_mut; +use std::collections::VecDeque; +use std::io; +use std::pin::Pin; +use std::task::Poll::*; +use std::task::{Context, Poll}; + +macro_rules! mock { + ($($x:expr,)*) => {{ + let mut v = VecDeque::new(); + v.extend(vec![$($x),*]); + Mock { calls: v } + }}; +} + +macro_rules! assert_next_eq { + ($io:ident, $expect:expr) => {{ + MockTask::new().enter(|cx| { + let res = assert_ready!($io.as_mut().poll_next(cx)); + match res { + Some(Ok(v)) => assert_eq!(v, $expect.as_ref()), + Some(Err(e)) => panic!("error = {:?}", e), + None => panic!("none"), + } + }); + }}; +} + +macro_rules! assert_next_pending { + ($io:ident) => {{ + MockTask::new().enter(|cx| match $io.as_mut().poll_next(cx) { + Ready(Some(Ok(v))) => panic!("value = {:?}", v), + Ready(Some(Err(e))) => panic!("error = {:?}", e), + Ready(None) => panic!("done"), + Pending => {} + }); + }}; +} + +macro_rules! assert_next_err { + ($io:ident) => {{ + MockTask::new().enter(|cx| match $io.as_mut().poll_next(cx) { + Ready(Some(Ok(v))) => panic!("value = {:?}", v), + Ready(Some(Err(_))) => {} + Ready(None) => panic!("done"), + Pending => panic!("pending"), + }); + }}; +} + +macro_rules! assert_done { + ($io:ident) => {{ + MockTask::new().enter(|cx| { + let res = assert_ready!($io.as_mut().poll_next(cx)); + match res { + Some(Ok(v)) => panic!("value = {:?}", v), + Some(Err(e)) => panic!("error = {:?}", e), + None => {} + } + }); + }}; +} + +#[test] +fn read_empty_io_yields_nothing() { + let io = Box::pin(FramedRead::new(mock!(), LengthDelimitedCodec::new())); + pin_mut!(io); + + assert_done!(io); +} + +#[test] +fn read_single_frame_one_packet() { + let io = FramedRead::new( + mock! { + data(b"\x00\x00\x00\x09abcdefghi"), + }, + LengthDelimitedCodec::new(), + ); + pin_mut!(io); + + assert_next_eq!(io, b"abcdefghi"); + assert_done!(io); +} + +#[test] +fn read_single_frame_one_packet_little_endian() { + let io = length_delimited::Builder::new() + .little_endian() + .new_read(mock! { + data(b"\x09\x00\x00\x00abcdefghi"), + }); + pin_mut!(io); + + assert_next_eq!(io, b"abcdefghi"); + assert_done!(io); +} + +#[test] +fn read_single_frame_one_packet_native_endian() { + let d = if cfg!(target_endian = "big") { + b"\x00\x00\x00\x09abcdefghi" + } else { + b"\x09\x00\x00\x00abcdefghi" + }; + let io = length_delimited::Builder::new() + .native_endian() + .new_read(mock! { + data(d), + }); + pin_mut!(io); + + assert_next_eq!(io, b"abcdefghi"); + assert_done!(io); +} + +#[test] +fn read_single_multi_frame_one_packet() { + let mut d: Vec = vec![]; + d.extend_from_slice(b"\x00\x00\x00\x09abcdefghi"); + d.extend_from_slice(b"\x00\x00\x00\x03123"); + d.extend_from_slice(b"\x00\x00\x00\x0bhello world"); + + let io = FramedRead::new( + mock! { + data(&d), + }, + LengthDelimitedCodec::new(), + ); + pin_mut!(io); + + assert_next_eq!(io, b"abcdefghi"); + assert_next_eq!(io, b"123"); + assert_next_eq!(io, b"hello world"); + assert_done!(io); +} + +#[test] +fn read_single_frame_multi_packet() { + let io = FramedRead::new( + mock! { + data(b"\x00\x00"), + data(b"\x00\x09abc"), + data(b"defghi"), + }, + LengthDelimitedCodec::new(), + ); + pin_mut!(io); + + assert_next_eq!(io, b"abcdefghi"); + assert_done!(io); +} + +#[test] +fn read_multi_frame_multi_packet() { + let io = FramedRead::new( + mock! { + data(b"\x00\x00"), + data(b"\x00\x09abc"), + data(b"defghi"), + data(b"\x00\x00\x00\x0312"), + data(b"3\x00\x00\x00\x0bhello world"), + }, + LengthDelimitedCodec::new(), + ); + pin_mut!(io); + + assert_next_eq!(io, b"abcdefghi"); + assert_next_eq!(io, b"123"); + assert_next_eq!(io, b"hello world"); + assert_done!(io); +} + +#[test] +fn read_single_frame_multi_packet_wait() { + let io = FramedRead::new( + mock! { + data(b"\x00\x00"), + Pending, + data(b"\x00\x09abc"), + Pending, + data(b"defghi"), + Pending, + }, + LengthDelimitedCodec::new(), + ); + pin_mut!(io); + + assert_next_pending!(io); + assert_next_pending!(io); + assert_next_eq!(io, b"abcdefghi"); + assert_next_pending!(io); + assert_done!(io); +} + +#[test] +fn read_multi_frame_multi_packet_wait() { + let io = FramedRead::new( + mock! { + data(b"\x00\x00"), + Pending, + data(b"\x00\x09abc"), + Pending, + data(b"defghi"), + Pending, + data(b"\x00\x00\x00\x0312"), + Pending, + data(b"3\x00\x00\x00\x0bhello world"), + Pending, + }, + LengthDelimitedCodec::new(), + ); + pin_mut!(io); + + assert_next_pending!(io); + assert_next_pending!(io); + assert_next_eq!(io, b"abcdefghi"); + assert_next_pending!(io); + assert_next_pending!(io); + assert_next_eq!(io, b"123"); + assert_next_eq!(io, b"hello world"); + assert_next_pending!(io); + assert_done!(io); +} + +#[test] +fn read_incomplete_head() { + let io = FramedRead::new( + mock! { + data(b"\x00\x00"), + }, + LengthDelimitedCodec::new(), + ); + pin_mut!(io); + + assert_next_err!(io); +} + +#[test] +fn read_incomplete_head_multi() { + let io = FramedRead::new( + mock! { + Pending, + data(b"\x00"), + Pending, + }, + LengthDelimitedCodec::new(), + ); + pin_mut!(io); + + assert_next_pending!(io); + assert_next_pending!(io); + assert_next_err!(io); +} + +#[test] +fn read_incomplete_payload() { + let io = FramedRead::new( + mock! { + data(b"\x00\x00\x00\x09ab"), + Pending, + data(b"cd"), + Pending, + }, + LengthDelimitedCodec::new(), + ); + pin_mut!(io); + + assert_next_pending!(io); + assert_next_pending!(io); + assert_next_err!(io); +} + +#[test] +fn read_max_frame_len() { + let io = length_delimited::Builder::new() + .max_frame_length(5) + .new_read(mock! { + data(b"\x00\x00\x00\x09abcdefghi"), + }); + pin_mut!(io); + + assert_next_err!(io); +} + +#[test] +fn read_update_max_frame_len_at_rest() { + let io = length_delimited::Builder::new().new_read(mock! { + data(b"\x00\x00\x00\x09abcdefghi"), + data(b"\x00\x00\x00\x09abcdefghi"), + }); + pin_mut!(io); + + assert_next_eq!(io, b"abcdefghi"); + io.decoder_mut().set_max_frame_length(5); + assert_next_err!(io); +} + +#[test] +fn read_update_max_frame_len_in_flight() { + let io = length_delimited::Builder::new().new_read(mock! { + data(b"\x00\x00\x00\x09abcd"), + Pending, + data(b"efghi"), + data(b"\x00\x00\x00\x09abcdefghi"), + }); + pin_mut!(io); + + assert_next_pending!(io); + io.decoder_mut().set_max_frame_length(5); + assert_next_eq!(io, b"abcdefghi"); + assert_next_err!(io); +} + +#[test] +fn read_one_byte_length_field() { + let io = length_delimited::Builder::new() + .length_field_length(1) + .new_read(mock! { + data(b"\x09abcdefghi"), + }); + pin_mut!(io); + + assert_next_eq!(io, b"abcdefghi"); + assert_done!(io); +} + +#[test] +fn read_header_offset() { + let io = length_delimited::Builder::new() + .length_field_length(2) + .length_field_offset(4) + .new_read(mock! { + data(b"zzzz\x00\x09abcdefghi"), + }); + pin_mut!(io); + + assert_next_eq!(io, b"abcdefghi"); + assert_done!(io); +} + +#[test] +fn read_single_multi_frame_one_packet_skip_none_adjusted() { + let mut d: Vec = vec![]; + d.extend_from_slice(b"xx\x00\x09abcdefghi"); + d.extend_from_slice(b"yy\x00\x03123"); + d.extend_from_slice(b"zz\x00\x0bhello world"); + + let io = length_delimited::Builder::new() + .length_field_length(2) + .length_field_offset(2) + .num_skip(0) + .length_adjustment(4) + .new_read(mock! { + data(&d), + }); + pin_mut!(io); + + assert_next_eq!(io, b"xx\x00\x09abcdefghi"); + assert_next_eq!(io, b"yy\x00\x03123"); + assert_next_eq!(io, b"zz\x00\x0bhello world"); + assert_done!(io); +} + +#[test] +fn read_single_multi_frame_one_packet_length_includes_head() { + let mut d: Vec = vec![]; + d.extend_from_slice(b"\x00\x0babcdefghi"); + d.extend_from_slice(b"\x00\x05123"); + d.extend_from_slice(b"\x00\x0dhello world"); + + let io = length_delimited::Builder::new() + .length_field_length(2) + .length_adjustment(-2) + .new_read(mock! { + data(&d), + }); + pin_mut!(io); + + assert_next_eq!(io, b"abcdefghi"); + assert_next_eq!(io, b"123"); + assert_next_eq!(io, b"hello world"); + assert_done!(io); +} + +#[test] +fn write_single_frame_length_adjusted() { + let io = length_delimited::Builder::new() + .length_adjustment(-2) + .new_write(mock! { + data(b"\x00\x00\x00\x0b"), + data(b"abcdefghi"), + flush(), + }); + pin_mut!(io); + + MockTask::new().enter(|cx| { + assert_ready_ok!(io.as_mut().poll_ready(cx)); + assert_ok!(io.as_mut().start_send(Bytes::from("abcdefghi"))); + assert_ready_ok!(io.as_mut().poll_flush(cx)); + assert!(io.get_ref().calls.is_empty()); + }); +} + +#[test] +fn write_nothing_yields_nothing() { + let io = FramedWrite::new(mock!(), LengthDelimitedCodec::new()); + pin_mut!(io); + + MockTask::new().enter(|cx| { + assert_ready_ok!(io.poll_flush(cx)); + }); +} + +#[test] +fn write_single_frame_one_packet() { + let io = FramedWrite::new( + mock! { + data(b"\x00\x00\x00\x09"), + data(b"abcdefghi"), + flush(), + }, + LengthDelimitedCodec::new(), + ); + pin_mut!(io); + + MockTask::new().enter(|cx| { + assert_ready_ok!(io.as_mut().poll_ready(cx)); + assert_ok!(io.as_mut().start_send(Bytes::from("abcdefghi"))); + assert_ready_ok!(io.as_mut().poll_flush(cx)); + assert!(io.get_ref().calls.is_empty()); + }); +} + +#[test] +fn write_single_multi_frame_one_packet() { + let io = FramedWrite::new( + mock! { + data(b"\x00\x00\x00\x09"), + data(b"abcdefghi"), + data(b"\x00\x00\x00\x03"), + data(b"123"), + data(b"\x00\x00\x00\x0b"), + data(b"hello world"), + flush(), + }, + LengthDelimitedCodec::new(), + ); + pin_mut!(io); + + MockTask::new().enter(|cx| { + assert_ready_ok!(io.as_mut().poll_ready(cx)); + assert_ok!(io.as_mut().start_send(Bytes::from("abcdefghi"))); + + assert_ready_ok!(io.as_mut().poll_ready(cx)); + assert_ok!(io.as_mut().start_send(Bytes::from("123"))); + + assert_ready_ok!(io.as_mut().poll_ready(cx)); + assert_ok!(io.as_mut().start_send(Bytes::from("hello world"))); + + assert_ready_ok!(io.as_mut().poll_flush(cx)); + assert!(io.get_ref().calls.is_empty()); + }); +} + +#[test] +fn write_single_multi_frame_multi_packet() { + let io = FramedWrite::new( + mock! { + data(b"\x00\x00\x00\x09"), + data(b"abcdefghi"), + flush(), + data(b"\x00\x00\x00\x03"), + data(b"123"), + flush(), + data(b"\x00\x00\x00\x0b"), + data(b"hello world"), + flush(), + }, + LengthDelimitedCodec::new(), + ); + pin_mut!(io); + + MockTask::new().enter(|cx| { + assert_ready_ok!(io.as_mut().poll_ready(cx)); + assert_ok!(io.as_mut().start_send(Bytes::from("abcdefghi"))); + + assert_ready_ok!(io.as_mut().poll_flush(cx)); + + assert_ready_ok!(io.as_mut().poll_ready(cx)); + assert_ok!(io.as_mut().start_send(Bytes::from("123"))); + + assert_ready_ok!(io.as_mut().poll_flush(cx)); + + assert_ready_ok!(io.as_mut().poll_ready(cx)); + assert_ok!(io.as_mut().start_send(Bytes::from("hello world"))); + + assert_ready_ok!(io.as_mut().poll_flush(cx)); + assert!(io.get_ref().calls.is_empty()); + }); +} + +#[test] +fn write_single_frame_would_block() { + let io = FramedWrite::new( + mock! { + Pending, + data(b"\x00\x00"), + Pending, + data(b"\x00\x09"), + data(b"abcdefghi"), + flush(), + }, + LengthDelimitedCodec::new(), + ); + pin_mut!(io); + + MockTask::new().enter(|cx| { + assert_ready_ok!(io.as_mut().poll_ready(cx)); + assert_ok!(io.as_mut().start_send(Bytes::from("abcdefghi"))); + + assert_pending!(io.as_mut().poll_flush(cx)); + assert_pending!(io.as_mut().poll_flush(cx)); + assert_ready_ok!(io.as_mut().poll_flush(cx)); + + assert!(io.get_ref().calls.is_empty()); + }); +} + +#[test] +fn write_single_frame_little_endian() { + let io = length_delimited::Builder::new() + .little_endian() + .new_write(mock! { + data(b"\x09\x00\x00\x00"), + data(b"abcdefghi"), + flush(), + }); + pin_mut!(io); + + MockTask::new().enter(|cx| { + assert_ready_ok!(io.as_mut().poll_ready(cx)); + assert_ok!(io.as_mut().start_send(Bytes::from("abcdefghi"))); + + assert_ready_ok!(io.as_mut().poll_flush(cx)); + assert!(io.get_ref().calls.is_empty()); + }); +} + +#[test] +fn write_single_frame_with_short_length_field() { + let io = length_delimited::Builder::new() + .length_field_length(1) + .new_write(mock! { + data(b"\x09"), + data(b"abcdefghi"), + flush(), + }); + pin_mut!(io); + + MockTask::new().enter(|cx| { + assert_ready_ok!(io.as_mut().poll_ready(cx)); + assert_ok!(io.as_mut().start_send(Bytes::from("abcdefghi"))); + + assert_ready_ok!(io.as_mut().poll_flush(cx)); + + assert!(io.get_ref().calls.is_empty()); + }); +} + +#[test] +fn write_max_frame_len() { + let io = length_delimited::Builder::new() + .max_frame_length(5) + .new_write(mock! {}); + pin_mut!(io); + + MockTask::new().enter(|cx| { + assert_ready_ok!(io.as_mut().poll_ready(cx)); + assert_err!(io.as_mut().start_send(Bytes::from("abcdef"))); + + assert!(io.get_ref().calls.is_empty()); + }); +} + +#[test] +fn write_update_max_frame_len_at_rest() { + let io = length_delimited::Builder::new().new_write(mock! { + data(b"\x00\x00\x00\x06"), + data(b"abcdef"), + flush(), + }); + pin_mut!(io); + + MockTask::new().enter(|cx| { + assert_ready_ok!(io.as_mut().poll_ready(cx)); + assert_ok!(io.as_mut().start_send(Bytes::from("abcdef"))); + + assert_ready_ok!(io.as_mut().poll_flush(cx)); + + io.encoder_mut().set_max_frame_length(5); + + assert_err!(io.as_mut().start_send(Bytes::from("abcdef"))); + + assert!(io.get_ref().calls.is_empty()); + }); +} + +#[test] +fn write_update_max_frame_len_in_flight() { + let io = length_delimited::Builder::new().new_write(mock! { + data(b"\x00\x00\x00\x06"), + data(b"ab"), + Pending, + data(b"cdef"), + flush(), + }); + pin_mut!(io); + + MockTask::new().enter(|cx| { + assert_ready_ok!(io.as_mut().poll_ready(cx)); + assert_ok!(io.as_mut().start_send(Bytes::from("abcdef"))); + + assert_pending!(io.as_mut().poll_flush(cx)); + + io.encoder_mut().set_max_frame_length(5); + + assert_ready_ok!(io.as_mut().poll_flush(cx)); + + assert_err!(io.as_mut().start_send(Bytes::from("abcdef"))); + assert!(io.get_ref().calls.is_empty()); + }); +} + +#[test] +fn write_zero() { + let io = length_delimited::Builder::new().new_write(mock! {}); + pin_mut!(io); + + MockTask::new().enter(|cx| { + assert_ready_ok!(io.as_mut().poll_ready(cx)); + assert_ok!(io.as_mut().start_send(Bytes::from("abcdef"))); + + assert_ready_err!(io.as_mut().poll_flush(cx)); + + assert!(io.get_ref().calls.is_empty()); + }); +} + +#[test] +fn encode_overflow() { + // Test reproducing tokio-rs/tokio#681. + let mut codec = length_delimited::Builder::new().new_codec(); + let mut buf = BytesMut::with_capacity(1024); + + // Put some data into the buffer without resizing it to hold more. + let some_as = std::iter::repeat(b'a').take(1024).collect::>(); + buf.put_slice(&some_as[..]); + + // Trying to encode the length header should resize the buffer if it won't fit. + codec.encode(Bytes::from("hello"), &mut buf).unwrap(); +} + +// ===== Test utils ===== + +struct Mock { + calls: VecDeque>>, +} + +enum Op { + Data(Vec), + Flush, +} + +use self::Op::*; + +impl AsyncRead for Mock { + fn poll_read( + mut self: Pin<&mut Self>, + _cx: &mut Context<'_>, + dst: &mut [u8], + ) -> Poll> { + match self.calls.pop_front() { + Some(Ready(Ok(Op::Data(data)))) => { + debug_assert!(dst.len() >= data.len()); + dst[..data.len()].copy_from_slice(&data[..]); + Ready(Ok(data.len())) + } + Some(Ready(Ok(_))) => panic!(), + Some(Ready(Err(e))) => Ready(Err(e)), + Some(Pending) => Pending, + None => Ready(Ok(0)), + } + } +} + +impl AsyncWrite for Mock { + fn poll_write( + mut self: Pin<&mut Self>, + _cx: &mut Context<'_>, + src: &[u8], + ) -> Poll> { + match self.calls.pop_front() { + Some(Ready(Ok(Op::Data(data)))) => { + let len = data.len(); + assert!(src.len() >= len, "expect={:?}; actual={:?}", data, src); + assert_eq!(&data[..], &src[..len]); + Ready(Ok(len)) + } + Some(Ready(Ok(_))) => panic!(), + Some(Ready(Err(e))) => Ready(Err(e)), + Some(Pending) => Pending, + None => Ready(Ok(0)), + } + } + + fn poll_flush(mut self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll> { + match self.calls.pop_front() { + Some(Ready(Ok(Op::Flush))) => Ready(Ok(())), + Some(Ready(Ok(_))) => panic!(), + Some(Ready(Err(e))) => Ready(Err(e)), + Some(Pending) => Pending, + None => Ready(Ok(())), + } + } + + fn poll_shutdown(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll> { + Ready(Ok(())) + } +} + +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) + } +} + +fn data(bytes: &[u8]) -> Poll> { + Ready(Ok(bytes.into())) +} + +fn flush() -> Poll> { + Ready(Ok(Flush)) +} diff --git a/tokio-fs/Cargo.toml b/tokio-fs/Cargo.toml index 64dbb7b41..ad2e798d1 100644 --- a/tokio-fs/Cargo.toml +++ b/tokio-fs/Cargo.toml @@ -24,7 +24,7 @@ categories = ["asynchronous", "network-programming", "filesystem"] publish = false [dependencies] -tokio-io = { version = "0.2.0", path = "../tokio-io" } +tokio-io = { version = "0.2.0", features = ["util"], path = "../tokio-io" } tokio-threadpool = { version = "0.2.0", path = "../tokio-threadpool" } futures-core-preview = "= 0.3.0-alpha.17" diff --git a/tokio-signal/src/lib.rs b/tokio-signal/src/lib.rs index 086c0346d..68bdd6860 100644 --- a/tokio-signal/src/lib.rs +++ b/tokio-signal/src/lib.rs @@ -51,6 +51,7 @@ //! //! ```rust,no_run //! #![feature(async_await)] +//! # #[cfg(unix)] { //! //! use futures_util::future; //! use futures_util::stream::StreamExt; @@ -79,6 +80,7 @@ //! println!("got signal {:?}", signal); //! Ok(()) //! } +//! # } //! ``` #[macro_use] diff --git a/tokio-sync/src/mpsc/bounded.rs b/tokio-sync/src/mpsc/bounded.rs index 7fc19c284..6c3a2fdea 100644 --- a/tokio-sync/src/mpsc/bounded.rs +++ b/tokio-sync/src/mpsc/bounded.rs @@ -124,7 +124,49 @@ impl Receiver { Receiver { chan } } - /// TODO: Dox + /// Receive the next value for this receiver. + /// + /// `None` is returned when all `Sender` halves have dropped, indicating + /// that no further values can be sent on the channel. + /// + /// # Examples + /// + /// ``` + /// #![feature(async_await)] + /// + /// use tokio::sync::mpsc; + /// + /// #[tokio::main] + /// async fn main() { + /// let (mut tx, mut rx) = mpsc::channel(100); + /// + /// tokio::spawn(async move { + /// tx.send("hello").await.unwrap(); + /// }); + /// + /// assert_eq!(Some("hello"), rx.recv().await); + /// assert_eq!(None, rx.recv().await); + /// } + /// ``` + /// + /// Values are buffered: + /// + /// ``` + /// #![feature(async_await)] + /// + /// use tokio::sync::mpsc; + /// + /// #[tokio::main] + /// async fn main() { + /// let (mut tx, mut rx) = mpsc::channel(100); + /// + /// tx.send("hello").await.unwrap(); + /// tx.send("world").await.unwrap(); + /// + /// assert_eq!(Some("hello"), rx.recv().await); + /// assert_eq!(Some("world"), rx.recv().await); + /// } + /// ``` #[allow(clippy::needless_lifetimes)] // false positive: https://github.com/rust-lang/rust-clippy/issues/3988 pub async fn recv(&mut self) -> Option { use futures_util::future::poll_fn; @@ -132,7 +174,7 @@ impl Receiver { poll_fn(|cx| self.poll_recv(cx)).await } - /// TODO: Dox + #[doc(hidden)] // TODO: remove pub fn poll_recv(&mut self, cx: &mut Context<'_>) -> Poll> { self.chan.recv(cx) } @@ -160,26 +202,7 @@ impl Sender { Sender { chan } } - /// Check if the `Sender` is ready to handle a value. - /// - /// Polls the channel to determine if there is guaranteed capacity to send - /// at least one item without waiting. - /// - /// When `poll_ready` returns `Ready`, the channel reserves capacity for one - /// message for this `Sender` instance. The capacity is held until a message - /// is send or the `Sender` instance is dropped. Callers should ensure a - /// message is sent in a timely fashion in order to not starve other - /// `Sender` instances. - /// - /// # Return value - /// - /// This method returns: - /// - /// - `Poll::Ready(Ok(_))` if capacity is reserved for a single message. - /// - `Poll::Pending` if the channel may not have capacity, in which - /// case the current task is queued to be notified once - /// capacity is available; - /// - `Poll::Ready(Err(SendError))` if the receiver has been dropped. + #[doc(hidden)] // TODO: remove pub fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll> { self.chan.poll_ready(cx).map_err(|_| SendError(())) } diff --git a/tokio-sync/src/mpsc/chan.rs b/tokio-sync/src/mpsc/chan.rs index 93e628240..4f228f09d 100644 --- a/tokio-sync/src/mpsc/chan.rs +++ b/tokio-sync/src/mpsc/chan.rs @@ -164,7 +164,6 @@ where } } - /// TODO: Docs pub(crate) fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll> { self.inner.semaphore.poll_acquire(cx, &mut self.permit) } diff --git a/tokio-sync/src/mpsc/unbounded.rs b/tokio-sync/src/mpsc/unbounded.rs index b9fbe08ed..60694f15e 100644 --- a/tokio-sync/src/mpsc/unbounded.rs +++ b/tokio-sync/src/mpsc/unbounded.rs @@ -87,12 +87,54 @@ impl UnboundedReceiver { UnboundedReceiver { chan } } - /// TODO: dox + #[doc(hidden)] // TODO: remove pub fn poll_recv(&mut self, cx: &mut Context<'_>) -> Poll> { self.chan.recv(cx) } - /// TODO: Dox + /// Receive the next value for this receiver. + /// + /// `None` is returned when all `Sender` halves have dropped, indicating + /// that no further values can be sent on the channel. + /// + /// # Examples + /// + /// ``` + /// #![feature(async_await)] + /// + /// use tokio::sync::mpsc; + /// + /// #[tokio::main] + /// async fn main() { + /// let (mut tx, mut rx) = mpsc::unbounded_channel(); + /// + /// tokio::spawn(async move { + /// tx.try_send("hello").unwrap(); + /// }); + /// + /// assert_eq!(Some("hello"), rx.recv().await); + /// assert_eq!(None, rx.recv().await); + /// } + /// ``` + /// + /// Values are buffered: + /// + /// ``` + /// #![feature(async_await)] + /// + /// use tokio::sync::mpsc; + /// + /// #[tokio::main] + /// async fn main() { + /// let (mut tx, mut rx) = mpsc::unbounded_channel(); + /// + /// tx.try_send("hello").unwrap(); + /// tx.try_send("world").unwrap(); + /// + /// assert_eq!(Some("hello"), rx.recv().await); + /// assert_eq!(Some("world"), rx.recv().await); + /// } + /// ``` #[allow(clippy::needless_lifetimes)] // false positive: https://github.com/rust-lang/rust-clippy/issues/3988 pub async fn recv(&mut self) -> Option { use futures_util::future::poll_fn; diff --git a/tokio-sync/src/oneshot.rs b/tokio-sync/src/oneshot.rs index b53e6d4df..7130118c8 100644 --- a/tokio-sync/src/oneshot.rs +++ b/tokio-sync/src/oneshot.rs @@ -157,18 +157,7 @@ impl Sender { Ok(()) } - /// Check if the associated [`Receiver`] handle has been dropped. - /// - /// # Return values - /// - /// If `Ready(Ok(_))` is returned then the associated `Receiver` has been - /// dropped, which means any work required for sending should be canceled. - /// - /// If `Pending` is returned then the associated `Receiver` is still - /// alive and may be able to receive a message if sent. The current task is - /// registered to receive a notification if the `Receiver` handle goes away. - /// - /// [`Receiver`]: struct.Receiver.html + #[doc(hidden)] // TODO: remove pub fn poll_closed(&mut self, cx: &mut Context<'_>) -> Poll<()> { let inner = self.inner.as_ref().unwrap(); diff --git a/tokio-tcp/tests/chain.rs b/tokio-tcp/tests/chain.rs deleted file mode 100644 index 3c9b9297d..000000000 --- a/tokio-tcp/tests/chain.rs +++ /dev/null @@ -1,49 +0,0 @@ -#![deny(warnings, rust_2018_idioms)] -#![cfg(feature = "broken")] - -use futures::stream::Stream; -use futures::Future; -use std::io::{Read, Write}; -use std::net::TcpStream; -use std::thread; -use tokio_io::io::read_to_end; -use tokio_tcp::TcpListener; - -macro_rules! t { - ($e:expr) => { - match $e { - Ok(e) => e, - Err(e) => panic!("{} failed with {:?}", stringify!($e), e), - } - }; -} - -#[test] -fn chain_clients() { - let srv = t!(TcpListener::bind(&t!("127.0.0.1:0".parse()))); - let addr = t!(srv.local_addr()); - - let t = thread::spawn(move || { - let mut s1 = TcpStream::connect(&addr).unwrap(); - s1.write_all(b"foo ").unwrap(); - let mut s2 = TcpStream::connect(&addr).unwrap(); - s2.write_all(b"bar ").unwrap(); - let mut s3 = TcpStream::connect(&addr).unwrap(); - s3.write_all(b"baz").unwrap(); - }); - - let clients = srv.incoming().take(3); - let copied = clients.collect().and_then(|clients| { - let mut clients = clients.into_iter(); - let a = clients.next().unwrap(); - let b = clients.next().unwrap(); - let c = clients.next().unwrap(); - - read_to_end(a.chain(b).chain(c), Vec::new()) - }); - - let (_, data) = t!(copied.wait()); - t.join().unwrap(); - - assert_eq!(data, b"foo bar baz"); -} diff --git a/tokio-tcp/tests/echo.rs b/tokio-tcp/tests/echo.rs index cebd366fb..cb7e8b02e 100644 --- a/tokio-tcp/tests/echo.rs +++ b/tokio-tcp/tests/echo.rs @@ -1,51 +1,43 @@ +#![feature(async_await)] #![deny(warnings, rust_2018_idioms)] -#![cfg(feature = "broken")] -use env_logger; -use futures::stream::Stream; -use futures::Future; -use std::io::{Read, Write}; -use std::net::TcpStream; -use std::thread; -use tokio_io::io::copy; -use tokio_io::AsyncRead; -use tokio_tcp::TcpListener; +use tokio::net::{TcpListener, TcpStream}; +use tokio::prelude::*; +use tokio::sync::oneshot; +use tokio_test::assert_ok; -macro_rules! t { - ($e:expr) => { - match $e { - Ok(e) => e, - Err(e) => panic!("{} failed with {:?}", stringify!($e), e), - } - }; -} +#[tokio::test] +async fn echo_server() { + const ITER: usize = 1024; -#[test] -fn echo_server() { - drop(env_logger::try_init()); + let (tx, rx) = oneshot::channel(); - let srv = t!(TcpListener::bind(&t!("127.0.0.1:0".parse()))); - let addr = t!(srv.local_addr()); + let addr = assert_ok!("127.0.0.1:0".parse()); + let mut srv = assert_ok!(TcpListener::bind(&addr)); + let addr = assert_ok!(srv.local_addr()); let msg = "foo bar baz"; - let t = thread::spawn(move || { - let mut s = TcpStream::connect(&addr).unwrap(); + tokio::spawn(async move { + let mut stream = assert_ok!(TcpStream::connect(&addr).await); - for _i in 0..1024 { - assert_eq!(t!(s.write(msg.as_bytes())), msg.len()); - let mut buf = [0; 1024]; - assert_eq!(t!(s.read(&mut buf)), msg.len()); - assert_eq!(&buf[..msg.len()], msg.as_bytes()); + for _ in 0..ITER { + // write + assert_ok!(stream.write_all(msg.as_bytes()).await); + + // read + let mut buf = [0; 11]; + assert_ok!(stream.read_exact(&mut buf).await); + assert_eq!(&buf[..], msg.as_bytes()); } + + assert_ok!(tx.send(())); }); - let clients = srv.incoming(); - let client = clients.into_future().map(|e| e.0.unwrap()).map_err(|e| e.0); - let halves = client.map(|s| s.split()); - let copied = halves.and_then(|(a, b)| copy(a, b)); + let (stream, _) = assert_ok!(srv.accept().await); + let (mut rd, mut wr) = stream.split(); - let (amt, _, _) = t!(copied.wait()); - t.join().unwrap(); + let n = assert_ok!(rd.copy(&mut wr).await); + assert_eq!(n, (ITER * msg.len()) as u64); - assert_eq!(amt, msg.len() as u64 * 1024); + assert_ok!(rx.await); } diff --git a/tokio-tcp/tests/limit.rs b/tokio-tcp/tests/limit.rs deleted file mode 100644 index 47fa56506..000000000 --- a/tokio-tcp/tests/limit.rs +++ /dev/null @@ -1,43 +0,0 @@ -#![deny(warnings, rust_2018_idioms)] -#![cfg(feature = "broken")] - -use futures::stream::Stream; -use futures::Future; -use std::io::{Read, Write}; -use std::net::TcpStream; -use std::thread; -use tokio_io::io::read_to_end; -use tokio_tcp::TcpListener; - -macro_rules! t { - ($e:expr) => { - match $e { - Ok(e) => e, - Err(e) => panic!("{} failed with {:?}", stringify!($e), e), - } - }; -} - -#[test] -fn limit() { - let srv = t!(TcpListener::bind(&t!("127.0.0.1:0".parse()))); - let addr = t!(srv.local_addr()); - - let t = thread::spawn(move || { - let mut s1 = TcpStream::connect(&addr).unwrap(); - s1.write_all(b"foo bar baz").unwrap(); - }); - - let clients = srv.incoming().take(1); - let copied = clients.collect().and_then(|clients| { - let mut clients = clients.into_iter(); - let a = clients.next().unwrap(); - - read_to_end(a.take(4), Vec::new()) - }); - - let (_, data) = t!(copied.wait()); - t.join().unwrap(); - - assert_eq!(data, b"foo "); -} diff --git a/tokio-tcp/tests/stream-buffered.rs b/tokio-tcp/tests/stream-buffered.rs deleted file mode 100644 index d30323036..000000000 --- a/tokio-tcp/tests/stream-buffered.rs +++ /dev/null @@ -1,55 +0,0 @@ -#![deny(warnings, rust_2018_idioms)] -#![cfg(feature = "broken")] - -use env_logger; -use futures::stream::Stream; -use futures::Future; -use std::io::{Read, Write}; -use std::net::TcpStream; -use std::thread; -use tokio_io::io::copy; -use tokio_io::AsyncRead; -use tokio_tcp::TcpListener; - -macro_rules! t { - ($e:expr) => { - match $e { - Ok(e) => e, - Err(e) => panic!("{} failed with {:?}", stringify!($e), e), - } - }; -} - -#[test] -fn echo_server() { - drop(env_logger::try_init()); - - let srv = t!(TcpListener::bind(&t!("127.0.0.1:0".parse()))); - let addr = t!(srv.local_addr()); - - let t = thread::spawn(move || { - let mut s1 = t!(TcpStream::connect(&addr)); - let mut s2 = t!(TcpStream::connect(&addr)); - - let msg = b"foo"; - assert_eq!(t!(s1.write(msg)), msg.len()); - assert_eq!(t!(s2.write(msg)), msg.len()); - let mut buf = [0; 1024]; - assert_eq!(t!(s1.read(&mut buf)), msg.len()); - assert_eq!(&buf[..msg.len()], msg); - assert_eq!(t!(s2.read(&mut buf)), msg.len()); - assert_eq!(&buf[..msg.len()], msg); - }); - - let future = srv - .incoming() - .map(|s| s.split()) - .map(|(a, b)| copy(a, b).map(|_| ())) - .buffered(10) - .take(2) - .collect(); - - t!(future.wait()); - - t.join().unwrap(); -} diff --git a/tokio-tcp/tests/tcp.rs b/tokio-tcp/tests/tcp.rs index 99211972f..e95768d84 100644 --- a/tokio-tcp/tests/tcp.rs +++ b/tokio-tcp/tests/tcp.rs @@ -1,114 +1,77 @@ +#![feature(async_await)] #![deny(warnings, rust_2018_idioms)] -#![cfg(feature = "broken")] -use env_logger; -use futures::{Future, Stream}; -use std::sync::mpsc::channel; -use std::{net, thread}; -use tokio_tcp::{TcpListener, TcpStream}; +use tokio::net::{TcpListener, TcpStream}; +use tokio::sync::oneshot; +use tokio_test::assert_ok; -macro_rules! t { - ($e:expr) => { - match $e { - Ok(e) => e, - Err(e) => panic!("{} failed with {:?}", stringify!($e), e), - } - }; -} - -#[test] -fn connect() { - drop(env_logger::try_init()); - let srv = t!(net::TcpListener::bind("127.0.0.1:0")); - let addr = t!(srv.local_addr()); - let t = thread::spawn(move || t!(srv.accept()).0); - - let stream = TcpStream::connect(&addr); - let mine = t!(stream.wait()); - let theirs = t.join().unwrap(); - - assert_eq!(t!(mine.local_addr()), t!(theirs.peer_addr())); - assert_eq!(t!(theirs.local_addr()), t!(mine.peer_addr())); -} - -#[test] -fn accept() { - drop(env_logger::try_init()); - let srv = t!(TcpListener::bind(&t!("127.0.0.1:0".parse()))); - let addr = t!(srv.local_addr()); - - let (tx, rx) = channel(); - let client = srv - .incoming() - .map(move |t| { - tx.send(()).unwrap(); - t - }) - .into_future() - .map_err(|e| e.0); - assert!(rx.try_recv().is_err()); - let t = thread::spawn(move || net::TcpStream::connect(&addr).unwrap()); - - let (mine, _remaining) = t!(client.wait()); - let mine = mine.unwrap(); - let theirs = t.join().unwrap(); - - assert_eq!(t!(mine.local_addr()), t!(theirs.peer_addr())); - assert_eq!(t!(theirs.local_addr()), t!(mine.peer_addr())); -} - -#[test] -fn accept2() { - drop(env_logger::try_init()); - let srv = t!(TcpListener::bind(&t!("127.0.0.1:0".parse()))); - let addr = t!(srv.local_addr()); - - let t = thread::spawn(move || net::TcpStream::connect(&addr).unwrap()); - - let (tx, rx) = channel(); - let client = srv - .incoming() - .map(move |t| { - tx.send(()).unwrap(); - t - }) - .into_future() - .map_err(|e| e.0); - assert!(rx.try_recv().is_err()); - - let (mine, _remaining) = t!(client.wait()); - mine.unwrap(); - t.join().unwrap(); +#[tokio::test] +async fn connect() { + let addr = assert_ok!("127.0.0.1:0".parse()); + let mut srv = assert_ok!(TcpListener::bind(&addr)); + let addr = assert_ok!(srv.local_addr()); + + let (tx, rx) = oneshot::channel(); + + tokio::spawn(async move { + let (socket, addr) = assert_ok!(srv.accept().await); + assert_eq!(addr, assert_ok!(socket.peer_addr())); + assert_ok!(tx.send(socket)); + }); + + let mine = assert_ok!(TcpStream::connect(&addr).await); + let theirs = assert_ok!(rx.await); + + assert_eq!( + assert_ok!(mine.local_addr()), + assert_ok!(theirs.peer_addr()) + ); + assert_eq!( + assert_ok!(theirs.local_addr()), + assert_ok!(mine.peer_addr()) + ); } +/* + * TODO: bring this back once TCP exposes HUP again + * #[cfg(target_os = "linux")] mod linux { - use tokio_tcp::TcpStream; + use tokio::net::{TcpListener, TcpStream}; + use tokio::prelude::*; + use tokio_test::assert_ok; - use env_logger; - use futures::{future, Future}; use mio::unix::UnixReady; - use net2::TcpStreamExt; - use tokio_io::AsyncRead; + use futures_util::future::poll_fn; use std::io::Write; use std::time::Duration; use std::{net, thread}; - #[test] + #[tokio::test] fn poll_hup() { - drop(env_logger::try_init()); + let addr = assert_ok!("127.0.0.1:0".parse()); + let mut srv = assert_ok!(TcpListener::bind(&addr)); + let addr = assert_ok!(srv.local_addr()); - let srv = t!(net::TcpListener::bind("127.0.0.1:0")); - let addr = t!(srv.local_addr()); + tokio::spawn(async move { + let (mut client, _) = assert_ok!(srv.accept().await); + assert_ok!(client.set_linger(Some(Duration::from_millis(0)))); + assert_ok!(client.write_all(b"hello world").await); + + // TODO: Drop? + }); + + /* let t = thread::spawn(move || { - let mut client = t!(srv.accept()).0; + let mut client = assert_ok!(srv.accept()).0; client.set_linger(Some(Duration::from_millis(0))).unwrap(); client.write(b"hello world").unwrap(); thread::sleep(Duration::from_millis(200)); }); + */ - let mut stream = t!(TcpStream::connect(&addr).wait()); + let mut stream = assert_ok!(TcpStream::connect(&addr).await); // Poll for HUP before reading. future::poll_fn(|| stream.poll_read_ready(UnixReady::hup().into())) @@ -132,3 +95,4 @@ mod linux { t.join().unwrap(); } } +*/ diff --git a/tokio-test/src/clock.rs b/tokio-test/src/clock.rs index a9badd369..942d9b6f6 100644 --- a/tokio-test/src/clock.rs +++ b/tokio-test/src/clock.rs @@ -2,21 +2,26 @@ //! //! # Example //! -//! ```ignore -//! use tokio_test::clock; -//! use tokio_test::{assert_ready, assert_not_ready}; +//! ``` +//! #![feature(async_await)] +//! +//! use tokio::clock; +//! use tokio_test::{assert_ready, assert_pending, task}; //! use tokio_timer::Delay; +//! //! use std::time::Duration; -//! use futures::Future; //! -//! clock::mock(|handle| { -//! let mut delay = Delay::new(handle.now() + Duration::from_secs(1)); +//! tokio_test::clock::mock(|handle| { +//! let mut task = task::spawn(async { +//! let delay = Delay::new(clock::now() + Duration::from_secs(1)); +//! delay.await +//! }); //! -//! assert_not_ready!(delay.poll()); +//! assert_pending!(task.poll()); //! //! handle.advance(Duration::from_secs(1)); //! -//! assert_ready!(delay.poll()); +//! assert_ready!(task.poll()); //! }); //! ``` diff --git a/tokio-test/src/lib.rs b/tokio-test/src/lib.rs index 5f81b556a..d28d19141 100644 --- a/tokio-test/src/lib.rs +++ b/tokio-test/src/lib.rs @@ -9,16 +9,6 @@ #![doc(test(no_crate_inject, attr(deny(rust_2018_idioms))))] //! Tokio and Futures based testing utilites -//! -//! # Example -//! -//! ```ignore -//! # use futures::{Future, future}; -//! use tokio_test::assert_ready; -//! -//! let mut fut = future::ok::<(), ()>(()); -//! assert_ready!(fut.poll()); -//! ``` pub mod clock; pub mod io; diff --git a/tokio-test/src/task.rs b/tokio-test/src/task.rs index e7d6c811d..91da41eb5 100644 --- a/tokio-test/src/task.rs +++ b/tokio-test/src/task.rs @@ -1,21 +1,4 @@ //! Futures task based helpers -//! -//! # Example -//! -//! This example will use the `MockTask` to set the current task on -//! poll. -//! -//! ```ignore -//! # use tokio_test::assert_ready_eq; -//! # use tokio_test::task::MockTask; -//! # use futures::{sync::mpsc, Stream, Sink, Future, Async}; -//! let mut task = MockTask::new(); -//! let (tx, mut rx) = mpsc::channel(5); -//! -//! tx.send(()).wait(); -//! -//! assert_ready_eq!(task.enter(|| rx.poll()), Some(())); -//! ``` use tokio_executor::enter; diff --git a/tokio-threadpool/Cargo.toml b/tokio-threadpool/Cargo.toml index 69630c2d0..ddaccf5c7 100644 --- a/tokio-threadpool/Cargo.toml +++ b/tokio-threadpool/Cargo.toml @@ -41,4 +41,3 @@ rand = "0.7" env_logger = "0.5" tokio = { version = "0.2.0", path = "../tokio" } tokio-test = { version = "0.2.0", path = "../tokio-test" } -futures-util-preview = "= 0.3.0-alpha.17" diff --git a/tokio-threadpool/examples/depth.rs b/tokio-threadpool/examples/depth.rs deleted file mode 100644 index bc1ca8a31..000000000 --- a/tokio-threadpool/examples/depth.rs +++ /dev/null @@ -1,50 +0,0 @@ -#![cfg(features = "broken")] - -extern crate env_logger; -extern crate futures; -extern crate tokio_threadpool; - -use futures::future::{self, Executor}; -use tokio_threadpool::*; - -use std::sync::mpsc; - -const ITER: usize = 2_000_000; -// const ITER: usize = 30; - -fn chained_spawn() { - let pool = ThreadPool::new(); - let tx = pool.sender().clone(); - - fn spawn(tx: Sender, res_tx: mpsc::Sender<()>, n: usize) { - if n == 0 { - res_tx.send(()).unwrap(); - } else { - let tx2 = tx.clone(); - tx.execute(future::lazy(move || { - spawn(tx2, res_tx, n - 1); - Ok(()) - })) - .ok() - .unwrap(); - } - } - - loop { - println!("~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~"); - let (res_tx, res_rx) = mpsc::channel(); - - for _ in 0..10 { - spawn(tx.clone(), res_tx.clone(), ITER); - } - - for _ in 0..10 { - res_rx.recv().unwrap(); - } - } -} - -pub fn main() { - let _ = ::env_logger::init(); - chained_spawn(); -} diff --git a/tokio-threadpool/examples/hello.rs b/tokio-threadpool/examples/hello.rs deleted file mode 100644 index 2b3b90e5f..000000000 --- a/tokio-threadpool/examples/hello.rs +++ /dev/null @@ -1,26 +0,0 @@ -#![cfg(features = "broken")] - -extern crate env_logger; -extern crate futures; -extern crate tokio_threadpool; - -use futures::sync::oneshot; -use futures::*; -use tokio_threadpool::*; - -pub fn main() { - let _ = ::env_logger::init(); - - let pool = ThreadPool::new(); - let tx = pool.sender().clone(); - - let res = oneshot::spawn( - future::lazy(|| { - println!("Running on the pool"); - Ok::<_, ()>("complete") - }), - &tx, - ); - - println!("Result: {:?}", res.wait()); -} diff --git a/tokio-threadpool/src/blocking.rs b/tokio-threadpool/src/blocking.rs index 720288367..0fad2d847 100644 --- a/tokio-threadpool/src/blocking.rs +++ b/tokio-threadpool/src/blocking.rs @@ -80,11 +80,11 @@ pub struct BlockingError { /// that needs to be performed. /// /// ```rust +/// #![feature(async_await)] +/// /// use tokio_threadpool::{ThreadPool, blocking}; /// -/// use futures::Future; -/// use futures::future::{lazy, poll_fn}; -/// +/// use futures_util::future::poll_fn; /// use std::sync::mpsc; /// use std::thread; /// use std::time::Duration; @@ -101,21 +101,21 @@ pub struct BlockingError { /// /// let pool = ThreadPool::new(); /// -/// pool.spawn(lazy(move || { +/// pool.spawn(async move { /// // Because `blocking` returns `Poll`, it is intended to be used /// // from the context of a `Future` implementation. Since we don't /// // have a complicated requirement, we can use `poll_fn` in this /// // case. -/// poll_fn(move || { +/// poll_fn(move |_| { /// blocking(|| { /// let msg = rx.recv().unwrap(); /// println!("message = {}", msg); /// }).map_err(|_| panic!("the threadpool shut down")) -/// }) -/// })); +/// }).await; +/// }); /// /// // Wait for the task we just spawned to complete. -/// pool.shutdown_on_idle().wait().unwrap(); +/// pool.shutdown_on_idle().wait(); /// } /// ``` pub fn blocking(f: F) -> Poll> diff --git a/tokio-threadpool/src/builder.rs b/tokio-threadpool/src/builder.rs index 0b8bf3b29..a7c019eb2 100644 --- a/tokio-threadpool/src/builder.rs +++ b/tokio-threadpool/src/builder.rs @@ -32,8 +32,10 @@ use tokio_executor::park::Park; /// # Examples /// /// ``` +/// #![feature(async_await)] +/// /// use tokio_threadpool::Builder; -/// use futures::future::{Future, lazy}; +/// /// use std::time::Duration; /// /// let thread_pool = Builder::new() @@ -41,13 +43,12 @@ use tokio_executor::park::Park; /// .keep_alive(Some(Duration::from_secs(30))) /// .build(); /// -/// thread_pool.spawn(lazy(|| { +/// thread_pool.spawn(async { /// println!("called from a worker thread"); -/// Ok(()) -/// })); +/// }); /// /// // Gracefully shutdown the threadpool -/// thread_pool.shutdown().wait().unwrap(); +/// thread_pool.shutdown().wait(); /// ``` pub struct Builder { /// Thread pool specific configuration values diff --git a/tokio-threadpool/src/sender.rs b/tokio-threadpool/src/sender.rs index b2803806d..74da374de 100644 --- a/tokio-threadpool/src/sender.rs +++ b/tokio-threadpool/src/sender.rs @@ -60,21 +60,19 @@ impl Sender { /// # Examples /// /// ```rust - /// # use tokio_threadpool::ThreadPool; - /// use futures::future::{Future, lazy}; + /// #![feature(async_await)] + /// + /// use tokio_threadpool::ThreadPool; /// - /// # pub fn main() { /// // Create a thread pool with default configuration values /// let thread_pool = ThreadPool::new(); /// - /// thread_pool.sender().spawn(lazy(|| { + /// thread_pool.sender().spawn(async { /// println!("called from a worker thread"); - /// Ok(()) - /// })).unwrap(); + /// }).unwrap(); /// /// // Gracefully shutdown the threadpool - /// thread_pool.shutdown().wait().unwrap(); - /// # } + /// thread_pool.shutdown().wait(); /// ``` pub fn spawn(&self, future: F) -> Result<(), SpawnError> where diff --git a/tokio-threadpool/src/thread_pool.rs b/tokio-threadpool/src/thread_pool.rs index 121b4dd50..6c6a574fc 100644 --- a/tokio-threadpool/src/thread_pool.rs +++ b/tokio-threadpool/src/thread_pool.rs @@ -51,19 +51,19 @@ impl ThreadPool { /// # Examples /// /// ```rust - /// # use tokio_threadpool::ThreadPool; - /// use futures::future::{Future, lazy}; + /// #![feature(async_await)] + /// + /// use tokio_threadpool::ThreadPool; /// /// // Create a thread pool with default configuration values /// let thread_pool = ThreadPool::new(); /// - /// thread_pool.spawn(lazy(|| { + /// thread_pool.spawn(async { /// println!("called from a worker thread"); - /// Ok(()) - /// })); + /// }); /// /// // Gracefully shutdown the threadpool - /// thread_pool.shutdown().wait().unwrap(); + /// thread_pool.shutdown().wait(); /// ``` /// /// # Panics diff --git a/tokio-timer/src/delay_queue.rs b/tokio-timer/src/delay_queue.rs index 531007738..db5ddc578 100644 --- a/tokio-timer/src/delay_queue.rs +++ b/tokio-timer/src/delay_queue.rs @@ -10,7 +10,6 @@ use crate::wheel::{self, Wheel}; use crate::{Delay, Error}; use futures_core::ready; -use futures_util::future::poll_fn; use slab::Slab; use std::cmp; use std::future::Future; @@ -72,8 +71,10 @@ use std::time::{Duration, Instant}; /// /// ```rust,no_run /// use tokio::timer::{delay_queue, DelayQueue, Error}; -/// use futures::{try_ready, Async, Poll, Stream}; +/// +/// use futures_core::ready; /// use std::collections::HashMap; +/// use std::task::{Context, Poll}; /// use std::time::Duration; /// # type CacheKey = String; /// # type Value = String; @@ -104,12 +105,13 @@ use std::time::{Duration, Instant}; /// } /// } /// -/// fn poll_purge(&mut self) -> Poll<(), Error> { -/// while let Some(entry) = try_ready!(self.expirations.poll()) { +/// fn poll_purge(&mut self, cx: &mut Context<'_>) -> Poll> { +/// while let Some(res) = ready!(self.expirations.poll_next(cx)) { +/// let entry = res?; /// self.entries.remove(entry.get_ref()); /// } /// -/// Ok(Async::Ready(())) +/// Poll::Ready(Ok(())) /// } /// } /// ``` @@ -349,7 +351,9 @@ impl DelayQueue { Key::new(key) } - /// TODO: Dox... also is the fn signature correct? + /// Attempt to pull out the next value of the delay queue, registering the + /// current task for wakeup if the value is not yet available, and returning + /// None if the queue is exhausted. pub fn poll_next( &mut self, cx: &mut task::Context<'_>, @@ -370,13 +374,6 @@ impl DelayQueue { })) } - /// TODO: Dox... also is the fn signature correct? - #[allow(clippy::needless_lifetimes)] // false positive: https://github.com/rust-lang/rust-clippy/issues/3988 - #[allow(clippy::should_implement_trait)] // false positive : https://github.com/rust-lang/rust-clippy/issues/4290 - pub async fn next(&mut self) -> Option, Error>> { - poll_fn(|cx| self.poll_next(cx)).await - } - /// Insert `value` into the queue set to expire after the requested duration /// elapses. /// diff --git a/tokio-timer/src/interval.rs b/tokio-timer/src/interval.rs index 3aa98b1aa..3b480f444 100644 --- a/tokio-timer/src/interval.rs +++ b/tokio-timer/src/interval.rs @@ -55,7 +55,7 @@ impl Interval { Interval { delay, duration } } - /// TODO: dox + #[doc(hidden)] // TODO: remove pub fn poll_next(&mut self, cx: &mut task::Context<'_>) -> Poll> { // Wait for the delay to be done ready!(Pin::new(&mut self.delay).poll(cx)); @@ -72,9 +72,30 @@ impl Interval { Poll::Ready(Some(now)) } - /// TODO: dox + /// Completes when the next instant in the interval has been reached. + /// + /// # Examples + /// + /// ``` + /// #![feature(async_await)] + /// + /// use tokio::timer::Interval; + /// + /// use std::time::Duration; + /// + /// #[tokio::main] + /// async fn main() { + /// let mut interval = Interval::new_interval(Duration::from_millis(10)); + /// + /// interval.next().await; + /// interval.next().await; + /// interval.next().await; + /// + /// // approximately 30ms have elapsed. + /// } + /// ``` #[allow(clippy::needless_lifetimes)] // false positive: https://github.com/rust-lang/rust-clippy/issues/3988 - #[allow(clippy::should_implement_trait)] // false positive : https://github.com/rust-lang/rust-clippy/issues/4290 + #[allow(clippy::should_implement_trait)] // TODO: rename (tokio-rs/tokio#1261) pub async fn next(&mut self) -> Option { poll_fn(|cx| self.poll_next(cx)).await } diff --git a/tokio-timer/src/timeout.rs b/tokio-timer/src/timeout.rs index 485547a39..88816da41 100644 --- a/tokio-timer/src/timeout.rs +++ b/tokio-timer/src/timeout.rs @@ -31,28 +31,33 @@ use std::time::{Duration, Instant}; /// then a timeout should be set on the future that processes the stream. For /// example: /// -/// ```rust -/// // import the `timeout` function, usually this is done -/// // with `use tokio::prelude::*` -/// use tokio::prelude::FutureExt; -/// use futures::Stream; -/// use futures::sync::mpsc; +/// ```rust,no_run +/// #![feature(async_await)] +/// +/// use tokio::prelude::*; +/// use tokio::sync::mpsc; +/// +/// use std::thread; /// use std::time::Duration; /// -/// let (tx, rx) = mpsc::unbounded(); -/// # tx.unbounded_send(()).unwrap(); -/// # drop(tx); +/// # async fn dox() { +/// let (mut tx, rx) = mpsc::unbounded_channel(); +/// +/// thread::spawn(move || { +/// tx.try_send(()).unwrap(); +/// thread::sleep(Duration::from_millis(10)); +/// tx.try_send(()).unwrap(); +/// }); /// /// let process = rx.for_each(|item| { /// // do something with `item` /// # drop(item); -/// # Ok(()) +/// # tokio::future::ready(()) /// }); /// -/// # tokio::runtime::current_thread::block_on_all( /// // Wrap the future with a `Timeout` set to expire in 10 milliseconds. -/// process.timeout(Duration::from_millis(10)) -/// # ).unwrap(); +/// process.timeout(Duration::from_millis(10)).await; +/// # } /// ``` /// /// # Cancelation @@ -91,18 +96,20 @@ impl Timeout { /// Create a new `Timeout` set to expire in 10 milliseconds. /// /// ```rust + /// #![feature(async_await)] + /// /// use tokio::timer::Timeout; - /// use futures::Future; - /// use futures::sync::oneshot; + /// use tokio::sync::oneshot; + /// /// use std::time::Duration; /// + /// # async fn dox() { /// let (tx, rx) = oneshot::channel(); /// # tx.send(()).unwrap(); /// - /// # tokio::runtime::current_thread::block_on_all( /// // Wrap the future with a `Timeout` set to expire in 10 milliseconds. - /// Timeout::new(rx, Duration::from_millis(10)) - /// # ).unwrap(); + /// Timeout::new(rx, Duration::from_millis(10)).await; + /// } /// ``` pub fn new(value: T, timeout: Duration) -> Timeout { let delay = Delay::new_timeout(now() + timeout, timeout); diff --git a/tokio-tls/examples/download-rust-lang.rs b/tokio-tls/examples_old/download-rust-lang.rs similarity index 100% rename from tokio-tls/examples/download-rust-lang.rs rename to tokio-tls/examples_old/download-rust-lang.rs diff --git a/tokio-tls/examples/echo.rs b/tokio-tls/examples_old/echo.rs similarity index 100% rename from tokio-tls/examples/echo.rs rename to tokio-tls/examples_old/echo.rs diff --git a/tokio-tls/examples/identity.p12 b/tokio-tls/examples_old/identity.p12 similarity index 100% rename from tokio-tls/examples/identity.p12 rename to tokio-tls/examples_old/identity.p12 diff --git a/tokio/Cargo.toml b/tokio/Cargo.toml index 6882772b5..c3e596607 100644 --- a/tokio/Cargo.toml +++ b/tokio/Cargo.toml @@ -29,19 +29,18 @@ default = [ "codec", "fs", "io", + "net", "reactor", "rt-full", "sync", - "tcp", "timer", - "udp", - "uds", ] codec = ["io", "tokio-codec", "bytes"] fs = ["tokio-fs"] io = ["tokio-io"] reactor = ["io", "tokio-reactor"] +net = ["reactor", "tcp", "udp", "uds"] rt-full = [ "num_cpus", "reactor", diff --git a/tokio/examples/udp-codec.rs b/tokio/examples/udp-codec.rs index 03be6562e..9c53ab953 100644 --- a/tokio/examples/udp-codec.rs +++ b/tokio/examples/udp-codec.rs @@ -7,6 +7,7 @@ //! "ping pong" pair where two sockets are sending messages back and forth. #![feature(async_await)] +#![cfg(feature = "rt-full")] #![deny(warnings, rust_2018_idioms)] use tokio::io; diff --git a/tokio/examples_old/chat-combinator-current-thread.rs b/tokio/examples_old/chat-combinator-current-thread.rs deleted file mode 100644 index c82c27118..000000000 --- a/tokio/examples_old/chat-combinator-current-thread.rs +++ /dev/null @@ -1,169 +0,0 @@ -//! A chat server that broadcasts a message to all connections. -//! -//! This is a line-based server which accepts connections, reads lines from -//! those connections, and broadcasts the lines to all other connected clients. -//! -//! This example is similar to chat.rs, but uses combinators and a much more -//! functional style. -//! -//! Because we are here running the reactor/executor on the same thread instead -//! of a threadpool, we can avoid full synchronization with Arc + Mutex and use -//! Rc + RefCell instead. The max performance is however limited to a CPU HW -//! thread. -//! -//! You can test this out by running: -//! -//! cargo run --example chat-combinator-current-thread -//! -//! And then in another window run: -//! -//! cargo run --example connect 127.0.0.1:8080 -//! -//! You can run the second command in multiple windows and then chat between the -//! two, seeing the messages from the other client as they're received. For all -//! connected clients they'll all join the same room and see everyone else's -//! messages. - -#![deny(warnings, rust_2018_idioms)] - -use futures; -use std::cell::RefCell; -use std::collections::HashMap; -use std::env; -use std::io::BufReader; -use std::iter; -use std::rc::Rc; -use tokio::io; -use tokio::net::TcpListener; -use tokio::prelude::*; -use tokio::runtime::current_thread::{Runtime, TaskExecutor}; - -fn main() -> Result<(), Box> { - let mut runtime = Runtime::new().unwrap(); - - // Create the TCP listener we'll accept connections on. - let addr = env::args().nth(1).unwrap_or("127.0.0.1:8080".to_string()); - let addr = addr.parse()?; - - let socket = TcpListener::bind(&addr)?; - println!("Listening on: {}", addr); - - // This is running on the Tokio current_thread runtime, so it will be single- - // threaded. The `Rc>` allows state to be shared across the tasks. - let connections = Rc::new(RefCell::new(HashMap::new())); - - // The server task asynchronously iterates over and processes each incoming - // connection. - let srv = socket - .incoming() - .map_err(|e| { - println!("failed to accept socket; error = {:?}", e); - e - }) - .for_each(move |stream| { - // The client's socket address - let addr = stream.peer_addr()?; - - println!("New Connection: {}", addr); - - // Split the TcpStream into two separate handles. One handle for reading - // and one handle for writing. This lets us use separate tasks for - // reading and writing. - let (reader, writer) = stream.split(); - - // Create a channel for our stream, which other sockets will use to - // send us messages. Then register our address with the stream to send - // data to us. - let (tx, rx) = futures::sync::mpsc::unbounded(); - let mut conns = connections.borrow_mut(); - conns.insert(addr, tx); - - // Define here what we do for the actual I/O. That is, read a bunch of - // lines from the socket and dispatch them while we also write any lines - // from other sockets. - let connections_inner = connections.clone(); - let reader = BufReader::new(reader); - - // Model the read portion of this socket by mapping an infinite - // iterator to each line off the socket. This "loop" is then - // terminated with an error once we hit EOF on the socket. - let iter = stream::iter_ok::<_, io::Error>(iter::repeat(())); - - let socket_reader = iter.fold(reader, move |reader, _| { - // Read a line off the socket, failing if we're at EOF - let line = io::read_until(reader, b'\n', Vec::new()); - let line = line.and_then(|(reader, vec)| { - if vec.len() == 0 { - Err(io::Error::new(io::ErrorKind::BrokenPipe, "broken pipe")) - } else { - Ok((reader, vec)) - } - }); - - // Convert the bytes we read into a string, and then send that - // string to all other connected clients. - let line = line.map(|(reader, vec)| (reader, String::from_utf8(vec))); - - // Move the connection state into the closure below. - let connections = connections_inner.clone(); - - line.map(move |(reader, message)| { - println!("{}: {:?}", addr, message); - let mut conns = connections.borrow_mut(); - - if let Ok(msg) = message { - // For each open connection except the sender, send the - // string via the channel. - let iter = conns - .iter_mut() - .filter(|&(&k, _)| k != addr) - .map(|(_, v)| v); - for tx in iter { - tx.unbounded_send(format!("{}: {}", addr, msg)).unwrap(); - } - } else { - let tx = conns.get_mut(&addr).unwrap(); - tx.unbounded_send("You didn't send valid UTF-8.".to_string()) - .unwrap(); - } - - reader - }) - }); - - // Whenever we receive a string on the Receiver, we write it to - // `WriteHalf`. - let socket_writer = rx.fold(writer, |writer, msg| { - let amt = io::write_all(writer, msg.into_bytes()); - let amt = amt.map(|(writer, _)| writer); - amt.map_err(|_| ()) - }); - - // Now that we've got futures representing each half of the socket, we - // use the `select` combinator to wait for either half to be done to - // tear down the other. Then we spawn off the result. - let connections = connections.clone(); - let socket_reader = socket_reader.map_err(|_| ()); - let connection = socket_reader.map(|_| ()).select(socket_writer.map(|_| ())); - - // Spawn locally a task to process the connection - TaskExecutor::current() - .spawn_local(Box::new(connection.then(move |_| { - let mut conns = connections.borrow_mut(); - conns.remove(&addr); - println!("Connection {} closed.", addr); - Ok(()) - }))) - .unwrap(); - - Ok(()) - }) - .map_err(|err| println!("error occurred: {:?}", err)); - - // Spawn srv itself - runtime.spawn(srv); - - // Execute server - runtime.run().unwrap(); - Ok(()) -} diff --git a/tokio/examples_old/chat-combinator.rs b/tokio/examples_old/chat-combinator.rs deleted file mode 100644 index 4c42cdfff..000000000 --- a/tokio/examples_old/chat-combinator.rs +++ /dev/null @@ -1,154 +0,0 @@ -//! A chat server that broadcasts a message to all connections. -//! -//! This is a line-based server which accepts connections, reads lines from -//! those connections, and broadcasts the lines to all other connected clients. -//! -//! This example is similar to chat.rs, but uses combinators and a much more -//! functional style. -//! -//! You can test this out by running: -//! -//! cargo run --example chat -//! -//! And then in another window run: -//! -//! cargo run --example connect 127.0.0.1:8080 -//! -//! You can run the second command in multiple windows and then chat between the -//! two, seeing the messages from the other client as they're received. For all -//! connected clients they'll all join the same room and see everyone else's -//! messages. - -#![deny(warnings, rust_2018_idioms)] - -use futures; -use std::collections::HashMap; -use std::env; -use std::io::BufReader; -use std::iter; -use std::sync::{Arc, Mutex}; -use tokio; -use tokio::io; -use tokio::net::TcpListener; -use tokio::prelude::*; - -fn main() -> Result<(), Box> { - // Create the TCP listener we'll accept connections on. - let addr = env::args().nth(1).unwrap_or("127.0.0.1:8080".to_string()); - let addr = addr.parse()?; - - let socket = TcpListener::bind(&addr)?; - println!("Listening on: {}", addr); - - // This is running on the Tokio runtime, so it will be multi-threaded. The - // `Arc>` allows state to be shared across the threads. - let connections = Arc::new(Mutex::new(HashMap::new())); - - // The server task asynchronously iterates over and processes each incoming - // connection. - let srv = socket - .incoming() - .map_err(|e| { - println!("failed to accept socket; error = {:?}", e); - e - }) - .for_each(move |stream| { - // The client's socket address - let addr = stream.peer_addr()?; - - println!("New Connection: {}", addr); - - // Split the TcpStream into two separate handles. One handle for reading - // and one handle for writing. This lets us use separate tasks for - // reading and writing. - let (reader, writer) = stream.split(); - - // Create a channel for our stream, which other sockets will use to - // send us messages. Then register our address with the stream to send - // data to us. - let (tx, rx) = futures::sync::mpsc::unbounded(); - connections.lock().unwrap().insert(addr, tx); - - // Define here what we do for the actual I/O. That is, read a bunch of - // lines from the socket and dispatch them while we also write any lines - // from other sockets. - let connections_inner = connections.clone(); - let reader = BufReader::new(reader); - - // Model the read portion of this socket by mapping an infinite - // iterator to each line off the socket. This "loop" is then - // terminated with an error once we hit EOF on the socket. - let iter = stream::iter_ok::<_, io::Error>(iter::repeat(())); - - let socket_reader = iter.fold(reader, move |reader, _| { - // Read a line off the socket, failing if we're at EOF - let line = io::read_until(reader, b'\n', Vec::new()); - let line = line.and_then(|(reader, vec)| { - if vec.len() == 0 { - Err(io::Error::new(io::ErrorKind::BrokenPipe, "broken pipe")) - } else { - Ok((reader, vec)) - } - }); - - // Convert the bytes we read into a string, and then send that - // string to all other connected clients. - let line = line.map(|(reader, vec)| (reader, String::from_utf8(vec))); - - // Move the connection state into the closure below. - let connections = connections_inner.clone(); - - line.map(move |(reader, message)| { - println!("{}: {:?}", addr, message); - let mut conns = connections.lock().unwrap(); - - if let Ok(msg) = message { - // For each open connection except the sender, send the - // string via the channel. - let iter = conns - .iter_mut() - .filter(|&(&k, _)| k != addr) - .map(|(_, v)| v); - for tx in iter { - tx.unbounded_send(format!("{}: {}", addr, msg)).unwrap(); - } - } else { - let tx = conns.get_mut(&addr).unwrap(); - tx.unbounded_send("You didn't send valid UTF-8.".to_string()) - .unwrap(); - } - - reader - }) - }); - - // Whenever we receive a string on the Receiver, we write it to - // `WriteHalf`. - let socket_writer = rx.fold(writer, |writer, msg| { - let amt = io::write_all(writer, msg.into_bytes()); - let amt = amt.map(|(writer, _)| writer); - amt.map_err(|_| ()) - }); - - // Now that we've got futures representing each half of the socket, we - // use the `select` combinator to wait for either half to be done to - // tear down the other. Then we spawn off the result. - let connections = connections.clone(); - let socket_reader = socket_reader.map_err(|_| ()); - let connection = socket_reader.map(|_| ()).select(socket_writer.map(|_| ())); - - // Spawn a task to process the connection - tokio::spawn(connection.then(move |_| { - connections.lock().unwrap().remove(&addr); - println!("Connection {} closed.", addr); - Ok(()) - })); - - Ok(()) - }) - .map_err(|err| println!("error occurred: {:?}", err)); - - // execute server - tokio::run(srv); - Ok(()) -} diff --git a/tokio/examples_old/chat.rs b/tokio/examples_old/chat.rs deleted file mode 100644 index eb2f51938..000000000 --- a/tokio/examples_old/chat.rs +++ /dev/null @@ -1,469 +0,0 @@ -//! A chat server that broadcasts a message to all connections. -//! -//! This example is explicitly more verbose than it has to be. This is to -//! illustrate more concepts. -//! -//! A chat server for telnet clients. After a telnet client connects, the first -//! line should contain the client's name. After that, all lines sent by a -//! client are broadcasted to all other connected clients. -//! -//! Because the client is telnet, lines are delimited by "\r\n". -//! -//! You can test this out by running: -//! -//! cargo run --example chat -//! -//! And then in another terminal run: -//! -//! telnet localhost 6142 -//! -//! You can run the `telnet` command in any number of additional windows. -//! -//! You can run the second command in multiple windows and then chat between the -//! two, seeing the messages from the other client as they're received. For all -//! connected clients they'll all join the same room and see everyone else's -//! messages. - -#![deny(warnings, rust_2018_idioms)] - -use bytes::{BufMut, Bytes, BytesMut}; -use futures::future::{self, Either}; -use futures::sync::mpsc; -use futures::try_ready; -use std::collections::HashMap; -use std::net::SocketAddr; -use std::sync::{Arc, Mutex}; -use tokio; -use tokio::io; -use tokio::net::{TcpListener, TcpStream}; -use tokio::prelude::*; - -/// Shorthand for the transmit half of the message channel. -type Tx = mpsc::UnboundedSender; - -/// Shorthand for the receive half of the message channel. -type Rx = mpsc::UnboundedReceiver; - -/// Data that is shared between all peers in the chat server. -/// -/// This is the set of `Tx` handles for all connected clients. Whenever a -/// message is received from a client, it is broadcasted to all peers by -/// iterating over the `peers` entries and sending a copy of the message on each -/// `Tx`. -struct Shared { - peers: HashMap, -} - -/// The state for each connected client. -struct Peer { - /// Name of the peer. - /// - /// When a client connects, the first line sent is treated as the client's - /// name (like alice or bob). The name is used to preface all messages that - /// arrive from the client so that we can simulate a real chat server: - /// - /// ```text - /// alice: Hello everyone. - /// bob: Welcome to telnet chat! - /// ``` - name: BytesMut, - - /// The TCP socket wrapped with the `Lines` codec, defined below. - /// - /// This handles sending and receiving data on the socket. When using - /// `Lines`, we can work at the line level instead of having to manage the - /// raw byte operations. - lines: Lines, - - /// Handle to the shared chat state. - /// - /// This is used to broadcast messages read off the socket to all connected - /// peers. - state: Arc>, - - /// Receive half of the message channel. - /// - /// This is used to receive messages from peers. When a message is received - /// off of this `Rx`, it will be written to the socket. - rx: Rx, - - /// Client socket address. - /// - /// The socket address is used as the key in the `peers` HashMap. The - /// address is saved so that the `Peer` drop implementation can clean up its - /// entry. - addr: SocketAddr, -} - -/// Line based codec -/// -/// This decorates a socket and presents a line based read / write interface. -/// -/// As a user of `Lines`, we can focus on working at the line level. So, we send -/// and receive values that represent entire lines. The `Lines` codec will -/// handle the encoding and decoding as well as reading from and writing to the -/// socket. -#[derive(Debug)] -struct Lines { - /// The TCP socket. - socket: TcpStream, - - /// Buffer used when reading from the socket. Data is not returned from this - /// buffer until an entire line has been read. - rd: BytesMut, - - /// Buffer used to stage data before writing it to the socket. - wr: BytesMut, -} - -impl Shared { - /// Create a new, empty, instance of `Shared`. - fn new() -> Self { - Shared { - peers: HashMap::new(), - } - } -} - -impl Peer { - /// Create a new instance of `Peer`. - fn new(name: BytesMut, state: Arc>, lines: Lines) -> Peer { - // Get the client socket address - let addr = lines.socket.peer_addr().unwrap(); - - // Create a channel for this peer - let (tx, rx) = mpsc::unbounded(); - - // Add an entry for this `Peer` in the shared state map. - state.lock().unwrap().peers.insert(addr, tx); - - Peer { - name, - lines, - state, - rx, - addr, - } - } -} - -/// This is where a connected client is managed. -/// -/// A `Peer` is also a future representing completely processing the client. -/// -/// When a `Peer` is created, the first line (representing the client's name) -/// has already been read. When the socket closes, the `Peer` future completes. -/// -/// While processing, the peer future implementation will: -/// -/// 1) Receive messages on its message channel and write them to the socket. -/// 2) Receive messages from the socket and broadcast them to all peers. -/// -impl Future for Peer { - type Item = (); - type Error = io::Error; - - fn poll(&mut self) -> Poll<(), io::Error> { - // Tokio (and futures) use cooperative scheduling without any - // preemption. If a task never yields execution back to the executor, - // then other tasks may be starved. - // - // To deal with this, robust applications should not have any unbounded - // loops. In this example, we will read at most `LINES_PER_TICK` lines - // from the client on each tick. - // - // If the limit is hit, the current task is notified, informing the - // executor to schedule the task again asap. - const LINES_PER_TICK: usize = 10; - - // Receive all messages from peers. - for i in 0..LINES_PER_TICK { - // Polling an `UnboundedReceiver` cannot fail, so `unwrap` here is - // safe. - match self.rx.poll().unwrap() { - Async::Ready(Some(v)) => { - // Buffer the line. Once all lines are buffered, they will - // be flushed to the socket (right below). - self.lines.buffer(&v); - - // If this is the last iteration, the loop will break even - // though there could still be lines to read. Because we did - // not reach `Async::NotReady`, we have to notify ourselves - // in order to tell the executor to schedule the task again. - if i + 1 == LINES_PER_TICK { - task::current().notify(); - } - } - _ => break, - } - } - - // Flush the write buffer to the socket - let _ = self.lines.poll_flush()?; - - // Read new lines from the socket - while let Async::Ready(line) = self.lines.poll()? { - println!("Received line ({:?}) : {:?}", self.name, line); - - if let Some(message) = line { - // Append the peer's name to the front of the line: - let mut line = self.name.clone(); - line.extend_from_slice(b": "); - line.extend_from_slice(&message); - line.extend_from_slice(b"\r\n"); - - // We're using `Bytes`, which allows zero-copy clones (by - // storing the data in an Arc internally). - // - // However, before cloning, we must freeze the data. This - // converts it from mutable -> immutable, allowing zero copy - // cloning. - let line = line.freeze(); - - // Now, send the line to all other peers - for (addr, tx) in &self.state.lock().unwrap().peers { - // Don't send the message to ourselves - if *addr != self.addr { - // The send only fails if the rx half has been dropped, - // however this is impossible as the `tx` half will be - // removed from the map before the `rx` is dropped. - tx.unbounded_send(line.clone()).unwrap(); - } - } - } else { - // EOF was reached. The remote client has disconnected. There is - // nothing more to do. - return Ok(Async::Ready(())); - } - } - - // As always, it is important to not just return `NotReady` without - // ensuring an inner future also returned `NotReady`. - // - // We know we got a `NotReady` from either `self.rx` or `self.lines`, so - // the contract is respected. - Ok(Async::NotReady) - } -} - -impl Drop for Peer { - fn drop(&mut self) { - self.state.lock().unwrap().peers.remove(&self.addr); - } -} - -impl Lines { - /// Create a new `Lines` codec backed by the socket - fn new(socket: TcpStream) -> Self { - Lines { - socket, - rd: BytesMut::new(), - wr: BytesMut::new(), - } - } - - /// Buffer a line. - /// - /// This writes the line to an internal buffer. Calls to `poll_flush` will - /// attempt to flush this buffer to the socket. - fn buffer(&mut self, line: &[u8]) { - // Ensure the buffer has capacity. Ideally this would not be unbounded, - // but to keep the example simple, we will not limit this. - self.wr.reserve(line.len()); - - // Push the line onto the end of the write buffer. - // - // The `put` function is from the `BufMut` trait. - self.wr.put(line); - } - - /// Flush the write buffer to the socket - fn poll_flush(&mut self) -> Poll<(), io::Error> { - // As long as there is buffered data to write, try to write it. - while !self.wr.is_empty() { - // Try to write some bytes to the socket - let n = try_ready!(self.socket.poll_write(&self.wr)); - - // As long as the wr is not empty, a successful write should - // never write 0 bytes. - assert!(n > 0); - - // This discards the first `n` bytes of the buffer. - let _ = self.wr.split_to(n); - } - - Ok(Async::Ready(())) - } - - /// Read data from the socket. - /// - /// This only returns `Ready` when the socket has closed. - fn fill_read_buf(&mut self) -> Poll<(), io::Error> { - loop { - // Ensure the read buffer has capacity. - // - // This might result in an internal allocation. - self.rd.reserve(1024); - - // Read data into the buffer. - let n = try_ready!(self.socket.read_buf(&mut self.rd)); - - if n == 0 { - return Ok(Async::Ready(())); - } - } - } -} - -impl Stream for Lines { - type Item = BytesMut; - type Error = io::Error; - - fn poll(&mut self) -> Poll, Self::Error> { - // First, read any new data that might have been received off the socket - let sock_closed = self.fill_read_buf()?.is_ready(); - - // Now, try finding lines - let pos = self - .rd - .windows(2) - .enumerate() - .find(|&(_, bytes)| bytes == b"\r\n") - .map(|(i, _)| i); - - if let Some(pos) = pos { - // Remove the line from the read buffer and set it to `line`. - let mut line = self.rd.split_to(pos + 2); - - // Drop the trailing \r\n - line.split_off(pos); - - // Return the line - return Ok(Async::Ready(Some(line))); - } - - if sock_closed { - Ok(Async::Ready(None)) - } else { - Ok(Async::NotReady) - } - } -} - -/// Spawn a task to manage the socket. -/// -/// This will read the first line from the socket to identify the client, then -/// add the client to the set of connected peers in the chat service. -fn process(socket: TcpStream, state: Arc>) { - // Wrap the socket with the `Lines` codec that we wrote above. - // - // By doing this, we can operate at the line level instead of doing raw byte - // manipulation. - let lines = Lines::new(socket); - - // The first line is treated as the client's name. The client is not added - // to the set of connected peers until this line is received. - // - // We use the `into_future` combinator to extract the first item from the - // lines stream. `into_future` takes a `Stream` and converts it to a future - // of `(first, rest)` where `rest` is the original stream instance. - let connection = lines - .into_future() - // `into_future` doesn't have the right error type, so map the error to - // make it work. - .map_err(|(e, _)| e) - // Process the first received line as the client's name. - .and_then(|(name, lines)| { - // If `name` is `None`, then the client disconnected without - // actually sending a line of data. - // - // Since the connection is closed, there is no further work that we - // need to do. So, we just terminate processing by returning - // `future::ok()`. - // - // The problem is that only a single future type can be returned - // from a combinator closure, but we want to return both - // `future::ok()` and `Peer` (below). - // - // This is a common problem, so the `futures` crate solves this by - // providing the `Either` helper enum that allows creating a single - // return type that covers two concrete future types. - let name = match name { - Some(name) => name, - None => { - // The remote client closed the connection without sending - // any data. - return Either::A(future::ok(())); - } - }; - - println!("`{:?}` is joining the chat", name); - - // Create the peer. - // - // This is also a future that processes the connection, only - // completing when the socket closes. - let peer = Peer::new(name, state, lines); - - // Wrap `peer` with `Either::B` to make the return type fit. - Either::B(peer) - }) - // Task futures have an error of type `()`, this ensures we handle the - // error. We do this by printing the error to STDOUT. - .map_err(|e| { - println!("connection error = {:?}", e); - }); - - // Spawn the task. Internally, this submits the task to a thread pool. - tokio::spawn(connection); -} - -pub fn main() -> Result<(), Box> { - // Create the shared state. This is how all the peers communicate. - // - // The server task will hold a handle to this. For every new client, the - // `state` handle is cloned and passed into the task that processes the - // client connection. - let state = Arc::new(Mutex::new(Shared::new())); - - let addr = "127.0.0.1:6142".parse()?; - - // Bind a TCP listener to the socket address. - // - // Note that this is the Tokio TcpListener, which is fully async. - let listener = TcpListener::bind(&addr)?; - - // The server task asynchronously iterates over and processes each - // incoming connection. - let server = listener - .incoming() - .for_each(move |socket| { - // Spawn a task to process the connection - process(socket, state.clone()); - Ok(()) - }) - .map_err(|err| { - // All tasks must have an `Error` type of `()`. This forces error - // handling and helps avoid silencing failures. - // - // In our example, we are only going to log the error to STDOUT. - println!("accept error = {:?}", err); - }); - - println!("server running on localhost:6142"); - - // Start the Tokio runtime. - // - // The Tokio is a pre-configured "out of the box" runtime for building - // asynchronous applications. It includes both a reactor and a task - // scheduler. This means applications are multithreaded by default. - // - // This function blocks until the runtime reaches an idle state. Idle is - // defined as all spawned tasks have completed and all I/O resources (TCP - // sockets in our case) have been dropped. - // - // In our example, we have not defined a shutdown strategy, so this will - // block until `ctrl-c` is pressed at the terminal. - tokio::run(server); - Ok(()) -} diff --git a/tokio/examples_old/echo.rs b/tokio/examples_old/echo.rs deleted file mode 100644 index 0b52490e3..000000000 --- a/tokio/examples_old/echo.rs +++ /dev/null @@ -1,113 +0,0 @@ -//! A "hello world" echo server with Tokio -//! -//! This server will create a TCP listener, accept connections in a loop, and -//! write back everything that's read off of each TCP connection. -//! -//! Because the Tokio runtime uses a thread pool, each TCP connection is -//! processed concurrently with all other TCP connections across multiple -//! threads. -//! -//! To see this server in action, you can run this in one terminal: -//! -//! cargo run --example echo -//! -//! and in another terminal you can run: -//! -//! cargo run --example connect 127.0.0.1:8080 -//! -//! Each line you type in to the `connect` terminal should be echo'd back to -//! you! If you open up multiple terminals running the `connect` example you -//! should be able to see them all make progress simultaneously. - -#![deny(warnings, rust_2018_idioms)] - -use std::env; -use std::net::SocketAddr; -use tokio; -use tokio::io; -use tokio::net::TcpListener; -use tokio::prelude::*; - -fn main() -> Result<(), Box> { - // Allow passing an address to listen on as the first argument of this - // program, but otherwise we'll just set up our TCP listener on - // 127.0.0.1:8080 for connections. - let addr = env::args().nth(1).unwrap_or("127.0.0.1:8080".to_string()); - let addr = addr.parse::()?; - - // Next up we create a TCP listener which will listen for incoming - // connections. This TCP listener is bound to the address we determined - // above and must be associated with an event loop, so we pass in a handle - // to our event loop. After the socket's created we inform that we're ready - // to go and start accepting connections. - let socket = TcpListener::bind(&addr)?; - println!("Listening on: {}", addr); - - // Here we convert the `TcpListener` to a stream of incoming connections - // with the `incoming` method. We then define how to process each element in - // the stream with the `for_each` method. - // - // This combinator, defined on the `Stream` trait, will allow us to define a - // computation to happen for all items on the stream (in this case TCP - // connections made to the server). The return value of the `for_each` - // method is itself a future representing processing the entire stream of - // connections, and ends up being our server. - let done = socket - .incoming() - .map_err(|e| println!("failed to accept socket; error = {:?}", e)) - .for_each(move |socket| { - // Once we're inside this closure this represents an accepted client - // from our server. The `socket` is the client connection (similar to - // how the standard library operates). - // - // We just want to copy all data read from the socket back onto the - // socket itself (e.g. "echo"). We can use the standard `io::copy` - // combinator in the `tokio-core` crate to do precisely this! - // - // The `copy` function takes two arguments, where to read from and where - // to write to. We only have one argument, though, with `socket`. - // Luckily there's a method, `Io::split`, which will split an Read/Write - // stream into its two halves. This operation allows us to work with - // each stream independently, such as pass them as two arguments to the - // `copy` function. - // - // The `copy` function then returns a future, and this future will be - // resolved when the copying operation is complete, resolving to the - // amount of data that was copied. - let (reader, writer) = socket.split(); - let amt = io::copy(reader, writer); - - // After our copy operation is complete we just print out some helpful - // information. - let msg = amt.then(move |result| { - match result { - Ok((amt, _, _)) => println!("wrote {} bytes", amt), - Err(e) => println!("error: {}", e), - } - - Ok(()) - }); - - // And this is where much of the magic of this server happens. We - // crucially want all clients to make progress concurrently, rather than - // blocking one on completion of another. To achieve this we use the - // `tokio::spawn` function to execute the work in the background. - // - // This function will transfer ownership of the future (`msg` in this - // case) to the Tokio runtime thread pool that. The thread pool will - // drive the future to completion. - // - // Essentially here we're executing a new task to run concurrently, - // which will allow all of our clients to be processed concurrently. - tokio::spawn(msg) - }); - - // And finally now that we've define what our server is, we run it! - // - // This starts the Tokio runtime, spawns the server task, and blocks the - // current thread until all tasks complete execution. Since the `done` task - // never completes (it just keeps accepting sockets), `tokio::run` blocks - // forever (until ctrl-c is pressed). - tokio::run(done); - Ok(()) -} diff --git a/tokio/examples_old/tinydb.rs b/tokio/examples_old/tinydb.rs deleted file mode 100644 index dde523553..000000000 --- a/tokio/examples_old/tinydb.rs +++ /dev/null @@ -1,225 +0,0 @@ -//! A "tiny database" and accompanying protocol -//! -//! This example shows the usage of shared state amongst all connected clients, -//! namely a database of key/value pairs. Each connected client can send a -//! series of GET/SET commands to query the current value of a key or set the -//! value of a key. -//! -//! This example has a simple protocol you can use to interact with the server. -//! To run, first run this in one terminal window: -//! -//! cargo run --example tinydb -//! -//! and next in another windows run: -//! -//! cargo run --example connect 127.0.0.1:8080 -//! -//! In the `connect` window you can type in commands where when you hit enter -//! you'll get a response from the server for that command. An example session -//! is: -//! -//! -//! $ cargo run --example connect 127.0.0.1:8080 -//! GET foo -//! foo = bar -//! GET FOOBAR -//! error: no key FOOBAR -//! SET FOOBAR my awesome string -//! set FOOBAR = `my awesome string`, previous: None -//! SET foo tokio -//! set foo = `tokio`, previous: Some("bar") -//! GET foo -//! foo = tokio -//! -//! Namely you can issue two forms of commands: -//! -//! * `GET $key` - this will fetch the value of `$key` from the database and -//! return it. The server's database is initially populated with the key `foo` -//! set to the value `bar` -//! * `SET $key $value` - this will set the value of `$key` to `$value`, -//! returning the previous value, if any. - -#![deny(warnings, rust_2018_idioms)] - -use std::collections::HashMap; -use std::env; -use std::io::BufReader; -use std::net::SocketAddr; -use std::sync::{Arc, Mutex}; -use tokio; -use tokio::io::{lines, write_all}; -use tokio::net::TcpListener; -use tokio::prelude::*; - -/// The in-memory database shared amongst all clients. -/// -/// This database will be shared via `Arc`, so to mutate the internal map we're -/// going to use a `Mutex` for interior mutability. -struct Database { - map: Mutex>, -} - -/// Possible requests our clients can send us -enum Request { - Get { key: String }, - Set { key: String, value: String }, -} - -/// Responses to the `Request` commands above -enum Response { - Value { - key: String, - value: String, - }, - Set { - key: String, - value: String, - previous: Option, - }, - Error { - msg: String, - }, -} - -fn main() -> Result<(), Box> { - // Parse the address we're going to run this server on - // and set up our TCP listener to accept connections. - let addr = env::args().nth(1).unwrap_or("127.0.0.1:8080".to_string()); - let addr = addr.parse::()?; - let listener = TcpListener::bind(&addr).map_err(|_| "failed to bind")?; - println!("Listening on: {}", addr); - - // Create the shared state of this server that will be shared amongst all - // clients. We populate the initial database and then create the `Database` - // structure. Note the usage of `Arc` here which will be used to ensure that - // each independently spawned client will have a reference to the in-memory - // database. - let mut initial_db = HashMap::new(); - initial_db.insert("foo".to_string(), "bar".to_string()); - let db = Arc::new(Database { - map: Mutex::new(initial_db), - }); - - let done = listener - .incoming() - .map_err(|e| println!("error accepting socket; error = {:?}", e)) - .for_each(move |socket| { - // As with many other small examples, the first thing we'll do is - // *split* this TCP stream into two separately owned halves. This'll - // allow us to work with the read and write halves independently. - let (reader, writer) = socket.split(); - - // Since our protocol is line-based we use `tokio_io`'s `lines` utility - // to convert our stream of bytes, `reader`, into a `Stream` of lines. - let lines = lines(BufReader::new(reader)); - - // Here's where the meat of the processing in this server happens. First - // we see a clone of the database being created, which is creating a - // new reference for this connected client to use. Also note the `move` - // keyword on the closure here which moves ownership of the reference - // into the closure, which we'll need for spawning the client below. - // - // The `map` function here means that we'll run some code for all - // requests (lines) we receive from the client. The actual handling here - // is pretty simple, first we parse the request and if it's valid we - // generate a response based on the values in the database. - let db = db.clone(); - let responses = lines.map(move |line| { - let request = match Request::parse(&line) { - Ok(req) => req, - Err(e) => return Response::Error { msg: e }, - }; - - let mut db = db.map.lock().unwrap(); - match request { - Request::Get { key } => match db.get(&key) { - Some(value) => Response::Value { - key, - value: value.clone(), - }, - None => Response::Error { - msg: format!("no key {}", key), - }, - }, - Request::Set { key, value } => { - let previous = db.insert(key.clone(), value.clone()); - Response::Set { - key, - value, - previous, - } - } - } - }); - - // At this point `responses` is a stream of `Response` types which we - // now want to write back out to the client. To do that we use - // `Stream::fold` to perform a loop here, serializing each response and - // then writing it out to the client. - let writes = responses.fold(writer, |writer, response| { - let mut response = response.serialize(); - response.push('\n'); - write_all(writer, response.into_bytes()).map(|(w, _)| w) - }); - - // Like with other small servers, we'll `spawn` this client to ensure it - // runs concurrently with all other clients, for now ignoring any errors - // that we see. - let msg = writes.then(move |_| Ok(())); - - tokio::spawn(msg) - }); - - tokio::run(done); - Ok(()) -} - -impl Request { - fn parse(input: &str) -> Result { - let mut parts = input.splitn(3, " "); - match parts.next() { - Some("GET") => { - let key = match parts.next() { - Some(key) => key, - None => return Err(format!("GET must be followed by a key")), - }; - if parts.next().is_some() { - return Err(format!("GET's key must not be followed by anything")); - } - Ok(Request::Get { - key: key.to_string(), - }) - } - Some("SET") => { - let key = match parts.next() { - Some(key) => key, - None => return Err(format!("SET must be followed by a key")), - }; - let value = match parts.next() { - Some(value) => value, - None => return Err(format!("SET needs a value")), - }; - Ok(Request::Set { - key: key.to_string(), - value: value.to_string(), - }) - } - Some(cmd) => Err(format!("unknown command: {}", cmd)), - None => Err(format!("empty input")), - } - } -} - -impl Response { - fn serialize(&self) -> String { - match *self { - Response::Value { ref key, ref value } => format!("{} = {}", key, value), - Response::Set { - ref key, - ref value, - ref previous, - } => format!("set {} = `{}`, previous: {:?}", key, value, previous), - Response::Error { ref msg } => format!("error: {}", msg), - } - } -} diff --git a/tokio/tests/buffered.rs b/tokio/tests/buffered.rs index 11f86f794..8bc458d78 100644 --- a/tokio/tests/buffered.rs +++ b/tokio/tests/buffered.rs @@ -1,63 +1,53 @@ -#![cfg(feature = "broken")] +#![feature(async_await)] #![deny(warnings, rust_2018_idioms)] +#![cfg(feature = "default")] -use env_logger; -use futures::stream::Stream; -use futures::Future; -use std::io::{BufReader, BufWriter, Read, Write}; +use tokio::net::TcpListener; +use tokio::prelude::*; +use tokio_test::assert_ok; + +use std::io::prelude::*; use std::net::TcpStream; use std::thread; -use tokio::net::TcpListener; -use tokio_io::io::copy; -macro_rules! t { - ($e:expr) => { - match $e { - Ok(e) => e, - Err(e) => panic!("{} failed with {:?}", stringify!($e), e), - } - }; -} - -#[test] -fn echo_server() { +#[tokio::test] +async fn echo_server() { const N: usize = 1024; - drop(env_logger::try_init()); - let srv = t!(TcpListener::bind(&t!("127.0.0.1:0".parse()))); - let addr = t!(srv.local_addr()); + let addr = assert_ok!("127.0.0.1:0".parse()); + let mut srv = assert_ok!(TcpListener::bind(&addr)); + let addr = assert_ok!(srv.local_addr()); let msg = "foo bar baz"; + let t = thread::spawn(move || { - let mut s = t!(TcpStream::connect(&addr)); + let mut s = assert_ok!(TcpStream::connect(&addr)); let t2 = thread::spawn(move || { - let mut s = t!(TcpStream::connect(&addr)); + let mut s = assert_ok!(TcpStream::connect(&addr)); let mut b = vec![0; msg.len() * N]; - t!(s.read_exact(&mut b)); + assert_ok!(s.read_exact(&mut b)); b }); let mut expected = Vec::::new(); for _i in 0..N { expected.extend(msg.as_bytes()); - assert_eq!(t!(s.write(msg.as_bytes())), msg.len()); + let res = assert_ok!(s.write(msg.as_bytes())); + assert_eq!(res, msg.len()); } + (expected, t2) }); - let clients = srv.incoming().take(2).collect(); - let copied = clients.and_then(|clients| { - let mut clients = clients.into_iter(); - let a = BufReader::new(clients.next().unwrap()); - let b = BufWriter::new(clients.next().unwrap()); - copy(a, b) - }); + let (mut a, _) = assert_ok!(srv.accept().await); + let (mut b, _) = assert_ok!(srv.accept().await); + + let n = assert_ok!(a.copy(&mut b).await); - let (amt, _, _) = t!(copied.wait()); let (expected, t2) = t.join().unwrap(); let actual = t2.join().unwrap(); assert!(expected == actual); - assert_eq!(amt, msg.len() as u64 * 1024); + assert_eq!(n, msg.len() as u64 * 1024); } diff --git a/tokio/tests/clock.rs b/tokio/tests/clock.rs index 2fa22c9d3..b7367804e 100644 --- a/tokio/tests/clock.rs +++ b/tokio/tests/clock.rs @@ -1,15 +1,15 @@ -#![cfg(feature = "broken")] +#![feature(async_await)] #![deny(warnings, rust_2018_idioms)] +#![cfg(feature = "default")] -use env_logger; -use std::sync::mpsc; -use std::time::{Duration, Instant}; -use tokio::prelude::*; use tokio::runtime::{self, current_thread}; use tokio::timer::*; use tokio_timer; use tokio_timer::clock::Clock; +use std::sync::mpsc; +use std::time::{Duration, Instant}; + struct MockNow(Instant); impl tokio_timer::clock::Now for MockNow { @@ -20,8 +20,6 @@ impl tokio_timer::clock::Now for MockNow { #[test] fn clock_and_timer_concurrent() { - let _ = env_logger::try_init(); - let when = Instant::now() + Duration::from_millis(5_000); let clock = Clock::new_with_now(MockNow(when)); @@ -29,14 +27,10 @@ fn clock_and_timer_concurrent() { let (tx, rx) = mpsc::channel(); - rt.spawn({ - Delay::new(when) - .map_err(|e| panic!("unexpected error; err={:?}", e)) - .and_then(move |_| { - assert!(Instant::now() < when); - tx.send(()).unwrap(); - Ok(()) - }) + rt.spawn(async move { + Delay::new(when).await; + assert!(Instant::now() < when); + tx.send(()).unwrap(); }); rx.recv().unwrap(); @@ -44,20 +38,13 @@ fn clock_and_timer_concurrent() { #[test] fn clock_and_timer_single_threaded() { - let _ = env_logger::try_init(); - let when = Instant::now() + Duration::from_millis(5_000); let clock = Clock::new_with_now(MockNow(when)); let mut rt = current_thread::Builder::new().clock(clock).build().unwrap(); - rt.block_on({ - Delay::new(when) - .map_err(|e| panic!("unexpected error; err={:?}", e)) - .and_then(move |_| { - assert!(Instant::now() < when); - Ok(()) - }) - }) - .unwrap(); + rt.block_on(async move { + Delay::new(when).await; + assert!(Instant::now() < when); + }); } diff --git a/tokio/tests/drop-core.rs b/tokio/tests/drop-core.rs index 98c136e82..fd451b139 100644 --- a/tokio/tests/drop-core.rs +++ b/tokio/tests/drop-core.rs @@ -1,41 +1,41 @@ -#![cfg(feature = "broken")] +#![feature(async_await)] #![deny(warnings, rust_2018_idioms)] +#![cfg(feature = "default")] -use futures::future; -use futures::prelude::*; -use futures::sync::oneshot; -use std::net; -use std::thread; use tokio::net::TcpListener; use tokio::reactor::Reactor; +use tokio_test::{assert_err, assert_pending, assert_ready, task}; #[test] fn tcp_doesnt_block() { - let core = Reactor::new().unwrap(); - let handle = core.handle(); - let listener = net::TcpListener::bind("127.0.0.1:0").unwrap(); - let listener = TcpListener::from_std(listener, &handle).unwrap(); - drop(core); - assert!(listener.incoming().wait().next().unwrap().is_err()); + let reactor = Reactor::new().unwrap(); + let handle = reactor.handle(); + let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + let mut listener = TcpListener::from_std(listener, &handle).unwrap(); + drop(reactor); + + let mut task = task::spawn(async move { + assert_err!(listener.accept().await); + }); + + assert_ready!(task.poll()); } #[test] fn drop_wakes() { - let core = Reactor::new().unwrap(); - let handle = core.handle(); - let listener = net::TcpListener::bind("127.0.0.1:0").unwrap(); - let listener = TcpListener::from_std(listener, &handle).unwrap(); - let (tx, rx) = oneshot::channel::<()>(); - let t = thread::spawn(move || { - let incoming = listener.incoming(); - let new_socket = incoming.into_future().map_err(|_| ()); - let drop_tx = future::lazy(|| { - drop(tx); - future::ok(()) - }); - assert!(new_socket.join(drop_tx).wait().is_err()); + let reactor = Reactor::new().unwrap(); + let handle = reactor.handle(); + let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + let mut listener = TcpListener::from_std(listener, &handle).unwrap(); + + let mut task = task::spawn(async move { + assert_err!(listener.accept().await); }); - drop(rx.wait()); - drop(core); - t.join().unwrap(); + + assert_pending!(task.poll()); + + drop(reactor); + + assert!(task.is_woken()); + assert_ready!(task.poll()); } diff --git a/tokio/tests/global.rs b/tokio/tests/global.rs deleted file mode 100644 index 52111a13a..000000000 --- a/tokio/tests/global.rs +++ /dev/null @@ -1,141 +0,0 @@ -#![cfg(feature = "broken")] -#![deny(warnings, rust_2018_idioms)] - -use env_logger; -use futures::prelude::*; -use std::sync::atomic::AtomicUsize; -use std::sync::atomic::Ordering::Relaxed; -use std::sync::Arc; -use std::{io, thread}; -use tokio; -use tokio::net::{TcpListener, TcpStream}; -use tokio::runtime::Runtime; -use tokio_io; - -macro_rules! t { - ($e:expr) => { - match $e { - Ok(e) => e, - Err(e) => panic!("{} failed with {:?}", stringify!($e), e), - } - }; -} - -#[test] -fn hammer_old() { - let _ = env_logger::try_init(); - - let threads = (0..10) - .map(|_| { - thread::spawn(|| { - let srv = t!(TcpListener::bind(&"127.0.0.1:0".parse().unwrap())); - let addr = t!(srv.local_addr()); - let mine = TcpStream::connect(&addr); - let theirs = srv - .incoming() - .into_future() - .map(|(s, _)| s.unwrap()) - .map_err(|(s, _)| s); - let (mine, theirs) = t!(mine.join(theirs).wait()); - - assert_eq!(t!(mine.local_addr()), t!(theirs.peer_addr())); - assert_eq!(t!(theirs.local_addr()), t!(mine.peer_addr())); - }) - }) - .collect::>(); - for thread in threads { - thread.join().unwrap(); - } -} - -struct Rd(Arc); -struct Wr(Arc); - -impl io::Read for Rd { - fn read(&mut self, dst: &mut [u8]) -> io::Result { - <&TcpStream>::read(&mut &*self.0, dst) - } -} - -impl tokio_io::AsyncRead for Rd {} - -impl io::Write for Wr { - fn write(&mut self, src: &[u8]) -> io::Result { - <&TcpStream>::write(&mut &*self.0, src) - } - - fn flush(&mut self) -> io::Result<()> { - Ok(()) - } -} - -impl tokio_io::AsyncWrite for Wr { - fn shutdown(&mut self) -> Poll<(), io::Error> { - Ok(().into()) - } -} - -#[test] -fn hammer_split() { - use tokio_io::io; - - const N: usize = 100; - const ITER: usize = 10; - - let _ = env_logger::try_init(); - - for _ in 0..ITER { - let srv = t!(TcpListener::bind(&"127.0.0.1:0".parse().unwrap())); - let addr = t!(srv.local_addr()); - - let cnt = Arc::new(AtomicUsize::new(0)); - - let rt = Runtime::new().unwrap(); - - fn split(socket: TcpStream, cnt: Arc) { - let socket = Arc::new(socket); - let rd = Rd(socket.clone()); - let wr = Wr(socket); - - let cnt2 = cnt.clone(); - - let rd = io::read(rd, vec![0; 1]) - .map(move |_| { - cnt2.fetch_add(1, Relaxed); - }) - .map_err(|e| panic!("read error = {:?}", e)); - - let wr = io::write_all(wr, b"1") - .map(move |_| { - cnt.fetch_add(1, Relaxed); - }) - .map_err(move |e| panic!("write error = {:?}", e)); - - tokio::spawn(rd); - tokio::spawn(wr); - } - - rt.spawn({ - let cnt = cnt.clone(); - srv.incoming() - .map_err(|e| panic!("accept error = {:?}", e)) - .take(N as u64) - .for_each(move |socket| { - split(socket, cnt.clone()); - Ok(()) - }) - }); - - for _ in 0..N { - rt.spawn({ - let cnt = cnt.clone(); - TcpStream::connect(&addr) - .map_err(move |e| panic!("connect error = {:?}", e)) - .map(move |socket| split(socket, cnt)) - }); - } - - rt.shutdown_on_idle().wait().unwrap(); - assert_eq!(N * 4, cnt.load(Relaxed)); - } -} diff --git a/tokio/tests/length_delimited.rs b/tokio/tests/length_delimited.rs deleted file mode 100644 index 2b2fe563c..000000000 --- a/tokio/tests/length_delimited.rs +++ /dev/null @@ -1,624 +0,0 @@ -#![cfg(feature = "broken")] -#![deny(warnings, rust_2018_idioms)] - -use bytes::{BufMut, Bytes, BytesMut}; -use futures::Async::*; -use futures::{Poll, Sink, Stream}; -use std::collections::VecDeque; -use std::io; -use tokio::codec::*; -use tokio::io::{AsyncRead, AsyncWrite}; - -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!(), LengthDelimitedCodec::new()); - - 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()), - }, - LengthDelimitedCodec::new(), - ); - - 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 = length_delimited::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_frame_one_packet_native_endian() { - let data = if cfg!(target_endian = "big") { - b"\x00\x00\x00\x09abcdefghi" - } else { - b"\x09\x00\x00\x00abcdefghi" - }; - let mut io = length_delimited::Builder::new() - .native_endian() - .new_read(mock! { - Ok(data[..].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()), - }, - LengthDelimitedCodec::new(), - ); - - 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()), - }, - LengthDelimitedCodec::new(), - ); - - 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()), - }, - LengthDelimitedCodec::new(), - ); - - 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()), - }, - LengthDelimitedCodec::new(), - ); - - 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()), - }, - LengthDelimitedCodec::new(), - ); - - 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()), - }, - LengthDelimitedCodec::new(), - ); - - 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()), - }, - LengthDelimitedCodec::new(), - ); - - 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()), - }, - LengthDelimitedCodec::new(), - ); - - 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 = length_delimited::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 = length_delimited::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.decoder_mut().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 = length_delimited::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.decoder_mut().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 = length_delimited::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 = length_delimited::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 = length_delimited::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 = length_delimited::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 = length_delimited::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(Bytes::from("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::new(mock!(), LengthDelimitedCodec::new()); - 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), - }, - LengthDelimitedCodec::new(), - ); - - assert!(io.start_send(Bytes::from("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), - }, - LengthDelimitedCodec::new(), - ); - - assert!(io.start_send(Bytes::from("abcdefghi")).unwrap().is_ready()); - assert!(io.start_send(Bytes::from("123")).unwrap().is_ready()); - assert!(io - .start_send(Bytes::from("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), - }, - LengthDelimitedCodec::new(), - ); - - assert!(io.start_send(Bytes::from("abcdefghi")).unwrap().is_ready()); - assert!(io.poll_complete().unwrap().is_ready()); - assert!(io.start_send(Bytes::from("123")).unwrap().is_ready()); - assert!(io.poll_complete().unwrap().is_ready()); - assert!(io - .start_send(Bytes::from("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), - }, - LengthDelimitedCodec::new(), - ); - - assert!(io.start_send(Bytes::from("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 = length_delimited::Builder::new() - .little_endian() - .new_write(mock! { - Ok(b"\x09\x00\x00\x00"[..].into()), - Ok(b"abcdefghi"[..].into()), - Ok(Flush), - }); - - assert!(io.start_send(Bytes::from("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 = length_delimited::Builder::new() - .length_field_length(1) - .new_write(mock! { - Ok(b"\x09"[..].into()), - Ok(b"abcdefghi"[..].into()), - Ok(Flush), - }); - - assert!(io.start_send(Bytes::from("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 = length_delimited::Builder::new() - .max_frame_length(5) - .new_write(mock! {}); - - assert_eq!( - io.start_send(Bytes::from("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 = length_delimited::Builder::new().new_write(mock! { - Ok(b"\x00\x00\x00\x06"[..].into()), - Ok(b"abcdef"[..].into()), - Ok(Flush), - }); - - assert!(io.start_send(Bytes::from("abcdef")).unwrap().is_ready()); - assert!(io.poll_complete().unwrap().is_ready()); - io.encoder_mut().set_max_frame_length(5); - assert_eq!( - io.start_send(Bytes::from("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 = length_delimited::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(Bytes::from("abcdef")).unwrap().is_ready()); - assert!(!io.poll_complete().unwrap().is_ready()); - io.encoder_mut().set_max_frame_length(5); - assert!(io.poll_complete().unwrap().is_ready()); - assert_eq!( - io.start_send(Bytes::from("abcdef")).unwrap_err().kind(), - io::ErrorKind::InvalidInput - ); - assert!(io.get_ref().calls.is_empty()); -} - -#[test] -fn write_zero() { - let mut io = length_delimited::Builder::new().new_write(mock! {}); - - assert!(io.start_send(Bytes::from("abcdef")).unwrap().is_ready()); - assert_eq!( - io.poll_complete().unwrap_err().kind(), - io::ErrorKind::WriteZero - ); - assert!(io.get_ref().calls.is_empty()); -} - -#[test] -fn encode_overflow() { - // Test reproducing tokio-rs/tokio#681. - let mut codec = length_delimited::Builder::new().new_codec(); - let mut buf = BytesMut::with_capacity(1024); - - // Put some data into the buffer without resizing it to hold more. - let some_as = std::iter::repeat(b'a').take(1024).collect::>(); - buf.put_slice(&some_as[..]); - - // Trying to encode the length header should resize the buffer if it won't fit. - codec.encode(Bytes::from("hello"), &mut buf).unwrap(); -} - -// ===== 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) - } -} diff --git a/tokio/tests/line-frames.rs b/tokio/tests/line-frames.rs deleted file mode 100644 index 42e884af8..000000000 --- a/tokio/tests/line-frames.rs +++ /dev/null @@ -1,85 +0,0 @@ -#![cfg(feature = "broken")] -#![deny(warnings, rust_2018_idioms)] - -use bytes::{BufMut, BytesMut}; -use env_logger; -use futures::{Future, Sink, Stream}; -use std::io; -use std::net::Shutdown; -use tokio::net::{TcpListener, TcpStream}; -use tokio_codec::{Decoder, Encoder}; -use tokio_io::io::{read, write_all}; -use tokio_threadpool::Builder; - -pub struct LineCodec; - -impl Decoder for LineCodec { - type Item = BytesMut; - type Error = io::Error; - - fn decode(&mut self, buf: &mut BytesMut) -> Result, io::Error> { - match buf.iter().position(|&b| b == b'\n') { - Some(i) => Ok(Some(buf.split_to(i + 1).into())), - None => Ok(None), - } - } - - fn decode_eof(&mut self, buf: &mut BytesMut) -> io::Result> { - if buf.len() == 0 { - Ok(None) - } else { - let amt = buf.len(); - Ok(Some(buf.split_to(amt))) - } - } -} - -impl Encoder for LineCodec { - type Item = BytesMut; - type Error = io::Error; - - fn encode(&mut self, item: BytesMut, into: &mut BytesMut) -> io::Result<()> { - into.put(&item[..]); - Ok(()) - } -} - -#[test] -fn echo() { - drop(env_logger::try_init()); - - let pool = Builder::new().pool_size(1).build(); - - let listener = TcpListener::bind(&"127.0.0.1:0".parse().unwrap()).unwrap(); - let addr = listener.local_addr().unwrap(); - let sender = pool.sender().clone(); - let srv = listener.incoming().for_each(move |socket| { - let (sink, stream) = LineCodec.framed(socket).split(); - sender - .spawn(sink.send_all(stream).map(|_| ()).map_err(|_| ())) - .unwrap(); - Ok(()) - }); - - pool.sender() - .spawn(srv.map_err(|e| panic!("srv error: {}", e))) - .unwrap(); - - let client = TcpStream::connect(&addr); - let client = client.wait().unwrap(); - let (client, _) = write_all(client, b"a\n").wait().unwrap(); - let (client, buf, amt) = read(client, vec![0; 1024]).wait().unwrap(); - assert_eq!(amt, 2); - assert_eq!(&buf[..2], b"a\n"); - - let (client, _) = write_all(client, b"\n").wait().unwrap(); - let (client, buf, amt) = read(client, buf).wait().unwrap(); - assert_eq!(amt, 1); - assert_eq!(&buf[..1], b"\n"); - - let (client, _) = write_all(client, b"b").wait().unwrap(); - client.shutdown(Shutdown::Write).unwrap(); - let (_client, buf, amt) = read(client, buf).wait().unwrap(); - assert_eq!(amt, 1); - assert_eq!(&buf[..1], b"b"); -} diff --git a/tokio/tests/pipe-hup.rs b/tokio/tests/pipe-hup.rs deleted file mode 100644 index 4d78ace4c..000000000 --- a/tokio/tests/pipe-hup.rs +++ /dev/null @@ -1,100 +0,0 @@ -#![cfg(feature = "broken")] -#![cfg(unix)] -#![deny(warnings, rust_2018_idioms)] - -use env_logger; -use futures::Future; -use libc; -use mio; -use mio::event::Evented; -use mio::unix::{EventedFd, UnixReady}; -use mio::{PollOpt, Ready, Token}; -use std::fs::File; -use std::io::{self, Write}; -use std::os::unix::io::{AsRawFd, FromRawFd}; -use std::thread; -use std::time::Duration; -use tokio::reactor::{Handle, PollEvented2}; -use tokio_io::io::read_to_end; - -macro_rules! t { - ($e:expr) => { - match $e { - Ok(e) => e, - Err(e) => panic!("{} failed with {:?}", stringify!($e), e), - } - }; -} - -struct MyFile(File); - -impl MyFile { - fn new(file: File) -> MyFile { - unsafe { - let r = libc::fcntl(file.as_raw_fd(), libc::F_SETFL, libc::O_NONBLOCK); - assert!(r != -1, "fcntl error: {}", io::Error::last_os_error()); - } - MyFile(file) - } -} - -impl io::Read for MyFile { - fn read(&mut self, bytes: &mut [u8]) -> io::Result { - self.0.read(bytes) - } -} - -impl Evented for MyFile { - fn register( - &self, - poll: &mio::Poll, - token: Token, - interest: Ready, - opts: PollOpt, - ) -> io::Result<()> { - let hup: Ready = UnixReady::hup().into(); - EventedFd(&self.0.as_raw_fd()).register(poll, token, interest | hup, opts) - } - fn reregister( - &self, - poll: &mio::Poll, - token: Token, - interest: Ready, - opts: PollOpt, - ) -> io::Result<()> { - let hup: Ready = UnixReady::hup().into(); - EventedFd(&self.0.as_raw_fd()).reregister(poll, token, interest | hup, opts) - } - fn deregister(&self, poll: &mio::Poll) -> io::Result<()> { - EventedFd(&self.0.as_raw_fd()).deregister(poll) - } -} - -#[test] -fn hup() { - drop(env_logger::try_init()); - - let handle = Handle::default(); - unsafe { - let mut pipes = [0; 2]; - assert!( - libc::pipe(pipes.as_mut_ptr()) != -1, - "pipe error: {}", - io::Error::last_os_error() - ); - let read = File::from_raw_fd(pipes[0]); - let mut write = File::from_raw_fd(pipes[1]); - let t = thread::spawn(move || { - write.write_all(b"Hello!\n").unwrap(); - write.write_all(b"Good bye!\n").unwrap(); - thread::sleep(Duration::from_millis(100)); - }); - - let source = PollEvented2::new_with_handle(MyFile::new(read), &handle).unwrap(); - - let reader = read_to_end(source, Vec::new()); - let (_, content) = t!(reader.wait()); - assert_eq!(&b"Hello!\nGood bye!\n"[..], &content[..]); - t.join().unwrap(); - } -} diff --git a/tokio/tests/reactor.rs b/tokio/tests/reactor.rs index 5c45cac86..03868e976 100644 --- a/tokio/tests/reactor.rs +++ b/tokio/tests/reactor.rs @@ -1,15 +1,35 @@ -#![cfg(feature = "broken")] +#![feature(async_await)] #![deny(warnings, rust_2018_idioms)] +#![cfg(feature = "default")] -use futures::executor::{spawn, Notify, Spawn}; -use futures::{Future, Stream}; -use std::mem; -use std::net::TcpStream; -use std::sync::{Arc, Mutex}; -use tokio_executor; -use tokio_reactor; use tokio_reactor::Reactor; use tokio_tcp::TcpListener; +use tokio_test::{assert_ok, assert_pending}; + +use futures_util::task::ArcWake; +use std::future::Future; +use std::net::TcpStream; +use std::pin::Pin; +use std::sync::{mpsc, Arc, Mutex}; +use std::task::Context; + +struct Task { + future: Mutex>>, +} + +impl ArcWake for Task { + fn wake_by_ref(_: &Arc) { + // Do nothing... + } +} + +impl Task { + fn new(future: T) -> Task { + Task { + future: Mutex::new(Box::pin(future)), + } + } +} #[test] fn test_drop_on_notify() { @@ -26,60 +46,35 @@ fn test_drop_on_notify() { // shutting down. Then, when the task handle is dropped, the task itself is // dropped. - struct MyNotify; - - type Task = Mutex>>>; - - impl Notify for MyNotify { - fn notify(&self, _: usize) { - // Do nothing - } - - fn clone_id(&self, id: usize) -> usize { - let ptr = id as *const Task; - let task = unsafe { Arc::from_raw(ptr) }; - - mem::forget(task.clone()); - mem::forget(task); - - id - } - - fn drop_id(&self, id: usize) { - let ptr = id as *const Task; - let _ = unsafe { Arc::from_raw(ptr) }; - } - } - - let addr = "127.0.0.1:0".parse().unwrap(); - let mut reactor = Reactor::new().unwrap(); - - // Create a listener - let listener = TcpListener::bind(&addr).unwrap(); - let addr = listener.local_addr().unwrap(); + let mut reactor = assert_ok!(Reactor::new()); + let (addr_tx, addr_rx) = mpsc::channel(); // Define a task that just drains the listener - let task = Box::new({ - listener - .incoming() - .for_each(|_| Ok(())) - .map_err(|_| panic!()) - }) as Box>; + let task = Arc::new(Task::new(async move { + let addr = assert_ok!("127.0.0.1:0".parse()); + // Create a listener + let mut listener = assert_ok!(TcpListener::bind(&addr)); - let task = Arc::new(Mutex::new(spawn(task))); - let notify = Arc::new(MyNotify); + // Send the address + let addr = listener.local_addr().unwrap(); + addr_tx.send(addr).unwrap(); - let mut enter = tokio_executor::enter().unwrap(); + loop { + let _ = listener.accept().await; + } + })); - tokio_reactor::with_default(&reactor.handle(), &mut enter, |_| { - let id = &*task as *const Task as usize; + let _enter = tokio_executor::enter().unwrap(); - task.lock() - .unwrap() - .poll_future_notify(¬ify, id) - .unwrap(); + tokio_reactor::with_default(&reactor.handle(), || { + let waker = task.clone().into_waker(); + let mut cx = Context::from_waker(&waker); + assert_pending!(task.future.lock().unwrap().as_mut().poll(&mut cx)); }); + // Get the address + let addr = addr_rx.recv().unwrap(); + drop(task); // Establish a connection to the acceptor diff --git a/tokio/tests/runtime_current_thread.rs b/tokio/tests/runtime_current_thread.rs index d5852bf97..021bce974 100644 --- a/tokio/tests/runtime_current_thread.rs +++ b/tokio/tests/runtime_current_thread.rs @@ -1,5 +1,6 @@ #![deny(warnings, rust_2018_idioms)] #![feature(async_await)] +#![cfg(feature = "default")] use tokio::io::{AsyncReadExt, AsyncWriteExt}; use tokio::net::{TcpListener, TcpStream}; diff --git a/tokio/tests/runtime_threaded.rs b/tokio/tests/runtime_threaded.rs index 6d04db76c..537cc79b9 100644 --- a/tokio/tests/runtime_threaded.rs +++ b/tokio/tests/runtime_threaded.rs @@ -1,5 +1,6 @@ #![deny(warnings, rust_2018_idioms)] #![feature(async_await)] +#![cfg(feature = "default")] use tokio; use tokio::io::{AsyncReadExt, AsyncWriteExt}; diff --git a/tokio/tests/timer.rs b/tokio/tests/timer.rs index 63994d18e..553a083d5 100644 --- a/tokio/tests/timer.rs +++ b/tokio/tests/timer.rs @@ -1,9 +1,8 @@ #![deny(warnings, rust_2018_idioms)] #![feature(async_await)] +#![cfg(feature = "default")] -use tokio; use tokio::prelude::*; -// use tokio::sync::mpsc; use tokio::timer::*; use std::sync::mpsc;