move http stuff to submodule in signal-gateway-bin

signal-gateway-bin/src/listen_http.rs

@@ -0,0 +1,136 @@
+use http::{Method, Request, Response, StatusCode};
+use http_body::Body;
+use http_body_util::BodyExt;
+use hyper::service::service_fn;
+use hyper_util::{rt::TokioIo, server::conn::auto};
+use signal_gateway::{Gateway, alertmanager::AlertPost};
+use std::{convert::Infallible, sync::Arc, time::Duration};
+use tokio::net::TcpListener;
+use tokio_util::bytes::Buf;
+use tracing::{error, info, warn};
+
+/// Start http listening task
+pub fn start_http_task(listener: TcpListener, gateway: Arc<Gateway>) -> tokio::task::JoinHandle<()> {
+    // Loop waiting for http incoming connections, and pass them to gateway
+    tokio::task::spawn(async move {
+        loop {
+            let Ok((stream, remote_addr)) = listener
+                .accept()
+                .await
+                .inspect_err(|err| error!("Error accepting connection: {err}"))
+            else {
+                tokio::time::sleep(Duration::from_secs(1)).await;
+                continue;
+            };
+            info!("New connection from: {}", remote_addr);
+
+            // Spawn a new task to handle each connection
+            let thread_gateway = gateway.clone();
+            tokio::spawn(async move {
+                let io = TokioIo::new(stream);
+
+                // Serve the connection using auto protocol detection (HTTP/1 or HTTP/2)
+                if let Err(err) = auto::Builder::new(hyper_util::rt::TokioExecutor::new())
+                    .serve_connection(
+                        io,
+                        service_fn(|req| handle_http_request(thread_gateway.clone(), req)),
+                    )
+                    .await
+                {
+                    error!("Error serving connection: {err}");
+                }
+            });
+        }
+    })
+}
+
+async fn handle_http_request(
+    gateway: Arc<Gateway>,
+    req: Request<hyper::body::Incoming>,
+) -> Result<Response<String>, Infallible> {
+    match handle_http_request_impl(gateway, req).await {
+        Ok(resp) => Ok(resp),
+        Err(resp) => Ok(resp),
+    }
+}
+
+async fn handle_http_request_impl<B>(
+    gateway: Arc<Gateway>,
+    req: Request<B>,
+) -> Result<Response<String>, Response<String>>
+where
+    B: Body + Send,
+    B::Data: Buf + Send,
+    B::Error: std::fmt::Display,
+{
+    info!(
+        "Received http request: {} {} (version: {:?})",
+        req.method(),
+        req.uri().path(),
+        req.version()
+    );
+
+    fn ok_resp() -> Response<String> {
+        Response::new("OK".into())
+    }
+    fn err_resp(code: StatusCode, text: impl Into<String>) -> Response<String> {
+        let mut resp = Response::new(text.into());
+        *resp.status_mut() = code;
+        resp
+    }
+
+    match req.uri().path() {
+        "/" | "/health" | "/ready" => {
+            if !matches!(req.method(), &Method::GET | &Method::HEAD) {
+                Ok(err_resp(
+                    StatusCode::NOT_IMPLEMENTED,
+                    "Use GET or HEAD with this route",
+                ))
+            } else {
+                Ok(ok_resp())
+            }
+        }
+        "/alert" => {
+            if !matches!(req.method(), &Method::POST) {
+                return Ok(err_resp(
+                    StatusCode::NOT_IMPLEMENTED,
+                    "Use POST with this route",
+                ));
+            }
+            let body_bytes = req
+                .into_body()
+                .collect()
+                .await
+                .map_err(|err| {
+                    err_resp(
+                        StatusCode::BAD_REQUEST,
+                        format!("When reading body bytes: {err}"),
+                    )
+                })?
+                .to_bytes()
+                .to_vec();
+
+            let body_text = str::from_utf8(&body_bytes).map_err(|err| {
+                warn!("When reading body bytes: {err}");
+                err_resp(StatusCode::BAD_REQUEST, "Request body was not utf-8")
+            })?;
+
+            let alert_msg: AlertPost = serde_json::from_str(body_text).map_err(|err| {
+                error!("Could not parse json: {err}:\n{body_text}");
+                err_resp(StatusCode::BAD_REQUEST, "Invalid Json")
+            })?;
+
+            if let Err(msg) = gateway.handle_alertmanager_post(alert_msg).await {
+                error!("gateway (handle_alertmanager_post): {msg}");
+                Ok(err_resp(StatusCode::INTERNAL_SERVER_ERROR, msg))
+            } else {
+                Ok(ok_resp())
+            }
+        }
+        _ => Ok(err_resp(
+            StatusCode::NOT_FOUND,
+            format!("Not found '{} {}'", req.method(), req.uri().path()),
+        )),
+    }
+}
+

signal-gateway-bin/src/main.rs

@@ -3,19 +3,12 @@
 #![deny(missing_docs)]
 
 use conf::Conf;
-use http::{Method, Request, Response, StatusCode};
-use http_body::Body;
-use http_body_util::BodyExt;
-use hyper::service::service_fn;
-use hyper_util::{rt::TokioIo, server::conn::auto};
-use signal_gateway::{CommandRouter, Gateway, GatewayConfig, Handling, alertmanager::AlertPost};
+use signal_gateway::{CommandRouter, Gateway, GatewayConfig, Handling};
 use signal_gateway_app_code::AppCodeTools;
 use signal_gateway_assistant_claude::{ClaudeAssistant, ClaudeConfig};
-use std::{
-    convert::Infallible, env, fs, net::SocketAddr, path::PathBuf, sync::Arc, time::Duration,
-};
+use std::{env, fs, net::SocketAddr, path::PathBuf, sync::Arc};
 use tokio::net::TcpListener;
