Merge branch 'master' into poll_next

.travis.yml

@@ -1,6 +1,16 @@
 language: rust
 
-env: RUSTFLAGS="-D warnings"
+env:
+  - RUSTFLAGS="-D warnings"
+
+# Cache the whole `~/.cargo` directory to keep `~/cargo/.crates.toml`.
+cache:
+  directories:
+    - /home/travis/.cargo
+
+# Don't cache the cargo registry because it's too big.
+before_cache:
+  - rm -rf /home/travis/.cargo/registry
 
 matrix:
   fast_finish: true
@@ -35,16 +45,15 @@ matrix:
       script:
         - cargo doc --features docs
 
-    # TODO(yoshuawuyts): re-enable mdbook
+    - name: book
-    # - name: book
+      rust: nightly
-    #   rust: nightly
+      os: linux
-    #   os: linux
+      before_script:
-    #   before_script:
+        - test -x $HOME/.cargo/bin/mdbook || ./ci/install-mdbook.sh
-    #     - test -x $HOME/.cargo/bin/mdbook || ./ci/install-mdbook.sh
+        - cargo build # to find 'extern crate async_std' by `mdbook test`
-    #     - cargo build # to find 'extern crate async_std' by `mdbook test`
+      script:
-    #   script:
+        - mdbook build docs
-    #     - mdbook build docs
+        - mdbook test -L ./target/debug/deps docs
-    #     - mdbook test -L ./target/debug/deps docs
 
 script:
   - cargo check --features unstable --all --benches --bins --examples --tests

Cargo.toml

@@ -44,6 +44,10 @@ femme = "1.2.0"
 surf = "1.0.2"
 tempdir = "0.3.7"
 
+# These are used by the book for examples
+futures-channel-preview = "0.3.0-alpha.18"
+futures-util-preview = "0.3.0-alpha.18"
+
 [dev-dependencies.futures-preview]
 version = "0.3.0-alpha.18"
 features = ["std", "nightly", "async-await"]

docs/src/tutorial/all_together.md

