component_future/

lib.rs

//! This crate implements the inner future protocol documented in [the tokio
//! docs](https://tokio.rs/docs/futures/getting_asynchronous/).
//!
//! # Overview
//!
//! If you are implementing a complicated future or stream which contains many
//! inner futures and streams, it can be difficult to keep track of when
//! looping is necessary, when it is safe to return `NotReady`, etc. This
//! provides an interface similar to the existing `poll` interface for futures
//! and streams, but extended to include additional state about how the inner
//! future or stream affected the outer future or stream's state. This allows
//! you to easily split your `poll` implementation into multiple methods, and
//! ensure that they interact properly.
//!
//! # Synopsis
//!
//! ```
//! enum OutputEvent {
//!     // ...
//! }
//!
//! struct Server {
//!     // ...
//! #   some_future: Box<
//! #       dyn futures::future::Future<Item = OutputEvent, Error = String>
//! #           + Send,
//! #   >,
//! #   other_future: Option<
//! #       Box<
//! #           dyn futures::future::Future<Item = OutputEvent, Error = String>
//! #               + Send,
//! #       >,
//! #   >,
//! }
//!
//! impl Server {
//!     fn process_thing(&self, thing: OutputEvent) {
//!         // ...
//!     }
//!
//!     fn process_other_thing(&self, thing: OutputEvent) -> OutputEvent {
//!         // ...
//! #       unimplemented!()
//!     }
//! }
//!
//! impl Server {
//!     const POLL_FNS:
//!         &'static [&'static dyn for<'a> Fn(
//!             &'a mut Self,
//!         )
//!             -> component_future::Poll<
//!             Option<OutputEvent>,
//!             String,
//!         >] = &[&Self::poll_thing, &Self::poll_other_thing];
//!
//!     fn poll_thing(
//!         &mut self,
//!     ) -> component_future::Poll<Option<OutputEvent>, String> {
//!         let thing = component_future::try_ready!(self.some_future.poll());
//!         self.process_thing(thing);
//!         Ok(component_future::Async::DidWork)
//!     }
//!
//!     fn poll_other_thing(
//!         &mut self,
//!     ) -> component_future::Poll<Option<OutputEvent>, String> {
//!         if let Some(other_future) = &mut self.other_future {
//!             let other_thing = component_future::try_ready!(
//!                 other_future.poll()
//!             );
//!             let processed_thing = self.process_other_thing(other_thing);
//!             self.other_future.take();
//!             Ok(component_future::Async::Ready(Some(processed_thing)))
//!         }
//!         else {
//!             Ok(component_future::Async::NothingToDo)
//!         }
//!     }
//! }
//!
//! impl futures::stream::Stream for Server {
//!     type Item = OutputEvent;
//!     type Error = String;
//!
//!     fn poll(&mut self) -> futures::Poll<Option<Self::Item>, Self::Error> {
//!         component_future::poll_stream(self, Self::POLL_FNS)
//!     }
//! }
//! ```

// XXX this is broken with ale
// #![warn(clippy::cargo)]
#![warn(clippy::pedantic)]
#![warn(clippy::nursery)]
#![allow(clippy::multiple_crate_versions)]
#![allow(clippy::type_complexity)]

const _DUMMY_DEPENDENCY: &str = include_str!("../Cargo.toml");

/// Return type of a component of a future or stream, indicating whether a
/// value is ready, or if not, what actions were taken.
#[derive(Copy, Clone, Debug, PartialEq, Eq)]
pub enum Async<Item> {
    /// We have a value for the main loop to return immediately.
    Ready(Item),

    /// One of our inner futures returned `futures::Async::NotReady`. If all
    /// of our other components return either `NothingToDo` or `NotReady`,
    /// then our overall future should return `NotReady` and wait to be polled
    /// again.
    NotReady,

    /// We did some work (moved our internal state closer to being ready to
    /// return a value), but we aren't ready to return a value yet. We should
    /// re-run all of the poll functions to see if the state modification made
    /// any of them also able to make progress.
    DidWork,

    /// We didn't poll any inner futures or otherwise change our internal
    /// state at all, so rerunning is unlikely to make progress. If all
    /// components return either `NothingToDo` or `NotReady` (and at least one
    /// returns `NotReady`), then we should just return `NotReady` and wait to
    /// be polled again. It is an error (panic) for all component poll methods
    /// to return `NothingToDo`.
    NothingToDo,
}

