feat: implement remove cache task in dfcache (#603)

Signed-off-by: Gaius <gaius.qi@gmail.com>

dragonfly-client/src/bin/dfcache/remove.rs

@@ -15,20 +15,178 @@
  */
 
 use clap::Parser;
-use dragonfly_client_core::Result;
+use dragonfly_api::dfdaemon::v2::DeleteCacheTaskRequest;
-use std::path::Path;
+use dragonfly_client::grpc::dfdaemon_download::DfdaemonDownloadClient;
+use dragonfly_client::grpc::health::HealthClient;
+use dragonfly_client_core::{Error, Result};
+use indicatif::{ProgressBar, ProgressStyle};
+use std::path::{Path, PathBuf};
+use std::time::Duration;
+use termion::{color, style};
+use tracing::error;
+
+// DEFAULT_PROGRESS_BAR_STEADY_TICK_INTERVAL is the default steady tick interval of progress bar.
+const DEFAULT_PROGRESS_BAR_STEADY_TICK_INTERVAL: Duration = Duration::from_millis(80);
 
 // RemoveCommand is the subcommand of remove.
 #[derive(Debug, Clone, Parser)]
 pub struct RemoveCommand {
     #[arg(help = "Specify the cache task ID to remove")]
     id: String,
+
+    #[arg(
+        long = "timeout",
+        value_parser= humantime::parse_duration,
+        default_value = "30m",
+        help = "Specify the timeout for removing a file"
+    )]
+    timeout: Duration,
 }
 
 // Implement the execute for RemoveCommand.
 impl RemoveCommand {
-    pub async fn execute(&self, _endpoint: &Path) -> Result<()> {
+    // execute executes the delete command.
-        println!("RemoveCommand is executed!");
+    pub async fn execute(&self, endpoint: &Path) -> Result<()> {
+        // Run delete sub command.
+        if let Err(err) = self.run(endpoint).await {
+            match err {
+                Error::TonicStatus(status) => {
+                    eprintln!(
+                        "{}{}{}Removing Failed!{}",
+                        color::Fg(color::Red),
+                        style::Italic,
+                        style::Bold,
+                        style::Reset,
+                    );
+
+                    eprintln!(
+                        "{}{}{}*********************************{}",
+                        color::Fg(color::Black),
+                        style::Italic,
+                        style::Bold,
+                        style::Reset
+                    );
+
+                    eprintln!(
+                        "{}{}{}Bad Code:{} {}",
+                        color::Fg(color::Red),
+                        style::Italic,
+                        style::Bold,
+                        style::Reset,
+                        status.code()
+                    );
+
+                    eprintln!(
+                        "{}{}{}Message:{} {}",
+                        color::Fg(color::Cyan),
+                        style::Italic,
+                        style::Bold,
+                        style::Reset,
+                        status.message()
+                    );
+
+                    eprintln!(
+                        "{}{}{}Details:{} {}",
+                        color::Fg(color::Cyan),
+                        style::Italic,
+                        style::Bold,
+                        style::Reset,
+                        std::str::from_utf8(status.details()).unwrap()
+                    );
+
+                    eprintln!(
+                        "{}{}{}*********************************{}",
+                        color::Fg(color::Black),
+                        style::Italic,
+                        style::Bold,
+                        style::Reset
+                    );
+                }
+                err => {
+                    eprintln!(
+                        "{}{}{}Removing Failed!{}",
+                        color::Fg(color::Red),
+                        style::Italic,
+                        style::Bold,
+                        style::Reset
+                    );
+
+                    eprintln!(
+                        "{}{}{}****************************************{}",
+                        color::Fg(color::Black),
+                        style::Italic,
+                        style::Bold,
+                        style::Reset
+                    );
+
+                    eprintln!(
+                        "{}{}{}Message:{} {}",
+                        color::Fg(color::Red),
+                        style::Italic,
+                        style::Bold,
+                        style::Reset,
+                        err
+                    );
+
+                    eprintln!(
+                        "{}{}{}****************************************{}",
+                        color::Fg(color::Black),
+                        style::Italic,
+                        style::Bold,
+                        style::Reset
+                    );
+                }
+            }
+
+            std::process::exit(1);
+        }
+
         Ok(())
     }
+
+    // run runs the delete command.
+    async fn run(&self, endpoint: &Path) -> Result<()> {
+        let dfdaemon_download_client = self
+            .get_dfdaemon_download_client(endpoint.to_path_buf())
+            .await
+            .map_err(|err| {
+                error!("initialize dfdaemon download client failed: {}", err);
+                err
+            })?;
+
+        let pb = ProgressBar::new_spinner();
+        pb.enable_steady_tick(DEFAULT_PROGRESS_BAR_STEADY_TICK_INTERVAL);
+        pb.set_style(
+            ProgressStyle::with_template("{spinner:.blue} {msg}")
+                .unwrap()
+                .tick_strings(&["⣾", "⣽", "⣻", "⢿", "⡿", "⣟", "⣯", "⣷"]),
+        );
+        pb.set_message("Removing...");
+
+        dfdaemon_download_client
+            .delete_cache_task(
+                DeleteCacheTaskRequest {
+                    task_id: self.id.clone(),
+                },
+                self.timeout,
+            )
+            .await?;
+
+        pb.finish_with_message("Done");
+        Ok(())
+    }
+
+    // get_and_check_dfdaemon_download_client gets a dfdaemon download client and checks its health.
+    async fn get_dfdaemon_download_client(
+        &self,
+        endpoint: PathBuf,
+    ) -> Result<DfdaemonDownloadClient> {
+        // Check dfdaemon's health.
+        let health_client = HealthClient::new_unix(endpoint.clone()).await?;
+        health_client.check_dfdaemon_download().await?;
+
+        // Get dfdaemon download client.
+        let dfdaemon_download_client = DfdaemonDownloadClient::new_unix(endpoint).await?;
+        Ok(dfdaemon_download_client)
+    }
 }