broker: switch rumqttd branch to asdf

broker/Cargo.lock

@@ -2826,7 +2826,7 @@ dependencies = [
 [[package]]
 name = "rumqttd"
 version = "0.15.0"
-source = "git+https://github.com/Evanfeenstra/rumqtt?branch=sphinx-3#c087e39bf297c157d96de986f85733bfa0bb9f39"
+source = "git+https://github.com/Evanfeenstra/rumqtt?branch=sphinx-asdf#3dd8727c13be54ee48f9e5efb664fe7f94a5cb25"
 dependencies = [
  "async-tungstenite",
  "axum",

broker/Cargo.toml

@@ -22,7 +22,7 @@ once_cell         = "1.12.0"
 pretty_env_logger = "0.4.0"
 rocket            = { version = "0.5.0-rc.2", features = ["json"] }
 rumqttc           = "0.12.0"
-rumqttd           = { git = "https://github.com/Evanfeenstra/rumqtt", branch = "sphinx-3", features = ["websocket"] }
+rumqttd           = { git = "https://github.com/Evanfeenstra/rumqtt", branch = "sphinx-asdf", features = ["websocket"] }
 secp256k1         = { version = "0.24.0", features = ["rand-std", "bitcoin_hashes"] }
 serde             = { version = "1.0", features = ["derive"] }
 serde_json        = "1.0"

broker/src/main.rs

@@ -1,4 +1,3 @@
-// #![feature(once_cell)]
 mod chain_tracker;
 mod conn;
 mod error_log;
@@ -20,7 +19,7 @@ use rocket::tokio::{
     self,
     sync::{broadcast, mpsc},
 };
-use rumqttd::{oneshot as std_oneshot, AuthMsg};
+use rumqttd::{oneshot as std_oneshot, AuthMsg, AuthType};
 use std::env;
 use std::sync::Arc;
 use url::Url;
@@ -133,11 +132,14 @@ pub fn broker_setup(
     std::thread::spawn(move || {
         while let Ok(am) = auth_rx.recv() {
             let pubkey = current_pubkey();
-            let (ok, new_pubkey) = check_auth(&am.username, &am.password, &pubkey);
+            let (ok, new_pubkey) = match am.msg {
+                AuthType::Login(login) => check_auth(&login.username, &login.password, &pubkey),
+                _ => (true, None),
+            };
             if let Some(np) = new_pubkey {
                 conns_set_pubkey(np);
             }
-            let _ = am.reply.send(ok);
+            let _ = am.tx.send(ok);
         }
     });
 

broker/src/mqtt.rs

@@ -20,18 +20,15 @@ pub fn start_broker(
     let conf = config(settings);
     // println!("CONF {:?}", conf);
 
-    let mut broker = Broker::new(conf);
+    let mut broker = Broker::new(conf, Some(auth_sender));
 
     let (mut link_tx, mut link_rx) = broker.link("localclient")?;
 
     let _ = link_tx.subscribe(format!("+/{}", topics::HELLO));
     let _ = link_tx.subscribe(format!("+/{}", topics::BYE));
 
-    let auth_sender_ = auth_sender;
     std::thread::spawn(move || {
-        broker
-            .start(Some(auth_sender_))
-            .expect("could not start broker");
+        broker.start().expect("could not start broker");
     });
 
     // connected/disconnected status alerts