Repository: bytebeamio/rumqtt (language: rust)

Issue:
`rumqttd` example not working
I'm trying to use the broker (`rumqttd`) but isn't able to work with the example code. Specifically I used the `singlenode` example and played with it a bit:

```rust
use rumqttd::{Broker, Config, Notification};
use std::thread;

fn main() {
    pretty_env_logger::init();

    let config = config::Config::builder()
        .add_source(config::File::with_name("rumqttd.toml"))
        .build()
        .unwrap();
    let config: Config = config.try_deserialize().unwrap();

    let mut broker = Broker::new(config);
    let (_link_tx, mut link_rx) = broker.link("singlenode").unwrap();
    thread::spawn(move || {
        broker.start().unwrap();
    });

    loop {
        println!("Waiting for notification");
        let notification = link_rx.recv();
        println!("Got notification");
        let notification = match notification.unwrap() {
            Some(v) => v,
            None => continue,
        };
        link_rx.ready().unwrap();

        match notification {
            Notification::Forward(forward) => {
                println!(
                    "Topic = {:?}, Payload = {} bytes",
                    forward.publish.topic,
                    forward.publish.payload.len()
                );
            }
            v => {
                println!("{v:?}");
            }
        }
    }
}
```

I then connected and published a single message: with the MQTTx tool.
```
galah92@Dell14:~/rust-mqtt$ mqttx pub -V 3.1.1 -t 'hello/%i/world' -m 'Hello from MQTTX CLI' -q 1
[2/27/2023] [10:57:43 PM] › …  Connecting...
[2/27/2023] [10:57:43 PM] › ✔  Connected
[2/27/2023] [10:57:43 PM] › …  Message publishing...
[2/27/2023] [10:57:43 PM] › ✔  Message published
```

But got in return:
```
galah92@Dell14:~/rust-mqtt$ cargo run
    Finished dev [unoptimized + debuginfo] target(s) in 0.13s
     Running `target/debug/rust-mqtt`
Waiting for notification
 ERROR rumqttd::server::broker > remote_link; tenant_id=None
 ERROR rumqttd::server::broker > Disconnected!! error=Network(Io(Custom { kind: ConnectionAborted, error: "connection closed by peer" }))
 ERROR rumqttd::router::routing > no-connection id 2 is already gone
```

Which looks like `link_rx.recv()` is still blocked, and the error logs don't make much sense.

What am I missing?

Thanks


Discussion / hints:
Thank you for reporting the issue!

- Its not getting blocked at `recv` call because it didn't subscribe to any topic so router doesn't forward this link any notifications. 

- And related to log messages we switched to using `tracing` and this example is still using `pretty_env_logger` thats why error messages are very terse. That `Disconnected!!` message is expected because `mqttx` connects to the broker publishes and then disconnects immediately.

Can you try using this example:

```rs
use rumqttd::{Broker, Config, Notification};

use std::thread;

fn main() {
    let builder = tracing_subscriber::fmt()
        .pretty()
        .with_line_number(false)
        .with_file(false)
        .with_thread_ids(false)
        .with_thread_names(false);

    builder
        .try_init()
        .expect("initialized subscriber succesfully");

    // As examples are compiled as seperate binary so this config is current path dependent. Run it
    // from root of this crate
    let config = config::Config::builder()
        .add_source(config::File::with_name("rumqttd.toml"))
        .build()
        .unwrap();

    let config: Config = config.try_deserialize().unwrap();

    dbg!(&config);

    let mut broker = Broker::new(config);
    let (mut link_tx, mut link_rx) = broker.link("singlenode").unwrap();
    thread::spawn(move || {
        broker.start().unwrap();
    });

    link_tx.subscribe("#").unwrap();

    let mut count = 0;
    loop {
        let notification = match link_rx.recv().unwrap() {
            Some(v) => v,
            None => continue,
        };

        match notification {
            Notification::Forward(forward) => {
                count += 1;
                println!(
                    "Topic = {:?}, Count = {}, Payload = {} bytes",
                    forward.publish.topic,
                    count,
                    forward.publish.payload.len()
                );
            }
            v => {
                println!("{v:?}");
            }
        }
    }
}

```
Yes, `broker.connect()` was missing indeed. It's impossible to understand from the documentation that this function is responsible to TCP binding. It's also odd that it's not part of the other examples there.
Thank you.
@galah92 which `broker.connect` are you talking about?

I will update the example file to include the subscription part.
I meant `subscribe`, sorry about that.

Fix patch:
diff --git a/rumqttd/examples/singlenode.rs b/rumqttd/examples/singlenode.rs
index 329b3cc0f..bcab216c3 100644
--- a/rumqttd/examples/singlenode.rs
+++ b/rumqttd/examples/singlenode.rs
@@ -3,7 +3,16 @@ use rumqttd::{Broker, Config, Notification};
 use std::thread;
 
 fn main() {
-    pretty_env_logger::init();
+    let builder = tracing_subscriber::fmt()
+        .pretty()
+        .with_line_number(false)
+        .with_file(false)
+        .with_thread_ids(false)
+        .with_thread_names(false);
+
+    builder
+        .try_init()
+        .expect("initialized subscriber succesfully");
 
     // As examples are compiled as seperate binary so this config is current path dependent. Run it
     // from root of this crate
@@ -17,11 +26,13 @@ fn main() {
     dbg!(&config);
 
     let mut broker = Broker::new(config);
-    let (_link_tx, mut link_rx) = broker.link("singlenode").unwrap();
+    let (mut link_tx, mut link_rx) = broker.link("singlenode").unwrap();
     thread::spawn(move || {
         broker.start().unwrap();
     });
 
+    link_tx.subscribe("#").unwrap();
+
     let mut count = 0;
     loop {
         let notification = match link_rx.recv().unwrap() {


Test patch: