futures-async-combinators

FOR LEARNING PURPOSES ONLY

This is a greatly simplified implementation of Future combinators like FutureExt::map, TryFutureExt::and_then...

Requirements

Rust nightly-2019-08-21 for async_await.

State

Future

• future::and_then
• future::err_into
• future::flatten
• future::flatten_stream
• future::inspect
• future::into_stream
• future::map
• future::map_err
• future::map_ok
• future::or_else
• future::poll_fn
• future::ready
• future::then
• future::unwrap_or_else

Stream

• stream::chain
• stream::collect
• stream::concat
• stream::filter
• stream::filter_map
• stream::flatten
• stream::fold
• stream::for_each
• stream::into_future
• stream::iter
• stream::map
• stream::next
• stream::poll_fn
• stream::repeat
• stream::skip
• stream::skip_while
• stream::take
• stream::take_while
• stream::then
• stream::unfold
• stream::zip

Why

To understand how combinators work by looking at clean source code. Compare:

pub async fn then<...>(future, f) -> ...
{
    let new_future = f(future.await);
    new_future.await
}

with original

impl Then
{
    fn poll(self, waker) -> Poll<...> {
        self.as_mut().chain().poll(waker, |output, f| f(output))
    }
}

impl Chain
{
    fn poll(self, waker, f) -> Poll<...>
    {
        let mut f = Some(f);

        // Safe to call `get_unchecked_mut` because we won't move the futures.
        let this = unsafe { Pin::get_unchecked_mut(self) };

        loop {
            let (output, data) = match this {
                Chain::First(fut1, data) => {
                    match unsafe { Pin::new_unchecked(fut1) }.poll(waker) {
                        Poll::Pending => return Poll::Pending,
                        Poll::Ready(output) => (output, data.take().unwrap()),
                    }
                }
                Chain::Second(fut2) => {
                    return unsafe { Pin::new_unchecked(fut2) }.poll(waker);
                }
                Chain::Empty => unreachable!()
            };

            *this = Chain::Empty; // Drop fut1
            let fut2 = (f.take().unwrap())(output, data);
            *this = Chain::Second(fut2)
        }
    }
}