forked from mirror/async-std
Merge branch 'master' into add_stream_timeout
k-nasa
5 years ago
commit
b17af61367
@ -1,46 +1,51 @@
|
||||
use std::mem;
|
||||
use std::pin::Pin;
|
||||
|
||||
use pin_project_lite::pin_project;
|
||||
|
||||
use super::read_until_internal;
|
||||
use crate::io::{self, BufRead};
|
||||
use crate::stream::Stream;
|
||||
use crate::task::{Context, Poll};
|
||||
|
||||
/// A stream over the contents of an instance of [`BufRead`] split on a particular byte.
|
||||
///
|
||||
/// This stream is created by the [`split`] method on types that implement [`BufRead`].
|
||||
///
|
||||
/// This type is an async version of [`std::io::Split`].
|
||||
///
|
||||
/// [`split`]: trait.BufRead.html#method.lines
|
||||
/// [`BufRead`]: trait.BufRead.html
|
||||
/// [`std::io::Split`]: https://doc.rust-lang.org/std/io/struct.Split.html
|
||||
#[derive(Debug)]
|
||||
pub struct Split<R> {
|
||||
pin_project! {
|
||||
/// A stream over the contents of an instance of [`BufRead`] split on a particular byte.
|
||||
///
|
||||
/// This stream is created by the [`split`] method on types that implement [`BufRead`].
|
||||
///
|
||||
/// This type is an async version of [`std::io::Split`].
|
||||
///
|
||||
/// [`split`]: trait.BufRead.html#method.lines
|
||||
/// [`BufRead`]: trait.BufRead.html
|
||||
/// [`std::io::Split`]: https://doc.rust-lang.org/std/io/struct.Split.html
|
||||
#[derive(Debug)]
|
||||
pub struct Split<R> {
|
||||
#[pin]
|
||||
pub(crate) reader: R,
|
||||
pub(crate) buf: Vec<u8>,
|
||||
pub(crate) read: usize,
|
||||
pub(crate) delim: u8,
|
||||
}
|
||||
}
|
||||
|
||||
impl<R: BufRead> Stream for Split<R> {
|
||||
type Item = io::Result<Vec<u8>>;
|
||||
|
||||
fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
|
||||
let Self {
|
||||
reader,
|
||||
buf,
|
||||
read,
|
||||
delim,
|
||||
} = unsafe { self.get_unchecked_mut() };
|
||||
let reader = unsafe { Pin::new_unchecked(reader) };
|
||||
let n = futures_core::ready!(read_until_internal(reader, cx, *delim, buf, read))?;
|
||||
if n == 0 && buf.is_empty() {
|
||||
let this = self.project();
|
||||
let n = futures_core::ready!(read_until_internal(
|
||||
this.reader,
|
||||
cx,
|
||||
*this.delim,
|
||||
this.buf,
|
||||
this.read
|
||||
))?;
|
||||
if n == 0 && this.buf.is_empty() {
|
||||
return Poll::Ready(None);
|
||||
}
|
||||
if buf[buf.len() - 1] == *delim {
|
||||
buf.pop();
|
||||
if this.buf[this.buf.len() - 1] == *this.delim {
|
||||
this.buf.pop();
|
||||
}
|
||||
Poll::Ready(Some(Ok(mem::replace(buf, vec![]))))
|
||||
Poll::Ready(Some(Ok(mem::replace(this.buf, vec![]))))
|
||||
}
|
||||
}
|
||||
|
||||
@ -0,0 +1,21 @@
|
||||
use std::pin::Pin;
|
||||
|
||||
use crate::future::Future;
|
||||
use crate::io::{self, Seek, SeekFrom};
|
||||
use crate::task::{Context, Poll};
|
||||
|
||||
#[doc(hidden)]
|
||||
#[allow(missing_debug_implementations)]
|
||||
pub struct SeekFuture<'a, T: Unpin + ?Sized> {
|
||||
pub(crate) seeker: &'a mut T,
|
||||
pub(crate) pos: SeekFrom,
|
||||
}
|
||||
|
||||
impl<T: Seek + Unpin + ?Sized> Future for SeekFuture<'_, T> {
|
||||
type Output = io::Result<u64>;
|
||||
|
||||
fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
|
||||
let pos = self.pos;
|
||||
Pin::new(&mut *self.seeker).poll_seek(cx, pos)
|
||||
}
|
||||
}
|
||||
@ -1,9 +1,9 @@
|
||||
//! OS-specific extensions.
|
||||
|
||||
#[cfg(any(unix, feature = "docs"))]
|
||||
#[cfg_attr(feature = "docs", doc(cfg(unix)))]
|
||||
pub mod unix;
|
||||
cfg_unix! {
|
||||
pub mod unix;
|
||||
}
|
||||
|
||||
#[cfg(any(windows, feature = "docs"))]
|
||||
#[cfg_attr(feature = "docs", doc(cfg(windows)))]
|
||||
pub mod windows;
|
||||
cfg_windows! {
|
||||
pub mod windows;
|
||||
}
|
||||
|
||||
@ -0,0 +1,100 @@
|
||||
use std::marker::PhantomData;
|
||||
use std::pin::Pin;
|
||||
|
||||
use pin_project_lite::pin_project;
|
||||
|
||||
use crate::future::Future;
|
||||
use crate::stream::Stream;
|
||||
use crate::task::{Context, Poll};
|
||||
|
||||
pin_project! {
|
||||
/// A stream that repeats elements of type `T` endlessly by applying a provided closure.
|
||||
///
|
||||
/// This stream is constructed by the [`repeat_with`] function.
|
||||
///
|
||||
/// [`repeat_with`]: fn.repeat_with.html
|
||||
#[derive(Debug)]
|
||||
pub struct RepeatWith<F, Fut, A> {
|
||||
f: F,
|
||||
#[pin]
|
||||
future: Option<Fut>,
|
||||
__a: PhantomData<A>,
|
||||
}
|
||||
}
|
||||
|
||||
/// Creates a new stream that repeats elements of type `A` endlessly by applying the provided closure.
|
||||
///
|
||||
/// # Examples
|
||||
///
|
||||
/// Basic usage:
|
||||
///
|
||||
/// ```
|
||||
/// # async_std::task::block_on(async {
|
||||
/// #
|
||||
/// use async_std::prelude::*;
|
||||
/// use async_std::stream;
|
||||
///
|
||||
/// let s = stream::repeat_with(|| async { 1 });
|
||||
///
|
||||
/// pin_utils::pin_mut!(s);
|
||||
///
|
||||
/// assert_eq!(s.next().await, Some(1));
|
||||
/// assert_eq!(s.next().await, Some(1));
|
||||
/// assert_eq!(s.next().await, Some(1));
|
||||
/// assert_eq!(s.next().await, Some(1));
|
||||
/// # })
|
||||
/// ```
|
||||
///
|
||||
/// Going finite:
|
||||
///
|
||||
/// ```
|
||||
/// # async_std::task::block_on(async {
|
||||
/// #
|
||||
/// use async_std::prelude::*;
|
||||
/// use async_std::stream;
|
||||
///
|
||||
/// let s = stream::repeat_with(|| async { 1u8 }).take(2);
|
||||
///
|
||||
/// pin_utils::pin_mut!(s);
|
||||
///
|
||||
/// assert_eq!(s.next().await, Some(1));
|
||||
/// assert_eq!(s.next().await, Some(1));
|
||||
/// assert_eq!(s.next().await, None);
|
||||
/// # })
|
||||
/// ```
|
||||
pub fn repeat_with<F, Fut, A>(repeater: F) -> RepeatWith<F, Fut, A>
|
||||
where
|
||||
F: FnMut() -> Fut,
|
||||
Fut: Future<Output = A>,
|
||||
{
|
||||
RepeatWith {
|
||||
f: repeater,
|
||||
future: None,
|
||||
__a: PhantomData,
|
||||
}
|
||||
}
|
||||
|
||||
impl<F, Fut, A> Stream for RepeatWith<F, Fut, A>
|
||||
where
|
||||
F: FnMut() -> Fut,
|
||||
Fut: Future<Output = A>,
|
||||
{
|
||||
type Item = A;
|
||||
|
||||
fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
|
||||
let mut this = self.project();
|
||||
loop {
|
||||
if this.future.is_some() {
|
||||
let res = futures_core::ready!(this.future.as_mut().as_pin_mut().unwrap().poll(cx));
|
||||
|
||||
this.future.set(None);
|
||||
|
||||
return Poll::Ready(Some(res));
|
||||
} else {
|
||||
let fut = (this.f)();
|
||||
|
||||
this.future.set(Some(fut));
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,45 @@
|
||||
use std::pin::Pin;
|
||||
|
||||
use pin_project_lite::pin_project;
|
||||
|
||||
use crate::future::Future;
|
||||
use crate::stream::Stream;
|
||||
use crate::task::{Context, Poll};
|
||||
|
||||
pin_project! {
|
||||
#[doc(hidden)]
|
||||
#[allow(missing_debug_implementations)]
|
||||
pub struct LastFuture<S, T> {
|
||||
#[pin]
|
||||
stream: S,
|
||||
last: Option<T>,
|
||||
}
|
||||
}
|
||||
|
||||
impl<S, T> LastFuture<S, T> {
|
||||
pub(crate) fn new(stream: S) -> Self {
|
||||
LastFuture { stream, last: None }
|
||||
}
|
||||
}
|
||||
|
||||
impl<S> Future for LastFuture<S, S::Item>
|
||||
where
|
||||
S: Stream + Unpin + Sized,
|
||||
S::Item: Copy,
|
||||
{
|
||||
type Output = Option<S::Item>;
|
||||
|
||||
fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
|
||||
let this = self.project();
|
||||
let next = futures_core::ready!(this.stream.poll_next(cx));
|
||||
|
||||
match next {
|
||||
Some(new) => {
|
||||
cx.waker().wake_by_ref();
|
||||
*this.last = Some(new);
|
||||
Poll::Pending
|
||||
}
|
||||
None => Poll::Ready(*this.last),
|
||||
}
|
||||
}
|
||||
}
|
||||
File diff suppressed because it is too large
Load Diff
@ -0,0 +1,350 @@
|
||||
use std::sync::atomic::{AtomicUsize, Ordering};
|
||||
use std::sync::Arc;
|
||||
use std::time::Duration;
|
||||
|
||||
use async_std::sync::channel;
|
||||
use async_std::task;
|
||||
use rand::{thread_rng, Rng};
|
||||
|
||||
fn ms(ms: u64) -> Duration {
|
||||
Duration::from_millis(ms)
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn smoke() {
|
||||
task::block_on(async {
|
||||
let (s, r) = channel(1);
|
||||
|
||||
s.send(7).await;
|
||||
assert_eq!(r.recv().await, Some(7));
|
||||
|
||||
s.send(8).await;
|
||||
assert_eq!(r.recv().await, Some(8));
|
||||
|
||||
drop(s);
|
||||
assert_eq!(r.recv().await, None);
|
||||
})
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn capacity() {
|
||||
for i in 1..10 {
|
||||
let (s, r) = channel::<()>(i);
|
||||
assert_eq!(s.capacity(), i);
|
||||
assert_eq!(r.capacity(), i);
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn len_empty_full() {
|
||||
task::block_on(async {
|
||||
let (s, r) = channel(2);
|
||||
|
||||
assert_eq!(s.len(), 0);
|
||||
assert_eq!(s.is_empty(), true);
|
||||
assert_eq!(s.is_full(), false);
|
||||
assert_eq!(r.len(), 0);
|
||||
assert_eq!(r.is_empty(), true);
|
||||
assert_eq!(r.is_full(), false);
|
||||
|
||||
s.send(()).await;
|
||||
|
||||
assert_eq!(s.len(), 1);
|
||||
assert_eq!(s.is_empty(), false);
|
||||
assert_eq!(s.is_full(), false);
|
||||
assert_eq!(r.len(), 1);
|
||||
assert_eq!(r.is_empty(), false);
|
||||
assert_eq!(r.is_full(), false);
|
||||
|
||||
s.send(()).await;
|
||||
|
||||
assert_eq!(s.len(), 2);
|
||||
assert_eq!(s.is_empty(), false);
|
||||
assert_eq!(s.is_full(), true);
|
||||
assert_eq!(r.len(), 2);
|
||||
assert_eq!(r.is_empty(), false);
|
||||
assert_eq!(r.is_full(), true);
|
||||
|
||||
r.recv().await;
|
||||
|
||||
assert_eq!(s.len(), 1);
|
||||
assert_eq!(s.is_empty(), false);
|
||||
assert_eq!(s.is_full(), false);
|
||||
assert_eq!(r.len(), 1);
|
||||
assert_eq!(r.is_empty(), false);
|
||||
assert_eq!(r.is_full(), false);
|
||||
})
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn recv() {
|
||||
task::block_on(async {
|
||||
let (s, r) = channel(100);
|
||||
|
||||
task::spawn(async move {
|
||||
assert_eq!(r.recv().await, Some(7));
|
||||
task::sleep(ms(1000)).await;
|
||||
assert_eq!(r.recv().await, Some(8));
|
||||
task::sleep(ms(1000)).await;
|
||||
assert_eq!(r.recv().await, Some(9));
|
||||
assert_eq!(r.recv().await, None);
|
||||
});
|
||||
|
||||
task::sleep(ms(1500)).await;
|
||||
s.send(7).await;
|
||||
s.send(8).await;
|
||||
s.send(9).await;
|
||||
})
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn send() {
|
||||
task::block_on(async {
|
||||
let (s, r) = channel(1);
|
||||
|
||||
task::spawn(async move {
|
||||
s.send(7).await;
|
||||
task::sleep(ms(1000)).await;
|
||||
s.send(8).await;
|
||||
task::sleep(ms(1000)).await;
|
||||
s.send(9).await;
|
||||
task::sleep(ms(1000)).await;
|
||||
s.send(10).await;
|
||||
});
|
||||
|
||||
task::sleep(ms(1500)).await;
|
||||
assert_eq!(r.recv().await, Some(7));
|
||||
assert_eq!(r.recv().await, Some(8));
|
||||
assert_eq!(r.recv().await, Some(9));
|
||||
})
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn recv_after_disconnect() {
|
||||
task::block_on(async {
|
||||
let (s, r) = channel(100);
|
||||
|
||||
s.send(1).await;
|
||||
s.send(2).await;
|
||||
s.send(3).await;
|
||||
|
||||
drop(s);
|
||||
|
||||
assert_eq!(r.recv().await, Some(1));
|
||||
assert_eq!(r.recv().await, Some(2));
|
||||
assert_eq!(r.recv().await, Some(3));
|
||||
assert_eq!(r.recv().await, None);
|
||||
})
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn len() {
|
||||
const COUNT: usize = 25_000;
|
||||
const CAP: usize = 1000;
|
||||
|
||||
task::block_on(async {
|
||||
let (s, r) = channel(CAP);
|
||||
|
||||
assert_eq!(s.len(), 0);
|
||||
assert_eq!(r.len(), 0);
|
||||
|
||||
for _ in 0..CAP / 10 {
|
||||
for i in 0..50 {
|
||||
s.send(i).await;
|
||||
assert_eq!(s.len(), i + 1);
|
||||
}
|
||||
|
||||
for i in 0..50 {
|
||||
r.recv().await;
|
||||
assert_eq!(r.len(), 50 - i - 1);
|
||||
}
|
||||
}
|
||||
|
||||
assert_eq!(s.len(), 0);
|
||||
assert_eq!(r.len(), 0);
|
||||
|
||||
for i in 0..CAP {
|
||||
s.send(i).await;
|
||||
assert_eq!(s.len(), i + 1);
|
||||
}
|
||||
|
||||
for _ in 0..CAP {
|
||||
r.recv().await.unwrap();
|
||||
}
|
||||
|
||||
assert_eq!(s.len(), 0);
|
||||
assert_eq!(r.len(), 0);
|
||||
|
||||
let child = task::spawn({
|
||||
let r = r.clone();
|
||||
async move {
|
||||
for i in 0..COUNT {
|
||||
assert_eq!(r.recv().await, Some(i));
|
||||
let len = r.len();
|
||||
assert!(len <= CAP);
|
||||
}
|
||||
}
|
||||
});
|
||||
|
||||
for i in 0..COUNT {
|
||||
s.send(i).await;
|
||||
let len = s.len();
|
||||
assert!(len <= CAP);
|
||||
}
|
||||
|
||||
child.await;
|
||||
|
||||
assert_eq!(s.len(), 0);
|
||||
assert_eq!(r.len(), 0);
|
||||
})
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn disconnect_wakes_receiver() {
|
||||
task::block_on(async {
|
||||
let (s, r) = channel::<()>(1);
|
||||
|
||||
let child = task::spawn(async move {
|
||||
assert_eq!(r.recv().await, None);
|
||||
});
|
||||
|
||||
task::sleep(ms(1000)).await;
|
||||
drop(s);
|
||||
|
||||
child.await;
|
||||
})
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn spsc() {
|
||||
const COUNT: usize = 100_000;
|
||||
|
||||
task::block_on(async {
|
||||
let (s, r) = channel(3);
|
||||
|
||||
let child = task::spawn(async move {
|
||||
for i in 0..COUNT {
|
||||
assert_eq!(r.recv().await, Some(i));
|
||||
}
|
||||
assert_eq!(r.recv().await, None);
|
||||
});
|
||||
|
||||
for i in 0..COUNT {
|
||||
s.send(i).await;
|
||||
}
|
||||
drop(s);
|
||||
|
||||
child.await;
|
||||
})
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn mpmc() {
|
||||
const COUNT: usize = 25_000;
|
||||
const TASKS: usize = 4;
|
||||
|
||||
task::block_on(async {
|
||||
let (s, r) = channel::<usize>(3);
|
||||
let v = (0..COUNT).map(|_| AtomicUsize::new(0)).collect::<Vec<_>>();
|
||||
let v = Arc::new(v);
|
||||
|
||||
let mut tasks = Vec::new();
|
||||
|
||||
for _ in 0..TASKS {
|
||||
let r = r.clone();
|
||||
let v = v.clone();
|
||||
tasks.push(task::spawn(async move {
|
||||
for _ in 0..COUNT {
|
||||
let n = r.recv().await.unwrap();
|
||||
v[n].fetch_add(1, Ordering::SeqCst);
|
||||
}
|
||||
}));
|
||||
}
|
||||
|
||||
for _ in 0..TASKS {
|
||||
let s = s.clone();
|
||||
tasks.push(task::spawn(async move {
|
||||
for i in 0..COUNT {
|
||||
s.send(i).await;
|
||||
}
|
||||
}));
|
||||
}
|
||||
|
||||
for t in tasks {
|
||||
t.await;
|
||||
}
|
||||
|
||||
for c in v.iter() {
|
||||
assert_eq!(c.load(Ordering::SeqCst), TASKS);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn oneshot() {
|
||||
const COUNT: usize = 10_000;
|
||||
|
||||
task::block_on(async {
|
||||
for _ in 0..COUNT {
|
||||
let (s, r) = channel(1);
|
||||
|
||||
let c1 = task::spawn(async move { r.recv().await.unwrap() });
|
||||
let c2 = task::spawn(async move { s.send(0).await });
|
||||
|
||||
c1.await;
|
||||
c2.await;
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn drops() {
|
||||
const RUNS: usize = 100;
|
||||
|
||||
static DROPS: AtomicUsize = AtomicUsize::new(0);
|
||||
|
||||
#[derive(Debug, PartialEq)]
|
||||
struct DropCounter;
|
||||
|
||||
impl Drop for DropCounter {
|
||||
fn drop(&mut self) {
|
||||
DROPS.fetch_add(1, Ordering::SeqCst);
|
||||
}
|
||||
}
|
||||
|
||||
let mut rng = thread_rng();
|
||||
|
||||
for _ in 0..RUNS {
|
||||
task::block_on(async {
|
||||
let steps = rng.gen_range(0, 10_000);
|
||||
let additional = rng.gen_range(0, 50);
|
||||
|
||||
DROPS.store(0, Ordering::SeqCst);
|
||||
let (s, r) = channel::<DropCounter>(50);
|
||||
|
||||
let child = task::spawn({
|
||||
let r = r.clone();
|
||||
async move {
|
||||
for _ in 0..steps {
|
||||
r.recv().await.unwrap();
|
||||
}
|
||||
}
|
||||
});
|
||||
|
||||
for _ in 0..steps {
|
||||
s.send(DropCounter).await;
|
||||
}
|
||||
|
||||
child.await;
|
||||
|
||||
for _ in 0..additional {
|
||||
s.send(DropCounter).await;
|
||||
}
|
||||
|
||||
assert_eq!(DROPS.load(Ordering::SeqCst), steps);
|
||||
drop(s);
|
||||
drop(r);
|
||||
assert_eq!(DROPS.load(Ordering::SeqCst), steps + additional);
|
||||
})
|
||||
}
|
||||
}
|
||||
Loading…
Reference in New Issue