@@ -4,17 +4,16 @@ At this point, we only need to start the broker to get a fully-functioning (in t
 
 ```rust,edition2018
 # extern crate async_std;
-# extern crate futures;
+# extern crate futures_channel;
+# extern crate futures_util;
 use async_std::{
     io::{self, BufReader},
     net::{TcpListener, TcpStream, ToSocketAddrs},
     prelude::*,
     task,
 };
-use futures::{
+use futures_channel::mpsc;
-    channel::mpsc,
+use futures_util::SinkExt;
-    SinkExt,
-};
 use std::{
     collections::hash_map::{HashMap, Entry},
     sync::Arc,

docs/src/tutorial/clean_shutdown.md

@@ -22,17 +22,16 @@ Let's add waiting to the server:
 
 ```rust,edition2018
 # extern crate async_std;
-# extern crate futures;
+# extern crate futures_channel;
+# extern crate futures_util;
 # use async_std::{
 #     io::{self, BufReader},
 #     net::{TcpListener, TcpStream, ToSocketAddrs},
 #     prelude::*,
 #     task,
 # };
-# use futures::{
+# use futures_channel::mpsc;
-#     channel::mpsc,
+# use futures_util::SinkExt;
-#     SinkExt,
-# };
 # use std::{
 #     collections::hash_map::{HashMap, Entry},
 #     sync::Arc,
@@ -156,17 +155,16 @@ And to the broker:
 
 ```rust,edition2018
 # extern crate async_std;
-# extern crate futures;
+# extern crate futures_channel;
+# extern crate futures_util;
 # use async_std::{
 #     io::{self, BufReader},
 #     net::{TcpListener, TcpStream, ToSocketAddrs},
 #     prelude::*,
 #     task,
 # };
-# use futures::{
+# use futures_channel::mpsc;
-#     channel::mpsc,
+# use futures_util::SinkExt;
-#     SinkExt,
-# };
 # use std::{
 #     collections::hash_map::{HashMap, Entry},
 #     sync::Arc,

docs/src/tutorial/connecting_readers_and_writers.md

@@ -12,15 +12,16 @@ The order of events "Bob sends message to Alice" and "Alice joins" is determined
 
 ```rust,edition2018
 # extern crate async_std;
-# extern crate futures;
+# extern crate futures_channel;
+# extern crate futures_util;
 # use async_std::{
 #     io::{Write},
 #     net::TcpStream,
 #     prelude::{Future, Stream},
 #     task,
 # };
-# use futures::channel::mpsc;
+# use futures_channel::mpsc;
-# use futures::sink::SinkExt;
+# use futures_util::sink::SinkExt;
 # use std::sync::Arc;
 #
 # type Result<T> = std::result::Result<T, Box<dyn std::error::Error + Send + Sync>>;

docs/src/tutorial/handling_disconnection.md

@@ -19,9 +19,11 @@ First, let's add a shutdown channel to the `client`:
 
 ```rust,edition2018
 # extern crate async_std;
-# extern crate futures;
+# extern crate futures_channel;
+# extern crate futures_util;
 # use async_std::net::TcpStream;
-# use futures::{channel::mpsc, SinkExt};
+# use futures_channel::mpsc;
+# use futures_util::SinkExt;
 # use std::sync::Arc;
 #
 # type Result<T> = std::result::Result<T, Box<dyn std::error::Error + Send + Sync>>;
@@ -68,9 +70,11 @@ We use the `select` macro for this purpose:
 
 ```rust,edition2018
 # extern crate async_std;
-# extern crate futures;
+# extern crate futures_channel;
+# extern crate futures_util;
 # use async_std::{io::Write, net::TcpStream};
-use futures::{channel::mpsc, select, FutureExt, StreamExt};
+use futures_channel::mpsc;
+use futures_util::{select, FutureExt, StreamExt};
 # use std::sync::Arc;
 
 # type Receiver<T> = mpsc::UnboundedReceiver<T>;
@@ -118,15 +122,18 @@ The final code looks like this:
 
 ```rust,edition2018
 # extern crate async_std;
-# extern crate futures;
+# extern crate futures_channel;
+# extern crate futures_util;
 use async_std::{
     io::{BufReader, BufRead, Write},
     net::{TcpListener, TcpStream, ToSocketAddrs},
     task,
 };
-use futures::{channel::mpsc, future::Future, select, FutureExt, SinkExt, StreamExt};
+use futures_channel::mpsc;
+use futures_util::{select, FutureExt, SinkExt, StreamExt};
 use std::{
     collections::hash_map::{Entry, HashMap},
+    future::Future,
     sync::Arc,
 };
 

docs/src/tutorial/implementing_a_client.md

@@ -16,13 +16,13 @@ With async, we can just use the `select!` macro.
 
 ```rust,edition2018
 # extern crate async_std;
-# extern crate futures;
+# extern crate futures_util;
 use async_std::{
     io::{stdin, BufRead, BufReader, Write},
     net::{TcpStream, ToSocketAddrs},
     task,
 };
-use futures::{select, FutureExt, StreamExt};
+use futures_util::{select, FutureExt, StreamExt};
 
 type Result<T> = std::result::Result<T, Box<dyn std::error::Error + Send + Sync>>;
 

docs/src/tutorial/sending_messages.md

@@ -13,14 +13,15 @@ if Alice and Charley send two messages to Bob at the same time, Bob will see the
 
 ```rust,edition2018
 # extern crate async_std;
-# extern crate futures;
+# extern crate futures_channel;
+# extern crate futures_util;
 # use async_std::{
 #     io::Write,
 #     net::TcpStream,
 #     prelude::Stream,
 # };
-use futures::channel::mpsc; // 1
+use futures_channel::mpsc; // 1
-use futures::sink::SinkExt;
+use futures_util::sink::SinkExt;
 use std::sync::Arc;
 
 # type Result<T> = std::result::Result<T, Box<dyn std::error::Error + Send + Sync>>;

src/io/buf_read/fill_buf.rs

@@ -0,0 +1,32 @@
+use std::pin::Pin;
+
+use futures_io::AsyncBufRead;
+
+use crate::future::Future;
+use crate::io;
+use crate::task::{Context, Poll};
+
+#[doc(hidden)]
+#[allow(missing_debug_implementations)]
+pub struct FillBufFuture<'a, R: ?Sized> {
+    reader: &'a mut R,
+}
+
+impl<'a, R: ?Sized> FillBufFuture<'a, R> {
+    pub(crate) fn new(reader: &'a mut R) -> Self {
+        Self { reader }
+    }
+}
+
+impl<'a, R: AsyncBufRead + Unpin + ?Sized> Future for FillBufFuture<'a, R> {
+    type Output = io::Result<&'a [u8]>;
+
+    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<&'a [u8]>> {
+        let Self { reader } = &mut *self;
+        let result = Pin::new(reader).poll_fill_buf(cx);
+        // This is safe because:
+        // 1. The buffer is valid for the lifetime of the reader.
+        // 2. Output is unrelated to the wrapper (Self).
+        result.map_ok(|buf| unsafe { std::mem::transmute::<&'_ [u8], &'a [u8]>(buf) })
+    }
+}

src/io/buf_read/mod.rs

@@ -1,7 +1,9 @@
+mod fill_buf;
 mod lines;
 mod read_line;
 mod read_until;
 
+use fill_buf::FillBufFuture;
 pub use lines::Lines;
 use read_line::ReadLineFuture;
 use read_until::ReadUntilFuture;
@@ -41,6 +43,26 @@ cfg_if! {
 /// [`futures::io::AsyncBufRead`]:
 /// https://docs.rs/futures-preview/0.3.0-alpha.17/futures/io/trait.AsyncBufRead.html
 pub trait BufRead {
+    /// Returns the contents of the internal buffer, filling it with more data from the inner
+    /// reader if it is empty.
+    ///
+    /// This function is a lower-level call. It needs to be paired with the [`consume`] method to
+    /// function properly. When calling this method, none of the contents will be "read" in the
+    /// sense that later calling `read` may return the same contents. As such, [`consume`] must be
+    /// called with the number of bytes that are consumed from this buffer to ensure that the bytes
+    /// are never returned twice.
+    ///
+    /// [`consume`]: #tymethod.consume
+    ///
+    /// An empty buffer returned indicates that the stream has reached EOF.
+    // TODO: write a proper doctest with `consume`
+    fn fill_buf<'a>(&'a mut self) -> ret!('a, FillBufFuture, io::Result<&'a [u8]>)
+    where
+        Self: Unpin,
+    {
+        FillBufFuture::new(self)
+    }
+
     /// Reads all bytes into `buf` until the delimiter `byte` or EOF is reached.
     ///
     /// This function will read bytes from the underlying stream until the delimiter or EOF is

src/io/write/mod.rs

@@ -127,6 +127,8 @@ pub trait Write {
     /// #
     /// # Ok(()) }) }
     /// ```
+    ///
+    /// [`write`]: #tymethod.write
     fn write_all<'a>(&'a mut self, buf: &'a [u8]) -> ret!('a, WriteAllFuture, io::Result<()>)
     where
         Self: Unpin,

