Merge pull request #12 from ivan770/cli

Better CLI

README.md

@@ -31,12 +31,9 @@
 ### Download binary from GitHub
 
 1. Download latest release from [GitHub](https://github.com/ivan770/spartan/releases/latest).
-2. Create Spartan.toml configuration file with example configuration below:
-```
-queues = ["default"]
-```
+2. Create Spartan.toml configuration file using `./spartan init`, add queues to it.
 3. Create empty directory with name `db` (you may change directory name using `Spartan.toml`).
-4. Run binary using `./spartan`, or `./spartan.exe` if you are using Windows.
+4. Start server with `./spartan start`.
 
 ### Build from source
 
@@ -45,14 +42,17 @@ Make sure you have Rust toolchain installed on your system.
 ```
 git clone https://github.com/ivan770/spartan
 cd spartan
-cargo run --release
+cargo build --release
 ```
 
 ## Configuration
 
-### Executable flags
+### Generic flags
 
 * `--config` - Change configuration file path (default: `Spartan.toml`).
+
+### `start` command flags
+
 * `--host` - Change server host (default: `127.0.0.1:5680`).
 
 ### Spartan.toml keys

spartan/Cargo.toml

@@ -30,7 +30,7 @@ version = "0.4"
 version = "1.0"
 
 [dependencies.bincode]
-version = "1.2"
+version = "1.3"
 
 [dependencies.toml]
 version = "0.5"

spartan/src/actions/mod.rs

@@ -24,7 +24,7 @@ enum QueueError {
     #[error("No message available")]
     NoMessageAvailable,
     #[error("Message not found")]
-    MessageNotFound
+    MessageNotFound,
 }
 
 impl ResponseError for QueueError {

spartan/src/cli/commands/init.rs

@@ -0,0 +1,32 @@
+use crate::{cli::Server, config::Config};
+use std::io::Error as IoError;
+use structopt::StructOpt;
+use thiserror::Error;
+use tokio::fs::write;
+use toml::to_vec;
+
+#[derive(Error, Debug)]
+pub enum InitCommandError {
+    #[error("Unable to serialize config")]
+    ConfigSerializationError,
+    #[error("Unable to write serialized config to file: {0}")]
+    ConfigWriteError(IoError),
+}
+
+#[derive(StructOpt)]
+pub struct InitCommand {}
+
+impl InitCommand {
+    pub async fn dispatch(&self, server: &Server) -> Result<(), InitCommandError> {
+        let config = Config::default();
+
+        write(
+            server.config_path(),
+            to_vec(&config).map_err(|_| InitCommandError::ConfigSerializationError)?,
+        )
+        .await
+        .map_err(InitCommandError::ConfigWriteError)?;
+
+        Ok(())
+    }
+}

spartan/src/cli/commands/mod.rs

@@ -0,0 +1,5 @@
+/// `start` command
+pub mod start;
+
+/// `init` command
+pub mod init;

spartan/src/cli/commands/start.rs

@@ -0,0 +1,98 @@
+use crate::{
+    cli::Server,
+    node::{
+        gc::spawn_gc, load_from_fs, persistence::PersistenceError, spawn_ctrlc_handler,
+        spawn_persistence, Manager,
+    },
+    routing::attach_routes,
+};
+use actix_rt::System;
+use actix_web::{web::Data, App, HttpServer};
+use std::{io::Error as IoError, net::SocketAddr};
+use structopt::StructOpt;
+use thiserror::Error;
+use tokio::{spawn, task::LocalSet};
+
+#[derive(Error, Debug)]
+pub enum StartCommandError {
+    #[error("Unable to restore DB from FS: {0}")]
+    RestoreDB(PersistenceError),
+    #[error("Unable to bind server to address: {0}")]
+    AddressBinding(IoError),
+    #[error("Internal server error: {0}")]
+    ServerError(IoError),
+    #[error("Unable to load configuration file")]
+    ConfigFileError,
+}
+
+#[derive(StructOpt)]
+pub struct StartCommand {
+    /// Server host
+    #[structopt(default_value = "127.0.0.1:5680", long)]
+    host: SocketAddr,
+}
+
+impl StartCommand {
+    pub fn host(&self) -> SocketAddr {
+        self.host
+    }
+
+    pub async fn dispatch(&self, server: &'static Server) -> Result<(), StartCommandError> {
+        debug!("Initializing runtime.");
+
+        let local_set = LocalSet::new();
+        let sys = System::run_in_tokio("server", &local_set);
+
+        debug!("Runtime initialized.");
+
+        info!("Initializing node.");
+
+        let mut manager = Manager::new(
+            server
+                .config()
+                .ok_or_else(|| StartCommandError::ConfigFileError)?,
+        );
+
+        info!("Node initialized.");
+
+        info!("Loading queues from FS.");
+
+        load_from_fs(&mut manager)
+            .await
+            .map_err(StartCommandError::RestoreDB)?;
+
+        info!("Queues loaded successfully.");
+
+        let manager = Data::new(manager);
+
+        debug!("Spawning GC handler.");
+
+        let cloned_manager = manager.clone();
+        spawn(async move { spawn_gc(&cloned_manager).await });
+
+        debug!("Spawning persistence job.");
+
+        let cloned_manager = manager.clone();
+        spawn(async move { spawn_persistence(&cloned_manager).await });
+
+        debug!("Spawning Ctrl-C handler");
+
+        let cloned_manager = manager.clone();
+        spawn(async move { spawn_ctrlc_handler(&cloned_manager).await });
+
+        HttpServer::new(move || {
+            App::new()
+                .app_data(manager.clone())
+                .configure(attach_routes)
+        })
+        .bind(self.host())
+        .map_err(StartCommandError::AddressBinding)?
+        .run()
+        .await
+        .map_err(StartCommandError::ServerError)?;
+
+        sys.await.map_err(StartCommandError::ServerError)?;
+
+        Ok(())
+    }
+}

spartan/src/cli/mod.rs

@@ -0,0 +1,58 @@
+/// CLI commands
+mod commands;
+
+use crate::config::Config;
+use commands::{init::InitCommand, start::StartCommand};
+use std::{
+    io::Error,
+    path::{Path, PathBuf},
+};
+use structopt::StructOpt;
+use tokio::fs::read;
+use toml::from_slice;
+
+#[derive(StructOpt)]
+pub enum Command {
+    #[structopt(about = "Start Spartan MQ server")]
+    Start(StartCommand),
+    #[structopt(about = "Initialize configuration file")]
+    Init(InitCommand),
+}
+
+#[derive(StructOpt)]
+pub struct Server {
+    /// Server configuration path
+    #[structopt(default_value = "Spartan.toml", long)]
+    config: PathBuf,
+
+    /// Loaded server configuration
+    #[structopt(skip = None)]
+    loaded_config: Option<Config>,
+
+    #[structopt(subcommand)]
+    command: Command,
+}
+
+impl Server {
+    /// Load configuration
+    pub async fn load_config(mut self) -> Result<Self, Error> {
+        match read(self.config.as_path()).await {
+            Ok(file) => self.loaded_config = Some(from_slice(&file)?),
+            Err(e) => info!("Unable to load configuration file: {}", e),
+        };
+
+        Ok(self)
+    }
+
+    pub fn config(&self) -> Option<&Config> {
+        self.loaded_config.as_ref()
+    }
+
+    pub fn config_path(&self) -> &Path {
+        self.config.as_path()
+    }
+
+    pub fn command(&self) -> &Command {
+        &self.command
+    }
+}

spartan/src/config.rs

@@ -0,0 +1,45 @@
+use serde::{Deserialize, Serialize};
+use std::path::PathBuf;
+
+/// Server configuration
+#[derive(Serialize, Deserialize)]
+pub struct Config {
+    /// Database path
+    #[serde(default)]
+    pub path: PathBuf,
+
+    /// Amount of seconds between persistence jobs
+    #[serde(default)]
+    pub persistence_timer: u64,
+
+    /// Amount of seconds between GC jobs
+    #[serde(default)]
+    pub gc_timer: u64,
+
+    /// Array of queues
+    pub queues: Vec<String>,
+}
+
+#[cfg(not(test))]
+impl Default for Config {
+    fn default() -> Config {
+        Config {
+            path: PathBuf::from("./db"),
+            persistence_timer: 900,
+            gc_timer: 300,
+            queues: Vec::new(),
+        }
+    }
+}
+
+#[cfg(test)]
+impl Default for Config {
+    fn default() -> Config {
+        Config {
+            path: PathBuf::from("./db"),
+            persistence_timer: 30,
+            gc_timer: 10,
+            queues: vec![String::from("test")],
+        }
+    }
+}