-use tokio_util::{bytes::Buf, sync::CancellationToken};
+use tokio_util::sync::CancellationToken;
 use tracing::{error, info, warn};
 use tracing_subscriber::EnvFilter;
 
@@ -25,6 +18,9 @@ use admin_http::AdminHttpConfig;
 mod app_code;
 use app_code::AppCodeConfigExt;
 
+mod listen_http;
+use listen_http::start_http_task;
+
 use signal_gateway_log_ingest::{JsonConfig, SyslogConfig};
 
 /// Top-level configuration for signal-gateway.
@@ -204,134 +200,11 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
     Ok(())
 }
 
-fn start_http_task(listener: TcpListener, gateway: Arc<Gateway>) -> tokio::task::JoinHandle<()> {
-    // Loop waiting for http incoming connections, and pass them to gateway
-    tokio::task::spawn(async move {
-        loop {
-            let Ok((stream, remote_addr)) = listener
-                .accept()
-                .await
-                .inspect_err(|err| error!("Error accepting connection: {err}"))
-            else {
-                tokio::time::sleep(Duration::from_secs(1)).await;
-                continue;
-            };
-            info!("New connection from: {}", remote_addr);
-
-            // Spawn a new task to handle each connection
-            let thread_gateway = gateway.clone();
-            tokio::spawn(async move {
-                let io = TokioIo::new(stream);
-
-                // Serve the connection using auto protocol detection (HTTP/1 or HTTP/2)
-                if let Err(err) = auto::Builder::new(hyper_util::rt::TokioExecutor::new())
-                    .serve_connection(
-                        io,
-                        service_fn(|req| handle_http_request(thread_gateway.clone(), req)),
-                    )
-                    .await
-                {
-                    error!("Error serving connection: {err}");
-                }
-            });
-        }
-    })
-}
-
-async fn handle_http_request(
-    gateway: Arc<Gateway>,
-    req: Request<hyper::body::Incoming>,
-) -> Result<Response<String>, Infallible> {
-    match handle_http_request_impl(gateway, req).await {
-        Ok(resp) => Ok(resp),
-        Err(resp) => Ok(resp),
-    }
-}
-
-async fn handle_http_request_impl<B>(
-    gateway: Arc<Gateway>,
-    req: Request<B>,
-) -> Result<Response<String>, Response<String>>
-where
-    B: Body + Send,
-    B::Data: Buf + Send,
-    B::Error: std::fmt::Display,
-{
-    info!(
-        "Received http request: {} {} (version: {:?})",
-        req.method(),
-        req.uri().path(),
-        req.version()
-    );
-
-    fn ok_resp() -> Response<String> {
-        Response::new("OK".into())
-    }
-    fn err_resp(code: StatusCode, text: impl Into<String>) -> Response<String> {
-        let mut resp = Response::new(text.into());
-        *resp.status_mut() = code;
-        resp
-    }
-
-    match req.uri().path() {
-        "/" | "/health" | "/ready" => {
-            if !matches!(req.method(), &Method::GET | &Method::HEAD) {
-                Ok(err_resp(
-                    StatusCode::NOT_IMPLEMENTED,
-                    "Use GET or HEAD with this route",
-                ))
-            } else {
-                Ok(ok_resp())
-            }
-        }
-        "/alert" => {
-            if !matches!(req.method(), &Method::POST) {
-                return Ok(err_resp(
-                    StatusCode::NOT_IMPLEMENTED,
-                    "Use POST with this route",
-                ));
-            }
-            let body_bytes = req
-                .into_body()
-                .collect()
-                .await
-                .map_err(|err| {
-                    err_resp(
-                        StatusCode::BAD_REQUEST,
-                        format!("When reading body bytes: {err}"),
-                    )
-                })?
-                .to_bytes()
-                .to_vec();
-
-            let body_text = str::from_utf8(&body_bytes).map_err(|err| {
-                warn!("When reading body bytes: {err}");
-                err_resp(StatusCode::BAD_REQUEST, "Request body was not utf-8")
-            })?;
-
-            let alert_msg: AlertPost = serde_json::from_str(body_text).map_err(|err| {
-                error!("Could not parse json: {err}:\n{body_text}");
-                err_resp(StatusCode::BAD_REQUEST, "Invalid Json")
-            })?;
-
-            if let Err(msg) = gateway.handle_alertmanager_post(alert_msg).await {
-                error!("gateway (handle_alertmanager_post): {msg}");
-                Ok(err_resp(StatusCode::INTERNAL_SERVER_ERROR, msg))
-            } else {
-                Ok(ok_resp())
-            }
-        }
-        _ => Ok(err_resp(
-            StatusCode::NOT_FOUND,
-            format!("Not found '{} {}'", req.method(), req.uri().path()),
-        )),
-    }
-}
-
 #[cfg(test)]
 mod tests {
     use super::*;
     use conf::Conf;
+    use std::time::Duration;
 
     #[test]
     fn test_toml_config() {

signal-gateway/src/gateway/log_handler.rs

@@ -59,7 +59,7 @@ pub struct LogHandlerConfig {
     #[conf(long, env, default_value = "64")]
     pub log_buffer_size: usize,
     /// Max age of an origin, after this we remove it to reclaim memory
-    #[conf(long, env, value_parser = conf_extra::parse_duration, default_value = "3d")]
+    #[conf(long, env, value_parser = conf_extra::parse_duration, default_value = "72h")]
     pub max_origin_age: Duration,
     /// Debug logging level for suppressed messages.
     /// 0 = no logging, 1 = log only overall limiter, 2 = log routes + overall, 3 = log all.