chunks_timeout.rs - mozsearch

Enable keyboard shortcuts

#![warn(rust_2018_idioms)]

#![cfg(all(feature = "time", feature = "sync", feature = "io-util"))]

use tokio::time;

use tokio_stream::{self as stream, StreamExt};

use tokio_test::assert_pending;

use tokio_test::task;

use futures::FutureExt;

use std::time::Duration;

#[tokio::test(start_paused = true)]

async fn usage() {

    let iter = vec![1, 2, 3].into_iter();

    let stream0 = stream::iter(iter);

    let iter = vec![4].into_iter();

    let stream1 =

        stream::iter(iter).then(move |n| time::sleep(Duration::from_secs(3)).map(move |_| n));

    let chunk_stream = stream0

        .chain(stream1)

        .chunks_timeout(4, Duration::from_secs(2));

    let mut chunk_stream = task::spawn(chunk_stream);

    assert_pending!(chunk_stream.poll_next());

    time::advance(Duration::from_secs(2)).await;

    assert_eq!(chunk_stream.next().await, Some(vec![1, 2, 3]));

    assert_pending!(chunk_stream.poll_next());

    time::advance(Duration::from_secs(2)).await;

    assert_eq!(chunk_stream.next().await, Some(vec![4]));

#[tokio::test(start_paused = true)]

async fn full_chunk_with_timeout() {

    let iter = vec![1, 2].into_iter();

    let stream0 = stream::iter(iter);

    let iter = vec![3].into_iter();

    let stream1 =

        stream::iter(iter).then(move |n| time::sleep(Duration::from_secs(1)).map(move |_| n));

    let iter = vec![4].into_iter();

    let stream2 =

        stream::iter(iter).then(move |n| time::sleep(Duration::from_secs(3)).map(move |_| n));

    let chunk_stream = stream0

        .chain(stream1)

        .chain(stream2)

        .chunks_timeout(3, Duration::from_secs(2));

    let mut chunk_stream = task::spawn(chunk_stream);

    assert_pending!(chunk_stream.poll_next());

    time::advance(Duration::from_secs(2)).await;

    assert_eq!(chunk_stream.next().await, Some(vec![1, 2, 3]));

    assert_pending!(chunk_stream.poll_next());

    time::advance(Duration::from_secs(2)).await;

    assert_eq!(chunk_stream.next().await, Some(vec![4]));

#[tokio::test]

#[ignore]

async fn real_time() {

    let iter = vec![1, 2, 3, 4].into_iter();

    let stream0 = stream::iter(iter);

    let iter = vec![5].into_iter();

    let stream1 =

        stream::iter(iter).then(move |n| time::sleep(Duration::from_secs(5)).map(move |_| n));

    let chunk_stream = stream0

        .chain(stream1)

        .chunks_timeout(3, Duration::from_secs(2));

    let mut chunk_stream = task::spawn(chunk_stream);

    assert_eq!(chunk_stream.next().await, Some(vec![1, 2, 3]));

    assert_eq!(chunk_stream.next().await, Some(vec![4]));

    assert_eq!(chunk_stream.next().await, Some(vec![5]));