ready_queue.rs - mozsearch

Enable keyboard shortcuts

use futures::channel::oneshot;

use futures::executor::{block_on, block_on_stream};

use futures::future;

use futures::stream::{FuturesUnordered, StreamExt};

use futures::task::Poll;

use futures_test::task::noop_context;

use std::panic::{self, AssertUnwindSafe};

use std::sync::{Arc, Barrier};

use std::thread;

#[test]

fn basic_usage() {

    block_on(future::lazy(move |cx| {

        let mut queue = FuturesUnordered::new();

        let (tx1, rx1) = oneshot::channel();

        let (tx2, rx2) = oneshot::channel();

        let (tx3, rx3) = oneshot::channel();

        queue.push(rx1);

        queue.push(rx2);

        queue.push(rx3);

        assert!(!queue.poll_next_unpin(cx).is_ready());

        tx2.send("hello").unwrap();

        assert_eq!(Poll::Ready(Some(Ok("hello"))), queue.poll_next_unpin(cx));

        assert!(!queue.poll_next_unpin(cx).is_ready());

        tx1.send("world").unwrap();

        tx3.send("world2").unwrap();

        assert_eq!(Poll::Ready(Some(Ok("world"))), queue.poll_next_unpin(cx));

        assert_eq!(Poll::Ready(Some(Ok("world2"))), queue.poll_next_unpin(cx));

        assert_eq!(Poll::Ready(None), queue.poll_next_unpin(cx));

    }));

#[test]

fn resolving_errors() {

    block_on(future::lazy(move |cx| {

        let mut queue = FuturesUnordered::new();

        let (tx1, rx1) = oneshot::channel();

        let (tx2, rx2) = oneshot::channel();

        let (tx3, rx3) = oneshot::channel();

        queue.push(rx1);

        queue.push(rx2);

        queue.push(rx3);

        assert!(!queue.poll_next_unpin(cx).is_ready());

        drop(tx2);

        assert_eq!(Poll::Ready(Some(Err(oneshot::Canceled))), queue.poll_next_unpin(cx));

        assert!(!queue.poll_next_unpin(cx).is_ready());

        drop(tx1);

        tx3.send("world2").unwrap();

        assert_eq!(Poll::Ready(Some(Err(oneshot::Canceled))), queue.poll_next_unpin(cx));

        assert_eq!(Poll::Ready(Some(Ok("world2"))), queue.poll_next_unpin(cx));

        assert_eq!(Poll::Ready(None), queue.poll_next_unpin(cx));

    }));

#[test]

fn dropping_ready_queue() {

    block_on(future::lazy(move |_| {

        let queue = FuturesUnordered::new();

        let (mut tx1, rx1) = oneshot::channel::<()>();

        let (mut tx2, rx2) = oneshot::channel::<()>();

        let (mut tx3, rx3) = oneshot::channel::<()>();

        queue.push(rx1);

        queue.push(rx2);

        queue.push(rx3);

            let cx = &mut noop_context();

            assert!(!tx1.poll_canceled(cx).is_ready());

            assert!(!tx2.poll_canceled(cx).is_ready());

            assert!(!tx3.poll_canceled(cx).is_ready());

            drop(queue);

            assert!(tx1.poll_canceled(cx).is_ready());

            assert!(tx2.poll_canceled(cx).is_ready());

            assert!(tx3.poll_canceled(cx).is_ready());

    }));

#[test]

fn stress() {

    const ITER: usize = if cfg!(miri) { 30 } else { 300 };

    for i in 0..ITER {

        let n = (i % 10) + 1;

        let mut queue = FuturesUnordered::new();

        for _ in 0..5 {

            let barrier = Arc::new(Barrier::new(n + 1));

            for num in 0..n {

                let barrier = barrier.clone();

                let (tx, rx) = oneshot::channel();

                queue.push(rx);

                thread::spawn(move || {

                    barrier.wait();

                    tx.send(num).unwrap();

});

            barrier.wait();

            let mut sync = block_on_stream(queue);

            let mut rx: Vec<_> = (&mut sync).take(n).map(|res| res.unwrap()).collect();

            assert_eq!(rx.len(), n);

            rx.sort_unstable();

            for (i, x) in rx.into_iter().enumerate() {

                assert_eq!(i, x);

            queue = sync.into_inner();

#[test]

fn panicking_future_dropped() {

    block_on(future::lazy(move |cx| {

        let mut queue = FuturesUnordered::new();

        queue.push(future::poll_fn(|_| -> Poll<Result<i32, i32>> { panic!() }));

        let r = panic::catch_unwind(AssertUnwindSafe(|| queue.poll_next_unpin(cx)));

        assert!(r.is_err());

        assert!(queue.is_empty());

        assert_eq!(Poll::Ready(None), queue.poll_next_unpin(cx));

    }));