gmt_dos-actors_dsl 0.2.0

GMT DOS Actors Domain Specific Language
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 std::collections::HashSet;

use syn::{
    parse::{Parse, ParseStream},
    LitInt, Token,
};

use crate::{client::SharedClient, Expand, Expanded};

mod chain;
use chain::Chain;

/// Data flow
///
/// A flow is characterized by a sampling rate and
/// a chain of actors i.e
///
/// rate: actor1[output1_of_actor1] -> actor2[output1_of_actor2] -> actor3
#[derive(Debug, Clone, Default)]
pub struct Flow {
    pub rate: usize,
    pub chain: Chain,
}

impl Flow {
    /// Iteration through the actors chain of each flow replacing duplicated client with shared reference
    pub fn dedup(&mut self, clients: &mut HashSet<SharedClient>) {
        self.chain.dedup(clients);
    }
    /// Iteration through the actors chain of each flow matching output/input rate or setting up a rate transition
    pub fn match_rates(&mut self) {
        self.chain.match_rates(self.rate);
    }
    /// Collect output special clients
    ///
    /// Special clients are [Sampler] and [Arrow]
    pub fn collect_clients(&self, clients: &mut HashSet<SharedClient>) {
        self.chain.iter().for_each(|client_output| {
            client_output
                .output
                .as_ref()
                .map(|output| output.collect(clients));
            self.chain
                .logger
                .as_ref()
                .map(|logger| clients.insert(logger.clone()));
        });
    }
}

impl Parse for Flow {
    fn parse(input: ParseStream) -> syn::Result<Self> {
        let rate = input.parse::<LitInt>()?.base10_parse::<usize>()?;
        let _: Token!(:) = input.parse()?;
        let chain = input.parse::<Chain>()?.logging(rate);
        Ok(Self { rate, chain })
    }
}

impl Expand for Flow {
    fn expand(&self) -> Expanded {
        self.chain.expand()
    }
}