refactor(cli): use SSE for streaming logs

Signed-off-by: Mateo Fernandez <[email protected]>

src/cli/Cargo.toml

@@ -3,15 +3,14 @@ name = "cli"
 version = "0.1.0"
 edition = "2021"
 
-# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
-
 [dependencies]
-clap = { version = "4.5.3", features = ["derive"] }
-toml = "0.8.12"
-tokio = { version = "1.36.0", features = ["full"] }
-serde = { version = "1.0.197", features = ["derive"] }
-serde_yaml = "0.9.34"
-schemars = "0.8.16"
-serde_json = "1.0.115"
-reqwest = "0.12.3"
-shared_models = { path="../shared-models" }
+clap = { version = "4.5", features = ["derive"] }
+crossterm = "0.27"
+futures = "0.3"
+reqwest = { version = "0.12", features = ["json"] }
+reqwest-eventsource = "0.6"
+serde = { version = "1.0", features = ["derive"] }
+serde_json = "1.0"
+shared_models = { path = "../shared-models" }
+tokio = { version = "1.38", features = ["full"] }
+toml = "0.8"

src/cli/src/api_client/execute.rs

@@ -0,0 +1,38 @@
+use super::Error;
+use reqwest_eventsource::EventSource;
+use serde::Deserialize;
+use shared_models::CloudletDtoRequest;
+
+pub async fn execute(base_url: &str, request: CloudletDtoRequest) -> Result<EventSource, Error> {
+    let client = reqwest::Client::new()
+        .post(format!("{base_url}/run"))
+        .json(&request);
+
+    EventSource::new(client).map_err(Error::CreateEventSource)
+}
+
+#[derive(Debug, Deserialize)]
+pub struct ExecuteJsonResponse {
+    pub stage: Stage,
+    pub stdout: Option<String>,
+    pub stderr: Option<String>,
+    pub exit_code: Option<i32>,
+}
+
+#[derive(Debug, Deserialize)]
+pub enum Stage {
+    Pending,
+    Building,
+    Running,
+    Done,
+    Failed,
+    Debug,
+}
+
+impl TryFrom<String> for ExecuteJsonResponse {
+    type Error = Error;
+
+    fn try_from(value: String) -> Result<Self, Self::Error> {
+        serde_json::from_str(&value).map_err(|_| Error::ExecuteResponseDeserialize)
+    }
+}

src/cli/src/api_client/mod.rs

@@ -0,0 +1,50 @@
+use crate::utils;
+use serde::Deserialize;
+use shared_models::{BuildConfig, CloudletDtoRequest, Language, ServerConfig};
+use std::{fs, path::PathBuf};
+
+pub mod execute;
+pub mod shutdown;
+
+pub use execute::*;
+pub use shutdown::*;
+
+#[derive(Debug)]
+pub enum Error {
+    ReadTomlConfigFile(std::io::Error),
+    TomlConfigParse(toml::de::Error),
+    ReadCodeFile(std::io::Error),
+    ExecuteRequestBody,
+    CreateEventSource(reqwest_eventsource::CannotCloneRequestError),
+    ExecuteResponseDeserialize,
+    ShutdownSendRequest(reqwest::Error),
+    ShutdownResponse(reqwest::Error),
+}
+
+#[derive(Debug, Deserialize)]
+#[serde(rename_all = "kebab-case")]
+struct TomlConfig {
+    workload_name: String,
+    language: Language,
+    action: String,
+    server: ServerConfig,
+    build: BuildConfig,
+}
+
+pub fn new_cloudlet_request(config_path: &PathBuf) -> Result<CloudletDtoRequest, Error> {
+    let toml_file = fs::read_to_string(config_path).map_err(Error::ReadTomlConfigFile)?;
+    let config: TomlConfig = toml::from_str(&toml_file).map_err(Error::TomlConfigParse)?;
+
+    let source_code_path = &config.build.source_code_path;
+    let code: String = utils::read_file(source_code_path).map_err(Error::ReadCodeFile)?;
+
+    Ok(CloudletDtoRequest {
+        workload_name: config.workload_name,
+        language: config.language,
+        code,
+        log_level: shared_models::LogLevel::INFO,
+        server: config.server,
+        build: config.build,
+        action: config.action,
+    })
+}

src/cli/src/api_client/shutdown.rs

@@ -0,0 +1,15 @@
+use super::Error;
+use shared_models::CloudletShutdownResponse;
+
+pub async fn shutdown(base_url: &str) -> Result<CloudletShutdownResponse, Error> {
+    let client = reqwest::Client::new();
+
+    client
+        .post(format!("{base_url}/shutdown"))
+        .send()
+        .await
+        .map_err(Error::ShutdownSendRequest)?
+        .json::<CloudletShutdownResponse>()
+        .await
+        .map_err(Error::ShutdownSendRequest)
+}

src/cli/src/args.rs

@@ -14,5 +14,5 @@ pub enum Commands {
         #[arg(short, long)]
         config_path: PathBuf,
     },
