catch SIGHUP/SIGTERM, optimize memory usage

Signed-off-by: Lance-Drane <[email protected]>

Author: Lance-Drane

proxy-http-client/src/configuration.rs

@@ -1,8 +1,8 @@
 /// FOR DEVOPS USERS:
 /// 1) The root struct is "Settings", follow logic from there
 /// 2) integers can be provided as a string in config files or environment variables
-/// 3) if using environment variables, see comment in "get_configuration()" as an example of how nesting works
-/// 4) if using ONLY a file variable, this is determined from the APP_CONFIG_FILE environment variable (environment variables have higher precedence)
+/// 3) if using environment variables, see comment in `get_configuration()` as an example of how nesting works
+/// 4) if using ONLY a file variable, this is determined from the `APP_CONFIG_FILE` environment variable (environment variables have higher precedence)
 /// 5) Additional logic can be found in shared-deps/src/configuration.rs
 use secrecy::SecretString;
 
@@ -21,7 +21,7 @@ pub struct ExternalProxy {
     pub password: SecretString,
 }
 
-#[derive(serde::Deserialize, Clone)]
+#[derive(serde::Deserialize, Clone, Debug)]
 pub struct Settings {
     /// configuration for the broker, which our applications are listening to
     pub broker: BrokerSettings, // TODO make this a Vec<BrokerSettings>

proxy-http-client/src/event_source.rs

@@ -10,10 +10,10 @@ use intersect_ingress_proxy_common::protocols::amqp::{
 use intersect_ingress_proxy_common::server_paths::SUBSCRIBE_URL;
 use intersect_ingress_proxy_common::signals::wait_for_os_signal;
 
-use crate::configuration::Settings;
+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, connection_pool: &Pool) -> Result<(), String> {
     let es_data_result = extract_eventsource_data(&message);
     if es_data_result.is_err() {
         return Ok(());
@@ -39,28 +39,28 @@ async fn send_message(message: String, connection_pool: Pool) -> Result<(), Stri
     })?;
 
     match amqp_publish_message(channel, &topic, data).await {
-        Ok(_) => Ok(()),
-        Err(_) => Err(
+        Ok(()) => Ok(()),
+        Err(()) => Err(
             "WARNING: message received from other proxy was NOT published on our own broker."
                 .into(),
         ),
     }
 }
 
 /// Return value - exit code to use
-pub async fn event_source_loop(configuration: &Settings, connection_pool: Pool) -> i32 {
+///
+/// # 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 {
     let mut es = EventSource::new(
         reqwest::Client::new()
-            .get(format!(
-                "{}{}",
-                &configuration.other_proxy.url, SUBSCRIBE_URL
-            ))
+            .get(format!("{}{}", &other_proxy.url, SUBSCRIBE_URL))
             .basic_auth(
-                &configuration.other_proxy.username,
-                Some(configuration.other_proxy.password.expose_secret()),
+                &other_proxy.username,
+                Some(&other_proxy.password.expose_secret()),
             ),
     )
-    .unwrap();
+    .expect("The event source request body was somehow a stream?");
     let mut rc = 0;
     loop {
         tokio::select! {
@@ -76,10 +76,10 @@ pub async fn event_source_loop(configuration: &Settings, connection_pool: Pool)
                     Some(event) => {
                         match event {
                             Ok(Event::Open) => {
-                                tracing::info!("connected to {}", &configuration.other_proxy.url);
+                                tracing::info!("connected to {}", &other_proxy.url);
                             },
                             Ok(Event::Message(message)) => {
-                                if let Err(e) = send_message(message.data, connection_pool.clone()).await {
+                                if let Err(e) = send_message(message.data, &connection_pool).await {
                                     tracing::error!(e);
                                 };
                             },
@@ -94,7 +94,7 @@ pub async fn event_source_loop(configuration: &Settings, connection_pool: Pool)
                 }
             },
             // OS kill signal
-            _ = wait_for_os_signal() => {
+            () = wait_for_os_signal() => {
                 break;
             },
         };

proxy-http-client/src/main.rs

@@ -32,7 +32,7 @@ pub async fn main() -> anyhow::Result<()> {
     }
 
     // set up broker connection pool
-    let pool = get_connection_pool(&configuration.broker).await;
+    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);
@@ -54,8 +54,11 @@ pub async fn main() -> anyhow::Result<()> {
         rx,
     );
 
+    let other_proxy = configuration.other_proxy.clone();
+    drop(configuration);
+
     // this will run until we get an event source error or we catch an OS signal
-    let rc = event_source_loop(&configuration, pool.clone()).await;
+    let rc = event_source_loop(other_proxy, pool).await;
 
     tracing::info!("Attempting graceful shutdown: No longer listening for events over HTTP");
     drop(tx);

proxy-http-client/src/poster.rs

@@ -10,6 +10,7 @@ pub struct Poster {
 }
 
 impl Poster {
+    #[must_use]
     pub fn new(proxy: &ExternalProxy) -> Self {
         let http_client = reqwest::Client::new()
             .post(format!("{}{}", proxy.url, PUBLISH_URL))

proxy-http-server/src/broadcaster.rs

@@ -18,13 +18,15 @@ pub struct Broadcaster {
 impl Broadcaster {
     /// Create the broadcaster. Note that it automatically wraps it in an Arc.
     /// The broadcaster manages its producer but does not manage its consumers
+    #[must_use]
     pub fn new() -> Arc<Self> {
         // use a fairly large channel capacity to account for potential receiver lags
         let (tx, _) = broadcast::channel(256);
         Arc::new(Broadcaster { fanout: tx })
     }
 
     /// Add a broadcaster consumer - the calling function is responsible for cleaning up the consumer
+    #[must_use]
     pub fn add_client(&self) -> broadcast::Receiver<Event> {
         self.fanout.subscribe()
     }
@@ -41,6 +43,7 @@ impl Broadcaster {
     /// 3) Somehow transfer these messages over to the other message broker, make the messages their responsibility.
     ///
     /// Once the messages are on the other message broker, proxy-http-server and proxy-http-client don't need to care, handling them will be the SDK's job.
+    #[must_use]
     pub fn broadcast(&self, event: &str) -> usize {
         self.fanout.send(Event::default().data(event)).unwrap_or(0)
     }

proxy-http-server/src/configuration.rs

@@ -1,8 +1,8 @@
 /// FOR DEVOPS USERS:
 /// 1) The root struct is "Settings", follow logic from there
 /// 2) integers can be provided as a string in config files or environment variables
-/// 3) if using environment variables, see comment in "get_configuration()" as an example of how nesting works
-/// 4) if using ONLY a file variable, this is determined from the APP_CONFIG_FILE environment variable (environment variables have higher precedence)
+/// 3) if using environment variables, see comment in `get_configuration()` as an example of how nesting works
+/// 4) if using ONLY a file variable, this is determined from the `APP_CONFIG_FILE` environment variable (environment variables have higher precedence)
 /// 5) Additional logic can be found in shared-deps/src/configuration.rs
 use secrecy::SecretString;
 use serde_aux::field_attributes::deserialize_number_from_string;
@@ -11,7 +11,7 @@ use intersect_ingress_proxy_common::configuration::{
     deserialize_enforce_topic_prefixes, BrokerSettings, LogLevel,
 };
 
-#[derive(serde::Deserialize, Clone)]
+#[derive(serde::Deserialize, Clone, Debug)]
 pub struct Settings {
     /// configuration for the broker, which our applications are listening to
     pub broker: BrokerSettings, // TODO make this a Vec<BrokerSettings>

proxy-http-server/src/main.rs

@@ -33,7 +33,7 @@ async fn main() -> anyhow::Result<()> {
     }
 
     // set up broker connection pool
-    let pool = get_connection_pool(&configuration.broker).await;
+    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);
@@ -54,10 +54,12 @@ async fn main() -> anyhow::Result<()> {
         pool,
         configuration.topic_prefix.clone(),
         APPLICATION_NAME.into(),
-        broadcaster.clone(),
+        broadcaster,
         rx,
     );
 
+    drop(configuration);
+
     application.run_until_stopped().await?;
 
     tracing::info!("Application shutting down, please wait for cleanups...");

proxy-http-server/src/routes/publish.rs

@@ -16,6 +16,11 @@ use intersect_ingress_proxy_common::protocols::amqp::{
 use crate::webapp::WebApplicationState;
 
 /// HTTP POST endpoint which will publish a message meeting the INTERSECT specification
+///
+/// # Errors
+///   - Sends back a 401 if authentication is incorrect
+///   - Sends back a 400 if the message body is improperly formatted
+///   - Sends back a 500 if the server was unable to send the message
 pub async fn publish_message(
     State(app_state): State<Arc<WebApplicationState>>,
     TypedHeader(authorization): TypedHeader<Authorization<Basic>>,
@@ -50,7 +55,7 @@ pub async fn publish_message(
         );
         return Err((
             StatusCode::BAD_REQUEST,
-            format!("{} is not a valid AMQP topic name", topic),
+            format!("{topic} is not a valid AMQP topic name"),
         ));
     }
     tracing::debug!("Publishing message with topic: {}", &topic);
@@ -72,7 +77,7 @@ pub async fn publish_message(
     })?;
     amqp_publish_message(channel, &topic, data)
         .await
-        .map_err(|_| {
+        .map_err(|()| {
             (
                 StatusCode::INTERNAL_SERVER_ERROR,
                 "server fault, message not published".to_string(),

proxy-http-server/src/routes/subscribe.rs

@@ -14,6 +14,7 @@ use std::sync::Arc;
 use crate::webapp::WebApplicationState;
 use intersect_ingress_proxy_common::signals::wait_for_os_signal;
 
+#[allow(clippy::needless_pass_by_value)]
 fn sse_response(
     app_state: Arc<WebApplicationState>,
 ) -> Sse<impl Stream<Item = Result<Event, Infallible>>> {
@@ -23,7 +24,7 @@ fn sse_response(
         loop {
             tokio::select! {
                 // if we catch an OS signal, disconnect the client
-                _ = wait_for_os_signal() => {
+                () = wait_for_os_signal() => {
                     break;
                 },
                 // send the broadcast message to the client, and continue listening for more messages
@@ -36,25 +37,25 @@ fn sse_response(
                         Err(e) => {
                             match e {
                                 tokio::sync::broadcast::error::RecvError::Closed => {
-                                    tracing::error!(error = ?e, "Broadcasting pipeline to SSE somehow closed, should not see this message!")
+                                    tracing::error!(error = ?e, "Broadcasting pipeline to SSE somehow closed, should not see this message!");
                                 },
                                 tokio::sync::broadcast::error::RecvError::Lagged(lag_count) => {
-                                    tracing::error!(error = ?e, "SSE has missed {} messages from broadcaster", lag_count)
+                                    tracing::error!(error = ?e, "SSE has missed {} messages from broadcaster", lag_count);
                                 },
-                            };
+                            }
                         },
                     }
                 },
-            };
-        };
+            }
+        }
     };
 
     Sse::new(stream).keep_alive(KeepAlive::default())
 }
 
 /// Resources:
-/// https://github.com/tokio-rs/axum/discussions/1670
-/// https://github.com/tokio-rs/axum/discussions/2264
+/// `<https://github.com/tokio-rs/axum/discussions/1670>`
+/// `<https://github.com/tokio-rs/axum/discussions/2264>`
 pub async fn sse_handler(
     State(app_state): State<Arc<WebApplicationState>>,
     TypedHeader(authorization): TypedHeader<Authorization<Basic>>,

proxy-http-server/src/webapp.rs

@@ -42,6 +42,9 @@ pub struct WebApplication {
 }
 
 impl WebApplication {
+    ///
+    /// # Errors
+    ///   - errors if unable to bind to provided TCP port
     pub async fn build(
         configuration: &Settings,
         broadcaster: Arc<Broadcaster>,
@@ -57,8 +60,8 @@ impl WebApplication {
             configuration.app_port
         );
         let listener = TcpListener::bind(address).await?;
-        let port = listener.local_addr().unwrap().port();
-        let server = run(listener, configuration, broadcaster, amqp_pool).await?;
+        let port = listener.local_addr()?.port();
+        let server = create_axum(listener, configuration, broadcaster, amqp_pool);
 
         tracing::info!("Web server is running on port {}", port);
 
@@ -69,6 +72,9 @@ impl WebApplication {
         self.port
     }
 
+    ///
+    /// # Errors
+    ///   - Errors if unable to initialize web server
     pub async fn run_until_stopped(self) -> Result<(), std::io::Error> {
         // the return type of "with_graceful_shutdown" is unstable, so set it up here
         self.server
@@ -77,12 +83,12 @@ impl WebApplication {
     }
 }
 
-async fn run(
+fn create_axum(
     listener: TcpListener,
     configuration: &Settings,
     broadcaster: Arc<Broadcaster>,
     amqp_pool: Pool,
-) -> Result<WebAppServer, anyhow::Error> {
+) -> WebAppServer {
     let middleware = ServiceBuilder::new()
         .set_x_request_id(MakeRequestUuid)
         .layer(
@@ -111,6 +117,5 @@ async fn run(
         .route("/healthcheck", get(health_check))
         .fallback(handler_404);
 
-    let server = axum::serve(listener, app);
-    Ok(server)
+    axum::serve(listener, app)
 }