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

Remove Pin API related unsafe code by using pin-project-lite cra… (#381)

Browse Source
yoshuawuyts-patch-1
Taiki Endo 5 years ago committed by Stjepan Glavina
parent ec23632f3e
commit 944e43d4bf
42 changed files with 894 additions and 825 deletions
Show Stats Download Patch File Download Diff File

1
Cargo.toml
Unescape Escape View File

@ -43,6 +43,7 @@ pin-utils = "0.1.0-alpha.4"
slab = "0.4.2" slab = "0.4.2"
kv-log-macro = "1.0.4" kv-log-macro = "1.0.4"
broadcaster = { version = "0.2.6", optional = true, default-features = false, features = ["default-channels"] } broadcaster = { version = "0.2.6", optional = true, default-features = false, features = ["default-channels"] }
pin-project-lite = "0.1"
[dev-dependencies] [dev-dependencies]
femme = "1.2.0" femme = "1.2.0"

25
src/future/timeout.rs
Unescape Escape View File

@ -4,6 +4,7 @@ use std::pin::Pin;
use std::time::Duration; use std::time::Duration;
use futures_timer::Delay; use futures_timer::Delay;
use pin_project_lite::pin_project;
use crate::future::Future; use crate::future::Future;
use crate::task::{Context, Poll}; use crate::task::{Context, Poll};
@ -39,24 +40,24 @@ where
f.await f.await
} }
/// A future that times out after a duration of time. pin_project! {
struct TimeoutFuture<F> { /// A future that times out after a duration of time.
future: F, struct TimeoutFuture<F> {
delay: Delay, #[pin]
} future: F,
#[pin]
impl<F> TimeoutFuture<F> { delay: Delay,
pin_utils::unsafe_pinned!(future: F); }
pin_utils::unsafe_pinned!(delay: Delay);
} }
impl<F: Future> Future for TimeoutFuture<F> { impl<F: Future> Future for TimeoutFuture<F> {
type Output = Result<F::Output, TimeoutError>; type Output = Result<F::Output, TimeoutError>;
fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> { fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
match self.as_mut().future().poll(cx) { let this = self.project();
match this.future.poll(cx) {
Poll::Ready(v) => Poll::Ready(Ok(v)), Poll::Ready(v) => Poll::Ready(Ok(v)),
Poll::Pending => match self.delay().poll(cx) { Poll::Pending => match this.delay.poll(cx) {
Poll::Ready(_) => Poll::Ready(Err(TimeoutError { _private: () })), Poll::Ready(_) => Poll::Ready(Err(TimeoutError { _private: () })),
Poll::Pending => Poll::Pending, Poll::Pending => Poll::Pending,
}, },

63
src/io/buf_read/lines.rs
Unescape Escape View File

@ -2,50 +2,55 @@ use std::mem;
use std::pin::Pin; use std::pin::Pin;
use std::str; use std::str;
use pin_project_lite::pin_project;
use super::read_until_internal; use super::read_until_internal;
use crate::io::{self, BufRead}; use crate::io::{self, BufRead};
use crate::stream::Stream; use crate::stream::Stream;
use crate::task::{Context, Poll}; use crate::task::{Context, Poll};
/// A stream of lines in a byte stream. pin_project! {
/// /// A stream of lines in a byte stream.
/// This stream is created by the [`lines`] method on types that implement [`BufRead`]. ///
/// /// This stream is created by the [`lines`] method on types that implement [`BufRead`].
/// This type is an async version of [`std::io::Lines`]. ///
/// /// This type is an async version of [`std::io::Lines`].
/// [`lines`]: trait.BufRead.html#method.lines ///
/// [`BufRead`]: trait.BufRead.html /// [`lines`]: trait.BufRead.html#method.lines
/// [`std::io::Lines`]: https://doc.rust-lang.org/std/io/struct.Lines.html /// [`BufRead`]: trait.BufRead.html
#[derive(Debug)] /// [`std::io::Lines`]: https://doc.rust-lang.org/std/io/struct.Lines.html
pub struct Lines<R> { #[derive(Debug)]
pub(crate) reader: R, pub struct Lines<R> {
pub(crate) buf: String, #[pin]
pub(crate) bytes: Vec<u8>, pub(crate) reader: R,
pub(crate) read: usize, pub(crate) buf: String,
pub(crate) bytes: Vec<u8>,
pub(crate) read: usize,
}
} }
impl<R: BufRead> Stream for Lines<R> { impl<R: BufRead> Stream for Lines<R> {
type Item = io::Result<String>; type Item = io::Result<String>;
fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> { fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
let Self { let this = self.project();
reader, let n = futures_core::ready!(read_line_internal(
buf, this.reader,
bytes, cx,
read, this.buf,
} = unsafe { self.get_unchecked_mut() }; this.bytes,
let reader = unsafe { Pin::new_unchecked(reader) }; this.read
let n = futures_core::ready!(read_line_internal(reader, cx, buf, bytes, read))?; ))?;
if n == 0 && buf.is_empty() { if n == 0 && this.buf.is_empty() {
return Poll::Ready(None); return Poll::Ready(None);
} }
if buf.ends_with('\n') { if this.buf.ends_with('\n') {
buf.pop(); this.buf.pop();
if buf.ends_with('\r') { if this.buf.ends_with('\r') {
buf.pop(); this.buf.pop();
} }
} }
Poll::Ready(Some(Ok(mem::replace(buf, String::new())))) Poll::Ready(Some(Ok(mem::replace(this.buf, String::new()))))
} }
} }

59
src/io/buf_read/split.rs
Unescape Escape View File

@ -1,46 +1,51 @@
use std::mem; use std::mem;
use std::pin::Pin; use std::pin::Pin;
use pin_project_lite::pin_project;
use super::read_until_internal; use super::read_until_internal;
use crate::io::{self, BufRead}; use crate::io::{self, BufRead};
use crate::stream::Stream; use crate::stream::Stream;
use crate::task::{Context, Poll}; use crate::task::{Context, Poll};
/// A stream over the contents of an instance of [`BufRead`] split on a particular byte. 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 stream is created by the [`split`] method on types that implement [`BufRead`].
/// This type is an async version of [`std::io::Split`]. ///
/// /// This type is an async version of [`std::io::Split`].
/// [`split`]: trait.BufRead.html#method.lines ///
/// [`BufRead`]: trait.BufRead.html /// [`split`]: trait.BufRead.html#method.lines
/// [`std::io::Split`]: https://doc.rust-lang.org/std/io/struct.Split.html /// [`BufRead`]: trait.BufRead.html
#[derive(Debug)] /// [`std::io::Split`]: https://doc.rust-lang.org/std/io/struct.Split.html
pub struct Split<R> { #[derive(Debug)]
pub(crate) reader: R, pub struct Split<R> {
pub(crate) buf: Vec<u8>, #[pin]
pub(crate) read: usize, pub(crate) reader: R,
pub(crate) delim: u8, pub(crate) buf: Vec<u8>,
pub(crate) read: usize,
pub(crate) delim: u8,
}
} }
impl<R: BufRead> Stream for Split<R> { impl<R: BufRead> Stream for Split<R> {
type Item = io::Result<Vec<u8>>; type Item = io::Result<Vec<u8>>;
fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> { fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
let Self { let this = self.project();
reader, let n = futures_core::ready!(read_until_internal(
buf, this.reader,
read, cx,
delim, *this.delim,
} = unsafe { self.get_unchecked_mut() }; this.buf,
let reader = unsafe { Pin::new_unchecked(reader) }; this.read
let n = futures_core::ready!(read_until_internal(reader, cx, *delim, buf, read))?; ))?;
if n == 0 && buf.is_empty() { if n == 0 && this.buf.is_empty() {
return Poll::Ready(None); return Poll::Ready(None);
} }
if buf[buf.len() - 1] == *delim { if this.buf[this.buf.len() - 1] == *this.delim {
buf.pop(); this.buf.pop();
} }
Poll::Ready(Some(Ok(mem::replace(buf, vec![])))) Poll::Ready(Some(Ok(mem::replace(this.buf, vec![]))))
} }
} }

141
src/io/buf_reader.rs
Unescape Escape View File

@ -2,51 +2,56 @@ use std::io::{IoSliceMut, Read as _};
use std::pin::Pin; use std::pin::Pin;
use std::{cmp, fmt}; use std::{cmp, fmt};
use pin_project_lite::pin_project;
use crate::io::{self, BufRead, Read, Seek, SeekFrom}; use crate::io::{self, BufRead, Read, Seek, SeekFrom};
use crate::task::{Context, Poll}; use crate::task::{Context, Poll};
const DEFAULT_CAPACITY: usize = 8 * 1024; const DEFAULT_CAPACITY: usize = 8 * 1024;
/// Adds buffering to any reader. pin_project! {
/// /// Adds buffering to any reader.
/// It can be excessively inefficient to work directly with a [`Read`] instance. A `BufReader` ///
/// performs large, infrequent reads on the underlying [`Read`] and maintains an in-memory buffer /// It can be excessively inefficient to work directly with a [`Read`] instance. A `BufReader`
/// of the incoming byte stream. /// performs large, infrequent reads on the underlying [`Read`] and maintains an in-memory buffer
/// /// of the incoming byte stream.
/// `BufReader` can improve the speed of programs that make *small* and *repeated* read calls to ///
/// the same file or network socket. It does not help when reading very large amounts at once, or /// `BufReader` can improve the speed of programs that make *small* and *repeated* read calls to
/// reading just one or a few times. It also provides no advantage when reading from a source that /// the same file or network socket. It does not help when reading very large amounts at once, or
/// is already in memory, like a `Vec<u8>`. /// reading just one or a few times. It also provides no advantage when reading from a source that
/// /// is already in memory, like a `Vec<u8>`.
/// When the `BufReader` is dropped, the contents of its buffer will be discarded. Creating ///
/// multiple instances of a `BufReader` on the same stream can cause data loss. /// When the `BufReader` is dropped, the contents of its buffer will be discarded. Creating
/// /// multiple instances of a `BufReader` on the same stream can cause data loss.
/// This type is an async version of [`std::io::BufReader`]. ///
/// /// This type is an async version of [`std::io::BufReader`].
/// [`Read`]: trait.Read.html ///
/// [`std::io::BufReader`]: https://doc.rust-lang.org/std/io/struct.BufReader.html /// [`Read`]: trait.Read.html
/// /// [`std::io::BufReader`]: https://doc.rust-lang.org/std/io/struct.BufReader.html
/// # Examples ///
/// /// # Examples
/// ```no_run ///
/// # fn main() -> std::io::Result<()> { async_std::task::block_on(async { /// ```no_run
/// # /// # fn main() -> std::io::Result<()> { async_std::task::block_on(async {
/// use async_std::fs::File; /// #
/// use async_std::io::BufReader; /// use async_std::fs::File;
/// use async_std::prelude::*; /// use async_std::io::BufReader;
/// /// use async_std::prelude::*;
/// let mut file = BufReader::new(File::open("a.txt").await?); ///
/// /// let mut file = BufReader::new(File::open("a.txt").await?);
/// let mut line = String::new(); ///
/// file.read_line(&mut line).await?; /// let mut line = String::new();
/// # /// file.read_line(&mut line).await?;
/// # Ok(()) }) } /// #
/// ``` /// # Ok(()) }) }
pub struct BufReader<R> { /// ```
inner: R, pub struct BufReader<R> {
buf: Box<[u8]>, #[pin]
pos: usize, inner: R,
cap: usize, buf: Box<[u8]>,
pos: usize,
cap: usize,
}
} }
impl<R: io::Read> BufReader<R> { impl<R: io::Read> BufReader<R> {
@ -95,10 +100,6 @@ impl<R: io::Read> BufReader<R> {
} }
impl<R> BufReader<R> { impl<R> BufReader<R> {
pin_utils::unsafe_pinned!(inner: R);
pin_utils::unsafe_unpinned!(pos: usize);
pin_utils::unsafe_unpinned!(cap: usize);
/// Gets a reference to the underlying reader. /// Gets a reference to the underlying reader.
/// ///
/// It is inadvisable to directly read from the underlying reader. /// It is inadvisable to directly read from the underlying reader.
@ -141,6 +142,13 @@ impl<R> BufReader<R> {
&mut self.inner &mut self.inner
} }
/// Gets a pinned mutable reference to the underlying reader.
///
/// It is inadvisable to directly read from the underlying reader.
fn get_pin_mut(self: Pin<&mut Self>) -> Pin<&mut R> {
self.project().inner
}
/// Returns a reference to the internal buffer. /// Returns a reference to the internal buffer.
/// ///
/// This function will not attempt to fill the buffer if it is empty. /// This function will not attempt to fill the buffer if it is empty.
@ -185,9 +193,10 @@ impl<R> BufReader<R> {
/// Invalidates all data in the internal buffer. /// Invalidates all data in the internal buffer.
#[inline] #[inline]
fn discard_buffer(mut self: Pin<&mut Self>) { fn discard_buffer(self: Pin<&mut Self>) {
*self.as_mut().pos() = 0; let this = self.project();
*self.cap() = 0; *this.pos = 0;
*this.cap = 0;
} }
} }
@ -201,7 +210,7 @@ impl<R: Read> Read for BufReader<R> {
// (larger than our internal buffer), bypass our internal buffer // (larger than our internal buffer), bypass our internal buffer
// entirely. // entirely.
if self.pos == self.cap && buf.len() >= self.buf.len() { if self.pos == self.cap && buf.len() >= self.buf.len() {
let res = futures_core::ready!(self.as_mut().inner().poll_read(cx, buf)); let res = futures_core::ready!(self.as_mut().get_pin_mut().poll_read(cx, buf));
self.discard_buffer(); self.discard_buffer();
return Poll::Ready(res); return Poll::Ready(res);
} }
@ -218,7 +227,8 @@ impl<R: Read> Read for BufReader<R> {
) -> Poll<io::Result<usize>> { ) -> Poll<io::Result<usize>> {
let total_len = bufs.iter().map(|b| b.len()).sum::<usize>(); let total_len = bufs.iter().map(|b| b.len()).sum::<usize>();
if self.pos == self.cap && total_len >= self.buf.len() { if self.pos == self.cap && total_len >= self.buf.len() {
let res = futures_core::ready!(self.as_mut().inner().poll_read_vectored(cx, bufs)); let res =
futures_core::ready!(self.as_mut().get_pin_mut().poll_read_vectored(cx, bufs));
self.discard_buffer(); self.discard_buffer();
return Poll::Ready(res); return Poll::Ready(res);
} }
@ -234,28 +244,23 @@ impl<R: Read> BufRead for BufReader<R> {
self: Pin<&'a mut Self>, self: Pin<&'a mut Self>,
cx: &mut Context<'_>, cx: &mut Context<'_>,
) -> Poll<io::Result<&'a [u8]>> { ) -> Poll<io::Result<&'a [u8]>> {
let Self { let mut this = self.project();
inner,
buf,
cap,
pos,
} = unsafe { self.get_unchecked_mut() };
let mut inner = unsafe { Pin::new_unchecked(inner) };
// If we've reached the end of our internal buffer then we need to fetch // If we've reached the end of our internal buffer then we need to fetch
// some more data from the underlying reader. // some more data from the underlying reader.
// Branch using `>=` instead of the more correct `==` // Branch using `>=` instead of the more correct `==`
// to tell the compiler that the pos..cap slice is always valid. // to tell the compiler that the pos..cap slice is always valid.
if *pos >= *cap { if *this.pos >= *this.cap {
debug_assert!(*pos == *cap); debug_assert!(*this.pos == *this.cap);
*cap = futures_core::ready!(inner.as_mut().poll_read(cx, buf))?; *this.cap = futures_core::ready!(this.inner.as_mut().poll_read(cx, this.buf))?;
*pos = 0; *this.pos = 0;
} }
Poll::Ready(Ok(&buf[*pos..*cap])) Poll::Ready(Ok(&this.buf[*this.pos..*this.cap]))
} }
fn consume(mut self: Pin<&mut Self>, amt: usize) { fn consume(self: Pin<&mut Self>, amt: usize) {
*self.as_mut().pos() = cmp::min(self.pos + amt, self.cap); let this = self.project();
*this.pos = cmp::min(*this.pos + amt, *this.cap);
} }
} }
@ -305,24 +310,26 @@ impl<R: Seek> Seek for BufReader<R> {
if let Some(offset) = n.checked_sub(remainder) { if let Some(offset) = n.checked_sub(remainder) {
result = futures_core::ready!( result = futures_core::ready!(
self.as_mut() self.as_mut()
.inner() .get_pin_mut()
.poll_seek(cx, SeekFrom::Current(offset)) .poll_seek(cx, SeekFrom::Current(offset))
)?; )?;
} else { } else {
// seek backwards by our remainder, and then by the offset // seek backwards by our remainder, and then by the offset
futures_core::ready!( futures_core::ready!(
self.as_mut() self.as_mut()
.inner() .get_pin_mut()
.poll_seek(cx, SeekFrom::Current(-remainder)) .poll_seek(cx, SeekFrom::Current(-remainder))
)?; )?;
self.as_mut().discard_buffer(); self.as_mut().discard_buffer();
result = futures_core::ready!( result = futures_core::ready!(
self.as_mut().inner().poll_seek(cx, SeekFrom::Current(n)) self.as_mut()
.get_pin_mut()
.poll_seek(cx, SeekFrom::Current(n))
)?; )?;
} }
} else { } else {
// Seeking with Start/End doesn't care about our buffer length. // Seeking with Start/End doesn't care about our buffer length.
result = futures_core::ready!(self.as_mut().inner().poll_seek(cx, pos))?; result = futures_core::ready!(self.as_mut().get_pin_mut().poll_seek(cx, pos))?;
} }
self.discard_buffer(); self.discard_buffer();
Poll::Ready(Ok(result)) Poll::Ready(Ok(result))

