Add once argument to amp-syncer

Cargo.toml

@@ -1,5 +1,5 @@
 [workspace.package]
-version = "0.6.3"
+version = "0.6.4"
 edition = "2021"
 license = "Apache-2.0"
 repository = "https://github.com/amphitheatre-app/amphitheatre"

syncer/src/config.rs

@@ -40,4 +40,7 @@ pub struct Config {
     /// The actor name.
     #[clap(long, env = "AMP_ACTOR")]
     pub actor: String,
+    // Exit after sync once (Overwrite).
+    #[clap(long, action = clap::ArgAction::Set, default_value = "false", env = "AMP_ONCE")]
+    pub once: bool,
 }

syncer/src/handle.rs

@@ -16,26 +16,27 @@ use std::path::Path;
 
 use amp_common::sync::{self, Synchronization};
 use tar::Archive;
-use tracing::{debug, error, trace, warn};
+use tracing::{debug, error, info, warn};
 
 type Result<T> = std::result::Result<T, Box<dyn std::error::Error>>;
 
 /// Overwrite workspace's files with payload tarball.
-pub fn overwrite(workspace: &Path, req: Synchronization) -> Result<()> {
+pub fn overwrite(workspace: &Path, req: &Synchronization) -> Result<()> {
     debug!("Received overwrite event, workspace: {:?}, req: {:?}", workspace, req);
 
-    if let Some(payload) = req.payload {
+    if let Some(payload) = &req.payload {
         let mut ar = Archive::new(payload.as_slice());
         ar.unpack(workspace)?;
+        info!("Received overwrite event and unpacked into workspace: {:?}", workspace);
     }
 
     Ok(())
 }
 
 /// Create new files or directories.
-pub fn create(workspace: &Path, req: Synchronization) -> Result<()> {
+pub fn create(workspace: &Path, req: &Synchronization) -> Result<()> {
     debug!("Received create event, workspace: {:?}, req: {:?}", workspace, req);
-    for path in req.paths {
+    for path in &req.paths {
         match path {
             sync::Path::File(file) => {
                 let path = workspace.join(file);
@@ -50,7 +51,8 @@ pub fn create(workspace: &Path, req: Synchronization) -> Result<()> {
                     }
                 }
 
-                std::fs::File::create(path)?;
+                std::fs::File::create(path.clone())?;
+                info!("Created file: {:?}", path);
             }
             sync::Path::Directory(path) => {
                 let path = workspace.join(path);
@@ -59,7 +61,8 @@ pub fn create(workspace: &Path, req: Synchronization) -> Result<()> {
                     continue;
                 }
 
-                std::fs::create_dir_all(path)?;
+                std::fs::create_dir_all(path.clone())?;
+                info!("Created directory: {:?}", path);
             }
         }
     }
@@ -68,28 +71,29 @@ pub fn create(workspace: &Path, req: Synchronization) -> Result<()> {
 }
 
 /// Modify existing files or directories.
-pub fn modify(workspace: &Path, req: Synchronization) -> Result<()> {
+pub fn modify(workspace: &Path, req: &Synchronization) -> Result<()> {
     debug!("Received modify event, workspace: {:?}, req: {:?}", workspace, req);
 
-    if let Some(payload) = req.payload {
+    if let Some(payload) = &req.payload {
         let mut ar = Archive::new(payload.as_slice());
         ar.unpack(workspace)?;
+        info!("Received modify event and unpacked into workspace: {:?}", workspace);
     }
 
     Ok(())
 }
 
 /// Rename existing files or directories.
-pub fn rename(_workspace: &Path, _req: Synchronization) -> Result<()> {
+pub fn rename(_workspace: &Path, _req: &Synchronization) -> Result<()> {
     error!("Received rename event, nothing to do!");
     Ok(())
 }
 
 /// Remove existing files or directories.
-pub fn remove(workspace: &Path, req: Synchronization) -> Result<()> {
+pub fn remove(workspace: &Path, req: &Synchronization) -> Result<()> {
     debug!("Received remove event, workspace: {:?}, req: {:?}", workspace, req);
 
-    for path in req.paths {
+    for path in &req.paths {
         match path {
             sync::Path::File(file) => {
                 let path = workspace.join(file);
@@ -98,7 +102,8 @@ pub fn remove(workspace: &Path, req: Synchronization) -> Result<()> {
                     continue;
                 }
 
-                std::fs::remove_file(path)?;
+                std::fs::remove_file(path.clone())?;
+                info!("Removed file: {:?}", path);
             }
             sync::Path::Directory(path) => {
                 let path = workspace.join(path);
@@ -107,8 +112,8 @@ pub fn remove(workspace: &Path, req: Synchronization) -> Result<()> {
                     continue;
                 }
 
-                trace!("Removing directory: {:?}", path);
-                std::fs::remove_dir_all(path)?;
+                std::fs::remove_dir_all(path.clone())?;
+                info!("Removed directory: {:?}", path);
             }
         }
     }

syncer/src/main.rs

@@ -62,11 +62,11 @@ async fn main() -> Result<(), async_nats::Error> {
 
         // Handle the message
         if let Err(err) = match req.kind {
-            Create => handle::create(workspace, req),
-            Modify => handle::modify(workspace, req),
-            Rename => handle::rename(workspace, req),
-            Remove => handle::remove(workspace, req),
-            Overwrite => handle::overwrite(workspace, req),
+            Create => handle::create(workspace, &req),
+            Modify => handle::modify(workspace, &req),
+            Rename => handle::rename(workspace, &req),
+            Remove => handle::remove(workspace, &req),
+            Overwrite => handle::overwrite(workspace, &req),
             Other => {
                 warn!("Received other event, nothing to do!");
                 Ok(())
@@ -83,6 +83,12 @@ async fn main() -> Result<(), async_nats::Error> {
         if let Err(err) = message.ack().await {
             error!("Failed to acknowledge message: {:?}", err);
         }
+
+        // If we're in once mode, exit after overwrite.
+        if config.once && req.kind == Overwrite {
+            info!("Exit after sync once (Overwrite), bye!");
+            std::process::exit(0);
+        }
     }
 
     Ok(())