Struct unicycle::Unordered

pub struct Unordered<F, S> { /* fields omitted */ }

A container for an unordered collection of Futures or Streams.

You should use one of the following type aliases to construct it:

• FuturesUnordered
• StreamsUnordered
• IndexedStreamsUnordered

Examples

use tokio::{stream::StreamExt as _, time};
use std::time::Duration;

#[tokio::main]
async fn main() {
    let mut futures = unicycle::FuturesUnordered::new();

    futures.push(time::delay_for(Duration::from_secs(2)));
    futures.push(time::delay_for(Duration::from_secs(3)));
    futures.push(time::delay_for(Duration::from_secs(1)));

    while let Some(_) = futures.next().await {
        println!("tick");
    }

    println!("done!");
}

Methods

impl<T> Unordered<T, Futures>

pub fn new() -> Self

Construct a new, empty FuturesUnordered.

Examples

use unicycle::FuturesUnordered;

let mut futures = FuturesUnordered::new();
assert!(futures.is_empty());

futures.push(async { 42 });

impl<T> Unordered<T, Streams>

pub fn new() -> Self

Construct a new, empty StreamsUnordered.

Examples

use unicycle::StreamsUnordered;
use tokio::stream::{StreamExt as _, iter};

#[tokio::main]
async fn main() {
    let mut streams = StreamsUnordered::new();
    assert!(streams.is_empty());

    streams.push(iter(vec![1, 2, 3, 4]));
    streams.push(iter(vec![5, 6, 7, 8]));

    let mut received = Vec::new();

    while let Some(value) = streams.next().await {
        received.push(value);
    }

    assert_eq!(vec![1, 5, 2, 6, 3, 7, 4, 8], received);
}

impl<F> Unordered<T, IndexedStreams>

pub fn new() -> Self

Construct a new, empty IndexedStreamsUnordered.

This is the same as StreamsUnordered, except that it yields the index of the stream who'se value was just yielded, alongside the yielded value.

Examples

use unicycle::IndexedStreamsUnordered;
use tokio::stream::{StreamExt as _, iter};

#[tokio::main]
async fn main() {
    let mut streams = IndexedStreamsUnordered::new();
    assert!(streams.is_empty());

    streams.push(iter(vec![1, 2]));
    streams.push(iter(vec![5, 6]));

    let mut received = Vec::new();

    while let Some(value) = streams.next().await {
        received.push(value);
    }

    assert_eq!(
        vec![
            (0, Some(1)),
            (1, Some(5)),
            (0, Some(2)),
            (1, Some(6)),
            (0, None),
            (1, None)
        ],
        received
    );
}

impl<T, S> Unordered<T, S>

pub fn is_empty(&self) -> bool

Test if the collection of futures is empty.

Examples

use unicycle::FuturesUnordered;

let mut futures = FuturesUnordered::<tokio::time::Delay>::new();
assert!(futures.is_empty());

pub fn push(&mut self, future: T) -> usize

Push the given future or stream to Unordered and return its task index.

Newly added futures are guaranteed to be polled, but there is no guarantee in which order this will happen.

Pushed tasks are pinned by the Unordered collection automatically.

Examples

use unicycle::FuturesUnordered;

let mut futures = FuturesUnordered::new();
assert!(futures.is_empty());
futures.push(async { 42 });
assert!(!futures.is_empty());

pub fn get_pin_mut(&mut self, index: usize) -> Option<Pin<&mut T>>

Get a pinned mutable reference to the stream or future at the given index.

Examples

use unicycle::FuturesUnordered;
use futures::future::poll_fn;
use std::future::Future as _;

#[tokio::main]
async fn main() {
    let mut futures = FuturesUnordered::new();
    let index = futures.push(async { 42 });

    let result = poll_fn(|cx| {
        futures.get_pin_mut(index).expect("expected future").poll(cx)
    }).await;

    assert_eq!(result, 42);
}

pub fn get_mut(&mut self, index: usize) -> Option<&mut T> where
    T: Unpin, 

Get a mutable reference to the stream or future at the given index. Requires that the stores stream or future is Unpin.

Examples

use unicycle::FuturesUnordered;
use futures::future::{ready, poll_fn};
use std::{pin::Pin, future::Future as _};

#[tokio::main]
async fn main() {
    let mut futures = FuturesUnordered::new();
    let index = futures.push(ready(42));

    let result = poll_fn(|cx| {
        Pin::new(futures.get_mut(index).expect("expected future")).poll(cx)
    }).await;

    assert_eq!(result, 42);
}

Trait Implementations

impl<T> Default for Unordered<T, Futures>

fn default() -> Self

Returns the "default value" for a type.

impl<T, S> Drop for Unordered<T, S>

fn drop(&mut self)

Executes the destructor for this type.

impl<T, S> Extend<T> for Unordered<T, S>

fn extend<I>(&mut self, iter: I) where
    I: IntoIterator<Item = T>, 

Extends a collection with the contents of an iterator.

impl<T, S> Send for Unordered<T, S>

impl<T> Stream for Unordered<T, Futures> where
    T: Future, 

type Item = T::Output

Values yielded by the stream.

fn poll_next(self: Pin<&mut Self>, cx: &mut Context) -> Poll<Option<Self::Item>>

Attempt to pull out the next value of this stream, registering the current task for wakeup if the value is not yet available, and returning None if the stream is exhausted.

fn size_hint(&self) -> (usize, Option<usize>)

Returns the bounds on the remaining length of the stream.

impl<T> Stream for Unordered<T, Streams> where
    T: Stream, 

type Item = T::Item

Values yielded by the stream.

fn poll_next(self: Pin<&mut Self>, cx: &mut Context) -> Poll<Option<Self::Item>>

Attempt to pull out the next value of this stream, registering the current task for wakeup if the value is not yet available, and returning None if the stream is exhausted.

fn size_hint(&self) -> (usize, Option<usize>)

Returns the bounds on the remaining length of the stream.

impl<T, S> Sync for Unordered<T, S>

impl<T, S> Unpin for Unordered<T, S>