src/stream/stream/filter_map.rs

@@ -0,0 +1,48 @@
+use std::marker::PhantomData;
+use std::pin::Pin;
+use std::task::{Context, Poll};
+
+/// A stream that both filters and maps.
+#[derive(Clone, Debug)]
+pub struct FilterMap<S, F, T, B> {
+    stream: S,
+    f: F,
+    __from: PhantomData<T>,
+    __to: PhantomData<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 {
+        FilterMap {
+            stream,
+            f,
+            __from: PhantomData,
+            __to: PhantomData,
+        }
+    }
+}
+
+impl<S, F, B> futures_core::stream::Stream for FilterMap<S, F, S::Item, B>
+where
+    S: futures_core::stream::Stream,
+    F: FnMut(S::Item) -> Option<B>,
+{
+    type Item = B;
+
+    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+        let next = futures_core::ready!(self.as_mut().stream().poll_next(cx));
+        match next {
+            Some(v) => match (self.as_mut().f())(v) {
+                Some(b) => Poll::Ready(Some(b)),
+                None => {
+                    cx.waker().wake_by_ref();
+                    Poll::Pending
+                }
+            },
+            None => Poll::Ready(None),
+        }
+    }
+}

src/stream/stream/find.rs

@@ -0,0 +1,49 @@
+use crate::task::{Context, Poll};
+use std::marker::PhantomData;
+use std::pin::Pin;
+
+#[doc(hidden)]
+#[allow(missing_debug_implementations)]
+pub struct FindFuture<'a, S, P, T> {
+    stream: &'a mut S,
+    p: P,
+    __t: PhantomData<T>,
+}
+
+impl<'a, S, P, T> FindFuture<'a, S, P, T> {
+    pin_utils::unsafe_pinned!(stream: &'a mut S);
+    pin_utils::unsafe_unpinned!(p: P);
+
+    pub(super) fn new(stream: &'a mut S, p: P) -> Self {
+        FindFuture {
+            stream,
+            p,
+            __t: PhantomData,
+        }
+    }
+}
+
+impl<'a, S, P> futures_core::future::Future for FindFuture<'a, S, P, S::Item>
+where
+    S: futures_core::stream::Stream + Unpin + Sized,
+    P: FnMut(&S::Item) -> bool,
+{
+    type Output = Option<S::Item>;
+
+    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        use futures_core::stream::Stream;
+
+        let item = futures_core::ready!(self.as_mut().stream().poll_next(cx));
+
+        match item {
+            Some(v) => match (self.as_mut().p())(&v) {
+                true => Poll::Ready(Some(v)),
+                false => {
+                    cx.waker().wake_by_ref();
+                    Poll::Pending
+                }
+            },
+            None => Poll::Ready(None),
+        }
+    }
+}