195
src/io/buf_writer.rs
Unescape Escape View File

@ -2,6 +2,7 @@ use std::fmt;
use std::pin::Pin; use std::pin::Pin;
use futures_core::ready; use futures_core::ready;
use pin_project_lite::pin_project;
use crate::io::write::WriteExt; use crate::io::write::WriteExt;
use crate::io::{self, Seek, SeekFrom, Write}; use crate::io::{self, Seek, SeekFrom, Write};
@ -9,88 +10,88 @@ use crate::task::{Context, Poll};
const DEFAULT_CAPACITY: usize = 8 * 1024; const DEFAULT_CAPACITY: usize = 8 * 1024;
/// Wraps a writer and buffers its output. pin_project! {
/// /// Wraps a writer and buffers its output.
/// It can be excessively inefficient to work directly with something that ///
/// implements [`Write`]. For example, every call to /// It can be excessively inefficient to work directly with something that
/// [`write`][`TcpStream::write`] on [`TcpStream`] results in a system call. A /// implements [`Write`]. For example, every call to
/// `BufWriter` keeps an in-memory buffer of data and writes it to an underlying /// [`write`][`TcpStream::write`] on [`TcpStream`] results in a system call. A
/// writer in large, infrequent batches. /// `BufWriter` keeps an in-memory buffer of data and writes it to an underlying
/// /// writer in large, infrequent batches.
/// `BufWriter` can improve the speed of programs that make *small* and ///
/// *repeated* write calls to the same file or network socket. It does not /// `BufWriter` can improve the speed of programs that make *small* and
/// help when writing very large amounts at once, or writing just one or a few /// *repeated* write calls to the same file or network socket. It does not
/// times. It also provides no advantage when writing to a destination that is /// help when writing very large amounts at once, or writing just one or a few
/// in memory, like a `Vec<u8>`. /// times. It also provides no advantage when writing to a destination that is
/// /// in memory, like a `Vec<u8>`.
/// When the `BufWriter` is dropped, the contents of its buffer will be written ///
/// out. However, any errors that happen in the process of flushing the buffer /// When the `BufWriter` is dropped, the contents of its buffer will be written
/// when the writer is dropped will be ignored. Code that wishes to handle such /// out. However, any errors that happen in the process of flushing the buffer
/// errors must manually call [`flush`] before the writer is dropped. /// when the writer is dropped will be ignored. Code that wishes to handle such
/// /// errors must manually call [`flush`] before the writer is dropped.
/// This type is an async version of [`std::io::BufReader`]. ///
/// /// This type is an async version of [`std::io::BufReader`].
/// [`std::io::BufReader`]: https://doc.rust-lang.org/std/io/struct.BufReader.html ///
/// /// [`std::io::BufReader`]: https://doc.rust-lang.org/std/io/struct.BufReader.html
/// # Examples ///
/// /// # Examples
/// Let's write the numbers one through ten to a [`TcpStream`]: ///
/// /// Let's write the numbers one through ten to a [`TcpStream`]:
/// ```no_run ///
/// # fn main() -> std::io::Result<()> { async_std::task::block_on(async { /// ```no_run
/// use async_std::net::TcpStream; /// # fn main() -> std::io::Result<()> { async_std::task::block_on(async {
/// use async_std::prelude::*; /// use async_std::net::TcpStream;
/// /// use async_std::prelude::*;
/// let mut stream = TcpStream::connect("127.0.0.1:34254").await?; ///
/// /// let mut stream = TcpStream::connect("127.0.0.1:34254").await?;
/// for i in 0..10 { ///
/// let arr = [i+1]; /// for i in 0..10 {
/// stream.write(&arr).await?; /// let arr = [i+1];
/// } /// stream.write(&arr).await?;
/// # /// }
/// # Ok(()) }) } /// #
/// ``` /// # Ok(()) }) }
/// /// ```
/// Because we're not buffering, we write each one in turn, incurring the ///
/// overhead of a system call per byte written. We can fix this with a /// Because we're not buffering, we write each one in turn, incurring the
/// `BufWriter`: /// overhead of a system call per byte written. We can fix this with a
/// /// `BufWriter`:
/// ```no_run ///
/// # fn main() -> std::io::Result<()> { async_std::task::block_on(async { /// ```no_run
/// use async_std::io::BufWriter; /// # fn main() -> std::io::Result<()> { async_std::task::block_on(async {
/// use async_std::net::TcpStream; /// use async_std::io::BufWriter;
/// use async_std::prelude::*; /// use async_std::net::TcpStream;
/// /// use async_std::prelude::*;
/// let mut stream = BufWriter::new(TcpStream::connect("127.0.0.1:34254").await?); ///
/// for i in 0..10 { /// let mut stream = BufWriter::new(TcpStream::connect("127.0.0.1:34254").await?);
/// let arr = [i+1]; /// for i in 0..10 {
/// stream.write(&arr).await?; /// let arr = [i+1];
/// }; /// stream.write(&arr).await?;
/// # /// };
/// # Ok(()) }) } /// #
/// ``` /// # Ok(()) }) }
/// /// ```
/// By wrapping the stream with a `BufWriter`, these ten writes are all grouped ///
/// together by the buffer, and will all be written out in one system call when /// By wrapping the stream with a `BufWriter`, these ten writes are all grouped
/// the `stream` is dropped. /// together by the buffer, and will all be written out in one system call when
/// /// the `stream` is dropped.
/// [`Write`]: trait.Write.html ///
/// [`TcpStream::write`]: ../net/struct.TcpStream.html#method.write /// [`Write`]: trait.Write.html
/// [`TcpStream`]: ../net/struct.TcpStream.html /// [`TcpStream::write`]: ../net/struct.TcpStream.html#method.write
/// [`flush`]: trait.Write.html#tymethod.flush /// [`TcpStream`]: ../net/struct.TcpStream.html
pub struct BufWriter<W> { /// [`flush`]: trait.Write.html#tymethod.flush
inner: W, pub struct BufWriter<W> {
buf: Vec<u8>, #[pin]
written: usize, inner: W,
buf: Vec<u8>,
written: usize,
}
} }
#[derive(Debug)] #[derive(Debug)]
pub struct IntoInnerError<W>(W, std::io::Error); pub struct IntoInnerError<W>(W, std::io::Error);
impl<W: Write> BufWriter<W> { impl<W: Write> BufWriter<W> {
pin_utils::unsafe_pinned!(inner: W);
pin_utils::unsafe_unpinned!(buf: Vec<u8>);
/// Creates a new `BufWriter` with a default buffer capacity. The default is currently 8 KB, /// Creates a new `BufWriter` with a default buffer capacity. The default is currently 8 KB,
/// but may change in the future. /// but may change in the future.
/// ///
@ -178,6 +179,13 @@ impl<W: Write> BufWriter<W> {
&mut self.inner &mut self.inner
} }
/// Gets a pinned mutable reference to the underlying writer.
///
/// It is inadvisable to directly write to the underlying writer.
fn get_pin_mut(self: Pin<&mut Self>) -> Pin<&mut W> {
self.project().inner
}
/// Consumes BufWriter, returning the underlying writer /// Consumes BufWriter, returning the underlying writer
/// ///
/// This method will not write leftover data, it will be lost. /// This method will not write leftover data, it will be lost.
@ -234,16 +242,15 @@ impl<W: Write> BufWriter<W> {
/// ///
/// [`LineWriter`]: struct.LineWriter.html /// [`LineWriter`]: struct.LineWriter.html
fn poll_flush_buf(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> { fn poll_flush_buf(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
let Self { let mut this = self.project();
inner, let len = this.buf.len();
buf,
written,
} = unsafe { Pin::get_unchecked_mut(self) };
let mut inner = unsafe { Pin::new_unchecked(inner) };
let len = buf.len();
let mut ret = Ok(()); let mut ret = Ok(());
while *written < len { while *this.written < len {
match inner.as_mut().poll_write(cx, &buf[*written..]) { match this
.inner
.as_mut()
.poll_write(cx, &this.buf[*this.written..])
{
Poll::Ready(Ok(0)) => { Poll::Ready(Ok(0)) => {
ret = Err(io::Error::new( ret = Err(io::Error::new(
io::ErrorKind::WriteZero, io::ErrorKind::WriteZero,
@ -251,7 +258,7 @@ impl<W: Write> BufWriter<W> {
)); ));
break; break;
} }
Poll::Ready(Ok(n)) => *written += n, Poll::Ready(Ok(n)) => *this.written += n,
Poll::Ready(Err(ref e)) if e.kind() == io::ErrorKind::Interrupted => {} Poll::Ready(Err(ref e)) if e.kind() == io::ErrorKind::Interrupted => {}
Poll::Ready(Err(e)) => { Poll::Ready(Err(e)) => {
ret = Err(e); ret = Err(e);
@ -260,10 +267,10 @@ impl<W: Write> BufWriter<W> {
Poll::Pending => return Poll::Pending, Poll::Pending => return Poll::Pending,
} }
} }
if *written > 0 { if *this.written > 0 {
buf.drain(..*written); this.buf.drain(..*this.written);
} }
*written = 0; *this.written = 0;
Poll::Ready(ret) Poll::Ready(ret)
} }
} }
@ -278,20 +285,20 @@ impl<W: Write> Write for BufWriter<W> {
ready!(self.as_mut().poll_flush_buf(cx))?; ready!(self.as_mut().poll_flush_buf(cx))?;
} }
if buf.len() >= self.buf.capacity() { if buf.len() >= self.buf.capacity() {
self.inner().poll_write(cx, buf) self.get_pin_mut().poll_write(cx, buf)
} else { } else {
Pin::new(&mut *self.buf()).poll_write(cx, buf) Pin::new(&mut *self.project().buf).poll_write(cx, buf)
} }
} }
fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> { fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
ready!(self.as_mut().poll_flush_buf(cx))?; ready!(self.as_mut().poll_flush_buf(cx))?;
self.inner().poll_flush(cx) self.get_pin_mut().poll_flush(cx)
} }
fn poll_close(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> { fn poll_close(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
ready!(self.as_mut().poll_flush_buf(cx))?; ready!(self.as_mut().poll_flush_buf(cx))?;
self.inner().poll_close(cx) self.get_pin_mut().poll_close(cx)
} }
} }
@ -314,6 +321,6 @@ impl<W: Write + Seek> Seek for BufWriter<W> {
pos: SeekFrom, pos: SeekFrom,
) -> Poll<io::Result<u64>> { ) -> Poll<io::Result<u64>> {
ready!(self.as_mut().poll_flush_buf(cx))?; ready!(self.as_mut().poll_flush_buf(cx))?;
self.inner().poll_seek(cx, pos) self.get_pin_mut().poll_seek(cx, pos)
} }
} }

