detection runner++

server/src/services/analytic_service/analytic_service.rs

@@ -2,7 +2,7 @@ use std::sync::Arc;
 
 use super::analytic_unit::types::{AnalyticUnitConfig, PatchConfig};
 use super::detection_runner::DetectionRunner;
-use super::types::{self, DetectionRunnerConfig, LearningWaiter, HSR};
+use super::types::{self, AnalyticUnitRF, DetectionRunnerConfig, HSR, LearningWaiter};
 use super::{
     analytic_client::AnalyticClient,
     types::{AnalyticServiceMessage, LearningStatus, RequestType, ResponseType},
@@ -29,7 +29,7 @@ pub struct AnalyticService {
 
     alerting: Option<AlertingConfig>,
 
-    analytic_unit: Option<Arc<RwLock<Box<dyn AnalyticUnit + Send + Sync>>>>,
+    analytic_unit: Option<AnalyticUnitRF>,
     analytic_unit_config: AnalyticUnitConfig,
     analytic_unit_learning_status: LearningStatus,
 
@@ -318,7 +318,7 @@ impl AnalyticService {
 
     async fn get_detections(
         tx: oneshot::Sender<anyhow::Result<Vec<Segment>>>,
-        analytic_unit: Arc<RwLock<Box<dyn AnalyticUnit + Send + Sync>>>,
+        analytic_unit: AnalyticUnitRF,
         ms: MetricService,
         from: u64,
         to: u64,
@@ -353,7 +353,7 @@ impl AnalyticService {
 
     async fn get_hsr(
         tx: oneshot::Sender<anyhow::Result<HSR>>,
-        analytic_unit: Arc<RwLock<Box<dyn AnalyticUnit + Send + Sync>>>,
+        analytic_unit: AnalyticUnitRF,
         ms: MetricService,
         from: u64,
         to: u64,

server/src/services/analytic_service/detection_runner.rs

@@ -6,27 +6,53 @@ use chrono::Utc;
 
 use tokio::sync::{mpsc, RwLock};
 
-use super::types::DetectionRunnerConfig;
+use super::types::{AnalyticUnitRF, DetectionRunnerConfig};
+use tokio::time::{sleep, Duration};
+
 
 pub struct DetectionRunner {
     config: DetectionRunnerConfig,
-    analytic_unit: Arc<RwLock<Box<dyn AnalyticUnit + Send + Sync>>>,
+    analytic_unit: AnalyticUnitRF,
+    running_handler: Option<tokio::task::JoinHandle<()>>,
 }
 
 impl DetectionRunner {
     pub fn new(
         config: DetectionRunnerConfig,
-        analytic_unit: Arc<RwLock<Box<dyn AnalyticUnit + Send + Sync>>>,
+        analytic_unit: AnalyticUnitRF,
     ) -> DetectionRunner {
         DetectionRunner {
             config,
             analytic_unit,
+            running_handler: None
         }
     }
 
-    pub async fn run() {
-        // TODO: await detection step
+    pub async fn run(&mut self) {
+        
         // TODO: get last detection timestamp from persistance
         // TODO: set lst detection from "now"
+        if self.running_handler.is_some() {
+            self.running_handler.as_mut().unwrap().abort();
+        }
+        self.running_handler = Some(tokio::spawn({
+            // TODO: clone channel
+            async move {
+                // AnalyticService::run_learning(tx, cfg, ms, ss).await;
+                // TODO: run detection
+                // TODO: await detection step
+
+                loop {
+                    // TODO: use interval
+                    sleep(Duration::from_secs(100)).await;
+                }
+
+            }
+        }));
+
+    }
+
+    pub async fn set_analytic_unit(analytic_unit: Arc<RwLock<Box<dyn AnalyticUnit + Send + Sync>>>) {
+        
     }
 }

server/src/services/analytic_service/types.rs

@@ -1,4 +1,4 @@
-use std::fmt;
+use std::{fmt, sync::Arc};
 
 use crate::services::segments_service::Segment;
 
@@ -11,10 +11,12 @@ use super::analytic_unit::types::PatchConfig;
 
 use anyhow::Result;
 use serde::Serialize;
-use tokio::sync::oneshot;
+use tokio::sync::{RwLock, oneshot};
 
 use crate::services::analytic_service::analytic_unit::types::AnalyticUnit;
 
+pub type AnalyticUnitRF = Arc<RwLock<Box<dyn AnalyticUnit + Send + Sync>>>;
+
 #[derive(Debug, Clone, PartialEq, Serialize)]
 pub enum LearningStatus {
     Initialization,