essential-relayer 0.4.0

Relay blocks from Essential builder to Essential node
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
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146

use super::*;
use error::RecoverableError;
use futures::FutureExt;
use std::sync::{atomic::AtomicUsize, Arc};

#[tokio::test]
async fn test_run_critical_err() {
    let handle =
        run(|_s| futures::future::ready::<InternalResult<()>>(Err(CriticalError::Overflow.into())))
            .unwrap();
    let e = handle.join().await.unwrap_err();
    assert!(matches!(e, CriticalError::Overflow));
}

#[tokio::test]
async fn test_run_recoverable_err() {
    let count = Arc::new(AtomicUsize::new(0));
    let c = count.clone();
    let handle = run(move |_shutdown| {
        if c.fetch_add(1, std::sync::atomic::Ordering::SeqCst) == 0 {
            futures::future::ready::<InternalResult<()>>(Err(RecoverableError::NonSequentialBlock(
                0, 2,
            )
            .into()))
        } else {
            futures::future::ready::<InternalResult<()>>(Ok(()))
        }
    })
    .unwrap();
    handle.join().await.unwrap();
    assert_eq!(count.load(std::sync::atomic::Ordering::SeqCst), 2);
}

#[tokio::test]
async fn test_run_recoverable_close() {
    let count = Arc::new(AtomicUsize::new(0));
    let c = count.clone();
    let handle = run(move |mut shutdown| {
        if c.fetch_add(1, std::sync::atomic::Ordering::SeqCst) == 0 {
            futures::future::ready::<InternalResult<()>>(Err(RecoverableError::NonSequentialBlock(
                0, 2,
            )
            .into()))
            .boxed()
        } else {
            async move {
                let _ = shutdown.changed().await;
                Ok(())
            }
            .boxed()
        }
    })
    .unwrap();
    tokio::time::sleep(std::time::Duration::from_millis(50)).await;
    handle.close().await.unwrap();
    assert_eq!(count.load(std::sync::atomic::Ordering::SeqCst), 2);
}

#[tokio::test]
async fn test_run_join() {
    let count = Arc::new(());
    let c = count.clone();
    let handle = run(move |mut shutdown| {
        let c = c.clone();
        async move {
            let _c = c;
            let _ = shutdown.changed().await;
            Ok(())
        }
    })
    .unwrap();
    tokio::time::timeout(std::time::Duration::from_millis(100), handle.join())
        .await
        .unwrap_err();
    tokio::time::sleep(std::time::Duration::from_millis(10)).await;
    // Once the join future is dropped both tasks should be end
    Arc::try_unwrap(count).unwrap();
}

#[tokio::test]
async fn test_run_close() {
    let count = Arc::new(());
    let c = count.clone();
    let handle = run(move |mut shutdown| {
        let c = c.clone();
        async move {
            let _c = c;
            let _ = shutdown.changed().await;
            Ok(())
        }
    })
    .unwrap();
    handle.close().await.unwrap();
    Arc::try_unwrap(count).unwrap();
}

#[tokio::test]
async fn test_run_join_immediate() {
    let count = Arc::new(());
    let c = count.clone();
    let handle = run(move |shutdown| {
        let c = c.clone();
        async move {
            let _c = c;
            let _s = shutdown;
            Ok(())
        }
    })
    .unwrap();
    handle.join().await.unwrap();
    Arc::try_unwrap(count).unwrap();
}

#[tokio::test]
async fn test_run_multiple_recoverable() {
    let count = Arc::new(AtomicUsize::new(0));
    let c = count.clone();
    let handle = run(move |mut shutdown| {
        if c.fetch_add(1, std::sync::atomic::Ordering::SeqCst) < 10 {
            futures::future::ready::<InternalResult<()>>(Err(RecoverableError::NonSequentialBlock(
                0, 2,
            )
            .into()))
            .boxed()
        } else {
            async move {
                let _ = shutdown.changed().await;
                Ok(())
            }
            .boxed()
        }
    })
    .unwrap();
    while count.load(std::sync::atomic::Ordering::SeqCst) < 10 {
        tokio::time::sleep(std::time::Duration::from_millis(10)).await;
    }
    assert_eq!(count.load(std::sync::atomic::Ordering::SeqCst), 11);

    // Still running
    let count = Arc::try_unwrap(count).unwrap_err();

    handle.close().await.unwrap();

    // Closed
    Arc::try_unwrap(count).unwrap();
}