subscribe/
subscribe.rs

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
use std::{thread, time::Duration};

use env_logger::{Builder, Env};
use log::{debug, info, warn};
use rmp_serde::from_slice;
use serde::Deserialize;
use tether_agent::{PlugOptionsBuilder, TetherAgentOptionsBuilder};

#[derive(Deserialize, Debug)]
struct CustomMessage {
    id: usize,
    name: String,
}
// Test this by sending a message like
// tether send --topic specific/any/two --message '{"id":1,"name":"boo"}'

fn main() {
    println!("Rust Tether Agent subscribe example");

    let mut builder = Builder::from_env(Env::default().default_filter_or("debug"));
    builder.filter_module("tether_agent", log::LevelFilter::Warn);
    builder.filter_module("rumqttc", log::LevelFilter::Warn);
    builder.init();

    debug!("Debugging is enabled; could be verbose");

    let mut tether_agent = TetherAgentOptionsBuilder::new("RustDemo")
        .id(Some("example"))
        .build()
        .expect("failed to init Tether agent");

    let input_one = PlugOptionsBuilder::create_input("one")
        .build(&mut tether_agent)
        .expect("failed to create input");
    info!("input one {} = {}", input_one.name(), input_one.topic());
    let input_two = PlugOptionsBuilder::create_input("two")
        .role(Some("specific"))
        .build(&mut tether_agent)
        .expect("failed to create input");
    info!("input two {} = {}", input_two.name(), input_two.topic());
    let input_empty = PlugOptionsBuilder::create_input("nothing")
        .build(&mut tether_agent)
        .expect("failed to create input");

    let input_everything = PlugOptionsBuilder::create_input("everything")
        .topic(Some("#"))
        .build(&mut tether_agent)
        .expect("failed to create input");

    let input_specify_id = PlugOptionsBuilder::create_input("groupMessages")
        .id(Some("someGroup"))
        .name(None)
        .build(&mut tether_agent)
        .expect("failed to create input");

    debug!(
        "input everything {} = {}",
        input_everything.name(),
        input_everything.topic()
    );

    info!("Checking messages every 1s, 10x...");

    loop {
        debug!("Checking for messages...");
        while let Some((topic, payload)) = tether_agent.check_messages() {
            // debug!(
            //     "........ Received a message topic {:?} => topic parts {:?}",
            //     topic, topic
            // );

            if input_one.matches(&topic) {
                info!(
                            "******** INPUT ONE:\n Received a message for plug named \"{}\" on topic {:?} with length {} bytes",
                            input_one.name(),
                            topic,
                            payload.len()
                        );
                // assert_eq!(parse_plug_name(topic.un), Some("one"));
            }
            if input_two.matches(&topic) {
                info!(
                        "******** INPUT TWO:\n Received a message for plug named \"{}\" on topic {:?} with length {} bytes",
                        input_two.name(),
                        topic,
                        payload.len()
                    );
                // assert_eq!(parse_plug_name(message.topic()), Some("two"));
                // assert_ne!(parse_plug_name(message.topic()), Some("one"));

                // Notice how you must give the from_slice function a type so it knows what to expect
                let decoded = from_slice::<CustomMessage>(&payload);
                match decoded {
                    Ok(d) => {
                        info!("Yes, we decoded the MessagePack payload as: {:?}", d);
                        let CustomMessage { name, id } = d;
                        debug!("Name is {} and ID is {}", name, id);
                    }
                    Err(e) => {
                        warn!("Failed to decode the payload: {}", e)
                    }
                };
            }
            if input_empty.matches(&topic) {
                info!(
                        "******** EMPTY MESSAGE:\n Received a message for plug named \"{}\" on topic {:?} with length {} bytes",
                        input_empty.name(),
                        topic,
                       payload.len()
                    );
                // assert_eq!(parse_plug_name(topic), Some("nothing"));
            }
            if input_everything.matches(&topic) {
                info!(
                    "******** EVERYTHING MATCHES HERE:\n Received a message for plug named \"{}\" on topic {:?} with length {} bytes",
                    input_everything.name(),
                    topic,
                   payload.len()
                );
            }
            if input_specify_id.matches(&topic) {
                info!("******** ID MATCH:\n Should match any role and plug name, but only messages with ID \"groupMessages\"");
                info!(
                    "\n Received a message from plug named \"{}\" on topic {:?} with length {} bytes",
                    input_specify_id.name(),
                    topic,
                    payload.len()
                );
                // assert_eq!(parse_agent_id(message.topic()), Some("groupMessages"));
            }
        }

        thread::sleep(Duration::from_millis(1000))
    }
}