rama-core 0.3.0

rama service core code, used by rama and service authors
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64

use super::{DelayStream, FutureExt as _, GracefulStream, StreamExt as _, stream};

use std::time::Duration;
use tokio::time;

#[tokio::test]
async fn delays_first_item_only() {
    time::pause();

    let dur = Duration::from_millis(10);
    let mut s = core::pin::pin!(DelayStream::new(dur, stream::iter([1u8, 2, 3])));

    tokio::time::sleep(Duration::from_micros(100)).await;
    assert_eq!(s.next().now_or_never(), None);

    tokio::time::sleep(Duration::from_millis(10)).await;
    assert_eq!(s.next().await, Some(1));
    assert_eq!(s.next().await, Some(2));
    assert_eq!(s.next().await, Some(3));
    assert_eq!(s.next().await, None);
}

#[tokio::test]
async fn immediate_when_duration_zero() {
    let mut s = core::pin::pin!(DelayStream::new(Duration::ZERO, stream::iter([10u8, 20])));

    assert_eq!(s.next().now_or_never().unwrap(), Some(10));
    assert_eq!(s.next().now_or_never().unwrap(), Some(20));
    assert_eq!(s.next().now_or_never().unwrap(), None);
}

#[tokio::test]
async fn graceful_stream_cancels_pending_stream() {
    time::pause();

    let (tx, rx) = tokio::sync::oneshot::channel::<()>();
    let cancel = async move {
        _ = rx.await;
    };

    let delayed = DelayStream::new(Duration::from_secs(10), stream::iter([1u8, 2, 3]));
    let mut s = core::pin::pin!(GracefulStream::new(cancel, delayed));

    tokio::time::sleep(Duration::from_millis(1)).await;
    assert_eq!(s.next().now_or_never(), None);

    _ = tx.send(());
    assert_eq!(s.next().await, None);
    assert_eq!(s.next().await, None);
}

#[tokio::test]
async fn graceful_stream_stops_after_cancel_even_if_stream_has_more_items() {
    let (tx, rx) = tokio::sync::oneshot::channel::<()>();
    let cancel = async move {
        _ = rx.await;
    };
    let mut s = core::pin::pin!(GracefulStream::new(cancel, stream::iter([1u8, 2, 3])));

    assert_eq!(s.next().await, Some(1));
    _ = tx.send(());
    assert_eq!(s.next().await, None);
    assert_eq!(s.next().await, None);
}