feat: Rewrite system architecture with new communication pattern

src/core/backup/backup_engine.rs

@@ -1,14 +1,16 @@
 use crate::core::backup::progress_tracker::ProgressTracker;
-use crate::core::gui::gui_message_handler::GuiMessageHandler;
-use crate::core::infrastructure::actor_system::ActorSystem;
 use crate::core::infrastructure::app_config::AppConfig;
+use crate::core::infrastructure::communication_manager::CommunicationManager;
 use crate::core::infrastructure::io_manager::IOManager;
-use crate::interface::file_system::FileSystemTrait;
-use crate::model::core::backup::backup_execution::*;
-use crate::model::core::gui::message::GuiMessage;
+use crate::interface::communication::command::CommandHandler;
+use crate::interface::communication::query::QueryHandler;
+use crate::interface::core::file_system::FileSystemTrait;
+use crate::model::core::backup::execution::*;
+use crate::model::core::backup::communication::*;
 use crate::model::error::system::SystemError;
 use crate::model::error::task::TaskError;
 use crate::model::error::Error;
+use async_trait::async_trait;
 use crossbeam_queue::SegQueue;
 use dashmap::DashMap;
 use futures::future::join_all;
@@ -20,33 +22,44 @@ use tokio::sync::oneshot;
 use tokio::task::JoinHandle;
 use tracing::error;
 use uuid::Uuid;
