Save the last status in sled and ws send it

Author: Kerollmops

.gitignore

@@ -1,2 +1,3 @@
 /target
+/spo2.db
 **/*.rs.bk

Cargo.lock

@@ -9,6 +9,7 @@ arraydeque = "0.4.5"
 futures-preview = "0.3.0-alpha.18"
 futures-timer = "0.4.0"
 once_cell = "1.0.2"
+serde = { version = "1.0.100", features = ["derive"] }
 serde_json = "1.0.40"
 sled = "0.28.0"
 surf = "1.0.2"

Cargo.toml

@@ -1,11 +1,12 @@
 use std::time::{Duration, Instant};
 
-use futures::sink::SinkExt;
 use futures::channel::mpsc::Sender;
+use futures::sink::SinkExt;
 use futures_timer::{Delay, TryFutureExt};
-
 use url::Url;
-use crate::ReportStatus;
+
+use crate::url_value::{Status, UrlValue};
+use crate::url_value::Status::{Healthy, Unhealthy, Unreacheable};
 
 const STILL_UNHEALTHY_TIMEOUT: Duration = Duration::from_secs(15 * 60); // 15 minutes
 const TIMEOUT:     Duration = Duration::from_secs(5);
@@ -14,48 +15,47 @@ const FAST_PING:   Duration = Duration::from_millis(800);
 
 type ArrayDeque10<T> = arraydeque::ArrayDeque<[T; 10], arraydeque::Wrapping>;
 
-#[derive(Debug, Clone, Copy, PartialEq, Eq)]
-enum Status {
-    Healthy,
-    Unhealthy,
-    Unreacheable,
-}
-
-use Status::{Healthy, Unhealthy, Unreacheable};
-
-impl Status {
-    fn is_good(&self) -> bool {
-        *self == Healthy
-    }
-}
-
 pub async fn health_checker(
     url: Url,
-    mut report_sender: Sender<(Url, ReportStatus)>,
+    mut report_sender: Sender<(Url, Status)>,
     event_sender: ws::Sender,
     database: sled::Db,
 )
 {
     let mut last_status = ArrayDeque10::new();
     let mut in_bad_state_since = None;
 
-    let message = format!("{},{:?}", url, ReportStatus::Healthy);
-    let _ = event_sender.send(message);
-
     loop {
-        let status = match surf::get(&url).timeout(TIMEOUT).await {
-            Ok(ref resp) if resp.status().is_success() => Healthy,
-            Ok(resp) => Unhealthy,
-            Err(e) => Unreacheable,
+        let (status, reason) = match surf::get(&url).timeout(TIMEOUT).await {
+            Ok(ref resp) if resp.status().is_success() => {
+                (Healthy, resp.status().to_string())
+            },
+            Ok(resp) => (Unhealthy, resp.status().to_string()),
+            Err(e) => (Unreacheable, e.to_string()),
         };
 
         last_status.push_front(status);
 
-        match database.get(url.as_str()) {
-            Ok(Some(_)) => (),
+        // update this value but do not erase the user custom data updates
+        let result = database.update_and_fetch(url.as_str(), |old| {
+            let old = old?;
+            let mut value: UrlValue = serde_json::from_slice(old).unwrap();
+            value.status = status;
+            value.reason = reason.clone();
+            Some(serde_json::to_vec(&value).unwrap())
+        });
+
+        // retrieve the new value and deserialize it
+        // assign the current url this way it can be send in notifications
+        let value = match result {
+            Ok(Some(value)) => {
+                let mut value: UrlValue = serde_json::from_slice(&value).unwrap();
+                value.url = Some(url.to_string());
+                value
+            },
             Ok(None) => break,
-            Err(e) => eprintln!("{}: {}", url, e),
-        }
+            Err(e) => { eprintln!("{}: {}", url, e); return },
+        };
 
         let cap = last_status.capacity() as f32;
         let bads = last_status.iter().filter(|s| !s.is_good()).count() as f32;
@@ -64,20 +64,20 @@ pub async fn health_checker(
         if ratio >= 0.5 && in_bad_state_since.is_none() {
             in_bad_state_since = Some(Instant::now());
 
-            let report = (url.clone(), ReportStatus::Unhealthy);
+            let report = (url.clone(), Status::Unhealthy);
             let _ = report_sender.send(report).await;
 
-            let message = format!("{},{:?}", url, ReportStatus::Unhealthy);
+            let message = serde_json::to_string(&value).unwrap();
             let _ = event_sender.send(message);
         }
 
         if ratio == 0.0 && in_bad_state_since.is_some() {
             in_bad_state_since = None;
 
-            let report = (url.clone(), ReportStatus::Healthy);
+            let report = (url.clone(), Status::Healthy);
             let _ = report_sender.send(report).await;
 
-            let message = format!("{},{:?}", url, ReportStatus::Healthy);
+            let message = serde_json::to_string(&value).unwrap();
             let _ = event_sender.send(message);
         }
 
@@ -89,17 +89,14 @@ pub async fn health_checker(
 
         if let Some(since) = in_bad_state_since {
             if since.elapsed() > STILL_UNHEALTHY_TIMEOUT {
-                let report = (url.clone(), ReportStatus::Unhealthy);
+                let report = (url.clone(), Status::Unhealthy);
                 let _ = report_sender.send(report).await;
 
-                let message = format!("{},{:?}", url, ReportStatus::Unhealthy);
+                let message = serde_json::to_string(&value).unwrap();
                 let _ = event_sender.send(message);
 
                 in_bad_state_since = Some(Instant::now());
             }
         }
     }
-
-    let message = format!("{},{}", url, "Removed");
-    let _ = event_sender.send(message);
 }

src/health_checker.rs

@@ -1,6 +1,7 @@
 mod health_checker;
 mod response;
 mod routes;
+mod url_value;
 
 use std::{env, io, str, thread};
 
@@ -11,21 +12,16 @@ use url::Url;
 
 use self::health_checker::health_checker;
 use self::routes::{update_url, read_url, delete_url, get_all_urls};
+use self::url_value::Status;
 
 const HTTP_LISTEN_ADDR: &str = "HTTP_LISTEN_ADDR";
 const WS_LISTEN_ADDR: &str = "WS_LISTEN_ADDR";
 const SLACK_HOOK_URL: &str = "SLACK_HOOK_URL";
 const DATABASE_PATH: &str = "DATABASE_PATH";
 
-#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
-pub enum ReportStatus {
-    Unhealthy,
-    Healthy,
-}
-
 pub struct State {
     thread_pool: ThreadPool,
-    notifier_sender: Sender<(Url, ReportStatus)>,
+    notifier_sender: Sender<(Url, Status)>,
     event_sender: ws::Sender,
     database: sled::Db,
 }

src/main.rs

@@ -8,6 +8,8 @@ use serde_json::Value;
 use crate::health_checker::health_checker;
 use crate::response::{Json, into_internal_error, into_bad_request, not_found};
 use crate::State;
+use crate::url_value::UrlValue;
+use crate::url_value::Status::{Healthy, Removed};
 
 pub async fn update_url(mut cx: Context<State>) -> Result<Json, WithStatus<String>> {
     let url = cx.param("url").map(urldecode::decode).unwrap();
@@ -16,27 +18,33 @@ pub async fn update_url(mut cx: Context<State>) -> Result<Json, WithStatus<Strin
     let body =  cx.body_bytes().await.map_err(into_bad_request)?;
     let body = if body.is_empty() { None } else { Some(body) };
 
-    let value = match body {
+    let user_data = match body {
         None => Value::Null,
         Some(body) => serde_json::from_slice(&body).map_err(into_bad_request)?,
     };
 
-    let value = serde_json::json!({ "data": value });
-    let value = serde_json::to_vec(&value).map_err(into_internal_error)?;
+    let mut value = UrlValue { url: None, status: Healthy, reason: String::new(), data: user_data };
+    let value_bytes = serde_json::to_vec(&value).map_err(into_internal_error)?;
 
     let pool = &cx.state().thread_pool;
     let database = cx.state().database.clone();
     let notifier_sender = cx.state().notifier_sender.clone();
     let event_sender = cx.state().event_sender.clone();
 
-    match database.insert(url.as_str(), value.as_slice()) {
+    match database.insert(url.as_str(), value_bytes.as_slice()) {
         Ok(None) => {
+            // send the initial healthy message when an url is added
+            value.url = Some(url.to_string());
+            let message = serde_json::to_string(&value).map_err(into_internal_error)?;
+            let _ = event_sender.send(message);
+
             pool.spawn_ok(async {
                 health_checker(url, notifier_sender, event_sender, database).await
             });
-            Ok(Json(value))
+
+            Ok(Json(value_bytes))
         },
-        Ok(Some(_)) => Ok(Json(value)),
+        Ok(Some(_)) => Ok(Json(value_bytes)),
         Err(e) => Err(into_internal_error(e)),
     }
 }
@@ -58,8 +66,19 @@ pub async fn delete_url(cx: Context<State>) -> Result<Json, WithStatus<String>>
     let url = Url::parse(&url).map_err(into_bad_request)?;
 
     let database = &cx.state().database;
+    let event_sender = &cx.state().event_sender;
+
     match database.remove(url.as_str()) {
-        Ok(Some(value)) => Ok(Json(value.to_vec())),
+        Ok(Some(value_bytes)) => {
+            let mut value: UrlValue = serde_json::from_slice(&value_bytes).map_err(into_internal_error)?;
+            value.status = Removed;
+            value.url = Some(url.to_string());
+
+            let message = serde_json::to_string(&value).map_err(into_internal_error)?;
+            let _ = event_sender.send(message);
+
+            Ok(Json(value_bytes.to_vec()))
+        },
         Ok(None) => Err(not_found()),
         Err(e) => Err(into_internal_error(e)),
     }
@@ -77,9 +96,10 @@ pub async fn get_all_urls(cx: Context<State>) -> Result<Json, WithStatus<String>
 
         let string = str::from_utf8(&key).map_err(into_internal_error)?;
         let url = Url::parse(&string).map_err(into_internal_error)?;
-        let value: Value = serde_json::from_slice(&value).map_err(into_internal_error)?;
 
-        let value = serde_json::json!({ "url": url.as_str(), "data": value });
+        let mut value: UrlValue = serde_json::from_slice(&value).map_err(into_internal_error)?;
+        value.url = Some(url.to_string());
+
         urls.push(value);
     }