-    Shutdown {},
+    Shutdown,
 }

src/cli/src/lib.rs

@@ -0,0 +1,69 @@
+use api_client::ExecuteJsonResponse;
+use args::{CliArgs, Commands};
+use crossterm::style::Stylize;
+use futures::TryStreamExt;
+use reqwest_eventsource::Event;
+use std::fmt::Display;
+
+mod api_client;
+pub mod args;
+mod utils;
+
+#[derive(Debug)]
+pub enum Error {
+    StdoutExecute(std::io::Error),
+    ApiClient(api_client::Error),
+    InvalidRequest(String),
+    ProgramFailed,
+}
+
+pub async fn run_cli(base_url: &str, args: CliArgs) -> Result<i32, Error> {
+    match args.command {
+        Commands::Run { config_path } => {
+            let body = api_client::new_cloudlet_request(&config_path).map_err(Error::ApiClient)?;
+            let mut es = api_client::execute(base_url, body)
+                .await
+                .map_err(Error::ApiClient)?;
+
+            let mut exit_code = 0;
+
+            while let Ok(Some(event)) = es.try_next().await {
+                match event {
+                    Event::Open => { /* skip */ }
+                    Event::Message(msg) => {
+                        let exec_response = ExecuteJsonResponse::try_from(msg.data);
+                        if let Ok(exec_response) = exec_response {
+                            if let Some(stdout) = exec_response.stdout {
+                                println!("{}", stylize(stdout, &exec_response.stage));
+                            }
+                            if let Some(stderr) = exec_response.stderr {
+                                println!("{}", stylize(stderr, &exec_response.stage));
+                            }
+                            if let Some(code) = exec_response.exit_code {
+                                exit_code = code;
+                            }
+                        }
+                    }
+                }
+            }
+
+            Ok(exit_code)
+        }
+        Commands::Shutdown {} => {
+            api_client::shutdown(base_url)
+                .await
+                .map_err(Error::ApiClient)?;
+
+            Ok(0)
+        }
+    }
+}
+
+fn stylize(output: String, stage: &api_client::Stage) -> impl Display {
+    match stage {
+        api_client::Stage::Building => output.yellow(),
+        api_client::Stage::Failed => output.dark_red(),
+        api_client::Stage::Debug => output.dark_blue(),
+        _ => output.stylize(),
+    }
+}

src/cli/src/main.rs

@@ -1,49 +1,20 @@
+use std::process::exit;
 use clap::Parser;
-
-use args::{CliArgs, Commands};
-
-use services::CloudletClient;
-use std::{fs, io, process::exit};
-
-mod args;
-mod services;
-mod utils;
+use cli::args::CliArgs;
 
 #[tokio::main]
-async fn main() -> io::Result<()> {
+async fn main() {
     let args = CliArgs::parse();
 
-    match args.command {
-        Commands::Run { config_path } => {
-            let toml_file = match fs::read_to_string(config_path.clone()) {
-                Ok(c) => c,
-                Err(_) => {
-                    eprintln!("Could not read file `{:?}`", config_path);
-                    exit(1);
-                }
-            };
-            let body = CloudletClient::new_cloudlet_config(toml_file);
-            let response = CloudletClient::run(body).await;
+    let api_url = std::env::var("API_URL").unwrap_or("localhost:3000".into());
+    let api_url = format!("http://{api_url}");
 
-            match response {
-                Ok(_) => println!("Request successful {:?}", response),
-                Err(e) => eprintln!("Error while making the request: {}", e),
-            }
-        }
-        Commands::Shutdown {} => {
-            let response = CloudletClient::shutdown().await;
-            match response {
-                Ok(bool) => {
-                    if bool {
-                        println!("Shutdown Request successful !")
-                    } else {
-                        println!("Shutdown Request Failed")
-                    }
-                }
-                Err(()) => println!("Cannot send shutdown Request"),
-            }
+    let result = cli::run_cli(&api_url, args).await;
+    match result {
+        Ok(exit_code) => exit(exit_code),
+        Err(e) => {
+            eprintln!("Could not execute the command:\n{:?}", e);
+            exit(1);
         }
     }
-
-    Ok(())
 }

src/cli/src/utils.rs

@@ -1,14 +1,11 @@
 use std::fs::File;
-use std::io::{self, Read};
+use std::io::Read;
 use std::path::PathBuf;
 
-pub struct ConfigFileHandler {}
+pub fn read_file(file_path: &PathBuf) -> std::io::Result<String> {
+    let mut file = File::open(file_path)?;
+    let mut contents = String::new();
+    file.read_to_string(&mut contents)?;
 
-impl ConfigFileHandler {
-    pub fn read_file(file_path: &PathBuf) -> io::Result<String> {
-        let mut file = File::open(file_path)?;
-        let mut contents = String::new();
-        file.read_to_string(&mut contents)?;
-        Ok(contents)
-    }
+    Ok(contents)
 }