pub fn run(&self) {
        let mut recv_buffer: Vec<u8> = vec![0; 1024];

        loop {
            habitat_common::sync::mark_thread_alive();

            if self.server.paused() {
                thread::sleep(Duration::from_millis(100));
                continue;
            }

            match self.socket.recv_from(&mut recv_buffer[..]) {
                Ok((length, addr)) => {
                    let swim_payload = match self.server.unwrap_wire(&recv_buffer[0..length]) {
                        Ok(swim_payload) => swim_payload,
                        Err(e) => {
                            // NOTE: In the future, we might want to block people who send us
                            // garbage all the time.
                            error!("Error unwrapping protocol message, {}", e);
                            let label_values = &["unwrap_wire", "failure"];
                            SWIM_BYTES_RECEIVED.with_label_values(label_values)
                                               .set(length.to_i64());
                            SWIM_MESSAGES_RECEIVED.with_label_values(label_values).inc();
                            continue;
                        }
                    };

                    let bytes_received = swim_payload.len();
                    let msg = match Swim::decode(&swim_payload) {
                        Ok(msg) => msg,
                        Err(e) => {
                            // NOTE: In the future, we might want to block people who send us
                            // garbage all the time.
                            error!("Error decoding protocol message, {}", e);
                            let label_values = &["undecodable", "failure"];
                            SWIM_BYTES_RECEIVED.with_label_values(label_values)
                                               .set(bytes_received.to_i64());
                            SWIM_MESSAGES_RECEIVED.with_label_values(label_values).inc();
                            continue;
                        }
                    };

                    // Setting a label_values variable here throws errors about moving borrowed
                    // content that I couldn't solve w/o clones. Leaving this for now. I'm sure
                    // there's a better way.
                    SWIM_BYTES_RECEIVED.with_label_values(&[msg.kind.as_str(), "success"])
                                       .set(bytes_received.to_i64());
                    SWIM_MESSAGES_RECEIVED.with_label_values(&[msg.kind.as_str(), "success"])
                                          .inc();

                    trace!("SWIM Message: {:?}", msg);
                    match msg.kind {
                        SwimKind::Ping(ping) => {
                            if self.server.is_member_blocked(&ping.from.id) {
                                debug!("Not processing message from {} - it is blocked",
                                       ping.from.id);
                                continue;
                            }
                            self.process_ping(addr, ping);
                        }
                        SwimKind::Ack(ack) => {
                            if self.server.is_member_blocked(&ack.from.id)
                               && ack.forward_to.is_none()
                            {
                                debug!("Not processing message from {} - it is blocked",
                                       ack.from.id);
                                continue;
                            }
                            self.process_ack(addr, ack);
                        }
                        SwimKind::PingReq(pingreq) => {
                            if self.server.is_member_blocked(&pingreq.from.id) {
                                debug!("Not processing message from {} - it is blocked",
                                       pingreq.from.id);
                                continue;
                            }
                            self.process_pingreq(addr, pingreq);
                        }
                    }
                }
                Err(e) => {
                    // TODO: We can't use magic numbers here because the Supervisor runs on more
                    // than one platform. I'm sure these were added as specific OS errors for Linux
                    // but we need to also handle Windows & Mac.
                    match e.raw_os_error() {
                        Some(35) | Some(11) | Some(10035) | Some(10060) => {
                            // This is the normal non-blocking result, or a timeout
                        }
                        Some(_) => {
                            error!("UDP Receive error: {}", e);
                            debug!("UDP Receive error debug: {:?}", e);
                        }
                        None => {
                            error!("UDP Receive error: {}", e);
                        }
                    }
                }
            }
        }
    }