/// Each component poll method should return a value of this type.
///
/// * `Ok(Async::Ready(t))` means that the overall future or stream is ready
///   to return a value.
/// * `Ok(Async::NotReady)` means that a poll method called by one of the
///   component futures or streams returned `NotReady`, and so it's safe for
///   the overall future or stream to also return `NotReady`.
/// * `Ok(Async::DidWork)` means that the overall future made progress by
///   updating its internal state, but isn't yet ready to return a value.
/// * `Ok(Async::NothingToDo)` means that no work was done at all.
/// * `Err(e)` means that the overall future or stream is ready to return an
///   error.
pub type Poll<Item, Error> = Result<Async<Item>, Error>;

/// A macro for extracting the successful type of a `futures::Poll<T, E>` and
/// turning it into a `component_future::Poll<T, E>`.
///
/// This macro propagates both errors and `NotReady` values by returning
/// early.
#[macro_export]
macro_rules! try_ready {
    ($e:expr) => {
        match $e {
            Ok(futures::Async::Ready(t)) => t,
            Ok(futures::Async::NotReady) => {
                return Ok($crate::Async::NotReady)
            }
            Err(e) => return Err(From::from(e)),
        }
    };
}

/// The body of a `futures::future::Future::poll` method.
///
/// It will repeatedly call the given component poll functions until none of
/// them returns `Ok(Async::Ready(t))`, `Ok(Async::DidWork)`, or `Err(e)` and
/// at least one of them returns `Ok(Async::NotReady)`.
///
/// # Panics
///
/// Panics if all component poll methods return `Ok(Async::NothingToDo)`.
///
/// # Examples
///
/// ```
/// # use futures::future::Future;
/// # struct Foo;
/// # impl Foo {
/// #     const POLL_FNS:
/// #         &'static [&'static dyn for<'a> Fn(
/// #             &'a mut Self,
/// #         ) -> component_future::Poll<(), ()>] = &[];
/// # }
/// impl Future for Foo {
///     type Item = ();
///     type Error = ();
///
///     fn poll(&mut self) -> futures::Poll<Self::Item, Self::Error> {
///         component_future::poll_future(self, Self::POLL_FNS)
///     }
/// }
/// ```
pub fn poll_future<'a, T, Item, Error>(
    future: &mut T,
    poll_fns: &'a [&'a dyn for<'b> Fn(&'b mut T) -> Poll<Item, Error>],
) -> futures::Poll<Item, Error>
where
    T: futures::future::Future<Item = Item, Error = Error>,
{
    loop {
        let mut not_ready = false;
        let mut did_work = false;

        for f in poll_fns {
            match f(future)? {
                Async::Ready(e) => return Ok(futures::Async::Ready(e)),
                Async::NotReady => not_ready = true,
                Async::NothingToDo => {}
                Async::DidWork => did_work = true,
            }
        }

        if !did_work {
            if not_ready {
                return Ok(futures::Async::NotReady);
            } else {
                unreachable!()
            }
        }
    }
}

/// The body of a `futures::stream::Stream::poll` method.
///
/// It will repeatedly call the given component poll functions until none of
/// them returns `Ok(Async::Ready(t))`, `Ok(Async::DidWork)`, or `Err(e)` and
/// at least one of them returns `Ok(Async::NotReady)`.
///
/// # Panics
///
/// Panics if all component poll methods return `Ok(Async::NothingToDo)`.
///
/// # Examples
///
/// ```
/// # use futures::stream::Stream;
/// # struct Foo;
/// # impl Foo {
/// #     const POLL_FNS:
/// #         &'static [&'static dyn for<'a> Fn(
/// #             &'a mut Self,
/// #         ) -> component_future::Poll<Option<()>, ()>] = &[];
/// # }
/// impl Stream for Foo {
///     type Item = ();
///     type Error = ();
///
///     fn poll(&mut self) -> futures::Poll<Option<Self::Item>, Self::Error> {
///         component_future::poll_stream(self, Self::POLL_FNS)
///     }
/// }
/// ```
pub fn poll_stream<'a, T, Item, Error>(
    stream: &mut T,
    poll_fns: &'a [&'a dyn for<'b> Fn(
        &'b mut T,
    ) -> Poll<Option<Item>, Error>],
) -> futures::Poll<Option<Item>, Error>
where
    T: futures::stream::Stream<Item = Item, Error = Error>,
{
    loop {
        let mut not_ready = false;
        let mut did_work = false;

        for f in poll_fns {
            match f(stream)? {
                Async::Ready(e) => return Ok(futures::Async::Ready(e)),
                Async::NotReady => not_ready = true,
                Async::NothingToDo => {}
                Async::DidWork => did_work = true,
            }
        }

        if !did_work {
            if not_ready {
                return Ok(futures::Async::NotReady);
            } else {
                unreachable!()
            }
        }
    }
}