feat: paxos support in server -> Implement broadcaster

src/db/loki_kv/control.rs

@@ -14,6 +14,8 @@ pub struct ControlFile {
     last_checkpoint_id: u64,
     checkpoint_directory_path: String,
     wal_directory_path: String,
+    current_leader_value: Option<u64>,
+    self_identifier: Option<u64>,
 }
 
 impl ControlFile {
@@ -23,12 +25,39 @@ impl ControlFile {
     pub fn get_next_timeline_id(&self) -> u64 {
         self.last_wal_timeline + 1
     }
+
     pub fn get_wal_directory_path(&self) -> &str {
         &self.wal_directory_path
     }
+
     pub fn get_checkpoint_directory_path(&self) -> &str {
         &self.checkpoint_directory_path
     }
+
+    pub fn get_current_leader_identifier(&self) -> Option<u64> {
+        self.current_leader_value.clone()
+    }
+
+    pub fn get_self_identifier(&self) -> Option<u64> {
+        self.self_identifier.clone()
+    }
+
+    pub fn set_current_leader_identifier(&mut self, current_leader_value: u64) {
+        self.current_leader_value = Some(current_leader_value.clone());
+    }
+
+    pub fn set_self_identifier(&mut self, id: u64) {
+        self.self_identifier = Some(id.clone());
+    }
+
+    pub fn is_leader(&self) -> bool {
+        if self.self_identifier.is_some() && self.current_leader_value.is_some() {
+            if self.self_identifier.unwrap() == self.current_leader_value.unwrap() {
+                return true;
+            }
+        }
+        return false;
+    }
     pub fn write(
         path: String,
         last_wal_timeline: u64,
@@ -53,6 +82,8 @@ impl ControlFile {
             last_checkpoint_id,
             checkpoint_directory_path,
             wal_directory_path,
+            current_leader_value: None,
+            self_identifier: Some(1 as u64),
         };
 
         // Take lock on control file

src/db/server_multithread/mod.rs

@@ -1 +1,2 @@
+pub mod paxos;
 pub mod server;

src/db/server_multithread/paxos.rs

@@ -0,0 +1,52 @@
+use std::net::SocketAddr;
+
+use tokio::net::UdpSocket;
+
+use crate::loki_kv::control::ControlFile;
+
+pub struct ServiceManager {
+    udp_socket: UdpSocket,
+    BROADCAST_ADDRESS: SocketAddr,
+}
+
+impl ServiceManager {
+    pub fn new() -> Self {
+        let listen_addr: SocketAddr = "0.0.0.0:8080".parse().unwrap();
+        let std_socket = std::net::UdpSocket::bind(listen_addr).unwrap();
+
+        std_socket.set_broadcast(true);
+
+        ServiceManager {
+            udp_socket: UdpSocket::from_std(std_socket).unwrap(),
+            BROADCAST_ADDRESS: "255.255.255.255:8080".parse().unwrap(),
+        }
+    }
+
+    pub async fn broadcast_message(&self, msg: &str) -> Result<(), String> {
+        self.udp_socket.send_to(msg.as_bytes(), self.BROADCAST_ADDRESS).await.unwrap();
+        Ok(())
+    }
+}
+
+pub struct PaxosNode {
+    ctrl_file: ControlFile,
+    service_manager: ServiceManager,
+}
+
+impl PaxosNode {
+    pub fn propose(&self) {
+        let value = self.ctrl_file.get_self_identifier();
+
+        // Broadcast to network
+        match value{
+           Some(val) => {
+               let msg = format!("PROPOSE {}", val);
+               self.service_manager.broadcast_message(msg.as_str());
+           },
+           None => panic!("No value to broadcast!")
+        }
+    }
+
+    pub fn accept() {}
+    pub fn learn() {}
+}