src/stream/stream/find_map.rs

@@ -0,0 +1,50 @@
+use std::marker::PhantomData;
+use std::pin::Pin;
+use std::task::{Context, Poll};
+
+#[allow(missing_debug_implementations)]
+pub struct FindMapFuture<'a, S, F, T, B> {
+    stream: &'a mut S,
+    f: F,
+    __b: PhantomData<B>,
+    __t: PhantomData<T>,
+}
+
+impl<'a, S, B, F, T> FindMapFuture<'a, S, F, T, B> {
+    pin_utils::unsafe_pinned!(stream: &'a mut S);
+    pin_utils::unsafe_unpinned!(f: F);
+
+    pub(super) fn new(stream: &'a mut S, f: F) -> Self {
+        FindMapFuture {
+            stream,
+            f,
+            __b: PhantomData,
+            __t: PhantomData,
+        }
+    }
+}
+
+impl<'a, S, B, F> futures_core::future::Future for FindMapFuture<'a, S, F, S::Item, B>
+where
+    S: futures_core::stream::Stream + Unpin + Sized,
+    F: FnMut(S::Item) -> Option<B>,
+{
+    type Output = Option<B>;
+
+    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        use futures_core::stream::Stream;
+
+        let item = futures_core::ready!(self.as_mut().stream().poll_next(cx));
+
+        match item {
+            Some(v) => match (self.as_mut().f())(v) {
+                Some(v) => Poll::Ready(Some(v)),
+                None => {
+                    cx.waker().wake_by_ref();
+                    Poll::Pending
+                }
+            },
+            None => Poll::Ready(None),
+        }
+    }
+}

src/stream/stream/min_by.rs

@@ -5,18 +5,19 @@ use crate::future::Future;
 use crate::stream::Stream;
 use crate::task::{Context, Poll};
 
