update broker to new rumqttd

broker/Cargo.toml

@@ -14,7 +14,7 @@ vls-protocol = { git = "https://gitlab.com/lightning-signer/validating-lightning
 vls-proxy = { git = "https://gitlab.com/lightning-signer/validating-lightning-signer.git" }
 vls-frontend = { git = "https://gitlab.com/lightning-signer/validating-lightning-signer.git" }
 vls-protocol-client = { git = "https://gitlab.com/lightning-signer/validating-lightning-signer.git" }
-rumqttd = { git = "https://github.com/Evanfeenstra/rumqtt", branch = "metrics" } 
+rumqttd = { git = "https://github.com/Evanfeenstra/rumqtt", branch = "sphinx" } 
 pretty_env_logger = "0.4.0"
 confy = "0.4.0"
 secp256k1 = { version = "0.24.0", features = ["rand-std", "bitcoin_hashes"] }

broker/src/mqtt.rs

@@ -1,183 +1,143 @@
 use crate::util::Settings;
 use crate::{ChannelReply, ChannelRequest};
-use librumqttd::{
+use rocket::tokio::task;
-    async_locallink,
-    consolelink::{self, ConsoleLink},
-    rumqttlog::router::ConnectionMetrics,
-    Config,
-};
 use rocket::tokio::time::timeout;
 use rocket::tokio::{self, sync::broadcast, sync::mpsc};
+use rumqttd::{Alert, AlertEvent, Broker, Config, Notification};
 use sphinx_signer::sphinx_glyph::topics;
-use std::sync::Arc;
-use std::sync::{LazyLock, Mutex};
 use std::time::Duration;
 
 // must get a reply within this time, or disconnects
 const REPLY_TIMEOUT_MS: u64 = 10000;
 
-static CONNECTED: LazyLock<Mutex<bool>> = LazyLock::new(|| Mutex::new(false));
-fn set_connected(b: bool) {
-    *CONNECTED.lock().unwrap() = b;
-}
-fn get_connected() -> bool {
-    *CONNECTED.lock().unwrap()
-}
-
 pub async fn start_broker(
     mut receiver: mpsc::Receiver<ChannelRequest>,
     status_sender: mpsc::Sender<bool>,
     error_sender: broadcast::Sender<Vec<u8>>,
     expected_client_id: &str,
     settings: Settings,
-) {
+) -> anyhow::Result<()> {
-    let config = config(settings);
+    let conf = config(settings);
     let client_id = expected_client_id.to_string();
 
-    let (mut router, servers, builder) = async_locallink::construct(config.clone());
+    let broker = Broker::new(conf);
+    let mut alerts = broker
+        .alerts(vec![
+            // "/alerts/error/+".to_string(),
+            "/alerts/event/connect/+".to_string(),
+            "/alerts/event/disconnect/+".to_string(),
+        ])
+        .unwrap();
 
-    // std thread for the router
+    // connected/disconnected status alerts
-    std::thread::spawn(move || {
+    let status_sender_ = status_sender.clone();
-        log::info!("start mqtt router");
+    let alerts_handle = tokio::spawn(async move {
-        router.start().expect("could not start router");
+        loop {
-    });
+            let alert = alerts.poll();
-
+            println!("Alert: {alert:?}");
-    tokio::spawn(async move {
+            match alert.1 {
-        log::info!("start mqtt relayer and localclient");
+                Alert::Event(cid, event) => {
-        let (msg_tx, mut msg_rx): (mpsc::Sender<Vec<u8>>, mpsc::Receiver<Vec<u8>>) =
+                    if cid == client_id {
-            mpsc::channel(1000);
+                        if let Some(status) = match event {
-        let (mut link_tx, mut link_rx) = builder.clone().connect("localclient", 200).await.unwrap();
+                            AlertEvent::Connect => Some(true),
-        link_tx
+                            AlertEvent::Disconnect => Some(false),
-            .subscribe([topics::VLS_RETURN, topics::CONTROL_RETURN, topics::ERROR])
+                            _ => None,
-            .await
+                        } {
-            .unwrap();
+                            let _ = status_sender_.send(status).await;
-
-        let router_tx = builder.router_tx();
-        let status_sender_ = status_sender.clone();
-        tokio::spawn(async move {
-            let config = config.clone().into();
-            let router_tx = router_tx.clone();
-            let console: Arc<ConsoleLink> = Arc::new(ConsoleLink::new(config, router_tx));
-            loop {
-                let metrics = consolelink::request_metrics(console.clone(), client_id.clone());
-                if let Some(c) = metrics_to_status(metrics, get_connected()) {
-                    set_connected(c);
-                    log::info!("connection status changed to: {}", c);
-                    let _ = status_sender_.send(c).await;
-                }
-                tokio::time::sleep(Duration::from_millis(500)).await;
-            }
-        });
-
-        let sub_task = tokio::spawn(async move {
-            while let Ok(message) = link_rx.recv().await {
-                for payload in message.payload {
-                    if message.topic == topics::ERROR {
-                        error_sender.send(payload.to_vec()).expect("asdfasdf");
-                    } else if let Err(e) = msg_tx.send(payload.to_vec()).await {
-                        log::warn!("pub err {:?}", e);
-                    }
-                }
-            }
-        });
-
-        let relay_task = tokio::spawn(async move {
-            while let Some(msg) = receiver.recv().await {
-                link_tx
-                    .publish(&msg.topic, false, msg.message)
-                    .await
-                    .expect("could not mqtt pub");
-                match timeout(Duration::from_millis(REPLY_TIMEOUT_MS), msg_rx.recv()).await {
-                    Ok(reply) => {
-                        if let Err(_) = msg.reply_tx.send(ChannelReply {
-                            reply: reply.unwrap(),
-                        }) {
-                            log::warn!("could not send on reply_tx");
                         }
                     }
-                    Err(e) => {
-                        log::warn!("reply_tx timed out {:?}", e);
-                        set_connected(false);
-                        let _ = status_sender.send(false).await;
-                    }
                 }
+                _ => (),
             }
-        });
+            tokio::time::sleep(Duration::from_millis(333)).await;
-
+        }
-        servers.await;
-        sub_task.await.unwrap();
-        relay_task.await.unwrap();
     });
 
-    // give one second for router to spawn listeners
+    // msg forwarding
-    tokio::time::sleep(std::time::Duration::from_secs(10)).await;
+    let (mut link_tx, mut link_rx) = broker.link("localclient")?;
-}
+    let (msg_tx, mut msg_rx): (mpsc::Sender<Vec<u8>>, mpsc::Receiver<Vec<u8>>) =
-
+        mpsc::channel(1000);
-fn metrics_to_status(metrics: ConnectionMetrics, client_connected: bool) -> Option<bool> {
+    link_tx.subscribe(topics::VLS_RETURN)?;
-    match metrics.tracker() {
+    link_tx.subscribe(topics::CONTROL_RETURN)?;
-        Some(t) => {
+    link_tx.subscribe(topics::ERROR)?;
-            // wait for subscription to be sure
+    let sub_task = tokio::spawn(async move {
-            if t.concrete_subscriptions_count() > 0 {
+        while let Ok(message) = link_rx.recv() {
-                if !client_connected {
+            if let Some(n) = message {
-                    Some(true) // changed to true
+                match n {
-                } else {
+                    Notification::Forward(f) => {
-                    None
+                        if f.publish.topic == topics::ERROR {
+                            let _ = error_sender.send(f.publish.topic.to_vec());
+                        } else {
+                            let _ = msg_tx.send(f.publish.payload.to_vec()).await;
+                        }
+                    }
+                    _ => (),
+                };
+            }
+        }
+    });
+    let relay_task = tokio::spawn(async move {
+        while let Some(msg) = receiver.recv().await {
+            link_tx.publish(msg.topic, msg.message);
+            match timeout(Duration::from_millis(REPLY_TIMEOUT_MS), msg_rx.recv()).await {
+                Ok(reply) => {
+                    if let Err(_) = msg.reply_tx.send(ChannelReply {
+                        reply: reply.unwrap(),
+                    }) {
+                        log::warn!("could not send on reply_tx");
+                    }
+                }
+                Err(e) => {
+                    log::warn!("reply_tx timed out {:?}", e);
+                    let _ = status_sender.send(false).await;
                 }
-            } else {
-                None
             }
         }
-        None => {
+    });
-            if client_connected {
+
-                Some(false)
+    alerts_handle.await?;
-            } else {
+    sub_task.await?;
-                None
+    relay_task.await?;
-            }
+    Ok(())
-        }
-    }
 }
 
 fn config(settings: Settings) -> Config {
-    use librumqttd::rumqttlog::Config as RouterConfig;
+    use rumqttd::{ConnectionSettings, ConsoleSettings, ServerSettings, SphinxLoginCredentials};
-    use librumqttd::{ConnectionSettings, ConsoleSettings, ServerSettings, SphinxLoginCredentials};
     use std::collections::HashMap;
     use std::net::{Ipv4Addr, SocketAddrV4};
-    use std::path::PathBuf;
+    let router = rumqttd::RouterConfig {
-    let id = 0;
+        instant_ack: true,
-    let router = RouterConfig {
-        id,
-        dir: PathBuf::from("/tmp/rumqttd"),
         max_segment_size: 10240,
         max_segment_count: 10,
         max_connections: 10001,
+        ..Default::default()
     };
     let mut servers = HashMap::new();
     servers.insert(
-        id.to_string(),
+        "sphinx-broker".to_string(),
         ServerSettings {
-            cert: None,
+            name: "sphinx-broker".to_string(),
             listen: SocketAddrV4::new(Ipv4Addr::new(0, 0, 0, 0), settings.mqtt_port).into(),
             next_connection_delay_ms: 1,
             connections: ConnectionSettings {
                 connection_timeout_ms: 5000,
-                max_client_id_len: 256,
                 throttle_delay_ms: 0,
                 max_payload_size: 5120,
                 max_inflight_count: 200,
                 max_inflight_size: 1024,
-                login_credentials: None,
+                auth: None,
                 sphinx_auth: Some(SphinxLoginCredentials { within: None }),
+                dynamic_filters: false,
             },
+            tls: None,
         },
     );
     Config {
-        id,
+        id: 0,
-        servers,
+        v4: servers,
         router,
-        console: ConsoleSettings {
+        console: ConsoleSettings::new("0.0.0.0:3030"),
-            listen: SocketAddrV4::new(Ipv4Addr::new(0, 0, 0, 0), 3030).into(),
-        },
         cluster: None,
-        replicator: None,
+        ..Default::default()
     }
 }