/// Run the thread. Listens for messages up to 1k in size, and then processes them accordingly.
Run the thread. Listens for messages up to 1k in size, and then processes them accordingly.
Run the thread . Listens for messages up to 1k in size and then processes them accordingly .
pub fn run(&self) {
        let mut recv_buffer: Vec<u8> = vec![0; 1024];
        loop {
            habitat_common::sync::mark_thread_alive();
            if self.server.paused() {
                thread::sleep(Duration::from_millis(100));
                continue;
            }
            match self.socket.recv_from(&mut recv_buffer[..]) {
                Ok((length, addr)) => {
                    let swim_payload = match self.server.unwrap_wire(&recv_buffer[0..length]) {
                        Ok(swim_payload) => swim_payload,
                        Err(e) => {
                            // NOTE: In the future, we might want to block people who send us
                            // garbage all the time.
                            error!("Error unwrapping protocol message, {}", e);
                            let label_values = &["unwrap_wire", "failure"];
                            SWIM_BYTES_RECEIVED.with_label_values(label_values)
                                               .set(length.to_i64());
                            SWIM_MESSAGES_RECEIVED.with_label_values(label_values).inc();
                            continue;
                        }
                    };
                    let bytes_received = swim_payload.len();
                    let msg = match Swim::decode(&swim_payload) {
                        Ok(msg) => msg,
                        Err(e) => {
                            // NOTE: In the future, we might want to block people who send us
                            // garbage all the time.
                            error!("Error decoding protocol message, {}", e);
                            let label_values = &["undecodable", "failure"];
                            SWIM_BYTES_RECEIVED.with_label_values(label_values)
                                               .set(bytes_received.to_i64());
                            SWIM_MESSAGES_RECEIVED.with_label_values(label_values).inc();
                            continue;
                        }
                    };
                    // Setting a label_values variable here throws errors about moving borrowed
                    // content that I couldn't solve w/o clones. Leaving this for now. I'm sure
                    // there's a better way.
                    SWIM_BYTES_RECEIVED.with_label_values(&[msg.kind.as_str(), "success"])
                                       .set(bytes_received.to_i64());
                    SWIM_MESSAGES_RECEIVED.with_label_values(&[msg.kind.as_str(), "success"])
                                          .inc();
                    trace!("SWIM Message: {:?}", msg);
                    match msg.kind {
                        SwimKind::Ping(ping) => {
                            if self.server.is_member_blocked(&ping.from.id) {
                                debug!("Not processing message from {} - it is blocked",
                                       ping.from.id);
                                continue;
                            }
                            self.process_ping(addr, ping);
                        }
                        SwimKind::Ack(ack) => {
                            if self.server.is_member_blocked(&ack.from.id)
                               && ack.forward_to.is_none()
                            {
                                debug!("Not processing message from {} - it is blocked",
                                       ack.from.id);
                                continue;
                            }
                            self.process_ack(addr, ack);
                        }
                        SwimKind::PingReq(pingreq) => {
                            if self.server.is_member_blocked(&pingreq.from.id) {
                                debug!("Not processing message from {} - it is blocked",
                                       pingreq.from.id);
                                continue;
                            }
                            self.process_pingreq(addr, pingreq);
                        }
                    }
                }
                Err(e) => {
                    // TODO: We can't use magic numbers here because the Supervisor runs on more
                    // than one platform. I'm sure these were added as specific OS errors for Linux
                    // but we need to also handle Windows & Mac.
                    match e.raw_os_error() {
                        Some(35) | Some(11) | Some(10035) | Some(10060) => {
                            // This is the normal non-blocking result, or a timeout
                        }
                        Some(_) => {
                            error!("UDP Receive error: {}", e);
                            debug!("UDP Receive error debug: {:?}", e);
                        }
                        None => {
                            error!("UDP Receive error: {}", e);
                        }
                    }
                }
            }
        }
    }
pub fn run ( & self ) { let mut recv_buffer : Vec < u8 > = vec ! [ 0 1024 ] ; loop { habitat_common :: sync :: mark_thread_alive ( ) ; if self . server . paused ( ) { thread :: sleep ( Duration :: from_millis ( 100 ) ) ; continue ; } match self . socket . recv_from ( & mut recv_buffer [ .. ] ) { Ok ( ( length , addr ) ) => { let swim_payload = match self . server . unwrap_wire ( & recv_buffer [ 0 .. length ] ) { Ok ( swim_payload ) => swim_payload , Err ( e ) => { // NOTE: In the future, we might want to block people who send us // garbage all the time. error ! ( " " e ) ; let label_values = & [ " " , " " ] ; SWIM_BYTES_RECEIVED . with_label_values ( label_values ) . set ( length . to_i64 ( ) ) ; SWIM_MESSAGES_RECEIVED . with_label_values ( label_values ) . inc ( ) ; continue ; } } ; let bytes_received = swim_payload . len ( ) ; let msg = match Swim :: decode ( & swim_payload ) { Ok ( msg ) => msg , Err ( e ) => { // NOTE: In the future, we might want to block people who send us // garbage all the time. error ! ( " " e ) ; let label_values = & [ " " , " " ] ; SWIM_BYTES_RECEIVED . with_label_values ( label_values ) . set ( bytes_received . to_i64 ( ) ) ; SWIM_MESSAGES_RECEIVED . with_label_values ( label_values ) . inc ( ) ; continue ; } } ; // Setting a label_values variable here throws errors about moving borrowed // content that I couldn't solve w/o clones. Leaving this for now. I'm sure // there's a better way. SWIM_BYTES_RECEIVED . with_label_values ( & [ msg . kind . as_str ( ) , " " ] ) . set ( bytes_received . to_i64 ( ) ) ; SWIM_MESSAGES_RECEIVED . with_label_values ( & [ msg . kind . as_str ( ) , " " ] ) . inc ( ) ; trace ! ( " " msg ) ; match msg . kind { SwimKind :: Ping ( ping ) => { if self . server . is_member_blocked ( & ping . from . id ) { debug ! ( " " ping from id ) ; continue ; } self . process_ping ( addr , ping ) ; } SwimKind :: Ack ( ack ) => { if self . server . is_member_blocked ( & ack . from . id ) && ack . forward_to . is_none ( ) { debug ! ( " " ack from id ) ; continue ; } self . process_ack ( addr , ack ) ; } SwimKind :: PingReq ( pingreq ) => { if self . server . is_member_blocked ( & pingreq . from . id ) { debug ! ( " " pingreq from id ) ; continue ; } self . process_pingreq ( addr , pingreq ) ; } } } Err ( e ) => { // TODO: We can't use magic numbers here because the Supervisor runs on more // than one platform. I'm sure these were added as specific OS errors for Linux // but we need to also handle Windows & Mac. match e . raw_os_error ( ) { Some ( 35 ) | Some ( 11 ) | Some ( 10035 ) | Some ( 10060 ) => { // This is the normal non-blocking result, or a timeout } Some ( _ ) => { error ! ( " " e ) ; debug ! ( " " e ) ; } None => { error ! ( " " e ) ; } } } } } }
Run the thread.
Run the thread .
&self
&self