Repository: tokio-rs/tokio
File: tokio-stream/tests/chunks_timeout.rs (Rust)

Code before:
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![]));

    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].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_eq!(chunk_stream.next().await, Some(vec![1, 2, 3]));
    assert_eq!(chunk_stream.next().await, Some(vec![4]));
}

Review comment (suggestion):
The documentation should explicitly point out that it can return an empty vector if no items show up for the duration.

Code after:
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]));
}