rotonda 0.4.0

composable, programmable BGP engine
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

use std::sync::Arc;

use futures::{future::Either, Future};

use crate::comms::{GateStatus, Terminated};

use super::status_reporter::UnitStatusReporter;

pub enum UnitActivity<O, E, F, X>
where
    F: Future<Output = X> + Unpin,
{
    Terminated,
    GateStatusChanged(GateStatus, F),
    InputError(E),
    InputReceived(O),
}

type ProcessGateResult = Result<GateStatus, Terminated>;

type SelectAllOutput<O, E, V> = (Result<O, E>, usize, Vec<V>);

type SelectResult<T, F, Z> = Either<(ProcessGateResult, F), (T, Z)>;

pub fn wait_for_activity<O, E, F, X, Z, USR>(
    status_reporter: &Arc<USR>,
    either: SelectResult<Result<O, E>, F, Z>,
) -> UnitActivity<O, E, F, X>
where
    F: Future<Output = X> + Unpin,
    USR: UnitStatusReporter,
{
    match either {
        Either::Left((Err(Terminated), _)) => {
            status_reporter.terminated();
            UnitActivity::Terminated
        }
        Either::Left((Ok(status), next_fut)) => {
            status_reporter.gate_status_announced(&status);
            if let GateStatus::Reconfiguring { .. } = status {
                status_reporter.reconfigured();
            }
            UnitActivity::GateStatusChanged(status, next_fut)
        }
        Either::Right((Err(err), _process_fut)) => {
            UnitActivity::InputError(err)
        }
        Either::Right((Ok(ok), _process_fut)) => {
            UnitActivity::InputReceived(ok)
        }
    }
}

impl<O, E, F, X, Z, USR> From<(&Arc<USR>, SelectResult<Result<O, E>, F, Z>)>
    for UnitActivity<O, E, F, X>
where
    F: Future<Output = X> + Unpin,
    USR: UnitStatusReporter,
{
    fn from(
        (status_reporter, either): (
            &Arc<USR>,
            SelectResult<Result<O, E>, F, Z>,
        ),
    ) -> Self {
        wait_for_activity(status_reporter, either)
    }
}

pub fn wait_for_first_activity<O, E, F, V, X, Z, USR>(
    status_reporter: &Arc<USR>,
    either: SelectResult<SelectAllOutput<O, E, V>, F, Z>,
) -> UnitActivity<O, E, F, X>
where
    F: Future<Output = X> + Unpin,
    USR: UnitStatusReporter,
{
    match either {
        Either::Left((Err(Terminated), _)) => {
            status_reporter.terminated();
            UnitActivity::Terminated
        }
        Either::Left((Ok(status), next_fut)) => {
            status_reporter.gate_status_announced(&status);
            UnitActivity::GateStatusChanged(status, next_fut)
        }
        Either::Right(((ready_fut, _fut_index, _other_futs), _)) => {
            match ready_fut {
                Err(err) => UnitActivity::InputError(err),
                Ok(ok) => UnitActivity::InputReceived(ok),
            }
        }
    }
}

impl<O, E, F, V, X, Z, USR>
    From<(&Arc<USR>, SelectResult<SelectAllOutput<O, E, V>, F, Z>)>
    for UnitActivity<O, E, F, X>
where
    F: Future<Output = X> + Unpin,
    USR: UnitStatusReporter,
{
    fn from(
        (status_reporter, res): (
            &Arc<USR>,
            SelectResult<SelectAllOutput<O, E, V>, F, Z>,
        ),
    ) -> Self {
        wait_for_first_activity(status_reporter, res)
    }
}