#4 - refactor application to support multiple protocols

MQTT currently just a stub

Signed-off-by: Lance-Drane <Lance-Drane@users.noreply.github.com>

Author: Lance-Drane

proxy-http-client/conf.yaml

@@ -9,6 +9,7 @@ broker:
   host: "127.0.0.1"
   # note: differs from other config file (two separate brokers used)
   port: 5673
+  protocol: "amqp"
 # 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"

proxy-http-client/src/event_source.rs

@@ -1,57 +1,33 @@
-use deadpool_amqprs::Pool;
 use futures::StreamExt;
 use reqwest_eventsource::{Event, EventSource};
 use secrecy::ExposeSecret;
 
 use intersect_ingress_proxy_common::intersect_messaging::extract_eventsource_data;
-use intersect_ingress_proxy_common::protocols::amqp::{
-    get_channel, is_routing_key_compliant, publish::amqp_publish_message,
-};
+use intersect_ingress_proxy_common::protocols::ProtoHandler;
 use intersect_ingress_proxy_common::server_paths::SUBSCRIBE_URL;
 use intersect_ingress_proxy_common::signals::wait_for_os_signal;
 
 use crate::configuration::ExternalProxy;
 
 /// Return Err only if we weren't able to publish a correct message to the broker, invalid messages are ignored
