bee-runtime 1.0.0

Runtime components and utilities for the bee framework
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
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97

// Copyright 2020-2021 IOTA Stiftung
// SPDX-License-Identifier: Apache-2.0

use std::time::Duration;

use bee_runtime::shutdown_stream::ShutdownStream;
use futures::{
    channel::{mpsc, oneshot},
    SinkExt, StreamExt,
};
use tokio::{task::spawn, time::sleep};

#[tokio::test]
async fn no_shutdown() {
    let (mut sender, receiver) = mpsc::unbounded::<usize>();
    let (_shutdown_sender, shutdown_receiver) = oneshot::channel::<()>();

    let handle = spawn(async move {
        let mut shutdown_stream = ShutdownStream::new(shutdown_receiver, receiver);

        let mut acc = 0;

        while let Some(item) = shutdown_stream.next().await {
            acc += item;
            sleep(Duration::from_millis(5)).await;
        }

        acc
    });

    for i in 0..=100 {
        assert!(sender.send(i).await.is_ok());
        sleep(Duration::from_millis(5)).await;
    }

    sleep(Duration::from_millis(5)).await;

    sender.disconnect();

    assert_eq!(handle.await.unwrap(), 5050);
}

#[tokio::test]
async fn early_shutdown() {
    let (sender, receiver) = mpsc::unbounded::<usize>();
    let (shutdown_sender, shutdown_receiver) = oneshot::channel::<()>();

    let handle = spawn(async move {
        let mut shutdown_stream = ShutdownStream::new(shutdown_receiver, receiver);

        let mut acc = 0;

        while let Some(item) = shutdown_stream.next().await {
            acc += item;
            sleep(Duration::from_millis(1)).await;
        }

        acc
    });

    for i in 0..=100 {
        assert!(sender.unbounded_send(i).is_ok());
    }

    assert!(shutdown_sender.send(()).is_ok());

    assert!(handle.await.unwrap() < 5050);
}

#[tokio::test]
async fn early_shutdown_split_from_fused() {
    let (sender, receiver) = mpsc::unbounded::<usize>();
    let (shutdown_sender, shutdown_receiver) = oneshot::channel::<()>();

    let handle = spawn(async move {
        let shutdown_stream = ShutdownStream::new(shutdown_receiver, receiver);
        let (shutdown, stream) = shutdown_stream.split();
        let mut shutdown_stream = ShutdownStream::from_fused(shutdown, stream);

        let mut acc = 0;

        while let Some(item) = shutdown_stream.next().await {
            acc += item;
            sleep(Duration::from_millis(1)).await;
        }

        acc
    });

    for i in 0..=100 {
        assert!(sender.unbounded_send(i).is_ok());
    }

    assert!(shutdown_sender.send(()).is_ok());

    assert!(handle.await.unwrap() < 5050);
}