-/// A future that yields the minimum item in a stream by a given comparison function.
 #[doc(hidden)]
 #[allow(missing_debug_implementations)]
-pub struct MinByFuture<S: Stream, F> {
+pub struct MinByFuture<S, F, T> {
     stream: S,
     compare: F,
-    min: Option<S::Item>,
+    min: Option<T>,
 }
 
-impl<S: Stream + Unpin, F> Unpin for MinByFuture<S, F> {}
+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>);
 
-impl<S: Stream + Unpin, F> MinByFuture<S, F> {
     pub(super) fn new(stream: S, compare: F) -> Self {
         MinByFuture {
             stream,
@@ -26,25 +27,25 @@ impl<S: Stream + Unpin, F> MinByFuture<S, F> {
     }
 }
 
-impl<S, F> Future for MinByFuture<S, F>
+impl<S, F> Future for MinByFuture<S, F, S::Item>
 where
-    S: Stream + Unpin,
+    S: Stream + Unpin + Sized,
     S::Item: Copy,
     F: FnMut(&S::Item, &S::Item) -> Ordering,
 {
     type Output = Option<S::Item>;
 
     fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
-        let next = futures_core::ready!(Pin::new(&mut self.stream).poll_next(cx));
+        let next = futures_core::ready!(self.as_mut().stream().poll_next(cx));
 
         match next {
             Some(new) => {
                 cx.waker().wake_by_ref();
-                match self.as_mut().min.take() {
+                match self.as_mut().min().take() {
-                    None => self.as_mut().min = Some(new),
+                    None => *self.as_mut().min() = Some(new),
-                    Some(old) => match (&mut self.as_mut().compare)(&new, &old) {
+                    Some(old) => match (&mut self.as_mut().compare())(&new, &old) {
-                        Ordering::Less => self.as_mut().min = Some(new),
+                        Ordering::Less => *self.as_mut().min() = Some(new),
-                        _ => self.as_mut().min = Some(old),
+                        _ => *self.as_mut().min() = Some(old),
                     },
                 }
                 Poll::Pending

src/stream/stream/mod.rs

@@ -23,16 +23,24 @@
 
 mod all;
 mod any;
+mod filter_map;
+mod find;
+mod find_map;
 mod min_by;
 mod next;
+mod nth;
 mod take;
 
 pub use take::Take;
 
 use all::AllFuture;
 use any::AnyFuture;
+use filter_map::FilterMap;
+use find::FindFuture;
+use find_map::FindMapFuture;
 use min_by::MinByFuture;
 use next::NextFuture;
+use nth::NthFuture;
 
 use std::cmp::Ordering;
 use std::marker::PhantomData;
@@ -51,6 +59,7 @@ cfg_if! {
             ($a:lifetime, $f:tt, $o:ty) => (ImplFuture<$a, $o>);
             ($a:lifetime, $f:tt, $o:ty, $t1:ty) => (ImplFuture<$a, $o>);
             ($a:lifetime, $f:tt, $o:ty, $t1:ty, $t2:ty) => (ImplFuture<$a, $o>);
+            ($a:lifetime, $f:tt, $o:ty, $t1:ty, $t2:ty, $t3:ty) => (ImplFuture<$a, $o>);
 
         }
     } else {
@@ -58,6 +67,7 @@ cfg_if! {
             ($a:lifetime, $f:tt, $o:ty) => ($f<$a, Self>);
             ($a:lifetime, $f:tt, $o:ty, $t1:ty) => ($f<$a, Self, $t1>);
             ($a:lifetime, $f:tt, $o:ty, $t1:ty, $t2:ty) => ($f<$a, Self, $t1, $t2>);
+            ($a:lifetime, $f:tt, $o:ty, $t1:ty, $t2:ty, $t3:ty) => ($f<$a, Self, $t1, $t2, $t3>);
         }
     }
 }
@@ -183,6 +193,43 @@ pub trait Stream {
         }
     }
 
+    /// Both filters and maps a stream.
+    ///
+    /// # Examples
+    ///
+    /// Basic usage:
+    ///
+    /// ```
+    /// # fn main() { async_std::task::block_on(async {
+    /// #
+    /// use std::collections::VecDeque;
+    /// use async_std::stream::Stream;
+    ///
+    /// let s: VecDeque<&str> = vec!["1", "lol", "3", "NaN", "5"].into_iter().collect();
+    ///
+    /// let mut parsed = s.filter_map(|a| a.parse::<u32>().ok());
+    ///
+    /// let one = parsed.next().await;
+    /// assert_eq!(one, Some(1));
+    ///
+    /// let three = parsed.next().await;
+    /// assert_eq!(three, Some(3));
+    ///
+    /// let five = parsed.next().await;
+    /// assert_eq!(five, Some(5));
+    ///
+    /// let end = parsed.next().await;
+    /// assert_eq!(end, None);
+    /// #
+    /// # }) }
+    fn filter_map<B, F>(self, f: F) -> FilterMap<Self, F, Self::Item, B>
+    where
+        Self: Sized,
+        F: FnMut(Self::Item) -> Option<B>,
+    {
+        FilterMap::new(self, f)
+    }
+
     /// Returns the element that gives the minimum value with respect to the
     /// specified comparison function. If several elements are equally minimum,
     /// the first element is returned. If the stream is empty, `None` is returned.
@@ -208,14 +255,72 @@ pub trait Stream {
     /// #
     /// # }) }
     /// ```
-    fn min_by<F>(self, compare: F) -> MinByFuture<Self, F>
+    fn min_by<F>(self, compare: F) -> MinByFuture<Self, F, Self::Item>
     where
-        Self: Sized + Unpin,
+        Self: Sized,
         F: FnMut(&Self::Item, &Self::Item) -> Ordering,
     {
         MinByFuture::new(self, compare)
     }
 
+    /// Returns the nth element of the stream.
+    ///
+    /// # Examples
+    ///
+    /// Basic usage:
+    ///
+    /// ```
+    /// # fn main() { async_std::task::block_on(async {
+    /// #
+    /// use std::collections::VecDeque;
+    /// use async_std::stream::Stream;
+    ///
+    /// let mut s: VecDeque<usize> = vec![1, 2, 3].into_iter().collect();
+    ///
+    /// let second = s.nth(1).await;
+    /// assert_eq!(second, Some(2));
+    /// #
+    /// # }) }
+    /// ```
+    /// Calling `nth()` multiple times:
+    ///
+    /// ```
+    /// # fn main() { async_std::task::block_on(async {
+    /// #
+    /// use std::collections::VecDeque;
+    /// use async_std::stream::Stream;
+    ///
+    /// let mut s: VecDeque<usize> = vec![1, 2, 3].into_iter().collect();
+    ///
+    /// let second = s.nth(0).await;
+    /// assert_eq!(second, Some(1));
+    ///
+    /// let second = s.nth(0).await;
+    /// assert_eq!(second, Some(2));
+    /// #
+    /// # }) }
+    /// ```
+    /// Returning `None` if the stream finished before returning `n` elements:
+    /// ```
+    /// # fn main() { async_std::task::block_on(async {
+    /// #
+    /// use std::collections::VecDeque;
+    /// use async_std::stream::Stream;
+    ///
+    /// let mut s: VecDeque<usize> = vec![1, 2, 3].into_iter().collect();
+    ///
+    /// let fourth = s.nth(4).await;
+    /// assert_eq!(fourth, None);
+    /// #
+    /// # }) }
+    /// ```
+    fn nth(&mut self, n: usize) -> ret!('_, NthFuture, Option<Self::Item>)
+    where
+        Self: Sized,
+    {
+        NthFuture::new(self, n)
+    }
+
     /// Tests if every element of the stream matches a predicate.
     ///
     /// `all()` takes a closure that returns `true` or `false`. It applies
@@ -273,6 +378,73 @@ pub trait Stream {
         }
     }
 
+    /// Searches for an element in a stream that satisfies a predicate.
+    ///
+    /// # Examples
+    ///
+    /// Basic usage:
+    ///
+    /// ```
+    /// # fn main() { async_std::task::block_on(async {
+    /// #
+    /// use async_std::prelude::*;
+    /// use std::collections::VecDeque;
+    ///
+    /// let mut s: VecDeque<usize> = vec![1, 2, 3].into_iter().collect();
+    /// let res = s.find(|x| *x == 2).await;
+    /// assert_eq!(res, Some(2));
+    /// #
+    /// # }) }
+    /// ```
+    ///
+    /// Resuming after a first find:
+    ///
+    /// ```
+    /// # fn main() { async_std::task::block_on(async {
+    /// #
+    /// use async_std::prelude::*;
+    /// use std::collections::VecDeque;
+    ///
+    /// let mut s: VecDeque<usize> = vec![1, 2, 3].into_iter().collect();
+    /// let res = s.find(|x| *x == 2).await;
+    /// assert_eq!(res, Some(2));
+    ///
+    /// let next = s.next().await;
+    /// assert_eq!(next, Some(3));
+    /// #
+    /// # }) }
+    /// ```
+    fn find<P>(&mut self, p: P) -> ret!('_, FindFuture, Option<Self::Item>, P, Self::Item)
+    where
+        Self: Sized,
+        P: FnMut(&Self::Item) -> bool,
+    {
+        FindFuture::new(self, p)
+    }
+
+    /// Applies function to the elements of stream and returns the first non-none result.
+    ///
+    /// ```
+    /// # fn main() { async_std::task::block_on(async {
+    /// #
+    /// use async_std::prelude::*;
+    /// use std::collections::VecDeque;
+    ///
+    /// let mut s: VecDeque<&str> = vec!["lol", "NaN", "2", "5"].into_iter().collect();
+    /// let first_number = s.find_map(|s| s.parse().ok()).await;
+    ///
+    /// assert_eq!(first_number, Some(2));
+    /// #
+    /// # }) }
+    /// ```
+    fn find_map<F, B>(&mut self, f: F) -> ret!('_, FindMapFuture, Option<B>, F, Self::Item, B)
+    where
+        Self: Sized,
+        F: FnMut(Self::Item) -> Option<B>,
+    {
+        FindMapFuture::new(self, f)
+    }
+
     /// Tests if any element of the stream matches a predicate.
     ///
     /// `any()` takes a closure that returns `true` or `false`. It applies

src/stream/stream/nth.rs

@@ -0,0 +1,41 @@
+use std::pin::Pin;
+use std::task::{Context, Poll};
+
+#[allow(missing_debug_implementations)]
+pub struct NthFuture<'a, S> {
+    stream: &'a mut S,
+    n: usize,
+}
+
+impl<'a, S> NthFuture<'a, S> {
+    pin_utils::unsafe_pinned!(stream: &'a mut S);
+    pin_utils::unsafe_unpinned!(n: usize);
+
+    pub(crate) fn new(stream: &'a mut S, n: usize) -> Self {
+        NthFuture { stream, n }
+    }
+}
+
+impl<'a, S> futures_core::future::Future for NthFuture<'a, S>
+where
+    S: futures_core::stream::Stream + Unpin + Sized,
+{
+    type Output = Option<S::Item>;
+
+    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        use futures_core::stream::Stream;
+
+        let next = futures_core::ready!(self.as_mut().stream().poll_next(cx));
+        match next {
+            Some(v) => match self.n {
+                0 => Poll::Ready(Some(v)),
+                _ => {
+                    *self.as_mut().n() -= 1;
+                    cx.waker().wake_by_ref();
+                    Poll::Pending
+                }
+            },
+            None => Poll::Ready(None),
+        }
+    }
+}