Merge pull request #60 from hastic/detection_runner_updade

update detections

server/src/services/analytic_service/analytic_service.rs

@@ -238,6 +238,9 @@ impl AnalyticService {
                     ResponseType::DetectionRunnerStarted(from) => {
                         println!("Detection runner started from {}", from)
                     }
+                    ResponseType::DetectionRunnerUpdate(id, timestamp) => {
+                        self.analytic_unit_service.set_last_detection(id, timestamp).unwrap();
+                    }
                     ResponseType::LearningStarted => {
                         self.analytic_unit_learning_status = LearningStatus::Learning
                     }

server/src/services/analytic_service/detection_runner.rs

@@ -1,4 +1,4 @@
-use chrono::Utc;
+use chrono::{Utc, DateTime};
 
 use tokio::sync::{mpsc, RwLock};
 
@@ -36,6 +36,7 @@ impl DetectionRunner {
             // TODO: clone channel
             let cfg = self.config.clone();
             let tx = self.tx.clone();
+            let au = self.analytic_unit.clone();
             async move {
                 // TODO: run detection "from" for big timespan
                 // TODO: parse detections to webhooks
@@ -54,9 +55,17 @@ impl DetectionRunner {
                 }
 
                 loop {
+                    // TODO: don't use DateTime, but count timestamp by steps
+                    let now: DateTime<Utc> = Utc::now();
+                    let to = now.timestamp() as u64;
                     // TODO: run detection periodically
                     sleep(Duration::from_secs(cfg.interval)).await;
-                    // TODO: use tx senf detection update
+                    match tx.send(AnalyticServiceMessage::Response(Ok(
+                        ResponseType::DetectionRunnerUpdate(au.as_ref().read().await.get_id(), to)
+                    ))).await {
+                        Ok(_) => {},
+                        Err(_e) => println!("Fail to send detection runner started notification"),
+                    }
                 }
             }
         }));

server/src/services/analytic_service/types.rs

@@ -44,6 +44,7 @@ impl Default for LearningTrain {
 
 pub enum ResponseType {
     DetectionRunnerStarted(u64),
+    DetectionRunnerUpdate(String, u64), // analytic_unit id and timestamp
     LearningStarted,
     LearningFinished(Box<dyn AnalyticUnit + Send + Sync>),
     LearningFinishedEmpty,

server/src/services/analytic_unit_service.rs

@@ -65,7 +65,12 @@ impl AnalyticUnitService {
     }
 
     // TODO: resolve with saving by id
-    pub fn set_last_detection(id: String, last_detection: u64) -> anyhow::Result<()> {
+    pub fn set_last_detection(&self, id: String, last_detection: u64) -> anyhow::Result<()> {
+        let conn = self.connection.lock().unwrap();
+        conn.execute(
+            "UPDATE analytic_unit SET last_detection = ?1 WHERE id = ?2",
+            params![last_detection, id]
+        )?;
         Ok(())
     }
 }