Fix/add so reuse

src/db/server_multithread/paxos.rs

@@ -1,4 +1,5 @@
 use local_ip_address::local_ip;
+use socket2::{Domain, SockAddr, Socket, Type};
 use std::{
     collections::{HashMap, HashSet},
     io::Split,
@@ -26,20 +27,34 @@ impl ServiceManager {
     pub fn new() -> Self {
         let control_file = ControlFile::read_from_file_path(get_control_file_path()).unwrap();
         let listen_addr: SocketAddr = control_file.get_send_addr().parse().unwrap();
-        let std_socket = std::net::UdpSocket::bind(listen_addr).unwrap();
+        let soc2_listen_socket = Socket::new(Domain::IPV4, Type::DGRAM, None).unwrap();
+        soc2_listen_socket.set_broadcast(true).unwrap();
+        soc2_listen_socket
+            .bind(&SockAddr::from(listen_addr))
+            .unwrap();
+        let std_socket: std::net::UdpSocket = soc2_listen_socket.into();
 
         let consume_addr: SocketAddr = control_file.get_consume_addr().parse().unwrap();
-        let std_consumer_socket = std::net::UdpSocket::bind(consume_addr).unwrap();
+        let soc2_raw_consumer_socket = match Socket::new(Domain::IPV4, Type::DGRAM, None) {
+            Ok(socket) => socket,
+            Err(e) => panic!("Failed to create socket: {}", e),
+        };
+        soc2_raw_consumer_socket.set_reuse_address(true).unwrap();
+        soc2_raw_consumer_socket
+            .bind(&SockAddr::from(consume_addr))
+            .unwrap();
+
+        let std_consumer_socket: std::net::UdpSocket = soc2_raw_consumer_socket.into();
 
-        let mut node_directory: HashSet<(String, String)> = HashSet::new();
+        let node_directory: HashSet<(String, String)> = HashSet::new();
 
-        std_socket.set_broadcast(true);
+        let broadcast_ip_string = format!("255.255.255.255:{}", consume_addr.port());
 
         ServiceManager {
             udp_socket_send: UdpSocket::from_std(std_socket).unwrap(),
             udp_socket_recv: UdpSocket::from_std(std_consumer_socket).unwrap(),
             node_directory: node_directory,
-            BROADCAST_ADDRESS: "255.255.255.255:8080".parse().unwrap(),
+            BROADCAST_ADDRESS: broadcast_ip_string.parse().unwrap(),
         }
     }
 
@@ -56,7 +71,10 @@ impl ServiceManager {
             // TODO: Add consumption logic
             // Somehitng like a go-routine treatment here?
             let mut msg_bytes: Vec<u8> = vec![];
-            self.udp_socket_recv.recv_from(&mut msg_bytes);
+            self.udp_socket_recv
+                .recv_from(&mut msg_bytes)
+                .await
+                .unwrap();
 
             tokio::spawn(async move {
                 // Log message