43
src/io/copy.rs
Unescape Escape View File

@ -1,5 +1,7 @@
use std::pin::Pin; use std::pin::Pin;
use pin_project_lite::pin_project;
use crate::future::Future; use crate::future::Future;
use crate::io::{self, BufRead, BufReader, Read, Write}; use crate::io::{self, BufRead, BufReader, Read, Write};
use crate::task::{Context, Poll}; use crate::task::{Context, Poll};
@ -46,47 +48,38 @@ where
R: Read + Unpin + ?Sized, R: Read + Unpin + ?Sized,
W: Write + Unpin + ?Sized, W: Write + Unpin + ?Sized,
{ {
pub struct CopyFuture<'a, R, W: ?Sized> { pin_project! {
reader: R, struct CopyFuture<R, W> {
writer: &'a mut W, #[pin]
amt: u64, reader: R,
} #[pin]
writer: W,
impl<R, W: Unpin + ?Sized> CopyFuture<'_, R, W> { amt: u64,
fn project(self: Pin<&mut Self>) -> (Pin<&mut R>, Pin<&mut W>, &mut u64) {
unsafe {
let this = self.get_unchecked_mut();
(
Pin::new_unchecked(&mut this.reader),
Pin::new(&mut *this.writer),
&mut this.amt,
)
}
} }
} }
impl<R, W> Future for CopyFuture<'_, R, W> impl<R, W> Future for CopyFuture<R, W>
where where
R: BufRead, R: BufRead,
W: Write + Unpin + ?Sized, W: Write + Unpin,
{ {
type Output = io::Result<u64>; type Output = io::Result<u64>;
fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> { fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
let (mut reader, mut writer, amt) = self.project(); let mut this = self.project();
loop { loop {
let buffer = futures_core::ready!(reader.as_mut().poll_fill_buf(cx))?; let buffer = futures_core::ready!(this.reader.as_mut().poll_fill_buf(cx))?;
if buffer.is_empty() { if buffer.is_empty() {
futures_core::ready!(writer.as_mut().poll_flush(cx))?; futures_core::ready!(this.writer.as_mut().poll_flush(cx))?;
return Poll::Ready(Ok(*amt)); return Poll::Ready(Ok(*this.amt));
} }
let i = futures_core::ready!(writer.as_mut().poll_write(cx, buffer))?; let i = futures_core::ready!(this.writer.as_mut().poll_write(cx, buffer))?;
if i == 0 { if i == 0 {
return Poll::Ready(Err(io::ErrorKind::WriteZero.into())); return Poll::Ready(Err(io::ErrorKind::WriteZero.into()));
} }
*amt += i as u64; *this.amt += i as u64;
reader.as_mut().consume(i); this.reader.as_mut().consume(i);
} }
} }
} }

89
src/io/read/chain.rs
Unescape Escape View File

