eater/async-std
1
0
Fork 0
forked from mirror/async-std
Code Issues Pull requests Releases Wiki Activity

Merge pull request #719 from k-nasa/remove_re_export

Browse source 
Stabilize most stream method and remove unnecessary macros
This commit is contained in:
Yoshua Wuyts 2020-03-06 14:31:06 +01:00 committed by GitHub
commit 61f9483cc5
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
34 changed files with 110 additions and 1192 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
Cargo.toml
View file

@ -26,7 +26,6 @@ default = [
"async-task",
"crossbeam-channel",
"crossbeam-deque",
"futures-timer",
"kv-log-macro",
"log",
"mio",
@ -35,13 +34,14 @@ default = [
"pin-project-lite",
]
docs = ["attributes", "unstable", "default"]
unstable = ["std", "broadcaster", "futures-timer"]
unstable = ["std", "broadcaster"]
attributes = ["async-attributes"]
std = [
"alloc",
"crossbeam-utils",
"futures-core/std",
"futures-io",
"futures-timer",
"memchr",
"once_cell",
"pin-utils",

10
examples/a-chat/client.rs
View file

@ -1,11 +1,12 @@
use futures::select;
use futures::FutureExt;
use std::io::{self, BufRead, BufReader as StdBufReader};
use async_std::{
io::{stdin, BufReader},
io::BufReader,
net::{TcpStream, ToSocketAddrs},
prelude::*,
task,
stream, task,
};
type Result<T> = std::result::Result<T, Box<dyn std::error::Error + Send + Sync>>;
@ -20,8 +21,9 @@ async fn try_main(addr: impl ToSocketAddrs) -> Result<()> {
let reader = BufReader::new(reader);
let mut lines_from_server = futures::StreamExt::fuse(reader.lines());
let stdin = BufReader::new(stdin());
let mut lines_from_stdin = futures::StreamExt::fuse(stdin.lines());
let stdin = StdBufReader::new(io::stdin());
let mut lines_from_stdin = stream::from_iter(stdin.lines());
loop {
select! {
line = lines_from_server.next().fuse() => match line {

7
examples/print-file.rs
View file

@ -1,6 +1,7 @@
//! Prints a file given as an argument to stdout.
use std::env::args;
use std::io::Write;
use async_std::fs::File;
use async_std::io;
@ -14,7 +15,7 @@ fn main() -> io::Result<()> {
task::block_on(async {
let mut file = File::open(&path).await?;
let mut stdout = io::stdout();
let mut stdout = std::io::stdout();
let mut buf = vec![0u8; LEN];
loop {
@ -23,12 +24,12 @@ fn main() -> io::Result<()> {
// If this is the end of file, clean up and return.
if n == 0 {
stdout.flush().await?;
stdout.flush()?;
return Ok(());
}
// Write the buffer into stdout.
stdout.write_all(&buf[..n]).await?;
stdout.write_all(&buf[..n])?;
}
})
}

12
examples/stdin-echo.rs
View file

@ -1,18 +1,18 @@
//! Echoes lines read on stdin to stdout.
use async_std::io;
use async_std::prelude::*;
use async_std::task;
use std::io::Write;
fn main() -> io::Result<()> {
task::block_on(async {
let stdin = io::stdin();
let mut stdout = io::stdout();
let stdin = std::io::stdin();
let mut stdout = std::io::stdout();
let mut line = String::new();
loop {
// Read a line from stdin.
let n = stdin.read_line(&mut line).await?;
let n = stdin.read_line(&mut line)?;
// If this is the end of stdin, return.
if n == 0 {
@ -20,8 +20,8 @@ fn main() -> io::Result<()> {
}
// Write the line to stdout.
stdout.write_all(line.as_bytes()).await?;
stdout.flush().await?;
stdout.write_all(line.as_bytes())?;
stdout.flush()?;
line.clear();
}
})

4
examples/stdin-timeout.rs
View file

@ -8,11 +8,11 @@ use async_std::task;
fn main() -> io::Result<()> {
// This async scope times out after 5 seconds.
task::block_on(io::timeout(Duration::from_secs(5), async {
let stdin = io::stdin();
let stdin = std::io::stdin();
// Read a line from the standard input and display it.
let mut line = String::new();
stdin.read_line(&mut line).await?;
stdin.read_line(&mut line)?;
dbg!(line);
Ok(())

3
src/future/into_future.rs
View file

@ -5,9 +5,10 @@ use std::future::Future;
/// # Examples
///
/// ```
/// use std::pin::Pin;
///
/// use async_std::future::{Future, IntoFuture};
/// use async_std::io;
/// use async_std::pin::Pin;
///
/// struct Client;
///

2
src/future/mod.rs
View file

@ -63,12 +63,14 @@ cfg_std! {
cfg_default! {
pub use timeout::{timeout, TimeoutError};
mod timeout;
}
cfg_unstable! {
pub use into_future::IntoFuture;
pub(crate) use maybe_done::MaybeDone;
mod into_future;
mod maybe_done;
}

10
src/io/copy.rs
View file

@ -32,13 +32,14 @@ use crate::utils::Context as _;
///
/// # Examples
///
/// ```
/// ```no_run
/// # fn main() -> std::io::Result<()> { async_std::task::block_on(async {
/// #
/// use async_std::io;
/// use async_std::fs::File;
///
/// let mut reader: &[u8] = b"hello";
/// let mut writer = io::stdout();
/// let mut writer = File::open("foo.txt").await?;
///
/// io::copy(&mut reader, &mut writer).await?;
/// #
@ -119,13 +120,14 @@ where
///
/// # Examples
///
/// ```
/// ```no_run
/// # fn main() -> std::io::Result<()> { async_std::task::block_on(async {
/// #
/// use async_std::io;
/// use async_std::fs::File;
///
/// let mut reader: &[u8] = b"hello";
/// let mut writer = io::stdout();
/// let mut writer = File::open("foo.txt").await?;
///
/// io::copy(&mut reader, &mut writer).await?;
/// #

84
src/io/mod.rs
View file

@ -122,56 +122,6 @@
//! # Ok(()) }) }
//! ```
//!
//! ## Standard input and output
//!
//! A very common source of input is standard input:
//!
//! ```no_run
//! use async_std::io;
//!
//! # fn main() -> std::io::Result<()> { async_std::task::block_on(async {
//! #
//! let mut input = String::new();
//!
//! io::stdin().read_line(&mut input).await?;
//!
//! println!("You typed: {}", input.trim());
//! #
//! # Ok(()) }) }
//! ```
//!
//! Note that you cannot use the [`?` operator] in functions that do not return
//! a [`Result<T, E>`][`Result`]. Instead, you can call [`.unwrap()`]
//! or `match` on the return value to catch any possible errors:
//!
//! ```no_run
//! use async_std::io;
//!
//! # fn main() -> std::io::Result<()> { async_std::task::block_on(async {
//! #
//! let mut input = String::new();
//!
//! io::stdin().read_line(&mut input).await.unwrap();
//! #
//! # Ok(()) }) }
//! ```
//!
//! And a very common source of output is standard output:
//!
//! ```no_run
//! use async_std::io;
//! use async_std::io::prelude::*;
//!
//! # fn main() -> std::io::Result<()> { async_std::task::block_on(async {
//! #
//! io::stdout().write(&[42]).await?;
//! #
//! # Ok(()) }) }
//! ```
//!
//! Of course, using [`io::stdout`] directly is less common than something like
//! [`println!`].
//!
//! ## Iterator types
//!
//! A large number of the structures provided by `std::io` are for various
@ -204,10 +154,14 @@
//!
//! ```no_run
//! use async_std::io;
//! use async_std::fs::File;
//!
//! # fn main() -> std::io::Result<()> { async_std::task::block_on(async {
//! #
//! io::copy(&mut io::stdin(), &mut io::stdout()).await?;
//! let mut reader: &[u8] = b"hello";
//! let mut writer = File::open("foo.txt").await?;
//!
//! io::copy(&mut reader, &mut writer).await?;
//! #
//! # Ok(()) }) }
//! ```
@ -224,13 +178,14 @@
//! ```
//! #![allow(dead_code)]
//! use async_std::io;
//! use std::time::Duration;
//!
//! async fn read_input() -> io::Result<()> {
//! let mut input = String::new();
//! let f = io::timeout(Duration::from_secs(5), async {
//! Ok(())
//! });
//!
//! io::stdin().read_line(&mut input).await?;
//!
//! println!("You typed: {}", input.trim());
//! assert_eq!(f.await?, ());
//!
//! Ok(())
//! }
@ -260,8 +215,6 @@
//! [`BufReader`]: struct.BufReader.html
//! [`BufWriter`]: struct.BufWriter.html
//! [`Write::write`]: trait.Write.html#tymethod.write
//! [`io::stdout`]: fn.stdout.html
//! [`println!`]: ../macro.println.html
//! [`Lines`]: struct.Lines.html
//! [`io::Result`]: type.Result.html
//! [`?` operator]: https://doc.rust-lang.org/stable/book/appendix-02-operators.html
@ -305,24 +258,7 @@ cfg_std! {
}
cfg_default! {
// For use in the print macros.
#[doc(hidden)]
pub use stdio::{_eprint, _print};
pub use stderr::{stderr, Stderr};
pub use stdin::{stdin, Stdin};
pub use stdout::{stdout, Stdout};
pub use timeout::timeout;
mod timeout;
mod stderr;
mod stdin;
mod stdio;
mod stdout;
}
cfg_unstable_default! {
pub use stderr::StderrLock;
pub use stdin::StdinLock;
pub use stdout::StdoutLock;
}

261
src/io/stderr.rs
View file

@ -1,261 +0,0 @@
use std::pin::Pin;
use std::sync::Mutex;
use std::future::Future;
use crate::io::{self, Write};
use crate::task::{spawn_blocking, Context, JoinHandle, Poll};
cfg_unstable! {
use once_cell::sync::Lazy;
use std::io::Write as _;
}
/// Constructs a new handle to the standard error of the current process.
///
/// This function is an async version of [`std::io::stderr`].
///
/// [`std::io::stderr`]: https://doc.rust-lang.org/std/io/fn.stderr.html
///
/// ### Note: Windows Portability Consideration
///
/// When operating in a console, the Windows implementation of this stream does not support
/// non-UTF-8 byte sequences. Attempting to write bytes that are not valid UTF-8 will return
/// an error.
///
/// # Examples
///
/// ```no_run
/// # fn main() -> std::io::Result<()> { async_std::task::block_on(async {
/// #
/// use async_std::io;
/// use async_std::prelude::*;
///
/// let mut stderr = io::stderr();
/// stderr.write_all(b"Hello, world!").await?;
/// #
/// # Ok(()) }) }
/// ```
pub fn stderr() -> Stderr {
Stderr(Mutex::new(State::Idle(Some(Inner {
stderr: std::io::stderr(),
buf: Vec::new(),
last_op: None,
}))))
}
/// A handle to the standard error of the current process.
///
/// This writer is created by the [`stderr`] function. See its documentation for
/// more.
///
/// ### Note: Windows Portability Consideration
///
/// When operating in a console, the Windows implementation of this stream does not support
/// non-UTF-8 byte sequences. Attempting to write bytes that are not valid UTF-8 will return
/// an error.
///
/// [`stderr`]: fn.stderr.html
#[derive(Debug)]
pub struct Stderr(Mutex<State>);
/// A locked reference to the Stderr handle.
///
/// This handle implements the [`Write`] traits, and is constructed via the [`Stderr::lock`]
/// method.
///
/// [`Write`]: trait.Read.html
/// [`Stderr::lock`]: struct.Stderr.html#method.lock
#[cfg(feature = "unstable")]
#[cfg_attr(feature = "docs", doc(cfg(unstable)))]
#[derive(Debug)]
pub struct StderrLock<'a>(std::io::StderrLock<'a>);
#[cfg(feature = "unstable")]
#[cfg_attr(feature = "docs", doc(cfg(unstable)))]
unsafe impl Send for StderrLock<'_> {}
/// The state of the asynchronous stderr.
///
/// The stderr can be either idle or busy performing an asynchronous operation.
#[derive(Debug)]
enum State {
/// The stderr is idle.
Idle(Option<Inner>),
/// The stderr is blocked on an asynchronous operation.
///
/// Awaiting this operation will result in the new state of the stderr.
Busy(JoinHandle<State>),
}
/// Inner representation of the asynchronous stderr.
#[derive(Debug)]
struct Inner {
/// The blocking stderr handle.
stderr: std::io::Stderr,
/// The write buffer.
buf: Vec<u8>,
/// The result of the last asynchronous operation on the stderr.
last_op: Option<Operation>,
}
/// Possible results of an asynchronous operation on the stderr.
#[derive(Debug)]
enum Operation {
Write(io::Result<usize>),
Flush(io::Result<()>),
}
impl Stderr {
/// Locks this handle to the standard error stream, returning a writable guard.
///
/// The lock is released when the returned lock goes out of scope. The returned guard also implements the Write trait for writing data.
///
/// # Examples
///
/// ```no_run
/// # fn main() -> std::io::Result<()> { async_std::task::block_on(async {
/// #
/// use async_std::io;
/// use async_std::prelude::*;
///
/// let stderr = io::stderr();
/// let mut handle = stderr.lock().await;
///
/// handle.write_all(b"hello world").await?;
/// #
/// # Ok(()) }) }
/// ```
#[cfg_attr(feature = "docs", doc(cfg(unstable)))]
#[cfg(any(feature = "unstable", feature = "docs"))]
pub async fn lock(&self) -> StderrLock<'static> {
static STDERR: Lazy<std::io::Stderr> = Lazy::new(std::io::stderr);
spawn_blocking(move || StderrLock(STDERR.lock())).await
}
}
impl Write for Stderr {
fn poll_write(
mut self: Pin<&mut Self>,
cx: &mut Context<'_>,
buf: &[u8],
) -> Poll<io::Result<usize>> {
let state = &mut *self.0.lock().unwrap();
loop {
match state {
State::Idle(opt) => {
let inner = opt.as_mut().unwrap();
// Check if the operation has completed.
if let Some(Operation::Write(res)) = inner.last_op.take() {
let n = res?;
// If more data was written than is available in the buffer, let's retry
// the write operation.
if n <= buf.len() {
return Poll::Ready(Ok(n));
}
} else {
let mut inner = opt.take().unwrap();
// Set the length of the inner buffer to the length of the provided buffer.
if inner.buf.len() < buf.len() {
inner.buf.reserve(buf.len() - inner.buf.len());
}
unsafe {
inner.buf.set_len(buf.len());
}
// Copy the data to write into the inner buffer.
inner.buf[..buf.len()].copy_from_slice(buf);
// Start the operation asynchronously.
*state = State::Busy(spawn_blocking(move || {
let res = std::io::Write::write(&mut inner.stderr, &inner.buf);
inner.last_op = Some(Operation::Write(res));
State::Idle(Some(inner))
}));
}
}
// Poll the asynchronous operation the stderr is currently blocked on.
State::Busy(task) => *state = futures_core::ready!(Pin::new(task).poll(cx)),
}
}
}
fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
let state = &mut *self.0.lock().unwrap();
loop {
match state {
State::Idle(opt) => {
let inner = opt.as_mut().unwrap();
// Check if the operation has completed.
if let Some(Operation::Flush(res)) = inner.last_op.take() {
return Poll::Ready(res);
} else {
let mut inner = opt.take().unwrap();
// Start the operation asynchronously.
*state = State::Busy(spawn_blocking(move || {
let res = std::io::Write::flush(&mut inner.stderr);
inner.last_op = Some(Operation::Flush(res));
State::Idle(Some(inner))
}));
}
}
// Poll the asynchronous operation the stderr is currently blocked on.
State::Busy(task) => *state = futures_core::ready!(Pin::new(task).poll(cx)),
}
}
}
fn poll_close(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
self.poll_flush(cx)
}
}
cfg_unix! {
use crate::os::unix::io::{AsRawFd, RawFd};
impl AsRawFd for Stderr {
fn as_raw_fd(&self) -> RawFd {
std::io::stderr().as_raw_fd()
}
}
}
cfg_windows! {
use crate::os::windows::io::{AsRawHandle, RawHandle};
impl AsRawHandle for Stderr {
fn as_raw_handle(&self) -> RawHandle {
std::io::stderr().as_raw_handle()
}
}
}
#[cfg(feature = "unstable")]
#[cfg_attr(feature = "docs", doc(cfg(unstable)))]
impl io::Write for StderrLock<'_> {
fn poll_write(
mut self: Pin<&mut Self>,
_cx: &mut Context<'_>,
buf: &[u8],
) -> Poll<io::Result<usize>> {
Poll::Ready(self.0.write(buf))
}
fn poll_flush(mut self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<io::Result<()>> {
Poll::Ready(self.0.flush())
}
fn poll_close(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
self.poll_flush(cx)
}
}

279
src/io/stdin.rs
View file

@ -1,279 +0,0 @@
use std::future::Future;
use std::pin::Pin;
use std::sync::Mutex;
use crate::future;
use crate::io::{self, Read};
use crate::task::{spawn_blocking, Context, JoinHandle, Poll};
use crate::utils::Context as _;
cfg_unstable! {
use once_cell::sync::Lazy;
use std::io::Read as _;
}
/// Constructs a new handle to the standard input of the current process.
///
/// This function is an async version of [`std::io::stdin`].
///
/// [`std::io::stdin`]: https://doc.rust-lang.org/std/io/fn.stdin.html
///
/// ### Note: Windows Portability Consideration
///
/// When operating in a console, the Windows implementation of this stream does not support
/// non-UTF-8 byte sequences. Attempting to write bytes that are not valid UTF-8 will return
/// an error.
///
/// # Examples
///
/// ```no_run
/// # fn main() -> std::io::Result<()> { async_std::task::block_on(async {
/// #
/// use async_std::io;
///
/// let stdin = io::stdin();
/// let mut line = String::new();
/// stdin.read_line(&mut line).await?;
/// #
/// # Ok(()) }) }
/// ```
pub fn stdin() -> Stdin {
Stdin(Mutex::new(State::Idle(Some(Inner {
stdin: std::io::stdin(),
line: String::new(),
buf: Vec::new(),
last_op: None,
}))))
}
/// A handle to the standard input of the current process.
///
/// This reader is created by the [`stdin`] function. See its documentation for
/// more.
///
/// ### Note: Windows Portability Consideration
///
/// When operating in a console, the Windows implementation of this stream does not support
/// non-UTF-8 byte sequences. Attempting to write bytes that are not valid UTF-8 will return
/// an error.
///
/// [`stdin`]: fn.stdin.html
#[derive(Debug)]
pub struct Stdin(Mutex<State>);
/// A locked reference to the Stdin handle.
///
/// This handle implements the [`Read`] traits, and is constructed via the [`Stdin::lock`] method.
///
/// [`Read`]: trait.Read.html
/// [`Stdin::lock`]: struct.Stdin.html#method.lock
#[cfg_attr(feature = "docs", doc(cfg(unstable)))]
#[cfg(feature = "unstable")]
#[derive(Debug)]
pub struct StdinLock<'a>(std::io::StdinLock<'a>);
#[cfg(feature = "unstable")]
#[cfg_attr(feature = "docs", doc(cfg(unstable)))]
unsafe impl Send for StdinLock<'_> {}
/// The state of the asynchronous stdin.
///
/// The stdin can be either idle or busy performing an asynchronous operation.
#[derive(Debug)]
enum State {
/// The stdin is idle.
Idle(Option<Inner>),
/// The stdin is blocked on an asynchronous operation.
///
/// Awaiting this operation will result in the new state of the stdin.
Busy(JoinHandle<State>),
}
/// Inner representation of the asynchronous stdin.
#[derive(Debug)]
struct Inner {
/// The blocking stdin handle.
stdin: std::io::Stdin,
/// The line buffer.
line: String,
/// The write buffer.
buf: Vec<u8>,
/// The result of the last asynchronous operation on the stdin.
last_op: Option<Operation>,
}
/// Possible results of an asynchronous operation on the stdin.
#[derive(Debug)]
enum Operation {
ReadLine(io::Result<usize>),
Read(io::Result<usize>),
}
impl Stdin {
/// Reads a line of input into the specified buffer.
///
/// # Examples
///
/// ```no_run
/// # fn main() -> std::io::Result<()> { async_std::task::block_on(async {
/// #
/// use async_std::io;
///
/// let stdin = io::stdin();
/// let mut line = String::new();
/// stdin.read_line(&mut line).await?;
/// #
/// # Ok(()) }) }
/// ```
pub async fn read_line(&self, buf: &mut String) -> io::Result<usize> {
future::poll_fn(|cx| {
let state = &mut *self.0.lock().unwrap();
loop {
match state {
State::Idle(opt) => {
let inner = opt.as_mut().unwrap();
// Check if the operation has completed.
if let Some(Operation::ReadLine(res)) = inner.last_op.take() {
let n = res?;
// Copy the read data into the buffer and return.
buf.push_str(&inner.line);
return Poll::Ready(Ok(n));
} else {
let mut inner = opt.take().unwrap();
// Start the operation asynchronously.
*state = State::Busy(spawn_blocking(move || {
inner.line.clear();
let res = inner.stdin.read_line(&mut inner.line);
inner.last_op = Some(Operation::ReadLine(res));
State::Idle(Some(inner))
}));
}
}
// Poll the asynchronous operation the stdin is currently blocked on.
State::Busy(task) => *state = futures_core::ready!(Pin::new(task).poll(cx)),
}
}
})
.await
.context(|| String::from("could not read line on stdin"))
}
/// Locks this handle to the standard input stream, returning a readable guard.
///
/// The lock is released when the returned lock goes out of scope. The returned guard also implements the Read trait for accessing the underlying data.
///
/// # Examples
///
/// ```no_run
/// # fn main() -> std::io::Result<()> { async_std::task::block_on(async {
/// #
/// use async_std::io;
/// use async_std::prelude::*;
///
/// let mut buffer = String::new();
///
/// let stdin = io::stdin();
/// let mut handle = stdin.lock().await;
///
/// handle.read_to_string(&mut buffer).await?;
/// #
/// # Ok(()) }) }
/// ```
#[cfg_attr(feature = "docs", doc(cfg(unstable)))]
#[cfg(any(feature = "unstable", feature = "docs"))]
pub async fn lock(&self) -> StdinLock<'static> {
static STDIN: Lazy<std::io::Stdin> = Lazy::new(std::io::stdin);
spawn_blocking(move || StdinLock(STDIN.lock())).await
}
}
impl Read for Stdin {
fn poll_read(
mut self: Pin<&mut Self>,
cx: &mut Context<'_>,
buf: &mut [u8],
) -> Poll<io::Result<usize>> {
let state = &mut *self.0.lock().unwrap();
loop {
match state {
State::Idle(opt) => {
let inner = opt.as_mut().unwrap();
// Check if the operation has completed.
if let Some(Operation::Read(res)) = inner.last_op.take() {
let n = res?;
// If more data was read than fits into the buffer, let's retry the read
// operation.
if n <= buf.len() {
// Copy the read data into the buffer and return.
buf[..n].copy_from_slice(&inner.buf[..n]);
return Poll::Ready(Ok(n));
}
} else {
let mut inner = opt.take().unwrap();
// Set the length of the inner buffer to the length of the provided buffer.
if inner.buf.len() < buf.len() {
inner.buf.reserve(buf.len() - inner.buf.len());
}
unsafe {
inner.buf.set_len(buf.len());
}
// Start the operation asynchronously.
*state = State::Busy(spawn_blocking(move || {
let res = std::io::Read::read(&mut inner.stdin, &mut inner.buf);
inner.last_op = Some(Operation::Read(res));
State::Idle(Some(inner))
}));
}
}
// Poll the asynchronous operation the stdin is currently blocked on.
State::Busy(task) => *state = futures_core::ready!(Pin::new(task).poll(cx)),
}
}
}
}
cfg_unix! {
use crate::os::unix::io::{AsRawFd, RawFd};
impl AsRawFd for Stdin {
fn as_raw_fd(&self) -> RawFd {
std::io::stdin().as_raw_fd()
}
}
}
cfg_windows! {
use crate::os::windows::io::{AsRawHandle, RawHandle};
impl AsRawHandle for Stdin {
fn as_raw_handle(&self) -> RawHandle {
std::io::stdin().as_raw_handle()
}
}
}
#[cfg(feature = "unstable")]
#[cfg_attr(feature = "docs", doc(cfg(unstable)))]
impl Read for StdinLock<'_> {
fn poll_read(
mut self: Pin<&mut Self>,
_cx: &mut Context<'_>,
buf: &mut [u8],
) -> Poll<io::Result<usize>> {
Poll::Ready(self.0.read(buf))
}
}

21
src/io/stdio.rs
View file

@ -1,21 +0,0 @@
//! Internal types for stdio.
//!
//! This module is a port of `libstd/io/stdio.rs`,and contains internal types for `print`/`eprint`.
use crate::io::{stderr, stdout};
use crate::prelude::*;
use std::fmt;
#[doc(hidden)]
pub async fn _print(args: fmt::Arguments<'_>) {
if let Err(e) = stdout().write_fmt(args).await {
panic!("failed printing to stdout: {}", e);
}
}
#[doc(hidden)]
pub async fn _eprint(args: fmt::Arguments<'_>) {
if let Err(e) = stderr().write_fmt(args).await {
panic!("failed printing to stderr: {}", e);
}
}

261
src/io/stdout.rs
View file

@ -1,261 +0,0 @@
use std::pin::Pin;
use std::sync::Mutex;
use std::future::Future;
use crate::io::{self, Write};
use crate::task::{spawn_blocking, Context, JoinHandle, Poll};
cfg_unstable! {
use once_cell::sync::Lazy;
use std::io::Write as _;
}
/// Constructs a new handle to the standard output of the current process.
///
/// This function is an async version of [`std::io::stdout`].
///
/// [`std::io::stdout`]: https://doc.rust-lang.org/std/io/fn.stdout.html
///
/// ### Note: Windows Portability Consideration
///
/// When operating in a console, the Windows implementation of this stream does not support
/// non-UTF-8 byte sequences. Attempting to write bytes that are not valid UTF-8 will return
/// an error.
///
/// # Examples
///
/// ```no_run
/// # fn main() -> std::io::Result<()> { async_std::task::block_on(async {
/// #
/// use async_std::io;
/// use async_std::prelude::*;
///
/// let mut stdout = io::stdout();
/// stdout.write_all(b"Hello, world!").await?;
/// #
/// # Ok(()) }) }
/// ```
pub fn stdout() -> Stdout {
Stdout(Mutex::new(State::Idle(Some(Inner {
stdout: std::io::stdout(),
buf: Vec::new(),
last_op: None,
}))))
}
/// A handle to the standard output of the current process.
///
/// This writer is created by the [`stdout`] function. See its documentation
/// for more.
///
/// ### Note: Windows Portability Consideration
///
/// When operating in a console, the Windows implementation of this stream does not support
/// non-UTF-8 byte sequences. Attempting to write bytes that are not valid UTF-8 will return
/// an error.
///
/// [`stdout`]: fn.stdout.html
#[derive(Debug)]
pub struct Stdout(Mutex<State>);
/// A locked reference to the Stderr handle.
///
/// This handle implements the [`Write`] traits, and is constructed via the [`Stdout::lock`]
/// method.
///
/// [`Write`]: trait.Read.html
/// [`Stdout::lock`]: struct.Stdout.html#method.lock
#[cfg(feature = "unstable")]
#[cfg_attr(feature = "docs", doc(cfg(unstable)))]
#[derive(Debug)]
pub struct StdoutLock<'a>(std::io::StdoutLock<'a>);
#[cfg(feature = "unstable")]
#[cfg_attr(feature = "docs", doc(cfg(unstable)))]
unsafe impl Send for StdoutLock<'_> {}
/// The state of the asynchronous stdout.
///
/// The stdout can be either idle or busy performing an asynchronous operation.
#[derive(Debug)]
enum State {
/// The stdout is idle.
Idle(Option<Inner>),
/// The stdout is blocked on an asynchronous operation.
///
/// Awaiting this operation will result in the new state of the stdout.
Busy(JoinHandle<State>),
}
/// Inner representation of the asynchronous stdout.
#[derive(Debug)]
struct Inner {
/// The blocking stdout handle.
stdout: std::io::Stdout,
/// The write buffer.
buf: Vec<u8>,
/// The result of the last asynchronous operation on the stdout.
last_op: Option<Operation>,
}
/// Possible results of an asynchronous operation on the stdout.
#[derive(Debug)]
enum Operation {
Write(io::Result<usize>),
Flush(io::Result<()>),
}
impl Stdout {
/// Locks this handle to the standard error stream, returning a writable guard.
///
/// The lock is released when the returned lock goes out of scope. The returned guard also implements the Write trait for writing data.
///
/// # Examples
///
/// ```no_run
/// # fn main() -> std::io::Result<()> { async_std::task::block_on(async {
/// #
/// use async_std::io;
/// use async_std::prelude::*;
///
/// let stdout = io::stdout();
/// let mut handle = stdout.lock().await;
///
/// handle.write_all(b"hello world").await?;
/// #
/// # Ok(()) }) }
/// ```
#[cfg_attr(feature = "docs", doc(cfg(unstable)))]
#[cfg(any(feature = "unstable", feature = "docs"))]
pub async fn lock(&self) -> StdoutLock<'static> {
static STDOUT: Lazy<std::io::Stdout> = Lazy::new(std::io::stdout);
spawn_blocking(move || StdoutLock(STDOUT.lock())).await
}
}
impl Write for Stdout {
fn poll_write(
mut self: Pin<&mut Self>,
cx: &mut Context<'_>,
buf: &[u8],
) -> Poll<io::Result<usize>> {
let state = &mut *self.0.lock().unwrap();
loop {
match state {
State::Idle(opt) => {
let inner = opt.as_mut().unwrap();
// Check if the operation has completed.
if let Some(Operation::Write(res)) = inner.last_op.take() {
let n = res?;
// If more data was written than is available in the buffer, let's retry
// the write operation.
if n <= buf.len() {
return Poll::Ready(Ok(n));
}
} else {
let mut inner = opt.take().unwrap();
// Set the length of the inner buffer to the length of the provided buffer.
if inner.buf.len() < buf.len() {
inner.buf.reserve(buf.len() - inner.buf.len());
}
unsafe {
inner.buf.set_len(buf.len());
}
// Copy the data to write into the inner buffer.
inner.buf[..buf.len()].copy_from_slice(buf);
// Start the operation asynchronously.
*state = State::Busy(spawn_blocking(move || {
let res = std::io::Write::write(&mut inner.stdout, &inner.buf);
inner.last_op = Some(Operation::Write(res));
State::Idle(Some(inner))
}));
}
}
// Poll the asynchronous operation the stdout is currently blocked on.
State::Busy(task) => *state = futures_core::ready!(Pin::new(task).poll(cx)),
}
}
}
fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
let state = &mut *self.0.lock().unwrap();
loop {
match state {
State::Idle(opt) => {
let inner = opt.as_mut().unwrap();
// Check if the operation has completed.
if let Some(Operation::Flush(res)) = inner.last_op.take() {
return Poll::Ready(res);
} else {
let mut inner = opt.take().unwrap();
// Start the operation asynchronously.
*state = State::Busy(spawn_blocking(move || {
let res = std::io::Write::flush(&mut inner.stdout);
inner.last_op = Some(Operation::Flush(res));
State::Idle(Some(inner))
}));
}
}
// Poll the asynchronous operation the stdout is currently blocked on.
State::Busy(task) => *state = futures_core::ready!(Pin::new(task).poll(cx)),
}
}
}
fn poll_close(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
self.poll_flush(cx)
}
}
cfg_unix! {
use crate::os::unix::io::{AsRawFd, RawFd};
impl AsRawFd for Stdout {
fn as_raw_fd(&self) -> RawFd {
std::io::stdout().as_raw_fd()
}
}
}
cfg_windows! {
use crate::os::windows::io::{AsRawHandle, RawHandle};
impl AsRawHandle for Stdout {
fn as_raw_handle(&self) -> RawHandle {
std::io::stdout().as_raw_handle()
}
}
}
#[cfg(feature = "unstable")]
#[cfg_attr(feature = "docs", doc(cfg(unstable)))]
impl Write for StdoutLock<'_> {
fn poll_write(
mut self: Pin<&mut Self>,
_cx: &mut Context<'_>,
buf: &[u8],
) -> Poll<io::Result<usize>> {
Poll::Ready(self.0.write(buf))
}
fn poll_flush(mut self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<io::Result<()>> {
Poll::Ready(self.0.flush())
}
fn poll_close(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
self.poll_flush(cx)
}
}

4
src/io/timeout.rs
View file

@ -23,9 +23,9 @@ use crate::io;
/// use async_std::io;
///
/// io::timeout(Duration::from_secs(5), async {
/// let stdin = io::stdin();
/// let stdin = std::io::stdin();
/// let mut line = String::new();
/// let n = stdin.read_line(&mut line).await?;
/// let n = stdin.read_line(&mut line)?;
/// Ok(())
/// })
/// .await?;

3
src/lib.rs
View file

@ -273,9 +273,6 @@ cfg_default! {
}
cfg_unstable! {
pub mod pin;
pub mod process;
mod unit;
mod vec;
mod result;

168
src/macros.rs
View file

@ -1,171 +1,3 @@
/// Prints to the standard output.
///
/// Equivalent to the [`println!`] macro except that a newline is not printed at
/// the end of the message.
///
/// Note that stdout is frequently line-buffered by default so it may be
/// necessary to use [`io::stdout().flush()`][flush] to ensure the output is emitted
/// immediately.
///
/// Use `print!` only for the primary output of your program. Use
/// [`eprint!`] instead to print error and progress messages.
///
/// [`println!`]: macro.println.html
/// [flush]: io/trait.Write.html#tymethod.flush
/// [`eprint!`]: macro.eprint.html
///
/// # Panics
///
/// Panics if writing to `io::stdout()` fails.
///
/// # Examples
///
/// ```
/// # async_std::task::block_on(async {
/// #
/// use async_std::io;
/// use async_std::prelude::*;
/// use async_std::print;
///
/// print!("this ").await;
/// print!("will ").await;
/// print!("be ").await;
/// print!("on ").await;
/// print!("the ").await;
/// print!("same ").await;
/// print!("line ").await;
///
/// io::stdout().flush().await.unwrap();
///
/// print!("this string has a newline, why not choose println! instead?\n").await;
///
/// io::stdout().flush().await.unwrap();
/// #
/// # })
/// ```
#[cfg(feature = "unstable")]
#[cfg_attr(feature = "docs", doc(cfg(unstable)))]
#[macro_export]
macro_rules! print {
($($arg:tt)*) => ($crate::io::_print(format_args!($($arg)*)))
}
/// Prints to the standard output, with a newline.
///
/// On all platforms, the newline is the LINE FEED character (`\n`/`U+000A`) alone
/// (no additional CARRIAGE RETURN (`\r`/`U+000D`)).
///
/// Use the [`format!`] syntax to write data to the standard output.
/// See [`std::fmt`] for more information.
///
/// Use `println!` only for the primary output of your program. Use
/// [`eprintln!`] instead to print error and progress messages.
///
/// [`format!`]: macro.format.html
/// [`std::fmt`]: https://doc.rust-lang.org/std/fmt/index.html
/// [`eprintln!`]: macro.eprintln.html
/// # Panics
///
/// Panics if writing to `io::stdout` fails.
///
/// # Examples
///
/// ```
/// # async_std::task::block_on(async {
/// #
/// use async_std::println;
///
/// println!().await; // prints just a newline
/// println!("hello there!").await;
/// println!("format {} arguments", "some").await;
/// #
/// # })
/// ```
#[cfg(feature = "unstable")]
#[cfg_attr(feature = "docs", doc(cfg(unstable)))]
#[macro_export]
macro_rules! println {
() => ($crate::print!("\n"));
($($arg:tt)*) => (async {
$crate::io::_print(format_args!($($arg)*)).await;
$crate::io::_print(format_args!("\n")).await;
})
}
/// Prints to the standard error.
///
/// Equivalent to the [`print!`] macro, except that output goes to
/// [`io::stderr`] instead of `io::stdout`. See [`print!`] for
/// example usage.
///
/// Use `eprint!` only for error and progress messages. Use `print!`
/// instead for the primary output of your program.
///
/// [`io::stderr`]: io/struct.Stderr.html
/// [`print!`]: macro.print.html
///
/// # Panics
///
/// Panics if writing to `io::stderr` fails.
///
/// # Examples
///
/// ```
/// # async_std::task::block_on(async {
/// #
/// use async_std::eprint;
///
/// eprint!("Error: Could not complete task").await;
/// #
/// # })
/// ```
#[cfg(feature = "unstable")]
#[cfg_attr(feature = "docs", doc(cfg(unstable)))]
#[macro_export]
macro_rules! eprint {
($($arg:tt)*) => ($crate::io::_eprint(format_args!($($arg)*)))
}
/// Prints to the standard error, with a newline.
///
/// Equivalent to the [`println!`] macro, except that output goes to
/// [`io::stderr`] instead of `io::stdout`. See [`println!`] for
/// example usage.
///
/// Use `eprintln!` only for error and progress messages. Use `println!`
/// instead for the primary output of your program.
///
/// [`io::stderr`]: io/struct.Stderr.html
/// [`println!`]: macro.println.html
///
/// # Panics
///
/// Panics if writing to `io::stderr` fails.
///
/// # Examples
///
/// ```
/// # async_std::task::block_on(async {
/// #
/// use async_std::eprintln;
///
/// eprintln!("Error: Could not complete task").await;
/// #
/// # })
/// ```
#[cfg(feature = "unstable")]
#[cfg_attr(feature = "docs", doc(cfg(unstable)))]
#[macro_export]
macro_rules! eprintln {
() => (async { $crate::eprint!("\n").await; });
($($arg:tt)*) => (
async {
$crate::io::_eprint(format_args!($($arg)*)).await;
$crate::io::_eprint(format_args!("\n")).await;
}
);
}
/// Declares task-local values.
///
/// The macro wraps any number of static declarations and makes them task-local. Attributes and

12
src/prelude.rs
View file

@ -38,16 +38,14 @@ cfg_std! {
pub use crate::io::prelude::SeekExt as _;
#[doc(no_inline)]
pub use crate::io::prelude::WriteExt as _;
#[doc(no_inline)]
pub use crate::stream::DoubleEndedStream;
#[doc(no_inline)]
pub use crate::stream::ExactSizeStream;
}
cfg_default! {
#[doc(no_inline)]
pub use crate::task_local;
}
cfg_unstable! {
#[doc(no_inline)]
pub use crate::stream::DoubleEndedStream;
#[doc(no_inline)]
pub use crate::stream::ExactSizeStream;
}

6
src/stream/double_ended_stream/mod.rs
View file

@ -1,7 +1,7 @@
use crate::stream::Stream;
use std::pin::Pin;
use std::task::{Context, Poll};
use core::pin::Pin;
use core::task::{Context, Poll};
mod next_back;
mod nth_back;
@ -22,8 +22,6 @@ use try_rfold::TryRFoldFuture;
/// `Item`s from the back, as well as the front.
///
/// [`Stream`]: trait.Stream.html
#[cfg(feature = "unstable")]
#[cfg_attr(feature = "docs", doc(cfg(unstable)))]
pub trait DoubleEndedStream: Stream {
#[doc = r#"
Attempts to receive the next item from the back of the stream.

2
src/stream/exact_size_stream.rs
View file

@ -76,8 +76,6 @@ pub use crate::stream::Stream;
/// # });
/// ```
#[allow(clippy::len_without_is_empty)] // ExactSizeIterator::is_empty is unstable
#[cfg(feature = "unstable")]
#[cfg_attr(feature = "docs", doc(cfg(unstable)))]
pub trait ExactSizeStream: Stream {
/// Returns the exact number of times the stream will iterate.
///

2
src/stream/extend.rs
View file

@ -27,8 +27,6 @@ use crate::stream::IntoStream;
/// #
/// # })
/// ```
#[cfg(feature = "unstable")]
#[cfg_attr(feature = "docs", doc(cfg(unstable)))]
pub trait Extend<A> {
/// Extends a collection with the contents of a stream.
fn extend<'a, T: IntoStream<Item = A> + 'a>(

2
src/stream/fused_stream.rs
View file

@ -14,8 +14,6 @@ use crate::stream::Stream;
/// [`None`]: https://doc.rust-lang.org/std/option/enum.Option.html#variant.None
/// [`Stream::fuse`]: trait.Stream.html#method.fuse
/// [`Fuse`]: struct.Fuse.html
#[cfg(feature = "unstable")]
#[cfg_attr(feature = "docs", doc(cfg(unstable)))]
pub trait FusedStream: Stream {}
impl<S: FusedStream + ?Sized + Unpin> FusedStream for &mut S {}

4
src/stream/interval.rs
View file

@ -41,8 +41,6 @@ use futures_timer::Delay;
/// #
/// # Ok(()) }) }
/// ```
#[cfg(feature = "unstable")]
#[cfg_attr(feature = "docs", doc(cfg(unstable)))]
pub fn interval(dur: Duration) -> Interval {
Interval {
delay: Delay::new(dur),
@ -56,8 +54,6 @@ pub fn interval(dur: Duration) -> Interval {
/// documentation for more.
///
/// [`interval`]: fn.interval.html
#[cfg(feature = "unstable")]
#[cfg_attr(feature = "docs", doc(cfg(unstable)))]
#[derive(Debug)]
pub struct Interval {
delay: Delay,

29
src/stream/mod.rs
View file

@ -317,29 +317,32 @@ mod once;
mod repeat;
mod repeat_with;
cfg_unstable! {
cfg_std! {
pub use double_ended_stream::DoubleEndedStream;
pub use exact_size_stream::ExactSizeStream;
pub use fused_stream::FusedStream;
pub use interval::{interval, Interval};
pub use pending::{pending, Pending};
pub use product::Product;
pub use successors::{successors, Successors};
pub use sum::Sum;
mod double_ended_stream;
mod exact_size_stream;
mod extend;
mod from_stream;
mod fused_stream;
mod interval;
mod into_stream;
mod pending;
mod product;
mod successors;
mod sum;
}
cfg_unstable! {
mod from_stream;
mod into_stream;
mod extend;
pub use double_ended_stream::DoubleEndedStream;
pub use exact_size_stream::ExactSizeStream;
pub use extend::{extend, Extend};
pub use from_stream::FromStream;
pub use fused_stream::FusedStream;
pub use interval::{interval, Interval};
pub use into_stream::IntoStream;
pub use pending::{pending, Pending};
pub use product::Product;
pub use stream::Merge;
pub use successors::{successors, Successors};
pub use sum::Sum;
}

6
src/stream/once.rs
View file

@ -5,7 +5,7 @@ use pin_project_lite::pin_project;
use crate::stream::Stream;
use crate::task::{Context, Poll};
#[cfg(feature = "unstable")]
#[cfg(feature = "std")]
use crate::stream::DoubleEndedStream;
/// Creates a stream that yields a single item.
@ -50,8 +50,8 @@ impl<T> Stream for Once<T> {
}
}
#[cfg(feature = "unstable")]
impl <T> DoubleEndedStream for Once<T> {
#[cfg(feature = "std")]
impl<T> DoubleEndedStream for Once<T> {
fn poll_next_back(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
Poll::Ready(self.project().value.take())
}

13
src/stream/product.rs
View file

@ -1,5 +1,6 @@
use core::pin::Pin;
use alloc::boxed::Box;
use core::future::Future;
use core::pin::Pin;
use crate::stream::Stream;
@ -13,8 +14,6 @@ use crate::stream::Stream;
/// [`product`]: trait.Product.html#tymethod.product
/// [`FromStream`]: trait.FromStream.html
/// [`Stream::product`]: trait.Stream.html#method.product
#[cfg(feature = "unstable")]
#[cfg_attr(feature = "docs", doc(cfg(unstable)))]
pub trait Product<A = Self>: Sized {
/// Method which takes a stream and generates `Self` from the elements by
/// multiplying the items.
@ -23,9 +22,9 @@ pub trait Product<A = Self>: Sized {
S: Stream<Item = A> + 'a;
}
use core::ops::Mul;
use core::num::Wrapping;
use crate::stream::stream::StreamExt;
use core::num::Wrapping;
use core::ops::Mul;
macro_rules! integer_product {
(@impls $one: expr, $($a:ty)*) => ($(
@ -75,5 +74,5 @@ macro_rules! float_product {
);
}
integer_product!{ i8 i16 i32 i64 i128 isize u8 u16 u32 u64 u128 usize }
float_product!{ f32 f64 }
integer_product! { i8 i16 i32 i64 i128 isize u8 u16 u32 u64 u128 usize }
float_product! { f32 f64 }

2
src/stream/stream/count.rs
View file

@ -9,8 +9,6 @@ use crate::task::{Context, Poll};
pin_project! {
#[doc(hidden)]
#[allow(missing_debug_implementations)]
#[cfg(feature = "unstable")]
#[cfg_attr(feature = "docs", doc(cfg(unstable)))]
pub struct CountFuture<S> {
#[pin]
stream: S,

2
src/stream/stream/merge.rs
View file

@ -16,8 +16,6 @@ pin_project! {
///
/// [`merge`]: trait.Stream.html#method.merge
/// [`Stream`]: trait.Stream.html
#[cfg(feature = "unstable")]
#[cfg_attr(feature = "docs", doc(cfg(unstable)))]
#[derive(Debug)]
pub struct Merge<L, R> {
#[pin]

64
src/stream/stream/mod.rs
View file

@ -112,35 +112,43 @@ pub use zip::Zip;
use core::cmp::Ordering;
cfg_unstable! {
cfg_std! {
use core::time::Duration;
use crate::stream::{Product, Sum};
use alloc::boxed::Box;
use core::future::Future;
use core::pin::Pin;
use core::time::Duration;
use unzip::UnzipFuture;
use count::CountFuture;
pub use throttle::Throttle;
pub use merge::Merge;
pub use delay::Delay;
pub use timeout::{Timeout, TimeoutError};
mod timeout;
mod throttle;
mod merge;
mod delay;
mod unzip;
mod count;
}
cfg_unstable! {
use crate::stream::FromStream;
use crate::stream::into_stream::IntoStream;
use crate::stream::{FromStream, Product, Sum};
use crate::stream::Extend;
use count::CountFuture;
use partition::PartitionFuture;
use unzip::UnzipFuture;
pub use merge::Merge;
pub use flatten::Flatten;
pub use flat_map::FlatMap;
pub use timeout::{TimeoutError, Timeout};
pub use throttle::Throttle;
pub use delay::Delay;
mod count;
mod merge;
mod flatten;
mod flat_map;
mod partition;
mod timeout;
mod throttle;
mod delay;
mod unzip;
}
extension_trait! {
@ -355,8 +363,7 @@ extension_trait! {
# }) }
```
"#]
#[cfg(feature = "unstable")]
#[cfg_attr(feature = "docs", doc(cfg(unstable)))]
#[cfg(feature = "std")]
fn throttle(self, d: Duration) -> Throttle<Self>
where
Self: Sized,
@ -598,8 +605,7 @@ extension_trait! {
# }) }
```
"#]
#[cfg(any(feature = "unstable", feature = "docs"))]
#[cfg_attr(feature = "docs", doc(cfg(unstable)))]
#[cfg(feature = "std")]
fn delay(self, dur: std::time::Duration) -> Delay<Self>
where
Self: Sized,
@ -1511,8 +1517,6 @@ extension_trait! {
# }) }
```
"#]
#[cfg(feature = "unstable")]
#[cfg_attr(feature = "docs", doc(cfg(unstable)))]
fn by_ref(&mut self) -> &mut Self {
self
}
@ -1656,8 +1660,7 @@ extension_trait! {
# Ok(()) }) }
```
"#]
#[cfg(any(feature = "unstable", feature = "docs"))]
#[cfg_attr(feature = "docs", doc(cfg(unstable)))]
#[cfg(feature = "std")]
fn timeout(self, dur: Duration) -> Timeout<Self>
where
Self: Stream + Sized,
@ -1822,8 +1825,7 @@ extension_trait! {
# }) }
```
"#]
#[cfg(feature = "unstable")]
#[cfg_attr(feature = "docs", doc(cfg(unstable)))]
#[cfg(feature = "std")]
fn unzip<A, B, FromA, FromB>(self) -> impl Future<Output = (FromA, FromB)> [UnzipFuture<Self, FromA, FromB>]
where
FromA: Default + Extend<A>,
@ -1921,8 +1923,7 @@ extension_trait! {
# });
```
"#]
#[cfg(feature = "unstable")]
#[cfg_attr(feature = "docs", doc(cfg(unstable)))]
#[cfg(feature = "std")]
fn merge<U>(self, other: U) -> Merge<Self, U>
where
Self: Sized,
@ -2068,8 +2069,7 @@ extension_trait! {
# }) }
```
"#]
#[cfg(feature = "unstable")]
#[cfg_attr(feature = "docs", doc(cfg(unstable)))]
#[cfg(feature = "std")]
fn count(self) -> impl Future<Output = usize> [CountFuture<Self>]
where
Self: Sized,
@ -2330,8 +2330,7 @@ extension_trait! {
# }) }
```
"#]
#[cfg(feature = "unstable")]
#[cfg_attr(feature = "docs", doc(cfg(unstable)))]
#[cfg(feature = "std")]
fn sum<'a, S>(
self,
) -> impl Future<Output = S> + 'a [Pin<Box<dyn Future<Output = S> + 'a>>]
@ -2376,8 +2375,7 @@ extension_trait! {
# }) }
```
"#]
#[cfg(feature = "unstable")]
#[cfg_attr(feature = "docs", doc(cfg(unstable)))]
#[cfg(feature = "std")]
fn product<'a, P>(
self,
) -> impl Future<Output = P> + 'a [Pin<Box<dyn Future<Output = P> + 'a>>]

2
src/stream/stream/timeout.rs
View file

@ -47,8 +47,6 @@ impl<S: Stream> Stream for Timeout<S> {
}
/// An error returned when a stream times out.
#[cfg_attr(feature = "docs", doc(cfg(unstable)))]
#[cfg(any(feature = "unstable", feature = "docs"))]
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub struct TimeoutError {
_private: (),

2
src/stream/stream/unzip.rs
View file

@ -8,8 +8,6 @@ use crate::task::{Context, Poll};
pin_project! {
#[derive(Clone, Debug)]
#[cfg(feature = "unstable")]
#[cfg_attr(feature = "docs", doc(cfg(unstable)))]
pub struct UnzipFuture<S, FromA, FromB> {
#[pin]
stream: S,

4
src/stream/successors.rs
View file

@ -27,8 +27,6 @@ use pin_project_lite::pin_project;
/// #
/// # }) }
/// ```
#[cfg(feature = "unstable")]
#[cfg_attr(feature = "docs", doc(cfg(unstable)))]
pub fn successors<F, T>(first: Option<T>, succ: F) -> Successors<F, T>
where
F: FnMut(&T) -> Option<T>,
@ -43,8 +41,6 @@ pin_project! {
/// This stream is constructed by [`successors`] function
///
/// [`successors`]: fn.succssors.html
#[cfg(feature = "unstable")]
#[cfg_attr(feature = "docs", doc(cfg(unstable)))]
#[derive(Debug)]
pub struct Successors<F, T>
where

3
src/stream/sum.rs
View file

@ -1,3 +1,4 @@
use alloc::boxed::Box;
use core::future::Future;
use core::pin::Pin;
@ -13,8 +14,6 @@ use crate::stream::Stream;
/// [`sum`]: trait.Sum.html#tymethod.sum
/// [`FromStream`]: trait.FromStream.html
/// [`Stream::sum`]: trait.Stream.html#method.sum
#[cfg(feature = "unstable")]
#[cfg_attr(feature = "docs", doc(cfg(unstable)))]
pub trait Sum<A = Self>: Sized {
/// Method which takes a stream and generates `Self` from the elements by
/// "summing up" the items.

7
src/utils.rs
View file

@ -21,7 +21,7 @@ pub fn abort_on_panic<T>(f: impl FnOnce() -> T) -> T {
}
/// Generates a random number in `0..n`.
#[cfg(any(feature = "unstable", feature = "default"))]
#[cfg(any(feature = "std", feature = "default"))]
pub fn random(n: u32) -> u32 {
use std::cell::Cell;
use std::num::Wrapping;
@ -257,11 +257,6 @@ macro_rules! extension_trait {
$(#[cfg(feature = "docs")] $imp)*
};
// Optimization: expand `$head` eagerly before starting a new method definition.
(@ext ($($head:tt)*) #[doc = $d:literal] $($tail:tt)*) => {
$($head)* extension_trait!(@ext (#[doc = $d]) $($tail)*);
};
// Parse the return type in an extension method.
(@doc ($($head:tt)*) -> impl Future<Output = $out:ty> $(+ $lt:lifetime)? [$f:ty] $($tail:tt)*) => {
extension_trait!(@doc ($($head)* -> owned::ImplFuture<$out>) $($tail)*);

7
tests/io_timeout.rs
View file

@ -7,10 +7,9 @@ use async_std::task;
#[should_panic(expected = "timed out")]
fn io_timeout_timedout() {
task::block_on(async {
io::timeout(Duration::from_secs(1), async {
let stdin = io::stdin();
let mut line = String::new();
let _n = stdin.read_line(&mut line).await?;
io::timeout(Duration::from_millis(100), async {
task::sleep(Duration::from_secs(1)).await;
Ok(())
})
.await