replace 'daddr' checks with 'is_valid_address()'

src/bin/client-demo.rs

@@ -396,7 +396,6 @@ fn converge(
     threads: &mut Vec<JoinHandle<()>>,
 ) -> Vec<NodeInfo> {
     //lets spy on the network
-    let daddr = "0.0.0.0:0".parse().unwrap();
     let (spy, spy_gossip) = spy_node();
     let mut spy_crdt = Crdt::new(spy).expect("Crdt::new");
     spy_crdt.insert(&leader);
@@ -420,7 +419,7 @@ fn converge(
             .table
             .values()
             .into_iter()
-            .filter(|x| x.contact_info.rpu != daddr)
+            .filter(|x| Crdt::is_valid_address(x.contact_info.rpu))
             .cloned()
             .collect();
         if v.len() >= num_nodes {

src/crdt.rs

@@ -524,15 +524,14 @@ impl Crdt {
     pub fn compute_broadcast_table(&self) -> Vec<NodeInfo> {
         let live: Vec<_> = self.alive.iter().collect();
         //thread_rng().shuffle(&mut live);
-        let daddr = "0.0.0.0:0".parse().unwrap();
         let me = &self.table[&self.me];
         let cloned_table: Vec<NodeInfo> = live.iter()
             .map(|x| &self.table[x.0])
             .filter(|v| {
                 if me.id == v.id {
                     //filter myself
                     false
-                } else if v.contact_info.tvu == daddr {
+                } else if !(Self::is_valid_address(v.contact_info.tvu)) {
                     trace!(
                         "{:x}:broadcast skip not listening {:x}",
                         me.debug_id(),
@@ -640,7 +639,6 @@ impl Crdt {
             .set_id(me.id)
             .expect("set_id in pub fn retransmit");
         let rblob = blob.read().unwrap();
-        let daddr = "0.0.0.0:0".parse().unwrap();
         let orders: Vec<_> = table
             .iter()
             .filter(|v| {
@@ -649,7 +647,7 @@ impl Crdt {
                 } else if me.leader_id == v.id {
                     trace!("skip retransmit to leader {:?}", v.id);
                     false
-                } else if v.contact_info.tvu == daddr {
+                } else if !(Self::is_valid_address(v.contact_info.tvu)) {
                     trace!("skip nodes that are not listening {:?}", v.id);
                     false
                 } else {
@@ -710,10 +708,9 @@ impl Crdt {
     }
 
     pub fn window_index_request(&self, ix: u64) -> Result<(SocketAddr, Vec<u8>)> {
-        let daddr = "0.0.0.0:0".parse().unwrap();
         let valid: Vec<_> = self.table
             .values()
-            .filter(|r| r.id != self.me && r.contact_info.tvu_window != daddr)
+            .filter(|r| r.id != self.me && Self::is_valid_address(r.contact_info.tvu_window))
             .collect();
         if valid.is_empty() {
             Err(CrdtError::NoPeers)?;
@@ -1143,6 +1140,10 @@ impl Crdt {
             })
             .unwrap()
     }
+
+    pub fn is_valid_address(addr: SocketAddr) -> bool {
+        (addr.port() != 0) && !(addr.ip().is_unspecified() || addr.ip().is_multicast())
+    }
 }
 
 pub struct Sockets {
@@ -1918,4 +1919,14 @@ mod tests {
         assert!(!me.alive.contains_key(&node_with_same_addr.id));
         assert!(me.alive[&node_with_diff_addr.id] > 0);
     }
+
+    #[test]
+    fn test_is_valid_address() {
+        let bad_address_port = "127.0.0.1:0".parse().unwrap();
+        assert!(!Crdt::is_valid_address(bad_address_port));
+        let bad_address_unspecified = "0.0.0.0:1234".parse().unwrap();
+        assert!(!Crdt::is_valid_address(bad_address_unspecified));
+        let bad_address_multicast = "224.254.0.0:1234".parse().unwrap();
+        assert!(!Crdt::is_valid_address(bad_address_multicast));
+    }
 }

tests/multinode.rs

@@ -54,7 +54,7 @@ fn converge(leader: &NodeInfo, num_nodes: usize) -> Vec<NodeInfo> {
             .values()
             .into_iter()
             .filter(|x| x.id != me)
-            .filter(|x| x.contact_info.rpu != daddr)
+            .filter(|x| Crdt::is_valid_address(x.contact_info.rpu))
             .cloned()
             .collect();
         if num >= num_nodes as u64 && v.len() >= num_nodes {
@@ -483,9 +483,8 @@ fn mk_client(leader: &NodeInfo) -> ThinClient {
         .set_read_timeout(Some(Duration::new(1, 0)))
         .unwrap();
     let transactions_socket = UdpSocket::bind("0.0.0.0:0").unwrap();
-    let daddr = "0.0.0.0:0".parse().unwrap();
-    assert!(leader.contact_info.rpu != daddr);
-    assert!(leader.contact_info.tpu != daddr);
+    assert!(Crdt::is_valid_address(leader.contact_info.rpu));
+    assert!(Crdt::is_valid_address(leader.contact_info.tpu));
     ThinClient::new(
         leader.contact_info.rpu,
         requests_socket,