@ -1,20 +1,25 @@
use crate::io::IoSliceMut;
use std::fmt; use std::fmt;
use std::pin::Pin; use std::pin::Pin;
use crate::io::{self, BufRead, Read}; use pin_project_lite::pin_project;
use crate::io::{self, BufRead, IoSliceMut, Read};
use crate::task::{Context, Poll}; use crate::task::{Context, Poll};
/// Adaptor to chain together two readers. pin_project! {
/// /// Adaptor to chain together two readers.
/// This struct is generally created by calling [`chain`] on a reader. ///
/// Please see the documentation of [`chain`] for more details. /// This struct is generally created by calling [`chain`] on a reader.
/// /// Please see the documentation of [`chain`] for more details.
/// [`chain`]: trait.Read.html#method.chain ///
pub struct Chain<T, U> { /// [`chain`]: trait.Read.html#method.chain
pub(crate) first: T, pub struct Chain<T, U> {
pub(crate) second: U, #[pin]
pub(crate) done_first: bool, pub(crate) first: T,
#[pin]
pub(crate) second: U,
pub(crate) done_first: bool,
}
} }
impl<T, U> Chain<T, U> { impl<T, U> Chain<T, U> {
@ -98,76 +103,64 @@ impl<T: fmt::Debug, U: fmt::Debug> fmt::Debug for Chain<T, U> {
} }
} }
impl<T: Read + Unpin, U: Read + Unpin> Read for Chain<T, U> { impl<T: Read, U: Read> Read for Chain<T, U> {
fn poll_read( fn poll_read(
mut self: Pin<&mut Self>, self: Pin<&mut Self>,
cx: &mut Context<'_>, cx: &mut Context<'_>,
buf: &mut [u8], buf: &mut [u8],
) -> Poll<io::Result<usize>> { ) -> Poll<io::Result<usize>> {
if !self.done_first { let this = self.project();
let rd = Pin::new(&mut self.first); if !*this.done_first {
match futures_core::ready!(this.first.poll_read(cx, buf)) {
match futures_core::ready!(rd.poll_read(cx, buf)) { Ok(0) if !buf.is_empty() => *this.done_first = true,
Ok(0) if !buf.is_empty() => self.done_first = true,
Ok(n) => return Poll::Ready(Ok(n)), Ok(n) => return Poll::Ready(Ok(n)),
Err(err) => return Poll::Ready(Err(err)), Err(err) => return Poll::Ready(Err(err)),
} }
} }
let rd = Pin::new(&mut self.second); this.second.poll_read(cx, buf)
rd.poll_read(cx, buf)
} }
fn poll_read_vectored( fn poll_read_vectored(
mut self: Pin<&mut Self>, self: Pin<&mut Self>,
cx: &mut Context<'_>, cx: &mut Context<'_>,
bufs: &mut [IoSliceMut<'_>], bufs: &mut [IoSliceMut<'_>],
) -> Poll<io::Result<usize>> { ) -> Poll<io::Result<usize>> {
if !self.done_first { let this = self.project();
let rd = Pin::new(&mut self.first); if !*this.done_first {
match futures_core::ready!(this.first.poll_read_vectored(cx, bufs)) {
match futures_core::ready!(rd.poll_read_vectored(cx, bufs)) { Ok(0) if !bufs.is_empty() => *this.done_first = true,
Ok(0) if !bufs.is_empty() => self.done_first = true,
Ok(n) => return Poll::Ready(Ok(n)), Ok(n) => return Poll::Ready(Ok(n)),
Err(err) => return Poll::Ready(Err(err)), Err(err) => return Poll::Ready(Err(err)),
} }
} }
let rd = Pin::new(&mut self.second); this.second.poll_read_vectored(cx, bufs)
rd.poll_read_vectored(cx, bufs)
} }
} }
impl<T: BufRead + Unpin, U: BufRead + Unpin> BufRead for Chain<T, U> { impl<T: BufRead, U: BufRead> BufRead for Chain<T, U> {
fn poll_fill_buf(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<&[u8]>> { fn poll_fill_buf(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<&[u8]>> {
let Self { let this = self.project();
first, if !*this.done_first {
second, match futures_core::ready!(this.first.poll_fill_buf(cx)) {
done_first,
} = unsafe { self.get_unchecked_mut() };
if !*done_first {
let first = unsafe { Pin::new_unchecked(first) };
match futures_core::ready!(first.poll_fill_buf(cx)) {
Ok(buf) if buf.is_empty() => { Ok(buf) if buf.is_empty() => {
*done_first = true; *this.done_first = true;
} }
Ok(buf) => return Poll::Ready(Ok(buf)), Ok(buf) => return Poll::Ready(Ok(buf)),
Err(err) => return Poll::Ready(Err(err)), Err(err) => return Poll::Ready(Err(err)),
} }
} }
let second = unsafe { Pin::new_unchecked(second) }; this.second.poll_fill_buf(cx)
second.poll_fill_buf(cx)
} }
fn consume(mut self: Pin<&mut Self>, amt: usize) { fn consume(self: Pin<&mut Self>, amt: usize) {
if !self.done_first { let this = self.project();
let rd = Pin::new(&mut self.first); if !*this.done_first {
rd.consume(amt) this.first.consume(amt)
} else { } else {
let rd = Pin::new(&mut self.second); this.second.consume(amt)
rd.consume(amt)
} }
} }
} }

54
src/io/read/take.rs
Unescape Escape View File

@ -1,19 +1,24 @@
use std::cmp; use std::cmp;
use std::pin::Pin; use std::pin::Pin;
use pin_project_lite::pin_project;
use crate::io::{self, BufRead, Read}; use crate::io::{self, BufRead, Read};
use crate::task::{Context, Poll}; use crate::task::{Context, Poll};
/// Reader adaptor which limits the bytes read from an underlying reader. pin_project! {
/// /// Reader adaptor which limits the bytes read from an underlying reader.
/// This struct is generally created by calling [`take`] on a reader. ///
/// Please see the documentation of [`take`] for more details. /// This struct is generally created by calling [`take`] on a reader.
/// /// Please see the documentation of [`take`] for more details.
/// [`take`]: trait.Read.html#method.take ///
#[derive(Debug)] /// [`take`]: trait.Read.html#method.take
pub struct Take<T> { #[derive(Debug)]
pub(crate) inner: T, pub struct Take<T> {
pub(crate) limit: u64, #[pin]
pub(crate) inner: T,
pub(crate) limit: u64,
}
} }
impl<T> Take<T> { impl<T> Take<T> {
@ -152,15 +157,15 @@ impl<T> Take<T> {
} }
} }
impl<T: Read + Unpin> Read for Take<T> { impl<T: Read> Read for Take<T> {
/// Attempt to read from the `AsyncRead` into `buf`. /// Attempt to read from the `AsyncRead` into `buf`.
fn poll_read( fn poll_read(
mut self: Pin<&mut Self>, self: Pin<&mut Self>,
cx: &mut Context<'_>, cx: &mut Context<'_>,
buf: &mut [u8], buf: &mut [u8],
) -> Poll<io::Result<usize>> { ) -> Poll<io::Result<usize>> {
let Self { inner, limit } = &mut *self; let this = self.project();
take_read_internal(Pin::new(inner), cx, buf, limit) take_read_internal(this.inner, cx, buf, this.limit)
} }
} }
@ -186,31 +191,30 @@ pub fn take_read_internal<R: Read + ?Sized>(
} }
} }
impl<T: BufRead + Unpin> BufRead for Take<T> { impl<T: BufRead> BufRead for Take<T> {
fn poll_fill_buf(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<&[u8]>> { fn poll_fill_buf(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<&[u8]>> {
let Self { inner, limit } = unsafe { self.get_unchecked_mut() }; let this = self.project();
let inner = unsafe { Pin::new_unchecked(inner) };
if *limit == 0 { if *this.limit == 0 {
return Poll::Ready(Ok(&[])); return Poll::Ready(Ok(&[]));
} }
match futures_core::ready!(inner.poll_fill_buf(cx)) { match futures_core::ready!(this.inner.poll_fill_buf(cx)) {
Ok(buf) => { Ok(buf) => {
let cap = cmp::min(buf.len() as u64, *limit) as usize; let cap = cmp::min(buf.len() as u64, *this.limit) as usize;
Poll::Ready(Ok(&buf[..cap])) Poll::Ready(Ok(&buf[..cap]))
} }
Err(e) => Poll::Ready(Err(e)), Err(e) => Poll::Ready(Err(e)),
} }
} }
fn consume(mut self: Pin<&mut Self>, amt: usize) { fn consume(self: Pin<&mut Self>, amt: usize) {
let this = self.project();
// Don't let callers reset the limit by passing an overlarge value // Don't let callers reset the limit by passing an overlarge value
let amt = cmp::min(amt as u64, self.limit) as usize; let amt = cmp::min(amt as u64, *this.limit) as usize;
self.limit -= amt as u64; *this.limit -= amt as u64;
let rd = Pin::new(&mut self.inner); this.inner.consume(amt);
rd.consume(amt);
} }
} }

37
src/io/timeout.rs
Unescape Escape View File

@ -3,7 +3,7 @@ use std::task::{Context, Poll};
use std::time::Duration; use std::time::Duration;
use futures_timer::Delay; use futures_timer::Delay;
use pin_utils::unsafe_pinned; use pin_project_lite::pin_project;
use crate::future::Future; use crate::future::Future;
use crate::io; use crate::io;
@ -43,22 +43,18 @@ where
.await .await
} }
/// Future returned by the `FutureExt::timeout` method. pin_project! {
#[derive(Debug)] /// Future returned by the `FutureExt::timeout` method.
pub struct Timeout<F, T> #[derive(Debug)]
where pub struct Timeout<F, T>
F: Future<Output = io::Result<T>>, where
{ F: Future<Output = io::Result<T>>,
future: F, {
timeout: Delay, #[pin]
} future: F,
#[pin]
impl<F, T> Timeout<F, T> timeout: Delay,
where }
F: Future<Output = io::Result<T>>,
{
unsafe_pinned!(future: F);
unsafe_pinned!(timeout: Delay);
} }
impl<F, T> Future for Timeout<F, T> impl<F, T> Future for Timeout<F, T>
@ -67,13 +63,14 @@ where
{ {
type Output = io::Result<T>; type Output = io::Result<T>;
fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> { fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
match self.as_mut().future().poll(cx) { let this = self.project();
match this.future.poll(cx) {
Poll::Pending => {} Poll::Pending => {}
other => return other, other => return other,
} }
if self.timeout().poll(cx).is_ready() { if this.timeout.poll(cx).is_ready() {
let err = Err(io::Error::new(io::ErrorKind::TimedOut, "future timed out").into()); let err = Err(io::Error::new(io::ErrorKind::TimedOut, "future timed out").into());
Poll::Ready(err) Poll::Ready(err)
} else { } else {

52
src/stream/from_fn.rs
Unescape Escape View File

@ -1,20 +1,25 @@
use std::marker::PhantomData; use std::marker::PhantomData;
use std::pin::Pin; use std::pin::Pin;
use pin_project_lite::pin_project;
use crate::future::Future; use crate::future::Future;
use crate::stream::Stream; use crate::stream::Stream;
use crate::task::{Context, Poll}; use crate::task::{Context, Poll};
/// A stream that yields elements by calling a closure. pin_project! {
/// /// A stream that yields elements by calling a closure.
/// This stream is constructed by [`from_fn`] function. ///
/// /// This stream is constructed by [`from_fn`] function.
/// [`from_fn`]: fn.from_fn.html ///
#[derive(Debug)] /// [`from_fn`]: fn.from_fn.html
pub struct FromFn<F, Fut, T> { #[derive(Debug)]
f: F, pub struct FromFn<F, Fut, T> {
future: Option<Fut>, f: F,
__t: PhantomData<T>, #[pin]
future: Option<Fut>,
__t: PhantomData<T>,
}
} }
/// Creates a new stream where to produce each new element a provided closure is called. /// Creates a new stream where to produce each new element a provided closure is called.
@ -68,11 +73,6 @@ where
} }
} }
impl<F, Fut, T> FromFn<F, Fut, T> {
pin_utils::unsafe_unpinned!(f: F);
pin_utils::unsafe_pinned!(future: Option<Fut>);
}
impl<F, Fut, T> Stream for FromFn<F, Fut, T> impl<F, Fut, T> Stream for FromFn<F, Fut, T>
where where
F: FnMut() -> Fut, F: FnMut() -> Fut,
@ -80,20 +80,18 @@ where
{ {
type Item = T; type Item = T;
fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> { fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
let mut this = self.project();
loop { loop {
match &self.future { if this.future.is_some() {
Some(_) => { let next =
let next = futures_core::ready!(this.future.as_mut().as_pin_mut().unwrap().poll(cx));
futures_core::ready!(self.as_mut().future().as_pin_mut().unwrap().poll(cx)); this.future.set(None);
self.as_mut().future().set(None);
return Poll::Ready(next); return Poll::Ready(next);
} } else {
None => { let fut = (this.f)();
let fut = (self.as_mut().f())(); this.future.set(Some(fut));
self.as_mut().future().set(Some(fut));
}
} }
} }
} }

8
src/stream/interval.rs
Unescape Escape View File

@ -4,8 +4,6 @@ use std::time::{Duration, Instant};
use futures_core::future::Future; use futures_core::future::Future;
use futures_core::stream::Stream; use futures_core::stream::Stream;
use pin_utils::unsafe_pinned;
use futures_timer::Delay; use futures_timer::Delay;
/// Creates a new stream that yields at a set interval. /// Creates a new stream that yields at a set interval.
@ -62,15 +60,11 @@ pub struct Interval {
interval: Duration, interval: Duration,
} }
impl Interval {
unsafe_pinned!(delay: Delay);
}
impl Stream for Interval { impl Stream for Interval {
type Item = (); type Item = ();
fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> { fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
if Pin::new(&mut *self).delay().poll(cx).is_pending() { if Pin::new(&mut self.delay).poll(cx).is_pending() {
return Poll::Pending; return Poll::Pending;
} }
let when = Instant::now(); let when = Instant::now();

24
src/stream/once.rs
Unescape Escape View File

@ -1,5 +1,7 @@
use std::pin::Pin; use std::pin::Pin;
use pin_project_lite::pin_project;
use crate::stream::Stream; use crate::stream::Stream;
use crate::task::{Context, Poll}; use crate::task::{Context, Poll};
@ -24,20 +26,22 @@ pub fn once<T>(t: T) -> Once<T> {
Once { value: Some(t) } Once { value: Some(t) }
} }
/// A stream that yields a single item. pin_project! {
/// /// A stream that yields a single item.
/// This stream is constructed by the [`once`] function. ///
/// /// This stream is constructed by the [`once`] function.
/// [`once`]: fn.once.html ///
#[derive(Debug)] /// [`once`]: fn.once.html
pub struct Once<T> { #[derive(Debug)]
value: Option<T>, pub struct Once<T> {
value: Option<T>,
}
} }
impl<T: Unpin> Stream for Once<T> { impl<T: Unpin> Stream for Once<T> {
type Item = T; type Item = T;
fn poll_next(mut self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<Option<T>> { fn poll_next(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<Option<T>> {
Poll::Ready(self.value.take()) Poll::Ready(self.project().value.take())
} }
} }

51
src/stream/repeat_with.rs
Unescape Escape View File

@ -1,20 +1,25 @@
use std::marker::PhantomData; use std::marker::PhantomData;
use std::pin::Pin; use std::pin::Pin;
use pin_project_lite::pin_project;
use crate::future::Future; use crate::future::Future;
use crate::stream::Stream; use crate::stream::Stream;
use crate::task::{Context, Poll}; use crate::task::{Context, Poll};
/// A stream that repeats elements of type `T` endlessly by applying a provided closure. 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. ///
/// /// This stream is constructed by the [`repeat_with`] function.
/// [`repeat_with`]: fn.repeat_with.html ///
#[derive(Debug)] /// [`repeat_with`]: fn.repeat_with.html
pub struct RepeatWith<F, Fut, A> { #[derive(Debug)]
f: F, pub struct RepeatWith<F, Fut, A> {
future: Option<Fut>, f: F,
__a: PhantomData<A>, #[pin]
future: Option<Fut>,
__a: PhantomData<A>,
}
} }
/// Creates a new stream that repeats elements of type `A` endlessly by applying the provided closure. /// Creates a new stream that repeats elements of type `A` endlessly by applying the provided closure.
@ -69,11 +74,6 @@ where
} }
} }
impl<F, Fut, A> RepeatWith<F, Fut, A> {
pin_utils::unsafe_unpinned!(f: F);
pin_utils::unsafe_pinned!(future: Option<Fut>);
}
impl<F, Fut, A> Stream for RepeatWith<F, Fut, A> impl<F, Fut, A> Stream for RepeatWith<F, Fut, A>
where where
F: FnMut() -> Fut, F: FnMut() -> Fut,
@ -81,22 +81,19 @@ where
{ {
type Item = A; type Item = A;
fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> { fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
let mut this = self.project();
loop { loop {
match &self.future { if this.future.is_some() {
Some(_) => { let res = futures_core::ready!(this.future.as_mut().as_pin_mut().unwrap().poll(cx));
let res =
futures_core::ready!(self.as_mut().future().as_pin_mut().unwrap().poll(cx));
self.as_mut().future().set(None); this.future.set(None);
return Poll::Ready(Some(res)); return Poll::Ready(Some(res));
} } else {
None => { let fut = (this.f)();
let fut = (self.as_mut().f())();
self.as_mut().future().set(Some(fut)); this.future.set(Some(fut));
}
} }
} }
} }

32
src/stream/stream/chain.rs
Unescape Escape View File

@ -1,20 +1,23 @@
use std::pin::Pin; use std::pin::Pin;
use pin_project_lite::pin_project;
use super::fuse::Fuse; use super::fuse::Fuse;
use crate::prelude::*; use crate::prelude::*;
use crate::task::{Context, Poll}; use crate::task::{Context, Poll};
/// Chains two streams one after another. pin_project! {
#[derive(Debug)] /// Chains two streams one after another.
pub struct Chain<S, U> { #[derive(Debug)]
first: Fuse<S>, pub struct Chain<S, U> {
second: Fuse<U>, #[pin]
first: Fuse<S>,
#[pin]
second: Fuse<U>,
}
} }
impl<S: Stream, U: Stream> Chain<S, U> { impl<S: Stream, U: Stream> Chain<S, U> {
pin_utils::unsafe_pinned!(first: Fuse<S>);
pin_utils::unsafe_pinned!(second: Fuse<U>);
pub(super) fn new(first: S, second: U) -> Self { pub(super) fn new(first: S, second: U) -> Self {
Chain { Chain {
first: first.fuse(), first: first.fuse(),
@ -26,22 +29,23 @@ impl<S: Stream, U: Stream> Chain<S, U> {
impl<S: Stream, U: Stream<Item = S::Item>> Stream for Chain<S, U> { impl<S: Stream, U: Stream<Item = S::Item>> Stream for Chain<S, U> {
type Item = S::Item; type Item = S::Item;
fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> { fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
if !self.first.done { let mut this = self.project();
let next = futures_core::ready!(self.as_mut().first().poll_next(cx)); if !this.first.done {
let next = futures_core::ready!(this.first.as_mut().poll_next(cx));
if let Some(next) = next { if let Some(next) = next {
return Poll::Ready(Some(next)); return Poll::Ready(Some(next));
} }
} }
if !self.second.done { if !this.second.done {
let next = futures_core::ready!(self.as_mut().second().poll_next(cx)); let next = futures_core::ready!(this.second.as_mut().poll_next(cx));
if let Some(next) = next { if let Some(next) = next {
return Poll::Ready(Some(next)); return Poll::Ready(Some(next));
} }
} }
if self.first.done && self.second.done { if this.first.done && this.second.done {
return Poll::Ready(None); return Poll::Ready(None);
} }

58
src/stream/stream/cmp.rs
Unescape Escape View File

@ -1,29 +1,30 @@
use std::cmp::Ordering; use std::cmp::Ordering;
use std::pin::Pin; use std::pin::Pin;
use pin_project_lite::pin_project;
use super::fuse::Fuse; use super::fuse::Fuse;
use crate::future::Future; use crate::future::Future;
use crate::prelude::*; use crate::prelude::*;
use crate::stream::Stream; use crate::stream::Stream;
use crate::task::{Context, Poll}; use crate::task::{Context, Poll};
// Lexicographically compares the elements of this `Stream` with those pin_project! {
// of another using `Ord`. // Lexicographically compares the elements of this `Stream` with those
#[doc(hidden)] // of another using `Ord`.
#[allow(missing_debug_implementations)] #[doc(hidden)]
pub struct CmpFuture<L: Stream, R: Stream> { #[allow(missing_debug_implementations)]
l: Fuse<L>, pub struct CmpFuture<L: Stream, R: Stream> {
r: Fuse<R>, #[pin]
l_cache: Option<L::Item>, l: Fuse<L>,
r_cache: Option<R::Item>, #[pin]
r: Fuse<R>,
l_cache: Option<L::Item>,
r_cache: Option<R::Item>,
}
} }
impl<L: Stream, R: Stream> CmpFuture<L, R> { impl<L: Stream, R: Stream> CmpFuture<L, R> {
pin_utils::unsafe_pinned!(l: Fuse<L>);
pin_utils::unsafe_pinned!(r: Fuse<R>);
pin_utils::unsafe_unpinned!(l_cache: Option<L::Item>);
pin_utils::unsafe_unpinned!(r_cache: Option<R::Item>);
pub(super) fn new(l: L, r: R) -> Self { pub(super) fn new(l: L, r: R) -> Self {
CmpFuture { CmpFuture {
l: l.fuse(), l: l.fuse(),
@ -42,11 +43,12 @@ where
{ {
type Output = Ordering; type Output = Ordering;
fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> { fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
let mut this = self.project();
loop { loop {
// Stream that completes earliest can be considered Less, etc // Stream that completes earliest can be considered Less, etc
let l_complete = self.l.done && self.as_mut().l_cache.is_none(); let l_complete = this.l.done && this.l_cache.is_none();
let r_complete = self.r.done && self.as_mut().r_cache.is_none(); let r_complete = this.r.done && this.r_cache.is_none();
if l_complete && r_complete { if l_complete && r_complete {
return Poll::Ready(Ordering::Equal); return Poll::Ready(Ordering::Equal);
@ -57,30 +59,30 @@ where
} }
// Get next value if possible and necesary // Get next value if possible and necesary
if !self.l.done && self.as_mut().l_cache.is_none() { if !this.l.done && this.l_cache.is_none() {
let l_next = futures_core::ready!(self.as_mut().l().poll_next(cx)); let l_next = futures_core::ready!(this.l.as_mut().poll_next(cx));
if let Some(item) = l_next { if let Some(item) = l_next {
*self.as_mut().l_cache() = Some(item); *this.l_cache = Some(item);
} }
} }
if !self.r.done && self.as_mut().r_cache.is_none() { if !this.r.done && this.r_cache.is_none() {
let r_next = futures_core::ready!(self.as_mut().r().poll_next(cx)); let r_next = futures_core::ready!(this.r.as_mut().poll_next(cx));
if let Some(item) = r_next { if let Some(item) = r_next {
*self.as_mut().r_cache() = Some(item); *this.r_cache = Some(item);
} }
} }
// Compare if both values are available. // Compare if both values are available.
if self.as_mut().l_cache.is_some() && self.as_mut().r_cache.is_some() { if this.l_cache.is_some() && this.r_cache.is_some() {
let l_value = self.as_mut().l_cache().take().unwrap(); let l_value = this.l_cache.take().unwrap();
let r_value = self.as_mut().r_cache().take().unwrap(); let r_value = this.r_cache.take().unwrap();
let result = l_value.cmp(&r_value); let result = l_value.cmp(&r_value);
if let Ordering::Equal = result { if let Ordering::Equal = result {
// Reset cache to prepare for next comparison // Reset cache to prepare for next comparison
*self.as_mut().l_cache() = None; *this.l_cache = None;
*self.as_mut().r_cache() = None; *this.r_cache = None;
} else { } else {
// Return non equal value // Return non equal value
return Poll::Ready(result); return Poll::Ready(result);

29
src/stream/stream/enumerate.rs
Unescape Escape View File

@ -1,19 +1,21 @@
use crate::task::{Context, Poll};
use std::pin::Pin; use std::pin::Pin;
use pin_project_lite::pin_project;
use crate::stream::Stream; use crate::stream::Stream;
use crate::task::{Context, Poll};
#[doc(hidden)] pin_project! {
#[allow(missing_debug_implementations)] #[doc(hidden)]
pub struct Enumerate<S> { #[allow(missing_debug_implementations)]
stream: S, pub struct Enumerate<S> {
i: usize, #[pin]
stream: S,
i: usize,
}
} }
impl<S> Enumerate<S> { impl<S> Enumerate<S> {
pin_utils::unsafe_pinned!(stream: S);
pin_utils::unsafe_unpinned!(i: usize);
pub(super) fn new(stream: S) -> Self { pub(super) fn new(stream: S) -> Self {
Enumerate { stream, i: 0 } Enumerate { stream, i: 0 }
} }
@ -25,13 +27,14 @@ where
{ {
type Item = (usize, S::Item); type Item = (usize, S::Item);
fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> { fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
let next = futures_core::ready!(self.as_mut().stream().poll_next(cx)); let this = self.project();
let next = futures_core::ready!(this.stream.poll_next(cx));
match next { match next {
Some(v) => { Some(v) => {
let ret = (self.i, v); let ret = (*this.i, v);
*self.as_mut().i() += 1; *this.i += 1;
Poll::Ready(Some(ret)) Poll::Ready(Some(ret))
} }
None => Poll::Ready(None), None => Poll::Ready(None),

27
src/stream/stream/filter.rs
Unescape Escape View File

@ -1,21 +1,23 @@
use std::marker::PhantomData; use std::marker::PhantomData;
use std::pin::Pin; use std::pin::Pin;
use pin_project_lite::pin_project;
use crate::stream::Stream; use crate::stream::Stream;
use crate::task::{Context, Poll}; use crate::task::{Context, Poll};
/// A stream to filter elements of another stream with a predicate. pin_project! {
#[derive(Debug)] /// A stream to filter elements of another stream with a predicate.
pub struct Filter<S, P, T> { #[derive(Debug)]
stream: S, pub struct Filter<S, P, T> {
predicate: P, #[pin]
__t: PhantomData<T>, stream: S,
predicate: P,
__t: PhantomData<T>,
}
} }
impl<S, P, T> Filter<S, P, T> { impl<S, P, T> Filter<S, P, T> {
pin_utils::unsafe_pinned!(stream: S);
pin_utils::unsafe_unpinned!(predicate: P);
pub(super) fn new(stream: S, predicate: P) -> Self { pub(super) fn new(stream: S, predicate: P) -> Self {
Filter { Filter {
stream, stream,
@ -32,11 +34,12 @@ where
{ {
type Item = S::Item; type Item = S::Item;
fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> { fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
let next = futures_core::ready!(self.as_mut().stream().poll_next(cx)); let this = self.project();
let next = futures_core::ready!(this.stream.poll_next(cx));
match next { match next {
Some(v) if (self.as_mut().predicate())(&v) => Poll::Ready(Some(v)), Some(v) if (this.predicate)(&v) => Poll::Ready(Some(v)),
Some(_) => { Some(_) => {
cx.waker().wake_by_ref(); cx.waker().wake_by_ref();
Poll::Pending Poll::Pending

29
src/stream/stream/filter_map.rs
Unescape Escape View File

@ -2,21 +2,23 @@ use std::marker::PhantomData;
use std::pin::Pin; use std::pin::Pin;
use std::task::{Context, Poll}; use std::task::{Context, Poll};
use pin_project_lite::pin_project;
use crate::stream::Stream; use crate::stream::Stream;
#[doc(hidden)] pin_project! {
#[allow(missing_debug_implementations)] #[doc(hidden)]
pub struct FilterMap<S, F, T, B> { #[allow(missing_debug_implementations)]
stream: S, pub struct FilterMap<S, F, T, B> {
f: F, #[pin]
__from: PhantomData<T>, stream: S,
__to: PhantomData<B>, f: F,
__from: PhantomData<T>,
__to: PhantomData<B>,
}
} }
impl<S, F, T, B> FilterMap<S, F, T, B> { impl<S, F, T, B> FilterMap<S, F, T, B> {
pin_utils::unsafe_pinned!(stream: S);
pin_utils::unsafe_unpinned!(f: F);
pub(crate) fn new(stream: S, f: F) -> Self { pub(crate) fn new(stream: S, f: F) -> Self {
FilterMap { FilterMap {
stream, stream,
@ -34,10 +36,11 @@ where
{ {
type Item = B; type Item = B;
fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> { fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
let next = futures_core::ready!(self.as_mut().stream().poll_next(cx)); let this = self.project();
let next = futures_core::ready!(this.stream.poll_next(cx));
match next { match next {
Some(v) => match (self.as_mut().f())(v) { Some(v) => match (this.f)(v) {
Some(b) => Poll::Ready(Some(b)), Some(b) => Poll::Ready(Some(b)),
None => { None => {
cx.waker().wake_by_ref(); cx.waker().wake_by_ref();

36
src/stream/stream/fold.rs
Unescape Escape View File

@ -1,24 +1,25 @@
use std::marker::PhantomData; use std::marker::PhantomData;
use std::pin::Pin; use std::pin::Pin;
use pin_project_lite::pin_project;
use crate::future::Future; use crate::future::Future;
use crate::stream::Stream; use crate::stream::Stream;
use crate::task::{Context, Poll}; use crate::task::{Context, Poll};
#[doc(hidden)] pin_project! {
#[allow(missing_debug_implementations)] #[doc(hidden)]
pub struct FoldFuture<S, F, T, B> { #[allow(missing_debug_implementations)]
stream: S, pub struct FoldFuture<S, F, T, B> {
f: F, #[pin]
acc: Option<B>, stream: S,
__t: PhantomData<T>, f: F,
acc: Option<B>,
__t: PhantomData<T>,
}
} }
impl<S, F, T, B> FoldFuture<S, F, T, B> { impl<S, F, T, B> FoldFuture<S, F, T, B> {
pin_utils::unsafe_pinned!(stream: S);
pin_utils::unsafe_unpinned!(f: F);
pin_utils::unsafe_unpinned!(acc: Option<B>);
pub(super) fn new(stream: S, init: B, f: F) -> Self { pub(super) fn new(stream: S, init: B, f: F) -> Self {
FoldFuture { FoldFuture {
stream, stream,
@ -36,17 +37,18 @@ where
{ {
type Output = B; type Output = B;
fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> { fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
let mut this = self.project();
loop { loop {
let next = futures_core::ready!(self.as_mut().stream().poll_next(cx)); let next = futures_core::ready!(this.stream.as_mut().poll_next(cx));
match next { match next {
Some(v) => { Some(v) => {
let old = self.as_mut().acc().take().unwrap(); let old = this.acc.take().unwrap();
let new = (self.as_mut().f())(old, v); let new = (this.f)(old, v);
*self.as_mut().acc() = Some(new); *this.acc = Some(new);
} }
None => return Poll::Ready(self.as_mut().acc().take().unwrap()), None => return Poll::Ready(this.acc.take().unwrap()),
} }
} }
} }

27
src/stream/stream/for_each.rs
Unescape Escape View File

@ -1,22 +1,24 @@
use std::marker::PhantomData; use std::marker::PhantomData;
use std::pin::Pin; use std::pin::Pin;
use pin_project_lite::pin_project;
use crate::future::Future; use crate::future::Future;
use crate::stream::Stream; use crate::stream::Stream;
use crate::task::{Context, Poll}; use crate::task::{Context, Poll};
#[doc(hidden)] pin_project! {
#[allow(missing_debug_implementations)] #[doc(hidden)]
pub struct ForEachFuture<S, F, T> { #[allow(missing_debug_implementations)]
stream: S, pub struct ForEachFuture<S, F, T> {
f: F, #[pin]
__t: PhantomData<T>, stream: S,
f: F,
__t: PhantomData<T>,
}
} }
impl<S, F, T> ForEachFuture<S, F, T> { impl<S, F, T> ForEachFuture<S, F, T> {
pin_utils::unsafe_pinned!(stream: S);
pin_utils::unsafe_unpinned!(f: F);
pub(super) fn new(stream: S, f: F) -> Self { pub(super) fn new(stream: S, f: F) -> Self {
ForEachFuture { ForEachFuture {
stream, stream,
@ -33,12 +35,13 @@ where
{ {
type Output = (); type Output = ();
fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> { fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
let mut this = self.project();
loop { loop {
let next = futures_core::ready!(self.as_mut().stream().poll_next(cx)); let next = futures_core::ready!(this.stream.as_mut().poll_next(cx));
match next { match next {
Some(v) => (self.as_mut().f())(v), Some(v) => (this.f)(v),
None => return Poll::Ready(()), None => return Poll::Ready(()),
} }
} }

33
src/stream/stream/fuse.rs
Unescape Escape View File

@ -1,33 +1,32 @@
use std::pin::Pin; use std::pin::Pin;
use pin_project_lite::pin_project;
use crate::stream::Stream; use crate::stream::Stream;
use crate::task::{Context, Poll}; use crate::task::{Context, Poll};
/// A `Stream` that is permanently closed once a single call to `poll` results in pin_project! {
/// `Poll::Ready(None)`, returning `Poll::Ready(None)` for all future calls to `poll`. /// A `Stream` that is permanently closed once a single call to `poll` results in
#[derive(Clone, Debug)] /// `Poll::Ready(None)`, returning `Poll::Ready(None)` for all future calls to `poll`.
pub struct Fuse<S> { #[derive(Clone, Debug)]
pub(crate) stream: S, pub struct Fuse<S> {
pub(crate) done: bool, #[pin]
} pub(crate) stream: S,
pub(crate) done: bool,
impl<S: Unpin> Unpin for Fuse<S> {} }
impl<S: Stream> Fuse<S> {
pin_utils::unsafe_pinned!(stream: S);
pin_utils::unsafe_unpinned!(done: bool);
} }
impl<S: Stream> Stream for Fuse<S> { impl<S: Stream> Stream for Fuse<S> {
type Item = S::Item; type Item = S::Item;
fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<S::Item>> { fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<S::Item>> {
if self.done { let this = self.project();
if *this.done {
Poll::Ready(None) Poll::Ready(None)
} else { } else {
let next = futures_core::ready!(self.as_mut().stream().poll_next(cx)); let next = futures_core::ready!(this.stream.poll_next(cx));
if next.is_none() { if next.is_none() {
*self.as_mut().done() = true; *this.done = true;
} }
Poll::Ready(next) Poll::Ready(next)
} }

27
src/stream/stream/ge.rs
Unescape Escape View File

@ -1,26 +1,29 @@
use std::cmp::Ordering; use std::cmp::Ordering;
use std::pin::Pin; use std::pin::Pin;
use pin_project_lite::pin_project;
use super::partial_cmp::PartialCmpFuture; use super::partial_cmp::PartialCmpFuture;
use crate::future::Future; use crate::future::Future;
use crate::prelude::*; use crate::prelude::*;
use crate::stream::Stream; use crate::stream::Stream;
use crate::task::{Context, Poll}; use crate::task::{Context, Poll};
// Determines if the elements of this `Stream` are lexicographically pin_project! {
// greater than or equal to those of another. // Determines if the elements of this `Stream` are lexicographically
#[doc(hidden)] // greater than or equal to those of another.
#[allow(missing_debug_implementations)] #[doc(hidden)]
pub struct GeFuture<L: Stream, R: Stream> { #[allow(missing_debug_implementations)]
partial_cmp: PartialCmpFuture<L, R>, pub struct GeFuture<L: Stream, R: Stream> {
#[pin]
partial_cmp: PartialCmpFuture<L, R>,
}
} }
impl<L: Stream, R: Stream> GeFuture<L, R> impl<L: Stream, R: Stream> GeFuture<L, R>
where where
L::Item: PartialOrd<R::Item>, L::Item: PartialOrd<R::Item>,
{ {
pin_utils::unsafe_pinned!(partial_cmp: PartialCmpFuture<L, R>);
pub(super) fn new(l: L, r: R) -> Self { pub(super) fn new(l: L, r: R) -> Self {
GeFuture { GeFuture {
partial_cmp: l.partial_cmp(r), partial_cmp: l.partial_cmp(r),
@ -30,14 +33,14 @@ where
impl<L: Stream, R: Stream> Future for GeFuture<L, R> impl<L: Stream, R: Stream> Future for GeFuture<L, R>
where where
L: Stream + Sized, L: Stream,
R: Stream + Sized, R: Stream,
L::Item: PartialOrd<R::Item>, L::Item: PartialOrd<R::Item>,
{ {
type Output = bool; type Output = bool;
fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> { fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
let result = futures_core::ready!(self.as_mut().partial_cmp().poll(cx)); let result = futures_core::ready!(self.project().partial_cmp.poll(cx));
match result { match result {
Some(Ordering::Greater) | Some(Ordering::Equal) => Poll::Ready(true), Some(Ordering::Greater) | Some(Ordering::Equal) => Poll::Ready(true),

23
src/stream/stream/gt.rs
Unescape Escape View File

@ -1,26 +1,29 @@
use std::cmp::Ordering; use std::cmp::Ordering;
use std::pin::Pin; use std::pin::Pin;
use pin_project_lite::pin_project;
use super::partial_cmp::PartialCmpFuture; use super::partial_cmp::PartialCmpFuture;
use crate::future::Future; use crate::future::Future;
use crate::prelude::*; use crate::prelude::*;
use crate::stream::Stream; use crate::stream::Stream;
use crate::task::{Context, Poll}; use crate::task::{Context, Poll};
// Determines if the elements of this `Stream` are lexicographically pin_project! {
// greater than those of another. // Determines if the elements of this `Stream` are lexicographically
#[doc(hidden)] // greater than those of another.
#[allow(missing_debug_implementations)] #[doc(hidden)]
pub struct GtFuture<L: Stream, R: Stream> { #[allow(missing_debug_implementations)]
partial_cmp: PartialCmpFuture<L, R>, pub struct GtFuture<L: Stream, R: Stream> {
#[pin]
partial_cmp: PartialCmpFuture<L, R>,
}
} }
impl<L: Stream, R: Stream> GtFuture<L, R> impl<L: Stream, R: Stream> GtFuture<L, R>
where where
L::Item: PartialOrd<R::Item>, L::Item: PartialOrd<R::Item>,
{ {
pin_utils::unsafe_pinned!(partial_cmp: PartialCmpFuture<L, R>);
pub(super) fn new(l: L, r: R) -> Self { pub(super) fn new(l: L, r: R) -> Self {
GtFuture { GtFuture {
partial_cmp: l.partial_cmp(r), partial_cmp: l.partial_cmp(r),
@ -36,8 +39,8 @@ where
{ {
type Output = bool; type Output = bool;
fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> { fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
let result = futures_core::ready!(self.as_mut().partial_cmp().poll(cx)); let result = futures_core::ready!(self.project().partial_cmp.poll(cx));
match result { match result {
Some(Ordering::Greater) => Poll::Ready(true), Some(Ordering::Greater) => Poll::Ready(true),

27
src/stream/stream/inspect.rs
Unescape Escape View File

@ -1,21 +1,23 @@
use std::marker::PhantomData; use std::marker::PhantomData;
use std::pin::Pin; use std::pin::Pin;
use pin_project_lite::pin_project;
use crate::stream::Stream; use crate::stream::Stream;
use crate::task::{Context, Poll}; use crate::task::{Context, Poll};
/// A stream that does something with each element of another stream. pin_project! {
#[derive(Debug)] /// A stream that does something with each element of another stream.
pub struct Inspect<S, F, T> { #[derive(Debug)]
stream: S, pub struct Inspect<S, F, T> {
f: F, #[pin]
__t: PhantomData<T>, stream: S,
f: F,
__t: PhantomData<T>,
}
} }
impl<S, F, T> Inspect<S, F, T> { impl<S, F, T> Inspect<S, F, T> {
pin_utils::unsafe_pinned!(stream: S);
pin_utils::unsafe_unpinned!(f: F);
pub(super) fn new(stream: S, f: F) -> Self { pub(super) fn new(stream: S, f: F) -> Self {
Inspect { Inspect {
stream, stream,
@ -32,11 +34,12 @@ where
{ {
type Item = S::Item; type Item = S::Item;
fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> { fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
let next = futures_core::ready!(self.as_mut().stream().poll_next(cx)); let mut this = self.project();
let next = futures_core::ready!(this.stream.as_mut().poll_next(cx));
Poll::Ready(next.and_then(|x| { Poll::Ready(next.and_then(|x| {
(self.as_mut().f())(&x); (this.f)(&x);
Some(x) Some(x)
})) }))
} }

27
src/stream/stream/last.rs
Unescape Escape View File

@ -1,20 +1,22 @@
use std::pin::Pin; use std::pin::Pin;
use pin_project_lite::pin_project;
use crate::future::Future; use crate::future::Future;
use crate::stream::Stream; use crate::stream::Stream;
use crate::task::{Context, Poll}; use crate::task::{Context, Poll};
#[doc(hidden)] pin_project! {
#[allow(missing_debug_implementations)] #[doc(hidden)]
pub struct LastFuture<S, T> { #[allow(missing_debug_implementations)]
stream: S, pub struct LastFuture<S, T> {
last: Option<T>, #[pin]
stream: S,
last: Option<T>,
}
} }
impl<S, T> LastFuture<S, T> { impl<S, T> LastFuture<S, T> {
pin_utils::unsafe_pinned!(stream: S);
pin_utils::unsafe_unpinned!(last: Option<T>);
pub(crate) fn new(stream: S) -> Self { pub(crate) fn new(stream: S) -> Self {
LastFuture { stream, last: None } LastFuture { stream, last: None }
} }
@ -27,16 +29,17 @@ where
{ {
type Output = Option<S::Item>; type Output = Option<S::Item>;
fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> { fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
let next = futures_core::ready!(self.as_mut().stream().poll_next(cx)); let this = self.project();
let next = futures_core::ready!(this.stream.poll_next(cx));
match next { match next {
Some(new) => { Some(new) => {
cx.waker().wake_by_ref(); cx.waker().wake_by_ref();
*self.as_mut().last() = Some(new); *this.last = Some(new);
Poll::Pending Poll::Pending
} }
None => Poll::Ready(self.last), None => Poll::Ready(*this.last),
} }
} }
} }

23
src/stream/stream/le.rs
Unescape Escape View File

@ -1,26 +1,29 @@
use std::cmp::Ordering; use std::cmp::Ordering;
use std::pin::Pin; use std::pin::Pin;
use pin_project_lite::pin_project;
use super::partial_cmp::PartialCmpFuture; use super::partial_cmp::PartialCmpFuture;
use crate::future::Future; use crate::future::Future;
use crate::prelude::*; use crate::prelude::*;
use crate::stream::Stream; use crate::stream::Stream;
use crate::task::{Context, Poll}; use crate::task::{Context, Poll};
/// Determines if the elements of this `Stream` are lexicographically pin_project! {
/// less or equal to those of another. /// Determines if the elements of this `Stream` are lexicographically
#[doc(hidden)] /// less or equal to those of another.
#[allow(missing_debug_implementations)] #[doc(hidden)]
pub struct LeFuture<L: Stream, R: Stream> { #[allow(missing_debug_implementations)]
partial_cmp: PartialCmpFuture<L, R>, pub struct LeFuture<L: Stream, R: Stream> {
#[pin]
partial_cmp: PartialCmpFuture<L, R>,
}
} }
impl<L: Stream, R: Stream> LeFuture<L, R> impl<L: Stream, R: Stream> LeFuture<L, R>
where where
L::Item: PartialOrd<R::Item>, L::Item: PartialOrd<R::Item>,
{ {
pin_utils::unsafe_pinned!(partial_cmp: PartialCmpFuture<L, R>);
pub(super) fn new(l: L, r: R) -> Self { pub(super) fn new(l: L, r: R) -> Self {
LeFuture { LeFuture {
partial_cmp: l.partial_cmp(r), partial_cmp: l.partial_cmp(r),
@ -36,8 +39,8 @@ where
{ {
type Output = bool; type Output = bool;
fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> { fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
let result = futures_core::ready!(self.as_mut().partial_cmp().poll(cx)); let result = futures_core::ready!(self.project().partial_cmp.poll(cx));
match result { match result {
Some(Ordering::Less) | Some(Ordering::Equal) => Poll::Ready(true), Some(Ordering::Less) | Some(Ordering::Equal) => Poll::Ready(true),

23
src/stream/stream/lt.rs
Unescape Escape View File

@ -1,26 +1,29 @@
use std::cmp::Ordering; use std::cmp::Ordering;
use std::pin::Pin; use std::pin::Pin;
use pin_project_lite::pin_project;
use super::partial_cmp::PartialCmpFuture; use super::partial_cmp::PartialCmpFuture;
use crate::future::Future; use crate::future::Future;
use crate::prelude::*; use crate::prelude::*;
use crate::stream::Stream; use crate::stream::Stream;
use crate::task::{Context, Poll}; use crate::task::{Context, Poll};
// Determines if the elements of this `Stream` are lexicographically pin_project! {
// less than those of another. // Determines if the elements of this `Stream` are lexicographically
#[doc(hidden)] // less than those of another.
#[allow(missing_debug_implementations)] #[doc(hidden)]
pub struct LtFuture<L: Stream, R: Stream> { #[allow(missing_debug_implementations)]
partial_cmp: PartialCmpFuture<L, R>, pub struct LtFuture<L: Stream, R: Stream> {
#[pin]
partial_cmp: PartialCmpFuture<L, R>,
}
} }
impl<L: Stream, R: Stream> LtFuture<L, R> impl<L: Stream, R: Stream> LtFuture<L, R>
where where
L::Item: PartialOrd<R::Item>, L::Item: PartialOrd<R::Item>,
{ {
pin_utils::unsafe_pinned!(partial_cmp: PartialCmpFuture<L, R>);
pub(super) fn new(l: L, r: R) -> Self { pub(super) fn new(l: L, r: R) -> Self {
LtFuture { LtFuture {
partial_cmp: l.partial_cmp(r), partial_cmp: l.partial_cmp(r),
@ -36,8 +39,8 @@ where
{ {
type Output = bool; type Output = bool;
fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> { fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
let result = futures_core::ready!(self.as_mut().partial_cmp().poll(cx)); let result = futures_core::ready!(self.project().partial_cmp.poll(cx));
match result { match result {
Some(Ordering::Less) => Poll::Ready(true), Some(Ordering::Less) => Poll::Ready(true),

29
src/stream/stream/map.rs
Unescape Escape View File

@ -1,22 +1,24 @@
use std::marker::PhantomData; use std::marker::PhantomData;
use std::pin::Pin; use std::pin::Pin;
use pin_project_lite::pin_project;
use crate::stream::Stream; use crate::stream::Stream;
use crate::task::{Context, Poll}; use crate::task::{Context, Poll};
#[doc(hidden)] pin_project! {
#[allow(missing_debug_implementations)] #[doc(hidden)]
pub struct Map<S, F, T, B> { #[allow(missing_debug_implementations)]
stream: S, pub struct Map<S, F, T, B> {
f: F, #[pin]
__from: PhantomData<T>, stream: S,
__to: PhantomData<B>, f: F,
__from: PhantomData<T>,
__to: PhantomData<B>,
}
} }
impl<S, F, T, B> Map<S, F, T, B> { impl<S, F, T, B> Map<S, F, T, B> {
pin_utils::unsafe_pinned!(stream: S);
pin_utils::unsafe_unpinned!(f: F);
pub(crate) fn new(stream: S, f: F) -> Self { pub(crate) fn new(stream: S, f: F) -> Self {
Map { Map {
stream, stream,
@ -34,8 +36,9 @@ where
{ {
type Item = B; type Item = B;
fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> { fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
let next = futures_core::ready!(self.as_mut().stream().poll_next(cx)); let this = self.project();
Poll::Ready(next.map(self.as_mut().f())) let next = futures_core::ready!(this.stream.poll_next(cx));
Poll::Ready(next.map(this.f))
} }
} }

40
src/stream/stream/merge.rs
Unescape Escape View File

@ -2,22 +2,25 @@ use std::pin::Pin;
use std::task::{Context, Poll}; use std::task::{Context, Poll};
use futures_core::Stream; use futures_core::Stream;
use pin_project_lite::pin_project;
/// A stream that merges two other streams into a single stream. pin_project! {
/// /// A stream that merges two other streams into a single stream.
/// This stream is returned by [`Stream::merge`]. ///
/// /// This stream is returned by [`Stream::merge`].
/// [`Stream::merge`]: trait.Stream.html#method.merge ///
#[cfg(feature = "unstable")] /// [`Stream::merge`]: trait.Stream.html#method.merge
#[cfg_attr(feature = "docs", doc(cfg(unstable)))] #[cfg(feature = "unstable")]
#[derive(Debug)] #[cfg_attr(feature = "docs", doc(cfg(unstable)))]
pub struct Merge<L, R> { #[derive(Debug)]
left: L, pub struct Merge<L, R> {
right: R, #[pin]
left: L,
#[pin]
right: R,
}
} }
impl<L, R> Unpin for Merge<L, R> {}
impl<L, R> Merge<L, R> { impl<L, R> Merge<L, R> {
pub(crate) fn new(left: L, right: R) -> Self { pub(crate) fn new(left: L, right: R) -> Self {
Self { left, right } Self { left, right }
@ -26,19 +29,20 @@ impl<L, R> Merge<L, R> {
impl<L, R, T> Stream for Merge<L, R> impl<L, R, T> Stream for Merge<L, R>
where where
L: Stream<Item = T> + Unpin, L: Stream<Item = T>,
R: Stream<Item = T> + Unpin, R: Stream<Item = T>,
{ {
type Item = T; type Item = T;
fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> { fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
if let Poll::Ready(Some(item)) = Pin::new(&mut self.left).poll_next(cx) { let this = self.project();
if let Poll::Ready(Some(item)) = this.left.poll_next(cx) {
// The first stream made progress. The Merge needs to be polled // The first stream made progress. The Merge needs to be polled
// again to check the progress of the second stream. // again to check the progress of the second stream.
cx.waker().wake_by_ref(); cx.waker().wake_by_ref();
Poll::Ready(Some(item)) Poll::Ready(Some(item))
} else { } else {
Pin::new(&mut self.right).poll_next(cx) this.right.poll_next(cx)
} }
} }
} }

38
src/stream/stream/min_by.rs
Unescape Escape View File

@ -1,23 +1,24 @@
use std::cmp::Ordering; use std::cmp::Ordering;
use std::pin::Pin; use std::pin::Pin;
use pin_project_lite::pin_project;
use crate::future::Future; use crate::future::Future;
use crate::stream::Stream; use crate::stream::Stream;
use crate::task::{Context, Poll}; use crate::task::{Context, Poll};
#[doc(hidden)] pin_project! {
#[allow(missing_debug_implementations)] #[doc(hidden)]
pub struct MinByFuture<S, F, T> { #[allow(missing_debug_implementations)]
stream: S, pub struct MinByFuture<S, F, T> {
compare: F, #[pin]
min: Option<T>, stream: S,
compare: F,
min: Option<T>,
}
} }
impl<S, F, T> MinByFuture<S, F, T> { impl<S, F, T> MinByFuture<S, F, T> {
pin_utils::unsafe_pinned!(stream: S);
pin_utils::unsafe_unpinned!(compare: F);
pin_utils::unsafe_unpinned!(min: Option<T>);
pub(super) fn new(stream: S, compare: F) -> Self { pub(super) fn new(stream: S, compare: F) -> Self {
MinByFuture { MinByFuture {
stream, stream,
@ -35,22 +36,23 @@ where
{ {
type Output = Option<S::Item>; type Output = Option<S::Item>;
fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> { fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
let next = futures_core::ready!(self.as_mut().stream().poll_next(cx)); let this = self.project();
let next = futures_core::ready!(this.stream.poll_next(cx));
match next { match next {
Some(new) => { Some(new) => {
cx.waker().wake_by_ref(); cx.waker().wake_by_ref();
match self.as_mut().min().take() { match this.min.take() {
None => *self.as_mut().min() = Some(new), None => *this.min = Some(new),
Some(old) => match (&mut self.as_mut().compare())(&new, &old) { Some(old) => match (this.compare)(&new, &old) {
Ordering::Less => *self.as_mut().min() = Some(new), Ordering::Less => *this.min = Some(new),
_ => *self.as_mut().min() = Some(old), _ => *this.min = Some(old),
}, },
} }
Poll::Pending Poll::Pending
} }
None => Poll::Ready(self.min), None => Poll::Ready(*this.min),
} }
} }
} }

58
src/stream/stream/partial_cmp.rs
Unescape Escape View File

@ -1,29 +1,30 @@
use std::cmp::Ordering; use std::cmp::Ordering;
use std::pin::Pin; use std::pin::Pin;
use pin_project_lite::pin_project;
use super::fuse::Fuse; use super::fuse::Fuse;
use crate::future::Future; use crate::future::Future;
use crate::prelude::*; use crate::prelude::*;
use crate::stream::Stream; use crate::stream::Stream;
use crate::task::{Context, Poll}; use crate::task::{Context, Poll};
// Lexicographically compares the elements of this `Stream` with those pin_project! {
// of another. // Lexicographically compares the elements of this `Stream` with those
#[doc(hidden)] // of another.
#[allow(missing_debug_implementations)] #[doc(hidden)]
pub struct PartialCmpFuture<L: Stream, R: Stream> { #[allow(missing_debug_implementations)]
l: Fuse<L>, pub struct PartialCmpFuture<L: Stream, R: Stream> {
r: Fuse<R>, #[pin]
l_cache: Option<L::Item>, l: Fuse<L>,
r_cache: Option<R::Item>, #[pin]
r: Fuse<R>,
l_cache: Option<L::Item>,
r_cache: Option<R::Item>,
}
} }
impl<L: Stream, R: Stream> PartialCmpFuture<L, R> { impl<L: Stream, R: Stream> PartialCmpFuture<L, R> {
pin_utils::unsafe_pinned!(l: Fuse<L>);
pin_utils::unsafe_pinned!(r: Fuse<R>);
pin_utils::unsafe_unpinned!(l_cache: Option<L::Item>);
pin_utils::unsafe_unpinned!(r_cache: Option<R::Item>);
pub(super) fn new(l: L, r: R) -> Self { pub(super) fn new(l: L, r: R) -> Self {
PartialCmpFuture { PartialCmpFuture {
l: l.fuse(), l: l.fuse(),
@ -42,12 +43,13 @@ where
{ {
type Output = Option<Ordering>; type Output = Option<Ordering>;
fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> { fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
let mut this = self.project();
loop { loop {
// Short circuit logic // Short circuit logic
// Stream that completes earliest can be considered Less, etc // Stream that completes earliest can be considered Less, etc
let l_complete = self.l.done && self.as_mut().l_cache.is_none(); let l_complete = this.l.done && this.l_cache.is_none();
let r_complete = self.r.done && self.as_mut().r_cache.is_none(); let r_complete = this.r.done && this.r_cache.is_none();
if l_complete && r_complete { if l_complete && r_complete {
return Poll::Ready(Some(Ordering::Equal)); return Poll::Ready(Some(Ordering::Equal));
@ -58,30 +60,30 @@ where
} }
// Get next value if possible and necesary // Get next value if possible and necesary
if !self.l.done && self.as_mut().l_cache.is_none() { if !this.l.done && this.l_cache.is_none() {
let l_next = futures_core::ready!(self.as_mut().l().poll_next(cx)); let l_next = futures_core::ready!(this.l.as_mut().poll_next(cx));
if let Some(item) = l_next { if let Some(item) = l_next {
*self.as_mut().l_cache() = Some(item); *this.l_cache = Some(item);
} }
} }
if !self.r.done && self.as_mut().r_cache.is_none() { if !this.r.done && this.r_cache.is_none() {
let r_next = futures_core::ready!(self.as_mut().r().poll_next(cx)); let r_next = futures_core::ready!(this.r.as_mut().poll_next(cx));
if let Some(item) = r_next { if let Some(item) = r_next {
*self.as_mut().r_cache() = Some(item); *this.r_cache = Some(item);
} }
} }
// Compare if both values are available. // Compare if both values are available.
if self.as_mut().l_cache.is_some() && self.as_mut().r_cache.is_some() { if this.l_cache.is_some() && this.r_cache.is_some() {
let l_value = self.as_mut().l_cache().take().unwrap(); let l_value = this.l_cache.as_mut().take().unwrap();
let r_value = self.as_mut().r_cache().take().unwrap(); let r_value = this.r_cache.as_mut().take().unwrap();
let result = l_value.partial_cmp(&r_value); let result = l_value.partial_cmp(&r_value);
if let Some(Ordering::Equal) = result { if let Some(Ordering::Equal) = result {
// Reset cache to prepare for next comparison // Reset cache to prepare for next comparison
*self.as_mut().l_cache() = None; *this.l_cache = None;
*self.as_mut().r_cache() = None; *this.r_cache = None;
} else { } else {
// Return non equal value // Return non equal value
return Poll::Ready(result); return Poll::Ready(result);

27
src/stream/stream/scan.rs
Unescape Escape View File

@ -1,13 +1,18 @@
use std::pin::Pin; use std::pin::Pin;
use pin_project_lite::pin_project;
use crate::stream::Stream; use crate::stream::Stream;
use crate::task::{Context, Poll}; use crate::task::{Context, Poll};
/// A stream to maintain state while polling another stream. pin_project! {
#[derive(Debug)] /// A stream to maintain state while polling another stream.
pub struct Scan<S, St, F> { #[derive(Debug)]
stream: S, pub struct Scan<S, St, F> {
state_f: (St, F), #[pin]
stream: S,
state_f: (St, F),
}
} }
impl<S, St, F> Scan<S, St, F> { impl<S, St, F> Scan<S, St, F> {
@ -17,13 +22,8 @@ impl<S, St, F> Scan<S, St, F> {
state_f: (initial_state, f), state_f: (initial_state, f),
} }
} }
pin_utils::unsafe_pinned!(stream: S);
pin_utils::unsafe_unpinned!(state_f: (St, F));
} }
impl<S: Unpin, St, F> Unpin for Scan<S, St, F> {}
impl<S, St, F, B> Stream for Scan<S, St, F> impl<S, St, F, B> Stream for Scan<S, St, F>
where where
S: Stream, S: Stream,
@ -31,11 +31,12 @@ where
{ {
type Item = B; type Item = B;
fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<B>> { fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<B>> {
let poll_result = self.as_mut().stream().poll_next(cx); let mut this = self.project();
let poll_result = this.stream.as_mut().poll_next(cx);
poll_result.map(|item| { poll_result.map(|item| {
item.and_then(|item| { item.and_then(|item| {
let (state, f) = self.as_mut().state_f(); let (state, f) = this.state_f;
f(state, item) f(state, item)
}) })
}) })

27
src/stream/stream/skip.rs
Unescape Escape View File

@ -1,19 +1,21 @@
use std::pin::Pin; use std::pin::Pin;
use std::task::{Context, Poll}; use std::task::{Context, Poll};
use pin_project_lite::pin_project;
use crate::stream::Stream; use crate::stream::Stream;
/// A stream to skip first n elements of another stream. pin_project! {
#[derive(Debug)] /// A stream to skip first n elements of another stream.
pub struct Skip<S> { #[derive(Debug)]
stream: S, pub struct Skip<S> {
n: usize, #[pin]
stream: S,
n: usize,
}
} }
impl<S> Skip<S> { impl<S> Skip<S> {
pin_utils::unsafe_pinned!(stream: S);
pin_utils::unsafe_unpinned!(n: usize);
pub(crate) fn new(stream: S, n: usize) -> Self { pub(crate) fn new(stream: S, n: usize) -> Self {
Skip { stream, n } Skip { stream, n }
} }
@ -25,14 +27,15 @@ where
{ {
type Item = S::Item; type Item = S::Item;
fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> { fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
let mut this = self.project();
loop { loop {
let next = futures_core::ready!(self.as_mut().stream().poll_next(cx)); let next = futures_core::ready!(this.stream.as_mut().poll_next(cx));
match next { match next {
Some(v) => match self.n { Some(v) => match *this.n {
0 => return Poll::Ready(Some(v)), 0 => return Poll::Ready(Some(v)),
_ => *self.as_mut().n() -= 1, _ => *this.n -= 1,
}, },
None => return Poll::Ready(None), None => return Poll::Ready(None),
} }

29
src/stream/stream/skip_while.rs
Unescape Escape View File

@ -1,21 +1,23 @@
use std::marker::PhantomData; use std::marker::PhantomData;
use std::pin::Pin; use std::pin::Pin;
use pin_project_lite::pin_project;
use crate::stream::Stream; use crate::stream::Stream;
use crate::task::{Context, Poll}; use crate::task::{Context, Poll};
/// A stream to skip elements of another stream based on a predicate. pin_project! {
#[derive(Debug)] /// A stream to skip elements of another stream based on a predicate.
pub struct SkipWhile<S, P, T> { #[derive(Debug)]
stream: S, pub struct SkipWhile<S, P, T> {
predicate: Option<P>, #[pin]
__t: PhantomData<T>, stream: S,
predicate: Option<P>,
__t: PhantomData<T>,
}
} }
impl<S, P, T> SkipWhile<S, P, T> { impl<S, P, T> SkipWhile<S, P, T> {
pin_utils::unsafe_pinned!(stream: S);
pin_utils::unsafe_unpinned!(predicate: Option<P>);
pub(crate) fn new(stream: S, predicate: P) -> Self { pub(crate) fn new(stream: S, predicate: P) -> Self {
SkipWhile { SkipWhile {
stream, stream,
@ -32,15 +34,16 @@ where
{ {
type Item = S::Item; type Item = S::Item;
fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> { fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
let mut this = self.project();
loop { loop {
let next = futures_core::ready!(self.as_mut().stream().poll_next(cx)); let next = futures_core::ready!(this.stream.as_mut().poll_next(cx));
match next { match next {
Some(v) => match self.as_mut().predicate() { Some(v) => match this.predicate {
Some(p) => { Some(p) => {
if !p(&v) { if !p(&v) {
*self.as_mut().predicate() = None; *this.predicate = None;
return Poll::Ready(Some(v)); return Poll::Ready(Some(v));
} }
} }

32
src/stream/stream/step_by.rs
Unescape Escape View File

@ -1,21 +1,22 @@
use std::pin::Pin; use std::pin::Pin;
use pin_project_lite::pin_project;
use crate::stream::Stream; use crate::stream::Stream;
use crate::task::{Context, Poll}; use crate::task::{Context, Poll};
/// A stream that steps a given amount of elements of another stream. pin_project! {
#[derive(Debug)] /// A stream that steps a given amount of elements of another stream.
pub struct StepBy<S> { #[derive(Debug)]
stream: S, pub struct StepBy<S> {
step: usize, #[pin]
i: usize, stream: S,
step: usize,
i: usize,
}
} }
impl<S> StepBy<S> { impl<S> StepBy<S> {
pin_utils::unsafe_pinned!(stream: S);
pin_utils::unsafe_unpinned!(step: usize);
pin_utils::unsafe_unpinned!(i: usize);
pub(crate) fn new(stream: S, step: usize) -> Self { pub(crate) fn new(stream: S, step: usize) -> Self {
StepBy { StepBy {
stream, stream,
@ -31,17 +32,18 @@ where
{ {
type Item = S::Item; type Item = S::Item;
fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> { fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
let mut this = self.project();
loop { loop {
let next = futures_core::ready!(self.as_mut().stream().poll_next(cx)); let next = futures_core::ready!(this.stream.as_mut().poll_next(cx));
match next { match next {
Some(v) => match self.i { Some(v) => match this.i {
0 => { 0 => {
*self.as_mut().i() = self.step; *this.i = *this.step;
return Poll::Ready(Some(v)); return Poll::Ready(Some(v));
} }
_ => *self.as_mut().i() -= 1, _ => *this.i -= 1,
}, },
None => return Poll::Ready(None), None => return Poll::Ready(None),
} }

33
src/stream/stream/take.rs
Unescape Escape View File

@ -1,33 +1,32 @@
use std::pin::Pin; use std::pin::Pin;
use pin_project_lite::pin_project;
use crate::stream::Stream; use crate::stream::Stream;
use crate::task::{Context, Poll}; use crate::task::{Context, Poll};
/// A stream that yields the first `n` items of another stream. pin_project! {
#[derive(Clone, Debug)] /// A stream that yields the first `n` items of another stream.
pub struct Take<S> { #[derive(Clone, Debug)]
pub(crate) stream: S, pub struct Take<S> {
pub(crate) remaining: usize, #[pin]
} pub(crate) stream: S,
pub(crate) remaining: usize,
impl<S: Unpin> Unpin for Take<S> {} }
impl<S: Stream> Take<S> {
pin_utils::unsafe_pinned!(stream: S);
pin_utils::unsafe_unpinned!(remaining: usize);
} }
impl<S: Stream> Stream for Take<S> { impl<S: Stream> Stream for Take<S> {
type Item = S::Item; type Item = S::Item;
fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<S::Item>> { fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<S::Item>> {
if self.remaining == 0 { let this = self.project();
if *this.remaining == 0 {
Poll::Ready(None) Poll::Ready(None)
} else { } else {
let next = futures_core::ready!(self.as_mut().stream().poll_next(cx)); let next = futures_core::ready!(this.stream.poll_next(cx));
match next { match next {
Some(_) => *self.as_mut().remaining() -= 1, Some(_) => *this.remaining -= 1,
None => *self.as_mut().remaining() = 0, None => *this.remaining = 0,
} }
Poll::Ready(next) Poll::Ready(next)
} }

27
src/stream/stream/take_while.rs
Unescape Escape View File

@ -1,21 +1,23 @@
use std::marker::PhantomData; use std::marker::PhantomData;
use std::pin::Pin; use std::pin::Pin;
use pin_project_lite::pin_project;
use crate::stream::Stream; use crate::stream::Stream;
use crate::task::{Context, Poll}; use crate::task::{Context, Poll};
/// A stream that yields elements based on a predicate. pin_project! {
#[derive(Debug)] /// A stream that yields elements based on a predicate.
pub struct TakeWhile<S, P, T> { #[derive(Debug)]
stream: S, pub struct TakeWhile<S, P, T> {
predicate: P, #[pin]
__t: PhantomData<T>, stream: S,
predicate: P,
__t: PhantomData<T>,
}
} }
impl<S, P, T> TakeWhile<S, P, T> { impl<S, P, T> TakeWhile<S, P, T> {
pin_utils::unsafe_pinned!(stream: S);
pin_utils::unsafe_unpinned!(predicate: P);
pub(super) fn new(stream: S, predicate: P) -> Self { pub(super) fn new(stream: S, predicate: P) -> Self {
TakeWhile { TakeWhile {
stream, stream,
@ -32,11 +34,12 @@ where
{ {
type Item = S::Item; type Item = S::Item;
fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> { fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
let next = futures_core::ready!(self.as_mut().stream().poll_next(cx)); let this = self.project();
let next = futures_core::ready!(this.stream.poll_next(cx));
match next { match next {
Some(v) if (self.as_mut().predicate())(&v) => Poll::Ready(Some(v)), Some(v) if (this.predicate)(&v) => Poll::Ready(Some(v)),
Some(_) => { Some(_) => {
cx.waker().wake_by_ref(); cx.waker().wake_by_ref();
Poll::Pending Poll::Pending

38
src/stream/stream/try_fold.rs
Unescape Escape View File

@ -1,24 +1,25 @@
use std::marker::PhantomData; use std::marker::PhantomData;
use std::pin::Pin; use std::pin::Pin;
use pin_project_lite::pin_project;
use crate::future::Future; use crate::future::Future;
use crate::stream::Stream; use crate::stream::Stream;
use crate::task::{Context, Poll}; use crate::task::{Context, Poll};
#[doc(hidden)] pin_project! {
#[allow(missing_debug_implementations)] #[doc(hidden)]
pub struct TryFoldFuture<S, F, T> { #[allow(missing_debug_implementations)]
stream: S, pub struct TryFoldFuture<S, F, T> {
f: F, #[pin]
acc: Option<T>, stream: S,
__t: PhantomData<T>, f: F,
acc: Option<T>,
__t: PhantomData<T>,
}
} }
impl<S, F, T> TryFoldFuture<S, F, T> { impl<S, F, T> TryFoldFuture<S, F, T> {
pin_utils::unsafe_pinned!(stream: S);
pin_utils::unsafe_unpinned!(f: F);
pin_utils::unsafe_unpinned!(acc: Option<T>);
pub(super) fn new(stream: S, init: T, f: F) -> Self { pub(super) fn new(stream: S, init: T, f: F) -> Self {
TryFoldFuture { TryFoldFuture {
stream, stream,
@ -36,23 +37,22 @@ where
{ {
type Output = Result<T, E>; type Output = Result<T, E>;
fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> { fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
let mut this = self.project();
loop { loop {
let next = futures_core::ready!(self.as_mut().stream().poll_next(cx)); let next = futures_core::ready!(this.stream.as_mut().poll_next(cx));
match next { match next {
Some(v) => { Some(v) => {
let old = self.as_mut().acc().take().unwrap(); let old = this.acc.take().unwrap();
let new = (self.as_mut().f())(old, v); let new = (this.f)(old, v);
match new { match new {
Ok(o) => { Ok(o) => *this.acc = Some(o),
*self.as_mut().acc() = Some(o);
}
Err(e) => return Poll::Ready(Err(e)), Err(e) => return Poll::Ready(Err(e)),
} }
} }
None => return Poll::Ready(Ok(self.as_mut().acc().take().unwrap())), None => return Poll::Ready(Ok(this.acc.take().unwrap())),
} }
} }
} }

29
src/stream/stream/try_for_each.rs
Unescape Escape View File

@ -1,23 +1,25 @@
use std::marker::PhantomData; use std::marker::PhantomData;
use std::pin::Pin; use std::pin::Pin;
use pin_project_lite::pin_project;
use crate::future::Future; use crate::future::Future;
use crate::stream::Stream; use crate::stream::Stream;
use crate::task::{Context, Poll}; use crate::task::{Context, Poll};
#[doc(hidden)] pin_project! {
#[allow(missing_debug_implementations)] #[doc(hidden)]
pub struct TryForEeachFuture<S, F, T, R> { #[allow(missing_debug_implementations)]
stream: S, pub struct TryForEeachFuture<S, F, T, R> {
f: F, #[pin]
__from: PhantomData<T>, stream: S,
__to: PhantomData<R>, f: F,
__from: PhantomData<T>,
__to: PhantomData<R>,
}
} }
impl<S, F, T, R> TryForEeachFuture<S, F, T, R> { impl<S, F, T, R> TryForEeachFuture<S, F, T, R> {
pin_utils::unsafe_pinned!(stream: S);
pin_utils::unsafe_unpinned!(f: F);
pub(crate) fn new(stream: S, f: F) -> Self { pub(crate) fn new(stream: S, f: F) -> Self {
TryForEeachFuture { TryForEeachFuture {
stream, stream,
@ -36,14 +38,15 @@ where
{ {
type Output = Result<(), E>; type Output = Result<(), E>;
fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> { fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
let mut this = self.project();
loop { loop {
let item = futures_core::ready!(self.as_mut().stream().poll_next(cx)); let item = futures_core::ready!(this.stream.as_mut().poll_next(cx));
match item { match item {
None => return Poll::Ready(Ok(())), None => return Poll::Ready(Ok(())),
Some(v) => { Some(v) => {
let res = (self.as_mut().f())(v); let res = (this.f)(v);
if let Err(e) = res { if let Err(e) = res {
return Poll::Ready(Err(e)); return Poll::Ready(Err(e));
} }

35
src/stream/stream/zip.rs
Unescape Escape View File

@ -1,14 +1,20 @@
use std::fmt; use std::fmt;
use std::pin::Pin; use std::pin::Pin;
use pin_project_lite::pin_project;
use crate::stream::Stream; use crate::stream::Stream;
use crate::task::{Context, Poll}; use crate::task::{Context, Poll};
/// An iterator that iterates two other iterators simultaneously. pin_project! {
pub struct Zip<A: Stream, B> { /// An iterator that iterates two other iterators simultaneously.
item_slot: Option<A::Item>, pub struct Zip<A: Stream, B> {
first: A, item_slot: Option<A::Item>,
second: B, #[pin]
first: A,
#[pin]
second: B,
}
} }
impl<A: Stream + fmt::Debug, B: fmt::Debug> fmt::Debug for Zip<A, B> { impl<A: Stream + fmt::Debug, B: fmt::Debug> fmt::Debug for Zip<A, B> {
@ -20,8 +26,6 @@ impl<A: Stream + fmt::Debug, B: fmt::Debug> fmt::Debug for Zip<A, B> {
} }
} }
impl<A: Stream + Unpin, B: Unpin> Unpin for Zip<A, B> {}
impl<A: Stream, B> Zip<A, B> { impl<A: Stream, B> Zip<A, B> {
pub(crate) fn new(first: A, second: B) -> Self { pub(crate) fn new(first: A, second: B) -> Self {
Zip { Zip {
@ -30,25 +34,22 @@ impl<A: Stream, B> Zip<A, B> {
second, second,
} }
} }
pin_utils::unsafe_unpinned!(item_slot: Option<A::Item>);
pin_utils::unsafe_pinned!(first: A);
pin_utils::unsafe_pinned!(second: B);
} }
impl<A: Stream, B: Stream> Stream for Zip<A, B> { impl<A: Stream, B: Stream> Stream for Zip<A, B> {
type Item = (A::Item, B::Item); type Item = (A::Item, B::Item);
fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> { fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
if self.as_mut().item_slot().is_none() { let this = self.project();
match self.as_mut().first().poll_next(cx) { if this.item_slot.is_none() {
match this.first.poll_next(cx) {
Poll::Pending => return Poll::Pending, Poll::Pending => return Poll::Pending,
Poll::Ready(None) => return Poll::Ready(None), Poll::Ready(None) => return Poll::Ready(None),
Poll::Ready(Some(item)) => *self.as_mut().item_slot() = Some(item), Poll::Ready(Some(item)) => *this.item_slot = Some(item),
} }
} }
let second_item = futures_core::ready!(self.as_mut().second().poll_next(cx)); let second_item = futures_core::ready!(this.second.poll_next(cx));
let first_item = self.as_mut().item_slot().take().unwrap(); let first_item = this.item_slot.take().unwrap();
Poll::Ready(second_item.map(|second_item| (first_item, second_item))) Poll::Ready(second_item.map(|second_item| (first_item, second_item)))
} }
} }

14
src/task/block_on.rs
Unescape Escape View File

@ -7,6 +7,7 @@ use std::task::{RawWaker, RawWakerVTable};
use std::thread; use std::thread;
use crossbeam_utils::sync::Parker; use crossbeam_utils::sync::Parker;
use pin_project_lite::pin_project;
use super::task; use super::task;
use super::task_local; use super::task_local;
@ -100,19 +101,18 @@ where
} }
} }
struct CatchUnwindFuture<F> { pin_project! {
future: F, struct CatchUnwindFuture<F> {
} #[pin]
future: F,
impl<F> CatchUnwindFuture<F> { }
pin_utils::unsafe_pinned!(future: F);
} }
impl<F: Future + UnwindSafe> Future for CatchUnwindFuture<F> { impl<F: Future + UnwindSafe> Future for CatchUnwindFuture<F> {
type Output = thread::Result<F::Output>; type Output = thread::Result<F::Output>;
fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> { fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
panic::catch_unwind(AssertUnwindSafe(|| self.future().poll(cx)))?.map(Ok) panic::catch_unwind(AssertUnwindSafe(|| self.project().future.poll(cx)))?.map(Ok)
} }
} }

Write Preview
Loading…
Cancel
Save