#4 - implement MQTT, without manual acknowledgments

Lance-Drane · Lance-Drane · commit 73f4735bd0f8 · 2025-09-08T17:32:49.000-04:00
Signed-off-by: Lance-Drane &lt;Lance-Drane@users.noreply.github.com&gt;
diff --git a/README.md b/README.md
@@ -84,15 +84,18 @@ Specific configuration structs are in `proxy-http-server/src/configuration.rs` a
 
 ## Setup
 
-### Using the RabbitMQ web management UIs
+### Using the RabbitMQ web management UIs (AMQP)
 
 These instructions assume you are using the docker compose configuration and the default `conf.yaml` configurations for each.
 
 1) Make sure that you run `docker compose up -d` from the root directory, to start both brokers and their associated management UIs.
 2) Make sure that you have both applications started (do NOT start more than 1 of each). Each application should be connected to a separate broker.
 3) To login to the broker that the server instance uses, go to localhost:15672, username `intersect_username`, password `intersect_password`
 4) To login to the broker that the client instance uses, go to localhost:15673, username `intersect_username`, password `intersect_password`
-5) On each application, click on the `Exchanges` tab, and click on the `intersect-messages` exchange.
+5)
+    - IF AMQP: On each application, click on the `Exchanges` tab, and click on the `intersect-messages` exchange.
+    - IF MQTT: On each application, click on the `Queues and Streams` tab, then click on the queue (it should look like `mqtt-subscription-proxy-http-clientqos1` or `mqtt-subscription-proxy-http-serverqos1`).
+
 6) Make sure that the `Publish message` dropdown is expanded, select the large text area which is labeled with `Payload:`
 
 For the application on `localhost:15672`, set the payload to below (no newlines):
diff --git a/proxy-http-client/conf.yaml b/proxy-http-client/conf.yaml
@@ -8,8 +8,12 @@ broker:
   password: intersect_password
   host: "127.0.0.1"
   # note: differs from other config file (two separate brokers used)
+  # -- IF AMQP --
   port: 5673
   protocol: "amqp"
+  # -- IF MQTT --
+  #port: 1884
+  #protocol: "mqtt"
 # use amqp topic notation, note that the system is different from the other configuration
 topic_prefix: "organization.facility.system2"  # CHANGE THIS PER DEPLOYMENT!!!
 log_level: "debug"
diff --git a/proxy-http-client/src/event_source.rs b/proxy-http-client/src/event_source.rs
@@ -20,6 +20,10 @@ async fn send_message(
     }
     let (topic, data) = es_data_result.unwrap();
 
+    if let Err(e) = proto_handler.preverify_publish(&topic) {
+        tracing::warn!("invalid topic name -- {e}");
+        return Ok(());
+    }
     proto_handler.publish_message(&topic, data).await
 }
 
diff --git a/proxy-http-server/conf.yaml b/proxy-http-server/conf.yaml
@@ -5,8 +5,12 @@ broker:
   password: intersect_password
   host: "127.0.0.1"
   # note: differs from other config file (two separate brokers used)
+  # -- IF AMQP --
   port: 5672
   protocol: "amqp"
+  # -- IF MQTT --
+  #port: 1883
+  #protocol: "mqtt"
 # use amqp topic notation
 topic_prefix: "organization.facility.system"  # CHANGE THIS PER DEPLOYMENT!!!
 log_level: "debug"
diff --git a/shared-deps/src/protocols/amqp/publish.rs b/shared-deps/src/protocols/amqp/publish.rs
@@ -1,7 +1,8 @@
 use amqprs::{channel::BasicPublishArguments, BasicProperties};
 use deadpool_amqprs::Pool;
 
-use crate::protocols::amqp::utils::{get_channel, is_routing_key_compliant};
+use crate::protocols::amqp::utils::get_channel;
+use crate::protocols::proxy::is_routing_key_compliant;
 use crate::{
     intersect_messaging::INTERSECT_MESSAGE_EXCHANGE, protocols::interfaces::PublishProtoHandler,
 };
