rumqttd 0.20.0

rumqttd is a high performance MQTT broker written in Rust which is light weight and embeddable
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

use config::FileFormat;
use rumqttd::Broker;

use clap::Parser;
use tracing::trace;

static RUMQTTD_DEFAULT_CONFIG: &str = include_str!("../rumqttd.toml");

#[derive(Parser)]
#[command(version)]
#[command(name = "rumqttd")]
#[command(about = "A high performance, lightweight and embeddable MQTT broker written in Rust.")]
#[command(author = "tekjar <raviteja@bytebeam.io>")]
struct CommandLine {
    /// path to config file
    #[arg(short, long)]
    config: Option<String>,
    #[command(subcommand)]
    command: Option<Command>,
    /// log level (v: info, vv: debug, vvv: trace)
    #[arg(short = 'v', long = "verbose", action = clap::ArgAction::Count)]
    verbose: u8,
    /// launch without printing banner
    #[arg(short, long)]
    quiet: bool,
}

#[derive(Parser)]
enum Command {
    /// Write default configuration file to stdout
    GenerateConfig,
}

fn main() {
    let commandline: CommandLine = CommandLine::parse();

    if let Some(Command::GenerateConfig) = commandline.command {
        println!("{RUMQTTD_DEFAULT_CONFIG}");
        return;
    }

    if !commandline.quiet {
        banner();
    }

    let level = match commandline.verbose {
        0 => "rumqttd=warn",
        1 => "rumqttd=info",
        2 => "rumqttd=debug",
        _ => "rumqttd=trace",
    };

    // tracing syntax ->
    let builder = tracing_subscriber::fmt()
        .pretty()
        .with_line_number(false)
        .with_file(false)
        .with_thread_ids(false)
        .with_thread_names(false)
        .with_env_filter(level)
        .with_filter_reloading();

    let reload_handle = builder.reload_handle();

    builder
        .try_init()
        .expect("initialized subscriber succesfully");

    let mut config_builder = config::Config::builder();

    config_builder = match &commandline.config {
        Some(config) => config_builder.add_source(config::File::with_name(config)),
        None => config_builder.add_source(config::File::from_str(
            RUMQTTD_DEFAULT_CONFIG,
            FileFormat::Toml,
        )),
    };

    let mut configs: rumqttd::Config = config_builder.build().unwrap().try_deserialize().unwrap();

    if let Some(console_config) = configs.console.as_mut() {
        console_config.set_filter_reload_handle(reload_handle)
    }

    validate_config(&configs);

    // println!("{:#?}", configs);

    let mut broker = Broker::new(configs);
    broker.start().unwrap();
}

// Do any extra validation that needs to be done before starting the broker here.
fn validate_config(configs: &rumqttd::Config) {
    if let Some(v4) = &configs.v4 {
        for (name, server_setting) in v4 {
            if let Some(tls_config) = &server_setting.tls {
                if !tls_config.validate_paths() {
                    panic!("Certificate path not valid for server v4.{name}.")
                }
                trace!("Validated certificate paths for server v4.{name}.");
            }
        }
    }

    if let Some(v5) = &configs.v5 {
        for (name, server_setting) in v5 {
            if let Some(tls_config) = &server_setting.tls {
                if !tls_config.validate_paths() {
                    panic!("Certificate path not valid for server v5.{name}.")
                }
                trace!("Validated certificate paths for server v5.{name}.");
            }
        }
    }

    if let Some(ws) = &configs.ws {
        for (name, server_setting) in ws {
            if let Some(tls_config) = &server_setting.tls {
                if !tls_config.validate_paths() {
                    panic!("Certificate path not valid for server ws.{name}.")
                }
                trace!("Validated certificate paths for server ws.{name}.");
            }
        }
    }
}

fn banner() {
    const B: &str = r"                                              
         ___ _   _ __  __  ___ _____ _____ ___  
        | _ \ | | |  \/  |/ _ \_   _|_   _|   \ 
        |   / |_| | |\/| | (_) || |   | | | |) |
        |_|_\\___/|_|  |_|\__\_\|_|   |_| |___/ 
    ";

    println!("{B}\n");
}