Adding notification for other nodes on new node join

src/server/cluster.rs

@@ -5,7 +5,7 @@ use std::net::{SocketAddr, TcpStream};
 use std::sync::{Arc, Mutex};
 use log::{error, info, warn};
 use crate::server::cache::Key;
-use crate::server::commands::CmdResponseEnum;
+use crate::server::commands::{CmdResponseEnum, CommandsEnum};
 use crate::server::commands::CommandsEnum::{GetClusterState, JoinCluster};
 
 pub type NodeId = String;
@@ -91,6 +91,17 @@ impl Cluster {
         }).collect()
     }
 
+    pub fn notify_cluster_nodes(&self, command: CommandsEnum) {
+        for (node_id, stream) in self.node_connections.lock().unwrap().iter() {
+            info!("Notifying {node_id}");
+            let mut writer = BufWriter::new(stream.try_clone().unwrap());
+            let mut command_str = serde_json::to_string(&command).unwrap();
+            command_str.push('\n');
+            writer.write_all(command_str.as_bytes()).unwrap();
+            writer.flush().unwrap();
+        }
+    }
+
     pub fn redistribute_buckets(&self) {
         let mut nodes: Vec<NodeId> = self.node_connections.lock().unwrap().keys().cloned().collect();
         nodes.push(self.self_node_id.to_string());

src/server/cluster_command_processing.rs

@@ -11,7 +11,12 @@ pub fn process_cluster_command(command: CommandsEnum, cluster: &mut Cluster, con
             nodes_to_ips.insert(cluster.self_node_id.to_string(), connection_stream.local_addr().unwrap());
             cluster.redistribute_buckets();
             let buckets_to_nodes = cluster.get_bucket_node_assignments();
-            // TODO: leader sends new ClusterState to all rest of nodes (to set new node responsible for those buckets)
+
+            let update_cluster_cmd = CommandsEnum::UpdateClusterState {
+                nodes_to_ips: nodes_to_ips.clone(),
+                buckets_to_nodes: buckets_to_nodes.clone(),
+            };
+            cluster.notify_cluster_nodes(update_cluster_cmd);
             CmdResponseEnum::ClusterState { nodes_to_ips, buckets_to_nodes }
         }
         CommandsEnum::GetClusterState {} => {

src/server/listener.rs

@@ -30,7 +30,8 @@ pub fn start_server(cache: Cache, cluster: Cluster, client_port: u32, server_por
         }
     });
     let server_threads = thread::spawn(move || {
-        let server_pool = ThreadPoolBuilder::new().num_threads(1).build().unwrap();
+        // if new nodes never disconnect, we need at least number of threads = number of expected nodes
+        let server_pool = ThreadPoolBuilder::new().num_threads(2).build().unwrap();
         for stream in server_listener.incoming() {
             let server_cluster_status_per_connection = Arc::clone(&server_cluster);
             server_pool.spawn(move || {