-async fn send_message(message: String, connection_pool: &Pool) -> Result<(), String> {
+async fn send_message(message: String, proto_handler: &impl ProtoHandler) -> Result<(), &str> {
     let es_data_result = extract_eventsource_data(&message);
     if es_data_result.is_err() {
         return Ok(());
     }
     let (topic, data) = es_data_result.unwrap();
-    if !is_routing_key_compliant(&topic) {
-        tracing::warn!(
-            "{} is not a valid AMQP topic name, will not attempt publish",
-            topic
-        );
-        return Ok(());
-    }
-    tracing::debug!("Publishing message with topic: {}", &topic);
-
-    let connection = connection_pool.get().await.map_err(|_| {
-        "WARNING: Couldn't get connection, message received from other proxy was NOT published on our own broker."
-            .to_string()
-    })?;
 
-    let channel = get_channel(&connection).await.map_err(|_| {
-        "WARNING: Couldn't get channel, message received from other proxy was NOT published on our own broker."
-            .to_string()
-    })?;
-
-    match amqp_publish_message(channel, &topic, data).await {
-        Ok(()) => Ok(()),
-        Err(()) => Err(
-            "WARNING: message received from other proxy was NOT published on our own broker."
-                .into(),
-        ),
-    }
+    proto_handler.publish_message(&topic, data).await
 }
 
 /// Return value - exit code to use
 ///
 /// # Panics
 ///   - Inner API could potentially panic but is currently not expected to do so
-pub async fn event_source_loop(other_proxy: ExternalProxy, connection_pool: Pool) -> i32 {
+pub async fn event_source_loop(
+    other_proxy: ExternalProxy,
+    connection_pool: &impl ProtoHandler,
+) -> i32 {
     let mut es = EventSource::new(
         reqwest::Client::new()
             .get(format!("{}{}", &other_proxy.url, SUBSCRIBE_URL))
@@ -79,7 +55,7 @@ pub async fn event_source_loop(other_proxy: ExternalProxy, connection_pool: Pool
                                 tracing::info!("connected to {}", &other_proxy.url);
                             },
                             Ok(Event::Message(message)) => {
-                                if let Err(e) = send_message(message.data, &connection_pool).await {
+                                if let Err(e) = send_message(message.data, connection_pool).await {
                                     tracing::error!(e);
                                 };
                             },

proxy-http-client/src/lib.rs

@@ -1,3 +1,5 @@
 pub mod configuration;
 pub mod event_source;
 pub mod poster;
+
+pub const APPLICATION_NAME: &str = "proxy-http-client";

proxy-http-client/src/main.rs

@@ -3,46 +3,26 @@ use std::sync::Arc;
 use tokio::sync::oneshot;
 
 use intersect_ingress_proxy_common::configuration::get_configuration;
-use intersect_ingress_proxy_common::protocols::amqp::{
-    get_connection_pool, subscribe::broker_consumer_loop, verify_connection_pool,
+use intersect_ingress_proxy_common::protocols::{
+    amqp::AmqpProtoHandler, mqtt::MqttProtoHandler, ProtoHandler,
 };
 use intersect_ingress_proxy_common::telemetry::{
     get_json_subscriber, get_pretty_subscriber, init_subscriber,
 };
 
-use proxy_http_client::{configuration::Settings, event_source::event_source_loop, poster::Poster};
-
-const APPLICATION_NAME: &str = "proxy-http-client";
+use proxy_http_client::{
+    configuration::Settings, event_source::event_source_loop, poster::Poster, APPLICATION_NAME,
+};
 
 // Muslc has a slow allocator, but we can only use jemalloc on 64-bit systems since jemalloc doesn't support i686.
 #[cfg(all(target_env = "musl", target_pointer_width = "64"))]
 #[global_allocator]
 static ALLOC: jemallocator::Jemalloc = jemallocator::Jemalloc;
 
-#[tokio::main]
-pub async fn main() -> anyhow::Result<()> {
-    let configuration = get_configuration::<Settings>().expect("Failed to read configuration");
-
-    // Start logging
-    if configuration.production {
-        let subscriber = get_json_subscriber(
-            APPLICATION_NAME.into(),
-            configuration.log_level.to_string(),
-            std::io::stderr,
-        );
-        init_subscriber(subscriber);
-    } else {
-        let subscriber = get_pretty_subscriber(configuration.log_level.to_string());
-        init_subscriber(subscriber);
-    }
-
-    // set up broker connection pool
-    let pool = get_connection_pool(&configuration.broker);
-    if let Err(msg) = verify_connection_pool(&pool, APPLICATION_NAME).await {
-        tracing::error!(msg);
-        std::process::exit(1);
-    }
-
+async fn begin_execution(
+    configuration: Settings,
+    proto_handler: impl ProtoHandler,
+) -> anyhow::Result<()> {
     // How this works:
     // - Pass in the receiver to the broker consumer loop
     // - In the broker consumer loop, use tokio::select! to wait for rx.recv() at key points
@@ -51,10 +31,8 @@ pub async fn main() -> anyhow::Result<()> {
     // - This allows us to "finish up" publishing a message to our broker before killing the application.
     let (tx, rx) = oneshot::channel();
 
-    let broker_join_handle = broker_consumer_loop(
-        pool.clone(),
+    let broker_join_handle = proto_handler.begin_subscribe_loop(
         configuration.topic_prefix.clone(),
-        APPLICATION_NAME.into(),
         Arc::new(Poster::new(&configuration.other_proxy)),
         rx,
     );
@@ -63,11 +41,50 @@ pub async fn main() -> anyhow::Result<()> {
     drop(configuration);
 
     // this will run until we get an event source error or we catch an OS signal
-    let rc = event_source_loop(other_proxy, pool).await;
+    let rc = event_source_loop(other_proxy, &proto_handler).await;
 
     tracing::info!("Attempting graceful shutdown: No longer listening for events over HTTP");
     drop(tx);
     broker_join_handle.await?;
 
     std::process::exit(rc);
 }
+
+#[tokio::main]
+async fn main() -> anyhow::Result<()> {
+    let configuration = get_configuration::<Settings>().expect("Failed to read configuration");
+
+    // Start logging
+    if configuration.production {
+        let subscriber = get_json_subscriber(
+            APPLICATION_NAME.into(),
+            configuration.log_level.to_string(),
+            std::io::stderr,
+        );
+        init_subscriber(subscriber);
+    } else {
+        let subscriber = get_pretty_subscriber(configuration.log_level.to_string());
+        init_subscriber(subscriber);
+    }
+
+    match configuration.broker.protocol {
+        intersect_ingress_proxy_common::configuration::Protocol::Amqp => {
+            let proto_handler =
+                AmqpProtoHandler::new(&configuration.broker, APPLICATION_NAME).await;
+            if proto_handler.is_err() {
+                tracing::error!("{}", proto_handler.unwrap_err());
+                std::process::exit(1);
+            }
+            begin_execution(configuration, proto_handler.unwrap()).await
+        }
+        intersect_ingress_proxy_common::configuration::Protocol::Mqtt => {
+            let proto_handler =
+                MqttProtoHandler::new(&configuration.broker, APPLICATION_NAME).await;
+            if proto_handler.is_err() {
+                tracing::error!("{}", proto_handler.unwrap_err());
+                std::process::exit(1);
+            }
+            begin_execution(configuration, proto_handler.unwrap()).await
+        }
+    }
+}

proxy-http-client/src/poster.rs

@@ -1,6 +1,6 @@
 use secrecy::ExposeSecret;
 
-use intersect_ingress_proxy_common::protocols::amqp::subscribe::HttpBroadcast;
+use intersect_ingress_proxy_common::protocols::HttpBroadcast;
 use intersect_ingress_proxy_common::server_paths::PUBLISH_URL;
 
 use crate::configuration::ExternalProxy;
@@ -29,10 +29,9 @@ impl HttpBroadcast for Poster {
             .body(event)
             .send()
             .await;
-        if result.is_ok() {
-            let result = result.unwrap();
-            let status = result.status().as_u16();
-            match result.bytes().await {
+        if let Ok(response) = result {
+            let status = response.status().as_u16();
+            match response.bytes().await {
                 Ok(bytes) => tracing::debug!("{:?}", bytes),
                 Err(err) => tracing::debug!("ERROR: {}", err.to_string()),
             }

proxy-http-server/conf.yaml

@@ -6,6 +6,7 @@ broker:
   host: "127.0.0.1"
   # note: differs from other config file (two separate brokers used)
   port: 5672
+  protocol: "amqp"
 # use amqp topic notation
 topic_prefix: "organization.facility.system"  # CHANGE THIS PER DEPLOYMENT!!!
 log_level: "debug"

proxy-http-server/src/broadcaster.rs

@@ -2,7 +2,7 @@ use axum::response::sse::Event;
 use std::sync::Arc;
 use tokio::sync::broadcast;
 
-use intersect_ingress_proxy_common::protocols::amqp::subscribe::HttpBroadcast;
+use intersect_ingress_proxy_common::protocols::HttpBroadcast;
 
 /// This broadcaster is an optimized implementation of a single-producer, multi-consumer channel.
 /// The Broadcaster is effectively the "link" between the broker and the HTTP gateway.

proxy-http-server/src/lib.rs

@@ -1,4 +1,7 @@
 pub mod broadcaster;
 pub mod configuration;
 pub mod routes;
-pub mod webapp;
+pub mod webapp_server;
+pub mod webapp_state;
+
+pub const APPLICATION_NAME: &str = "proxy-http-server";

proxy-http-server/src/main.rs

@@ -1,25 +1,54 @@
+use std::sync::Arc;
+
 use tokio::sync::oneshot;
 
 use intersect_ingress_proxy_common::configuration::get_configuration;
-use intersect_ingress_proxy_common::protocols::amqp::subscribe::broker_consumer_loop;
-use intersect_ingress_proxy_common::protocols::amqp::{
-    get_connection_pool, verify_connection_pool,
+use intersect_ingress_proxy_common::protocols::{
+    amqp::AmqpProtoHandler, mqtt::MqttProtoHandler, ProtoHandler,
 };
 use intersect_ingress_proxy_common::telemetry::{
     get_json_subscriber, get_pretty_subscriber, init_subscriber,
 };
 
 use proxy_http_server::{
-    broadcaster::Broadcaster, configuration::Settings, webapp::WebApplication,
+    broadcaster::Broadcaster,
+    configuration::Settings,
+    webapp_server::{AmqpWebApplication, MqttWebApplication, WebApplication},
+    APPLICATION_NAME,
 };
 
-const APPLICATION_NAME: &str = "proxy-http-server";
-
 // Muslc has a slow allocator, but we can only use jemalloc on 64-bit systems since jemalloc doesn't support i686.
 #[cfg(all(target_env = "musl", target_pointer_width = "64"))]
 #[global_allocator]
 static ALLOC: jemallocator::Jemalloc = jemallocator::Jemalloc;
 
+async fn begin_execution(
+    configuration: Settings,
+    proto_handler: impl ProtoHandler,
+    application: impl WebApplication,
+    broadcaster: Arc<Broadcaster>,
+) -> anyhow::Result<()> {
+    // How this works:
+    // - Pass in the receiver to the broker consumer loop
+    // - In the broker consumer loop, use tokio::select! to wait for rx.recv() at key points
+    // - After the HTTP server has been shut down, drop the sender from memory, which will trigger an rx.recv() command
+    // - This allows us to "finish up" publishing a message to our broker before killing the application.
+    let (tx, rx) = oneshot::channel();
+
+    let broker_join_handle =
+        proto_handler.begin_subscribe_loop(configuration.topic_prefix.clone(), broadcaster, rx);
+
+    drop(configuration);
+
+    application.run_until_stopped().await?;
+
+    tracing::info!("Application shutting down, please wait for cleanups...");
+    drop(tx);
+    broker_join_handle.await?;
+
+    Ok(())
+}
+
 #[tokio::main]
 async fn main() -> anyhow::Result<()> {
     let configuration = get_configuration::<Settings>().expect("Failed to read configuration");
@@ -37,39 +66,39 @@ async fn main() -> anyhow::Result<()> {
         init_subscriber(subscriber);
     }
 
-    // set up broker connection pool
-    let pool = get_connection_pool(&configuration.broker);
-    if let Err(msg) = verify_connection_pool(&pool, APPLICATION_NAME).await {
-        tracing::error!(msg);
-        std::process::exit(1);
-    }
-
     let broadcaster = Broadcaster::new();
-    let application =
-        WebApplication::build(&configuration, broadcaster.clone(), pool.clone()).await?;
-
-    // How this works:
-    // - Pass in the receiver to the broker consumer loop
-    // - In the broker consumer loop, use tokio::select! to wait for rx.recv() at key points
-    // - After the HTTP server has been shut down, drop the sender from memory, which will trigger an rx.recv() command
-    // - This allows us to "finish up" publishing a message to our broker before killing the application.
-    let (tx, rx) = oneshot::channel();
-
-    let broker_join_handle = broker_consumer_loop(
-        pool,
-        configuration.topic_prefix.clone(),
-        APPLICATION_NAME.into(),
-        broadcaster,
-        rx,
-    );
-
-    drop(configuration);
-
-    application.run_until_stopped().await?;
-
-    tracing::info!("Application shutting down, please wait for cleanups...");
-    drop(tx);
-    broker_join_handle.await?;
-
-    Ok(())
+    match configuration.broker.protocol {
+        intersect_ingress_proxy_common::configuration::Protocol::Amqp => {
+            let proto_handler =
+                AmqpProtoHandler::new(&configuration.broker, APPLICATION_NAME).await;
+            if proto_handler.is_err() {
+                tracing::error!("{}", proto_handler.unwrap_err());
+                std::process::exit(1);
+            }
+            let proto_handler = proto_handler.unwrap();
+            let web_server = AmqpWebApplication::build(
+                &configuration,
+                broadcaster.clone(),
+                proto_handler.clone(),
+            )
+            .await?;
+            begin_execution(configuration, proto_handler, web_server, broadcaster).await
+        }
+        intersect_ingress_proxy_common::configuration::Protocol::Mqtt => {
+            let proto_handler =
+                MqttProtoHandler::new(&configuration.broker, APPLICATION_NAME).await;
+            if proto_handler.is_err() {
+                tracing::error!("{}", proto_handler.unwrap_err());
+                std::process::exit(1);
+            }
+            let proto_handler = proto_handler.unwrap();
+            let web_server = MqttWebApplication::build(
+                &configuration,
+                broadcaster.clone(),
+                proto_handler.clone(),
+            )
+            .await?;
+            begin_execution(configuration, proto_handler, web_server, broadcaster).await
+        }
+    }
 }