+use crate::model::core::gui::communication::ExecutionErrors;
 
 pub struct BackupEngine {
     app_config: Arc<AppConfig>,
     io_manager: Arc<IOManager>,
-    actor_system: Arc<ActorSystem>,
+    communication_manager: Arc<CommunicationManager>,
     progress_tracker: Arc<ProgressTracker>,
-    executions: Arc<DashMap<Uuid, BackupExecution>>,
+    executions: Arc<DashMap<Uuid, Execution>>,
     running_executions: Arc<DashMap<Uuid, (oneshot::Sender<()>, JoinHandle<()>)>>,
 }
 
 impl BackupEngine {
     pub fn new(
         app_config: Arc<AppConfig>,
         io_manager: Arc<IOManager>,
-        actor_system: Arc<ActorSystem>,
+        communication_manager: Arc<CommunicationManager>,
         progress_tracker: Arc<ProgressTracker>,
     ) -> Self {
         Self {
             app_config,
             io_manager,
-            actor_system,
+            communication_manager,
             progress_tracker,
             executions: Arc::new(DashMap::new()),
             running_executions: Arc::new(DashMap::new()),
         }
     }
 
+    pub async fn register_services(self: Arc<Self>) {
+        let communication_manager = self.communication_manager.clone();
+        communication_manager
+            .with_service(self)
+            .command::<BackupCommand>()
+            .query::<BackupQuery>()
+            .event::<ExecutionErrors>()
+            .build();
+    }
+
     pub async fn stop_all_executions(&self) {
         let keys: Vec<Uuid> = self
             .running_executions
@@ -66,29 +79,29 @@ impl BackupEngine {
         }
     }
 
-    pub fn get_all_executions(&self) -> Vec<(Uuid, BackupExecution)> {
+    pub fn get_all_executions(&self) -> Vec<(Uuid, Execution)> {
         self.executions
             .iter()
             .map(|entry| (*entry.key(), entry.value().clone()))
             .collect()
     }
 
-    pub async fn add_execution(&self, execution: BackupExecution) {
+    pub async fn add_execution(&self, execution: Execution) {
         self.executions.insert(execution.uuid, execution);
     }
 
     pub async fn remove_execution(&self, uuid: &Uuid) {
         self.executions.remove(uuid);
     }
 
-    pub async fn start_execution(&self, uuid: Uuid) -> Result<(), Error> {
-        if self.running_executions.contains_key(&uuid) {
+    pub async fn start_execution(&self, uuid: &Uuid) -> Result<(), Error> {
+        if self.running_executions.contains_key(uuid) {
             Err(TaskError::IllegalRunState)?
         }
 
         let mut ref_mut = self
             .executions
-            .get_mut(&uuid)
+            .get_mut(uuid)
             .ok_or(TaskError::ExecutionNotFound)?;
         let execution = ref_mut.value_mut();
         if execution.state != BackupState::Pending {
@@ -100,14 +113,14 @@ impl BackupEngine {
         let execution = execution.clone();
         let (tx, rx) = oneshot::channel();
         let handle = tokio::spawn(async move { execution_runner.run(execution, rx, false).await });
-        self.running_executions.insert(uuid, (tx, handle));
+        self.running_executions.insert(*uuid, (tx, handle));
         Ok(())
     }
 
-    pub async fn suspend_execution(&self, uuid: Uuid) -> Result<(), Error> {
+    pub async fn suspend_execution(&self, uuid: &Uuid) -> Result<(), Error> {
         let mut ref_mut = self
             .executions
-            .get_mut(&uuid)
+            .get_mut(uuid)
             .ok_or(TaskError::ExecutionNotFound)?;
         let execution = ref_mut.value_mut();
         if execution.state != BackupState::Running {
@@ -118,7 +131,7 @@ impl BackupEngine {
 
         let (_, (shutdown, handle)) = self
             .running_executions
-            .remove(&uuid)
+            .remove(uuid)
             .ok_or(TaskError::ExecutionNotFound)?;
         shutdown
             .send(())
@@ -127,14 +140,14 @@ impl BackupEngine {
         Ok(())
     }
 
-    pub async fn resume_execution(&self, uuid: Uuid) -> Result<(), Error> {
+    pub async fn resume_execution(&self, uuid: &Uuid) -> Result<(), Error> {
         if self.running_executions.contains_key(&uuid) {
             Err(TaskError::IllegalRunState)?
         }
 
         let mut ref_mut = self
             .executions
-            .get_mut(&uuid)
+            .get_mut(uuid)
             .ok_or(TaskError::ExecutionNotFound)?;
         let execution = ref_mut.value_mut();
         if execution.state != BackupState::Suspended {
@@ -146,21 +159,21 @@ impl BackupEngine {
         let execution = execution.clone();
         let (tx, rx) = oneshot::channel();
         let handle = tokio::spawn(async move { execution_runner.run(execution, rx, true).await });
-        self.running_executions.insert(uuid, (tx, handle));
+        self.running_executions.insert(*uuid, (tx, handle));
         Ok(())
     }
 
     fn to_execution_runner(&self) -> ExecutionRunner {
         let config = self.app_config.clone();
         let io_manager = self.io_manager.clone();
-        let actor_system = self.actor_system.clone();
+        let communication_manager = self.communication_manager.clone();
         let progress_tracker = self.progress_tracker.clone();
         let executions = self.executions.clone();
         let running_executions = self.running_executions.clone();
         ExecutionRunner::new(
             config,
             io_manager,
-            actor_system,
+            communication_manager,
             progress_tracker,
             executions,
             running_executions,
@@ -171,37 +184,32 @@ impl BackupEngine {
 struct ExecutionRunner {
     app_config: Arc<AppConfig>,
     io_manager: Arc<IOManager>,
-    actor_system: Arc<ActorSystem>,
+    communication_manager: Arc<CommunicationManager>,
     progress_tracker: Arc<ProgressTracker>,
-    executions: Arc<DashMap<Uuid, BackupExecution>>,
+    executions: Arc<DashMap<Uuid, Execution>>,
     running_executions: Arc<DashMap<Uuid, (oneshot::Sender<()>, JoinHandle<()>)>>,
 }
 
 impl ExecutionRunner {
     pub fn new(
         app_config: Arc<AppConfig>,
         io_manager: Arc<IOManager>,
-        actor_system: Arc<ActorSystem>,
+        communication_manager: Arc<CommunicationManager>,
         progress_tracker: Arc<ProgressTracker>,
-        executions: Arc<DashMap<Uuid, BackupExecution>>,
+        executions: Arc<DashMap<Uuid, Execution>>,
         running_executions: Arc<DashMap<Uuid, (oneshot::Sender<()>, JoinHandle<()>)>>,
     ) -> Self {
         Self {
             app_config,
             io_manager,
-            actor_system,
+            communication_manager,
             progress_tracker,
             executions,
             running_executions,
         }
     }
 
-    async fn run(
-        &self,
-        execution: BackupExecution,
-        mut shutdown: oneshot::Receiver<()>,
-        resume: bool,
-    ) {
+    async fn run(&self, execution: Execution, mut shutdown: oneshot::Receiver<()>, resume: bool) {
         let config = &self.app_config;
         let progress_tracker = &self.progress_tracker;
 
@@ -253,17 +261,16 @@ impl ExecutionRunner {
                         next_level.extend(worker_next_level);
                         if !worker_errors.is_empty() {
                             errors.extend(worker_errors.clone());
-                            if let Some(gui_ref) = self.actor_system.actor_of::<GuiMessageHandler>()
+                            let event = ExecutionErrors {
+                                uuid: execution.uuid,
+                                errors: worker_errors,
+                            };
+                            if let Err(err) = self
+                                .communication_manager
+                                .publish_event::<ExecutionErrors>(event)
+                                .await
                             {
-                                if let Err(err) = gui_ref
-                                    .tell(GuiMessage::ExecutionErrors {
-                                        uuid: execution.uuid,
-                                        errors: worker_errors,
-                                    })
-                                    .await
-                                {
-                                    error!("{}", err);
-                                }
+                                error!("{}", err);
                             }
                         }
                     }
@@ -317,7 +324,7 @@ impl Worker {
 
     async fn run(
         &self,
-        execution: BackupExecution,
+        execution: Execution,
         global_queue: Arc<SegQueue<PathBuf>>,
         mut shutdown: oneshot::Receiver<()>,
     ) -> (Vec<PathBuf>, Vec<Error>) {
@@ -382,7 +389,7 @@ impl Worker {
 
     async fn process_entry(
         &self,
-        execution: &BackupExecution,
+        execution: &Execution,
         current_path: &Path,
     ) -> Result<Option<PathBuf>, Error> {
         let io_manager = &self.io_manager;
@@ -391,7 +398,8 @@ impl Worker {
         let destination_root = &execution.destination_path;
 
         let source_path = current_path;
-        let destination_path = self.calculate_destination_path(source_path, source_root, destination_root)?;
+        let destination_path =
+            self.calculate_destination_path(source_path, source_root, destination_root)?;
         let destination_path = destination_path.as_path();
 
         let is_symlink = io_manager.is_symlink(source_path).await.unwrap_or(false);
@@ -413,7 +421,7 @@ impl Worker {
 
     async fn backup_directory(
         &self,
-        execution: &BackupExecution,
+        execution: &Execution,
         source_path: &Path,
         destination_path: &Path,
     ) -> Result<Option<PathBuf>, Error> {
@@ -438,7 +446,7 @@ impl Worker {
 
     async fn backup_file(
         &self,
-        execution: &BackupExecution,
+        execution: &Execution,
         source_path: &Path,
         destination_path: &Path,
     ) -> Result<Option<PathBuf>, Error> {
@@ -469,7 +477,7 @@ impl Worker {
     #[inline(always)]
     async fn process_symlink(
         &self,
-        execution: &BackupExecution,
+        execution: &Execution,
         source_path: &Path,
         destination_path: &Path,
     ) -> Result<(), Error> {
@@ -484,7 +492,7 @@ impl Worker {
 
     async fn follow_symlink(
         &self,
-        execution: &BackupExecution,
+        execution: &Execution,
         source_path: &Path,
         destination_path: &Path,
     ) -> Result<(), Error> {
@@ -541,7 +549,7 @@ impl Worker {
 
     async fn copy_symlink(
         &self,
-        execution: &BackupExecution,
+        execution: &Execution,
         source_path: &Path,
         destination_path: &Path,
     ) -> Result<(), Error> {
@@ -646,3 +654,39 @@ impl Worker {
         Ok(destination_root.join(relative_path))
     }
 }
+
+#[async_trait]
+impl CommandHandler<BackupCommand> for BackupEngine {
+    async fn handle_command(&self, command: BackupCommand) -> Result<(), Error> {
+        match command {
+            BackupCommand::AddExecution(execution) => {
+                self.add_execution(execution).await;
+            }
+            BackupCommand::RemoveExecution(uuid) => {
+                self.remove_execution(&uuid).await;
+            }
+            BackupCommand::StartExecution(uuid) => {
+                self.start_execution(&uuid).await?;
+            }
+            BackupCommand::SuspendExecution(uuid) => {
+                self.suspend_execution(&uuid).await?;
+            }
+            BackupCommand::ResumeExecution(uuid) => {
+                self.resume_execution(&uuid).await?;
+            }
+        }
+        Ok(())
+    }
+}
+
+#[async_trait]
+impl QueryHandler<BackupQuery> for BackupEngine {
+    async fn handle_query(&self, query: BackupQuery) -> Result<BackupQueryResponse, Error> {
+        match query {
+            BackupQuery::GetExecutions => {
+                let executions = self.get_all_executions();
+                Ok(BackupQueryResponse::GetExecutions(executions))
+            }
+        }
+    }
+}