Add channel and cache health check endpoints; update state initialization

Author: Neko-Nik

src/main.rs

@@ -13,13 +13,14 @@ mod db;
 
 #[actix_web::main]
 async fn main() -> std::io::Result<()> {
-    let (pg_pool, in_mem_cache) = state::init().await;
+    let (pg_pool, in_mem_cache, tx) = state::init().await;
 
     // Start the Actix web server
     HttpServer::new(move || {
         App::new()
             .app_data(pg_pool.clone())
             .app_data(in_mem_cache.clone())
+            .app_data(tx.clone())
             .wrap(Cors::default()
                 .allow_any_origin()
                 .allowed_methods(vec!["GET", "POST"])
@@ -30,6 +31,8 @@ async fn main() -> std::io::Result<()> {
                 actix_scope("/health")
                 .service(health::api_health_check)
                 .service(health::db_health_check)
+                .service(health::cache_health_check)
+                .service(health::channel_health_check)
             )
             .service(
                 actix_scope("/sample_db")

src/routes/health.rs

@@ -1,6 +1,8 @@
 use crate::db::pgsql_handlers::health_check as health_check_pgsql;
-use actix_web::{get, web, HttpResponse, Responder};
+use actix_web::{get, post, web, HttpResponse, Responder};
+use crate::types::{AppCache, make_key};
 use deadpool_postgres::Pool as PgPool;
+use std::sync::mpsc::Sender;
 
 
 // Health check endpoint
@@ -15,6 +17,33 @@ async fn api_health_check() -> impl Responder {
 async fn db_health_check(state: web::Data<PgPool>) -> impl Responder {
     match health_check_pgsql(&state).await {
         Ok(_) => HttpResponse::Ok().body("Database is running!"),
-        Err(err) => HttpResponse::InternalServerError().json(format!("Failed: {}", err)),
+        Err(err) => HttpResponse::PreconditionFailed().json(format!("Failed: {}", err)),
     }
 }
+
+
+// Cache health check
+#[get("/cache")]
+async fn cache_health_check(cache: web::Data<AppCache>) -> impl Responder {
+    const CACHE_KEY: &str = "health_check";
+    const CACHE_VALUE: &str = "Cache is running!";
+    let key = make_key(CACHE_KEY);
+
+    cache.insert(key, CACHE_VALUE.to_string()).await;
+
+    if let Some(cached_value) = cache.get(&make_key(CACHE_KEY)).await {
+        if cached_value == CACHE_VALUE {
+            return HttpResponse::Ok().body(cached_value);
+        }
+    }
+    HttpResponse::PreconditionFailed().body("Cache health check failed!")
+}
+
+
+// Channel Health check
+#[post("/channel")]
+async fn channel_health_check(state: web::Data<Sender<u8>>) -> impl Responder {
+    // Send a 7 int to the channel
+    let _ = state.send(7);
+    HttpResponse::Ok().body("Channel health check initiated!")
+}

src/state.rs

@@ -2,11 +2,12 @@ use deadpool_postgres::{Manager, RecyclingMethod, Pool as PgPool};
 use deadpool::{managed::Timeouts, Runtime};
 use actix_web::web::Data as webData;
 use tokio_postgres::{Config, NoTls};
+use crate::types::process_channel;
 use std::env::var as env_var;
+use std::sync::mpsc::Sender;
 use super::types::AppCache;
 use std::time::Duration;
 use log::{info, warn};
-use actix_web::web;
 
 
 struct PgSettings {
@@ -215,7 +216,7 @@ fn init_cache(cache_settings: &MokaSettings) -> AppCache {
 }
 
 
-pub async fn init() -> (webData<PgPool>, web::Data<AppCache>) {
+pub async fn init() -> (webData<PgPool>, webData<AppCache>, webData<Sender<u8>>) {
     // Preparing to start the server by collecting environment variables
     let app_settings: AppSettings = AppSettings::from_env();
 
@@ -233,6 +234,10 @@ pub async fn init() -> (webData<PgPool>, web::Data<AppCache>) {
     // Initialize the in-memory cache (Moka)
     let in_mem_cache = init_cache(&app_settings.cache_settings);
 
+    // Initialize the channel
+    let (tx, rx) = std::sync::mpsc::channel::<u8>();
+    process_channel(rx);
+
     // Wrap the state of the application and share it
-    (webData::new(postgres_state), webData::new(in_mem_cache))
+    (webData::new(postgres_state), webData::new(in_mem_cache), webData::new(tx))
 }

src/types.rs

@@ -1,3 +1,4 @@
+use std::sync::mpsc::Receiver;
 use moka::future::Cache;
 use std::sync::Arc;
 
@@ -31,3 +32,22 @@ pub async fn cache_data<T, S>(key: S, value: &T, cache_conn: &AppCache)
     // Insert the JSON string into the cache
     cache_conn.insert(make_key(key), json).await;
 }
+
+
+/// Process the channel
+pub fn process_channel(rx: Receiver<u8>) {
+    std::thread::spawn(move || {
+        // Keep reading
+        loop {
+            match rx.recv() {
+                Ok(val) => {
+                    log::info!("Received: {}", val);
+                }
+                Err(_) => {
+                    log::warn!("Channel closed");
+                    break;
+                }
+            }
+        }
+    });
+}