#async #await #futures

nightly futures-async-combinators

toy futures combinators

5 unstable releases

0.3.2 Oct 3, 2019
0.3.1 Aug 22, 2019
0.3.0 Aug 2, 2019
0.2.0 Feb 24, 2019
0.1.0 Feb 21, 2019

#315 in Asynchronous

23 downloads per month

MIT license

57KB
860 lines

futures-async-combinators

Build Status Coverage Status Docs

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)
        }
    }
}

Dependencies

~54KB