Add replica management

Author: farost

src/main.rs

@@ -40,6 +40,7 @@ use crate::{
     },
     runtime::BackgroundRuntime,
 };
+use crate::operations::{replica_deregister, replica_list, replica_primary, replica_register, server_version};
 
 mod cli;
 mod completions;
@@ -326,6 +327,9 @@ fn execute_commands(context: &mut ConsoleContext, mut input: &str, must_log_comm
 }
 
 fn entry_repl(driver: Arc<TypeDBDriver>, runtime: BackgroundRuntime) -> Repl<ConsoleContext> {
+    let server_commands = Subcommand::new("server")
+        .add(CommandLeaf::new("version", "Retrieve server version.", server_version));
+    
     let database_commands = Subcommand::new("database")
         .add(CommandLeaf::new("list", "List databases on the server.", database_list))
         .add(CommandLeaf::new_with_input(
@@ -410,6 +414,25 @@ fn entry_repl(driver: Arc<TypeDBDriver>, runtime: BackgroundRuntime) -> Repl<Con
             user_update_password,
         ));
 
+    let replica_commands = Subcommand::new("replica")
+        .add(CommandLeaf::new("list", "List replicas.", replica_list))
+        .add(CommandLeaf::new("primary", "Get current primary replica.", replica_primary))
+        .add(CommandLeaf::new_with_inputs(
+            "register",
+            "Register new replica.",
+            vec![
+                CommandInput::new("replica id", get_word, None, None),
+                CommandInput::new("address", get_word, None, None),
+            ],
+            replica_register,
+        ))
+        .add(CommandLeaf::new_with_input(
+            "deregister",
+            "Deregister existing replica.",
+            CommandInput::new("replica id", get_word, None, None),
+            replica_deregister,
+        ));
+
     let transaction_commands = Subcommand::new("transaction")
         .add(CommandLeaf::new_with_input(
             "read",
@@ -445,8 +468,10 @@ fn entry_repl(driver: Arc<TypeDBDriver>, runtime: BackgroundRuntime) -> Repl<Con
     let history_path = home_dir().unwrap_or_else(|| temp_dir()).join(ENTRY_REPL_HISTORY);
 
     let repl = Repl::new(PROMPT.to_owned(), history_path, false, None)
+        .add(server_commands)
         .add(database_commands)
         .add(user_commands)
+        .add(replica_commands)
         .add(transaction_commands);
 
     repl

src/operations.rs

@@ -21,6 +21,16 @@ use crate::{
     transaction_repl, ConsoleContext,
 };
 
+pub(crate) fn server_version(context: &mut ConsoleContext, _input: &[String]) -> CommandResult {
+    let driver = context.driver.clone();
+    let server_version = context
+        .background_runtime
+        .run(async move { driver.server_version().await })
+        .map_err(|err| Box::new(err) as Box<dyn Error + Send>)?;
+    println!("{}", server_version);
+    Ok(())
+}
+
 pub(crate) fn database_list(context: &mut ConsoleContext, _input: &[String]) -> CommandResult {
     let driver = context.driver.clone();
     let databases = context
@@ -201,6 +211,55 @@ pub(crate) fn user_update_password(context: &mut ConsoleContext, input: &[String
     Ok(())
 }
 
+pub(crate) fn replica_list(context: &mut ConsoleContext, _input: &[String]) -> CommandResult {
+    let driver = context.driver.clone();
+    let replicas = driver.replicas();
+    if replicas.is_empty() {
+        println!("No replicas are present.");
+    } else {
+        for replica in replicas {
+            println!("{}", replica.address());
+        }
+    }
+    Ok(())
+}
+
+pub(crate) fn replica_primary(context: &mut ConsoleContext, _input: &[String]) -> CommandResult {
+    let driver = context.driver.clone();
+    let primary_replica = driver.primary_replica();
+    if let Some(primary_replica) = primary_replica {
+        println!("{}", primary_replica.address());
+    } else {
+        println!("No primary replica is present.");
+    }
+    Ok(())
+}
+
+pub(crate) fn replica_register(context: &mut ConsoleContext, input: &[String]) -> CommandResult {
+    let driver = context.driver.clone();
+    let replica_id: u64 = input[0].parse().map_err(|_| Box::new(ReplError { message: format!("Replica id '{}' must be a number.", input[0]) })
+        as Box<dyn Error + Send>)?;
+    let address = input[1].clone();
+    context
+        .background_runtime
+        .run(async move { driver.register_replica(replica_id, address).await })
+        .map_err(|err| Box::new(err) as Box<dyn Error + Send>)?;
+    println!("Successfully registered replica.");
+    Ok(())
+}
+
+pub(crate) fn replica_deregister(context: &mut ConsoleContext, input: &[String]) -> CommandResult {
+    let driver = context.driver.clone();
+    let replica_id: u64 = input[0].parse().map_err(|_| Box::new(ReplError { message: format!("Replica id '{}' must be a number.", input[0]) })
+        as Box<dyn Error + Send>)?;
+    context
+        .background_runtime
+        .run(async move { driver.deregister_replica(replica_id).await })
+        .map_err(|err| Box::new(err) as Box<dyn Error + Send>)?;
+    println!("Successfully deregistered replica.");
+    Ok(())
+}
+
 pub(crate) fn transaction_read(context: &mut ConsoleContext, input: &[String]) -> CommandResult {
     let driver = context.driver.clone();
     let db_name = &input[0];