tether-agent 4.1.0-alpha

Standardised use of MQTT and MessagePack for inter-process communication
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
147
148
149
150
151
152
153
154
155
156

use crate::{
    tether_compliant_topic::{TetherCompliantTopic, TetherOrCustomTopic},
    TetherAgent,
};

use log::*;

use super::{ChannelDefBuilder, ChannelSenderDef};

pub struct ChannelSenderDefBuilder {
    channel_name: String,
    qos: Option<i32>,
    override_publish_role: Option<String>,
    override_publish_id: Option<String>,
    override_topic: Option<String>,
    retain: Option<bool>,
}

impl ChannelDefBuilder for ChannelSenderDefBuilder {
    /// Use the ChannelSenderDefBuilder for creating a **custom** Definition that can
    /// be passed to Tether Agent `.create_sender_with_defintion`.
    ///
    /// If you don't need a custom Definition, simply use Tether Agent `.create_sender` instead.
    ///
    /// First call .new(), finalise with .build() to get the ChannelSenderDefinition.
    fn new(name: &str) -> Self {
        ChannelSenderDefBuilder {
            channel_name: String::from(name),
            override_publish_id: None,
            override_publish_role: None,
            override_topic: None,
            retain: None,
            qos: None,
        }
    }

    fn qos(self, qos: Option<i32>) -> Self {
        ChannelSenderDefBuilder { qos, ..self }
    }

    fn role(self, role: Option<&str>) -> Self {
        if self.override_topic.is_some() {
            error!("Override topic was also provided; this will take precedence");
            self
        } else {
            let override_publish_role = role.map(|s| s.into());
            ChannelSenderDefBuilder {
                override_publish_role,
                ..self
            }
        }
    }

    fn id(self, id: Option<&str>) -> Self {
        if self.override_topic.is_some() {
            error!("Override topic was also provided; this will take precedence");
            self
        } else {
            let override_publish_id = id.map(|s| s.into());
            ChannelSenderDefBuilder {
                override_publish_id,
                ..self
            }
        }
    }

    fn override_name(self, override_channel_name: Option<&str>) -> Self {
        if self.override_topic.is_some() {
            error!("Override topic was also provided; this will take precedence");
            return self;
        }
        match override_channel_name {
            Some(n) => ChannelSenderDefBuilder {
                channel_name: n.into(),
                ..self
            },
            None => {
                debug!("Override Channel name set to None; will use original name \"{}\" given in ::create_receiver constructor", self.channel_name);
                self
            }
        }
    }

    fn override_topic(self, override_topic: Option<&str>) -> Self {
        match override_topic {
            Some(t) => {
                if TryInto::<TetherCompliantTopic>::try_into(t).is_ok() {
                    info!("Custom topic passes Tether Compliant Topic validation");
                } else if t == "#" {
                    info!("Wildcard \"#\" custom topics are not Tether Compliant Topics but are valid");
                } else {
                    warn!(
                        "Could not convert \"{}\" into Tether Compliant Topic; presumably you know what you're doing!",
                        t
                    );
                }
                ChannelSenderDefBuilder {
                    override_topic: Some(t.into()),
                    ..self
                }
            }
            None => ChannelSenderDefBuilder {
                override_topic: None,
                ..self
            },
        }
    }
}

impl ChannelSenderDefBuilder {
    pub fn retain(self, should_retain: Option<bool>) -> Self {
        ChannelSenderDefBuilder {
            retain: should_retain,
            ..self
        }
    }

    pub fn build(self, tether_agent: &TetherAgent) -> ChannelSenderDef {
        let tpt: TetherOrCustomTopic = match self.override_topic {
            Some(custom) => {
                warn!(
                    "Custom topic override: \"{}\" - all other options ignored",
                    custom
                );
                TetherOrCustomTopic::Custom(custom)
            }
            None => {
                let optional_id_part = match self.override_publish_id {
                    Some(id) => {
                        debug!("Publish ID was overriden at Channel options level. The Agent ID will be ignored.");
                        Some(id)
                    }
                    None => {
                        debug!("Publish ID was not overriden at Channel options level. The Agent ID will be used instead, if specified in Agent creation.");
                        tether_agent.id().map(String::from)
                    }
                };

                TetherOrCustomTopic::Tether(TetherCompliantTopic::new_for_publish(
                    tether_agent,
                    &self.channel_name,
                    self.override_publish_role.as_deref(),
                    optional_id_part.as_deref(),
                ))
            }
        };

        ChannelSenderDef {
            name: self.channel_name,
            generated_topic: tpt.full_topic_string(),
            topic: tpt,
            qos: self.qos.unwrap_or(1),
            retain: self.retain.unwrap_or(false),
        }
    }
}