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

//! Null target.

use non_empty_vec::NonEmpty;
use serde::Deserialize;
use serde_with::{formats::PreferOne, serde_as, OneOrMany};
use tokio::sync::mpsc;

use crate::comms::Terminated;
use crate::manager::{Component, WaitPoint};
use crate::{comms::Link, manager::TargetCommand};

#[serde_as]
#[derive(Debug, Deserialize)]
pub struct Target {
    #[allow(dead_code)]
    #[serde(alias = "source")]
    #[serde_as(deserialize_as = "OneOrMany<_, PreferOne>")]
    sources: Vec<Link>,
}

impl Target {
    /// Runs the target.
    pub async fn run(
        mut self,
        _: Component,
        mut cmd_rx: mpsc::Receiver<TargetCommand>,
        waitpoint: WaitPoint,
    ) -> Result<(), Terminated> {
        // Prevent wasted time and effort on sending us updates that we won't
        // use by suspending our link to upstream units. Don't just exit the
        // function because that will cause the upstream units to terminate if
        // we are their only downstream.
        for source in &mut self.sources {
            source.suspend().await;
        }

        // Wait for other components to be, and signal to other components
        // that we are, ready to start. All units and targets start together,
        // otherwise data passed from one component to another may be lost if
        // the receiving component is not yet ready to accept it.
        waitpoint.running().await;

        while let Some(cmd) = cmd_rx.recv().await {
            match cmd {
                TargetCommand::Reconfigure { new_config } => {
                    if let crate::targets::Target::Null(Target {
                        sources: new_sources,
                        ..
                        // config
                    }) = new_config
                    {
                        self.sources = new_sources;
                        for source in &mut self.sources {
                            source.suspend().await;
                        }
                    }
                }

                TargetCommand::ReportLinks { report } => {
                    if let Ok(non_empty) =
                        NonEmpty::try_from(self.sources.clone())
                    {
                        report.set_sources(&non_empty);
                    }
                }

                TargetCommand::Terminate => {
                    break;
                }
            }
        }

        Err(Terminated)
    }
}