mosuka · mosuka · Nov 18, 2019 · Nov 18, 2019
diff --git a/proto/indexpb.proto b/proto/indexpb.proto
@@ -17,4 +17,6 @@ service Index {
     }
     rpc RaftConfChange (indexrpcpb.ConfChangeReq) returns (indexrpcpb.RaftDone) {
     }
+    rpc Status (indexrpcpb.IndexReq) returns (indexrpcpb.StatusResp) {
+    }
 }
diff --git a/proto/indexrpcpb.proto b/proto/indexrpcpb.proto
@@ -14,7 +14,9 @@ enum ReqType {
     Put = 1;
     Delete = 2;
     Search = 3;
-    PeerAddr = 4;
+    Join = 4;
+    Leave = 5;
+    Status = 6;
 }
 
 message IndexReq {
@@ -34,6 +36,11 @@ message ConfChangeReq {
     uint32 port = 3;
 }
 
+message StatusResp {
+    string value = 1;
+    RespErr err = 2;
+}
+
 message GetResp {
     string value = 1;
     RespErr err = 2;

diff --git a/src/client/client.rs b/src/client/client.rs
@@ -4,10 +4,12 @@ use std::time::Duration;
 
 use grpcio::{ChannelBuilder, EnvBuilder};
 use log::*;
+use raft::eraftpb::{ConfChange, ConfChangeType};
 
 use crate::proto::indexpb_grpc::IndexClient;
 use crate::proto::indexrpcpb::{
-    DeleteResp, GetResp, IndexReq, PutResp, ReqType, RespErr, SearchResp,
+    ConfChangeReq, DeleteResp, GetResp, IndexReq, PutResp, RaftDone, ReqType, RespErr, SearchResp,
+    StatusResp,
 };
 
 pub fn create_client(addr: &str) -> IndexClient {
@@ -135,4 +137,85 @@ impl Clerk {
             thread::sleep(Duration::from_millis(100));
         }
     }
+
+    pub fn join(&mut self, id: u64, ip: &str, port: u16) {
+        let mut cc = ConfChange::new();
+        cc.set_id(id);
+        cc.set_node_id(id);
+        cc.set_change_type(ConfChangeType::AddNode);
+        let mut req = ConfChangeReq::new();
+        req.set_cc(cc);
+        req.set_ip(ip.to_string());
+        req.set_port(port as u32);
+
+        loop {
+            let reply = self.servers[self.leader_id]
+                .raft_conf_change(&req)
+                .unwrap_or_else(|e| {
+                    error!("{:?}", e);
+                    let mut resp = RaftDone::new();
+                    resp.set_err(RespErr::ErrWrongLeader);
+                    resp
+                });
+            match reply.err {
+                RespErr::OK => return,
+                RespErr::ErrWrongLeader => (),
+                RespErr::ErrNoKey => return,
+            }
+            self.leader_id = (self.leader_id + 1) % self.servers.len();
+            thread::sleep(Duration::from_millis(100));
+        }
+    }
+
+    pub fn leave(&mut self, id: u64) {
+        let mut cc = ConfChange::new();
+        cc.set_id(id);
+        cc.set_node_id(id);
+        cc.set_change_type(ConfChangeType::RemoveNode);
+        let mut req = ConfChangeReq::new();
+        req.set_cc(cc);
+
+        loop {
+            let reply = self.servers[self.leader_id]
+                .raft_conf_change(&req)
+                .unwrap_or_else(|e| {
+                    error!("{:?}", e);
+                    let mut resp = RaftDone::new();
+                    resp.set_err(RespErr::ErrWrongLeader);
+                    resp
+                });
+            match reply.err {
+                RespErr::OK => return,
+                RespErr::ErrWrongLeader => (),
+                RespErr::ErrNoKey => return,
+            }
+            self.leader_id = (self.leader_id + 1) % self.servers.len();
+            thread::sleep(Duration::from_millis(100));
+        }
+    }
+
+    pub fn status(&mut self) -> String {
+        let mut req = IndexReq::new();
+        req.set_client_id(self.client_id);
+        req.set_req_type(ReqType::Status);
+        req.set_seq(self.request_seq);
+        self.request_seq += 1;
+
+        loop {
+            let reply = self.servers[self.leader_id]
+                .status(&req)
+                .unwrap_or_else(|_e| {
+                    let mut resp = StatusResp::new();
+                    resp.set_err(RespErr::ErrWrongLeader);
+                    resp
+                });
+            match reply.err {
+                RespErr::OK => return reply.value,
+                RespErr::ErrWrongLeader => (),
+                RespErr::ErrNoKey => return String::from(""),
+            }
+            self.leader_id = (self.leader_id + 1) % self.servers.len();
+            thread::sleep(Duration::from_millis(100));
+        }
+    }
 }
diff --git a/src/cmd.rs b/src/cmd.rs
@@ -4,3 +4,4 @@ pub mod leave;
 pub mod search;
 pub mod serve;
 pub mod set;
+pub mod status;
diff --git a/src/cmd/delete.rs b/src/cmd/delete.rs
@@ -1,18 +1,16 @@
 use clap::ArgMatches;
 
 use crate::client::client::{create_client, Clerk};
-use crate::util::log::set_log_level;
+use crate::util::log::set_logger;
 
 pub fn run_delete_cli(matches: &ArgMatches) -> Result<(), String> {
-    set_log_level();
-    env_logger::init();
+    set_logger();
 
     let servers: Vec<_> = matches
         .values_of("SERVERS")
         .unwrap()
         .map(|addr| create_client(addr))
         .collect();
-
     let key = matches.value_of("KEY").unwrap();
 
     let client_id = rand::random();

diff --git a/src/cmd/get.rs b/src/cmd/get.rs
@@ -1,18 +1,16 @@
 use clap::ArgMatches;
 
 use crate::client::client::{create_client, Clerk};
-use crate::util::log::set_log_level;
+use crate::util::log::set_logger;
 
 pub fn run_get_cli(matches: &ArgMatches) -> Result<(), String> {
-    set_log_level();
-    env_logger::init();
+    set_logger();
 
     let servers: Vec<_> = matches
         .values_of("SERVERS")
         .unwrap()
         .map(|addr| create_client(addr))
         .collect();
-
     let key = matches.value_of("KEY").unwrap();
 
     let client_id = rand::random();

diff --git a/src/cmd/leave.rs b/src/cmd/leave.rs
@@ -1,45 +1,22 @@
-use std::collections::HashMap;
-
 use clap::ArgMatches;
-use log::*;
-use raft::eraftpb::{ConfChange, ConfChangeType};
 
-use crate::client::client::create_client;
-use crate::proto::indexrpcpb::ConfChangeReq;
-use crate::server::util::conf_change;
-use crate::util::log::set_log_level;
+use crate::client::client::{create_client, Clerk};
+use crate::util::log::set_logger;
 
 pub fn run_leave_cli(matches: &ArgMatches) -> Result<(), String> {
-    set_log_level();
-    env_logger::init();
+    set_logger();
 
-    let host = matches.value_of("HOST").unwrap();
-    let port = matches.value_of("PORT").unwrap().parse::<u16>().unwrap();
+    let servers: Vec<_> = matches
+        .values_of("SERVERS")
+        .unwrap()
+        .map(|addr| create_client(addr))
+        .collect();
     let id = matches.value_of("ID").unwrap().parse::<u64>().unwrap();
-    let mut peers = HashMap::new();
-    if let Some(peers_vec) = matches.values_of("PEERS") {
-        peers_vec
-            .map(|s| {
-                let mut parts = s.split('=');
-                let id = parts.next().unwrap().parse::<u64>().unwrap();
-                let addr = parts.next().unwrap();
-                peers.insert(id, create_client(addr));
-            })
-            .count();
-    }
-    if let Some(leader_id_str) = matches.value_of("LEADER_ID") {
-        let leader_id = leader_id_str.parse::<u64>().unwrap();
-        let mut cc = ConfChange::new();
-        cc.set_id(id);
-        cc.set_node_id(id);
-        cc.set_change_type(ConfChangeType::RemoveNode);
-        let mut req = ConfChangeReq::new();
-        req.set_cc(cc);
-        req.set_ip(host.to_owned());
-        req.set_port(port as u32);
-        conf_change(id, leader_id, &peers, req);
-        info!("the node was successfully removed from the cluster");
-    }
+
+    let client_id = rand::random();
+
+    let mut client = Clerk::new(&servers, client_id);
+    client.leave(id);
 
     Ok(())
 }
diff --git a/src/cmd/search.rs b/src/cmd/search.rs
@@ -1,18 +1,16 @@
 use clap::ArgMatches;
 
 use crate::client::client::{create_client, Clerk};
-use crate::util::log::set_log_level;
+use crate::util::log::set_logger;
 
 pub fn run_search_cli(matches: &ArgMatches) -> Result<(), String> {
-    set_log_level();
-    env_logger::init();
+    set_logger();
 
     let servers: Vec<_> = matches
         .values_of("SERVERS")
         .unwrap()
         .map(|addr| create_client(addr))
         .collect();
-
     let query = matches.value_of("QUERY").unwrap();
 
     let client_id = rand::random();

diff --git a/src/cmd/serve.rs b/src/cmd/serve.rs
@@ -5,20 +5,17 @@ use std::sync::Arc;
 
 use clap::ArgMatches;
 use log::*;
-use raft::eraftpb::{ConfChange, ConfChangeType};
 use serde_json;
 use tantivy::schema::Schema;
 use tantivy::Index;
 
-use crate::client::client::create_client;
-use crate::proto::indexrpcpb::ConfChangeReq;
+use crate::client::client::{create_client, Clerk};
+use crate::proto::indexpb_grpc::IndexClient;
 use crate::server::server::IndexServer;
-use crate::server::util::conf_change;
-use crate::util::log::set_log_level;
+use crate::util::log::set_logger;
 
 pub fn run_serve_cli(matches: &ArgMatches) -> Result<(), String> {
-    set_log_level();
-    env_logger::init();
+    set_logger();
 
     let host = matches.value_of("HOST").unwrap();
     let port = matches.value_of("PORT").unwrap().parse::<u16>().unwrap();
@@ -34,19 +31,11 @@ pub fn run_serve_cli(matches: &ArgMatches) -> Result<(), String> {
             })
             .count();
     }
-    if let Some(leader_id_str) = matches.value_of("LEADER_ID") {
-        let leader_id = leader_id_str.parse::<u64>().unwrap();
-        let mut cc = ConfChange::new();
-        cc.set_id(id);
-        cc.set_node_id(id);
-        cc.set_change_type(ConfChangeType::AddNode);
-        let mut req = ConfChangeReq::new();
-        req.set_cc(cc);
-        req.set_ip(host.to_owned());
-        req.set_port(port as u32);
-        conf_change(id, leader_id, &peers, req);
-        info!("the node was successfully added to the cluster");
-    }
+    let leader_id = matches
+        .value_of("LEADER_ID")
+        .unwrap_or("0")
+        .parse::<u64>()
+        .unwrap();
     let data_directory = matches.value_of("DATA_DIRECTORY").unwrap();
     let schema_file = matches.value_of("SCHEMA_FILE").unwrap();
     let unique_key_field_name = matches.value_of("UNIQUE_KEY_FIELD_NAME").unwrap();
@@ -79,6 +68,7 @@ pub fn run_serve_cli(matches: &ArgMatches) -> Result<(), String> {
         id,
         host,
         port,
+        leader_id,
         peers,
         Arc::new(index),
         unique_key_field_name,

diff --git a/src/cmd/set.rs b/src/cmd/set.rs
@@ -1,18 +1,16 @@
 use clap::ArgMatches;
 
 use crate::client::client::{create_client, Clerk};
-use crate::util::log::set_log_level;
+use crate::util::log::set_logger;
 
 pub fn run_set_cli(matches: &ArgMatches) -> Result<(), String> {
-    set_log_level();
-    env_logger::init();
+    set_logger();
 
     let servers: Vec<_> = matches
         .values_of("SERVERS")
         .unwrap()
         .map(|addr| create_client(addr))
         .collect();
-
     let key = matches.value_of("KEY").unwrap();
     let value = matches.value_of("VALUE").unwrap();
 

diff --git a/src/cmd/status.rs b/src/cmd/status.rs
@@ -0,0 +1,22 @@
+use clap::ArgMatches;
+
+use crate::client::client::{create_client, Clerk};
+use crate::util::log::set_logger;
+
+pub fn run_status_cli(matches: &ArgMatches) -> Result<(), String> {
+    set_logger();
+
+    let servers: Vec<_> = matches
+        .values_of("SERVERS")
+        .unwrap()
+        .map(|addr| create_client(addr))
+        .collect();
+
+    let client_id = rand::random();
+
+    let mut client = Clerk::new(&servers, client_id);
+    let value = client.status();
+    print!("{}", value);
+
+    Ok(())
+}
-Original file line number
+Diff line change
@@ Expand Up / @@ -17,4 +17,6 @@ service Index { @@
         }
         rpc RaftConfChange (indexrpcpb.ConfChangeReq) returns (indexrpcpb.RaftDone) {
         }
+        rpc Status (indexrpcpb.IndexReq) returns (indexrpcpb.StatusResp) {
+        }
     }