@@ -34,20 +35,13 @@ impl PublishProtoHandler for AmqpPublishProtoHandler {
     fn preverify_publish(&self, topic: &str) -> Result<(), String> {
         if !is_routing_key_compliant(topic) {
             return Err(format!(
-                "'{topic}' does not meet the AMQP routing key specification"
+                "'{topic}' does not meet the INTERSECT proxy-app routing key specification."
             ));
         }
         Ok(())
     }
 
     async fn publish_message(&self, topic: &str, data: String) -> Result<(), &str> {
-        if !is_routing_key_compliant(topic) {
-            tracing::warn!(
-                "{} is not a valid AMQP topic name, will not attempt publish",
-                topic
-            );
-            return Ok(());
-        }
         let connection = self.pool.get().await.map_err(|e| {
             tracing::error!(error = ?e, "cannot connect to broker");
             "cannot connect to broker"
diff --git a/shared-deps/src/protocols/amqp/subscribe.rs b/shared-deps/src/protocols/amqp/subscribe.rs
@@ -12,7 +12,6 @@ use crate::intersect_messaging::{make_eventsource_data, should_message_passthrou
 use crate::protocols::amqp::utils::{get_channel, verify_connection_pool};
 use crate::protocols::interfaces::{HttpBroadcast, SubscribeProtoHandler};
 
-#[derive(Clone)]
 pub struct AmqpSubscribeProtoHandler {
     pool: Pool,
     /// application_name is used for the hardcoded queue name and for debugging purposes
diff --git a/shared-deps/src/protocols/amqp/utils.rs b/shared-deps/src/protocols/amqp/utils.rs
@@ -12,7 +12,7 @@ use crate::{configuration::BrokerSettings, intersect_messaging::INTERSECT_MESSAG
 ///
 /// NOTE: calling this function cannot fail on its own, you have to call pool.get().await for it to fail
 #[must_use]
-pub fn get_connection_pool(connection_details: &BrokerSettings) -> Pool {
+pub(crate) fn get_connection_pool(connection_details: &BrokerSettings) -> Pool {
     let mut args = OpenConnectionArguments::new(
         &connection_details.host,
         connection_details.port,
@@ -29,7 +29,10 @@ pub fn get_connection_pool(connection_details: &BrokerSettings) -> Pool {
 ///
 /// # Errors
 ///   - Errors if can't connect to server or has invalid permissions
-pub async fn verify_connection_pool(pool: &Pool, queue_name_src: &str) -> Result<(), String> {
+pub(crate) async fn verify_connection_pool(
+    pool: &Pool,
+    queue_name_src: &str,
+) -> Result<(), String> {
     let connection = pool
         .get()
         .await
@@ -83,7 +86,7 @@ pub async fn verify_connection_pool(pool: &Pool, queue_name_src: &str) -> Result
 ///
 /// # Errors
 ///   - Errors if failure to communicate with broker server
-pub async fn get_channel(connection: &Connection) -> Result<Channel, amqprs::error::Error> {
+pub(crate) async fn get_channel(connection: &Connection) -> Result<Channel, amqprs::error::Error> {
     let channel = connection.open_channel(None).await?;
     channel.register_callback(DefaultChannelCallback).await?;
     Ok(channel)
@@ -105,12 +108,3 @@ async fn make_exchange(channel: &Channel) -> Result<(), amqprs::error::Error> {
         )
         .await
 }
-
-/// make sure that the routing key is valid for AMQP
-///
-/// we do not permit publishing on wildcards
-#[must_use]
-pub(crate) fn is_routing_key_compliant(key: &str) -> bool {
-    !key.chars()
-        .any(|c| !c.is_alphanumeric() && c != '-' && c != '_' && c != '.' && c != ':')
-}
diff --git a/shared-deps/src/protocols/mod.rs b/shared-deps/src/protocols/mod.rs
@@ -1,3 +1,4 @@
 pub mod amqp;
 pub mod interfaces;
 pub mod mqtt;
+pub mod proxy;
diff --git a/shared-deps/src/protocols/mqtt/init.rs b/shared-deps/src/protocols/mqtt/init.rs
@@ -5,7 +5,10 @@ use secrecy::ExposeSecret;
 
 use crate::{
     configuration::BrokerSettings,
-    protocols::mqtt::{publish::MqttPublishProtoHandler, subscribe::MqttSubscribeProtoHandler},
+    protocols::mqtt::{
+        publish::MqttPublishProtoHandler, subscribe::MqttSubscribeProtoHandler,
+        utils::subscribe_all,
+    },
 };
 
 /// Sets up the MQTT proto handlers, and verifies that we can connect to the MQTT broker.
@@ -25,7 +28,8 @@ pub async fn init_mqtt_proto_handlers(
     // TODO may want to handle message sizes >= 4 GiB
     mqtt_options.set_max_packet_size(1 << 32, 1 << 32);
     mqtt_options.set_clean_session(false);
-    mqtt_options.set_manual_acks(true);
+    // TODO - implement manual acks
+    // mqtt_options.set_manual_acks(true);
     mqtt_options.set_keep_alive(Duration::from_secs(60));
 
     let (mqtt_client, mut mqtt_event_loop) = AsyncClient::new(mqtt_options, 256);
@@ -58,6 +62,9 @@ pub async fn init_mqtt_proto_handlers(
         }
     }
 
+    // listen for every single message on the exchange, we must do this due to the way userspace messages work
+    subscribe_all(&mqtt_client).await?;
+
     let publish_handler = MqttPublishProtoHandler::new(application_name, mqtt_client.clone());
     let subscribe_handler =
         MqttSubscribeProtoHandler::new(application_name, mqtt_client, mqtt_event_loop);
diff --git a/shared-deps/src/protocols/mqtt/mod.rs b/shared-deps/src/protocols/mqtt/mod.rs
@@ -1,3 +1,4 @@
 pub mod init;
 pub mod publish;
 pub mod subscribe;
+pub mod utils;
diff --git a/shared-deps/src/protocols/mqtt/publish.rs b/shared-deps/src/protocols/mqtt/publish.rs
@@ -1,6 +1,7 @@
-use rumqttc::AsyncClient;
+use rumqttc::{AsyncClient, QoS};
 
 use crate::protocols::interfaces::PublishProtoHandler;
+use crate::protocols::proxy::is_routing_key_compliant;
 
 #[derive(Clone)]
 pub struct MqttPublishProtoHandler {
@@ -28,9 +29,20 @@ impl MqttPublishProtoHandler {
 
 impl PublishProtoHandler for MqttPublishProtoHandler {
     fn preverify_publish(&self, topic: &str) -> Result<(), String> {
+        if !is_routing_key_compliant(topic) {
+            return Err(format!(
+                "'{topic}' does not meet the INTERSECT proxy-app routing key specification."
+            ));
+        }
         Ok(())
     }
     async fn publish_message(&self, topic: &str, data: String) -> Result<(), &str> {
-        Err("yippee")
+        self.mqtt_client
+            .publish(topic, QoS::AtLeastOnce, true, data.into_bytes())
+            .await
+            .map_err(|err| {
+                tracing::error!("Could not publish message -- {err}");
+                "could not publish message"
+            })
     }
 }
diff --git a/shared-deps/src/protocols/mqtt/subscribe.rs b/shared-deps/src/protocols/mqtt/subscribe.rs
@@ -2,9 +2,13 @@ use std::sync::Arc;
 
 use tokio::sync::oneshot::Receiver;
 
-use rumqttc::{AsyncClient, EventLoop};
+use rumqttc::{AsyncClient, EventLoop, Publish};
 
-use crate::protocols::interfaces::{HttpBroadcast, SubscribeProtoHandler};
+use crate::{
+    intersect_messaging::{make_eventsource_data, should_message_passthrough},
+    protocols::interfaces::{HttpBroadcast, SubscribeProtoHandler},
+    protocols::mqtt::utils::mqtt_topic_to_proxy_topic,
+};
 
 pub struct MqttSubscribeProtoHandler {
     mqtt_client: AsyncClient,
@@ -47,7 +51,6 @@ impl SubscribeProtoHandler for MqttSubscribeProtoHandler {
                 self.mqtt_client,
                 self.mqtt_event_loop,
                 config_topic,
-                self.application_name,
                 broadcaster,
                 killswitch,
             )
@@ -58,13 +61,133 @@ impl SubscribeProtoHandler for MqttSubscribeProtoHandler {
 
 async fn broker_consumer_loop_inner(
     mqtt_client: AsyncClient,
-    mqtt_event_loop: EventLoop,
+    mut mqtt_event_loop: EventLoop,
     config_topic: String,
-    queue_name_src: &str,
     broadcaster: Arc<impl HttpBroadcast + Send + Sync + 'static>,
     // calls recv() once the EventSource loop or HTTP server catches an OS signal
     mut killswitch: Receiver<()>,
 ) {
-    let mut needs_reverification = false;
-    'connection_loop: loop {}
+    let still_connected = loop {
+        tokio::select! {
+            _ = &mut killswitch => {
+                break true;
+            },
+            event_loop_pool = mqtt_event_loop.poll() => {
+                match event_loop_pool {
+                    Ok(event) => {
+                        match event {
+                            rumqttc::Event::Incoming(packet) => {
+                                match packet {
+                                    rumqttc::Packet::Publish(publish_packet) => {
+                                        consume_message(
+                                            publish_packet,
+                                            &mqtt_client,
+                                            &config_topic,
+                                            &broadcaster,
+                                            &mut killswitch,
+                                        ).await;
+                                    },
+                                    packet => {
+                                        tracing::debug!("Incoming packet -- {packet:?}");
+                                    },
+                                }
+                            },
+                            rumqttc::Event::Outgoing(outgoing) => {
+                                tracing::debug!("Outgoing packet -- {outgoing:?}");
+                            },
+                        }
+                    },
+                    Err(conn_err) => {
+                        tracing::warn!("Consumer lost connection to broker, retrying in 5 seconds. Specifics: {conn_err}");
+                        let retry_wait = tokio::time::sleep(std::time::Duration::from_secs(5));
+                        tokio::pin!(retry_wait);
+                        tokio::select! {
+                            _ = &mut killswitch => {
+                                // HTTP component has been shut down, no point in waiting
+                                tracing::warn!("Shutting down while attempting to reconnect to broker");
+                                break false;
+                            },
+                            () = &mut retry_wait => {
+                                continue;
+                            },
+                        }
+                    },
+                }
+            },
+        }
+    };
+    if still_connected {
+        let _ = mqtt_client.disconnect().await;
+    }
+}
+
+async fn consume_message(
+    publish_packet: Publish,
+    // TODO - the client will eventually manually ACK messages
+    _mqtt_client: &AsyncClient,
+    config_topic: &str,
+    broadcaster: &Arc<impl HttpBroadcast + Send + Sync + 'static>,
+    killswitch: &mut Receiver<()>,
+) {
+    let mut should_ack = true;
+    if publish_packet.dup {
+        tracing::warn!("message was redelivered");
+    }
+
+    match String::from_utf8(publish_packet.payload.to_vec()) {
+        Ok(utf8_data) => {
+            tracing::debug!("got raw message data from broker: {}", &utf8_data);
+            match should_message_passthrough(&utf8_data, config_topic) {
+                Err(e) => {
+                    // This should generally not be seen, so log it as a warning
+                    tracing::warn!(error = ?e, "message is valid UTF-8 but not INTERSECT JSON");
+                }
+                Ok(false) => {
+                    tracing::debug!(
+                        "message source is not from this system, will not broadcast it"
+                    );
+                }
+                Ok(true) => {
+                    let topic = mqtt_topic_to_proxy_topic(&publish_packet.topic);
+                    match make_eventsource_data(&topic, &utf8_data) {
+                        Err(_) => {}
+                        Ok(event) => {
+                            tracing::debug!(
+                                "Consume message {}, data: {}",
+                                publish_packet.pkid,
+                                event
+                            );
+                            // TODO handle this better later, see broadcast() documentation for details.
+                            tokio::select! {
+                                _ = killswitch => {
+                                    // WARNING: in the client implementation, this may happen while waiting on a response, resulting in us rejecting a message we actually passed through successfully
+                                    // this would only happen if we actually call publish_event_to_http(), if the killswitch was toggled before reaching here we will always do the killswitch branch.
+                                    tracing::warn!("Got message from broker but did not send it over HTTP, the message will be rejected.");
+                                    should_ack = false;
+                                },
+                                http_result = broadcaster.publish_event_to_http(event) => {
+                                    if !http_result {
+                                        tracing::warn!("Some clients may not have gotten a message, the message will be rejected.");
+                                        should_ack = false;
+                                    }
+                                },
+                            }
+                        }
+                    }
+                }
+            }
+        }
+        Err(e) => {
+            // this should generally not be seen, so log as an error
+            tracing::error!(error = ?e, "message data is not UTF-8, cannot be forwarded over SSE");
+        }
+    }
+
+    // TODO implement acknowledgments
+    if !should_ack {
+        tracing::warn!(
+            "We SHOULD be rejecting message {}, ACK is not implemented yet though",
+            publish_packet.pkid
+        );
+    }
 }
diff --git a/shared-deps/src/protocols/mqtt/utils.rs b/shared-deps/src/protocols/mqtt/utils.rs
@@ -0,0 +1,15 @@
+use rumqttc::{AsyncClient, QoS};
+
+pub(crate) async fn subscribe_all(mqtt_client: &AsyncClient) -> Result<(), String> {
+    mqtt_client
+        .subscribe("#", QoS::AtLeastOnce)
+        .await
+        .map_err(|e| e.to_string())
+}
+
+/// convert an MQTT topic representation to the proxy's expected topic representation (which mirrors AMQP)
+///
+/// this should generally happen at some point in the subscribe loop
+pub(crate) fn mqtt_topic_to_proxy_topic(topic: &str) -> String {
+    topic.replace("/", ".")
+}
diff --git a/shared-deps/src/protocols/proxy.rs b/shared-deps/src/protocols/proxy.rs

Original file line number	Diff line number	Diff line change
`@@ -20,6 +20,10 @@ async fn send_message(`
`20`	`20`	`}`
`21`	`21`	`let (topic, data) = es_data_result.unwrap();`
`22`	`22`
	`23`	`+ if let Err(e) = proto_handler.preverify_publish(&topic) {`
	`24`	`+ tracing::warn!("invalid topic name -- {e}");`
	`25`	`+ return Ok(());`
	`26`	`+ }`
`23`	`27`	`proto_handler.publish_message(&topic, data).await`
`24`	`28`	`}`
`25`	`29`