async fn start_heartbeat(&mut self) -> Result<(), ServiceError> {
        assert!(self.hb_channel.is_none(), "Heartbeat loop already started");

        let (sender, mut receiver) = channel::<HeartbeatLoopCommand>(1);

        self.hb_channel = Some(sender);

        let ttl: u64 = self.ttl as u64;
        let service = self.service.clone();
        let instance = self.instance.clone();
        let descr = Arc::clone(&self.descriptor);

        tracing::info!(
            "Service {}:{} starting heartbeat loop with interval {} secs.",
            self.service,
            self.instance,
            ttl
        );

        let mut heartbeat_lost_time = None;

        let h = tokio::spawn(async move {
            loop {
                match descr.lock().await.as_mut() {
                    Some(d) => {
                        // We care only about TTL loss reported explicitly, as it's
                        // the only reliable trigger for heartbeat-loss related actions.
                        // Direct I/O errors upon heartbeat update don't count as a valid
                        // reason to judge about heartbeat loss unless all attempts to report
                        // heartbeat failed within TTL time (measured since the moment of
                        // the first heartbeat update failure).
                        if let Err(e) = d.send_heartbeat().await {
                            match e {
                                ServiceError::HeartbeatLost { .. } => {
                                    tracing::error!(
                                        "Service {}:{} lost heartbeat",
                                        service,
                                        instance,
                                    );
                                    return Err(e);
                                }
                                _ => {
                                    if heartbeat_lost_time.is_none() {
                                        heartbeat_lost_time = Some(std::time::Instant::now());

                                        tracing::error!(
                                            "Service {}:{} failed to send heartbeat, error={}",
                                            service,
                                            instance,
                                            e,
                                        );
                                    } else {
                                        // Check whether TTL is elapsed and report error in case
                                        // all heartbeat update attemps failed within TTL interval.
                                        let elapsed =
                                            heartbeat_lost_time.as_ref().unwrap().elapsed();
                                        if elapsed.as_secs() > ttl {
                                            tracing::error!(
                                                "Service {}:{} lost heartbeat due to exceeding heartbeat TTL ({} sec)",
                                                service,
                                                instance,
                                                ttl,
                                            );
                                            return Err(ServiceError::HeartbeatLost {
                                                service,
                                                instance,
                                            });
                                        }
                                    }
                                }
                            }
                        } else {
                            // Heartbeat succeeded, reset the failure time.
                            if heartbeat_lost_time.is_some() {
                                heartbeat_lost_time.take();
                                tracing::info!(
                                    "Service {}:{} successfully recovered heartbeat within TTL interval",
                                    service,
                                    instance,
                                );
                            }
                        }
                    }
                    // Service is being unregistered, break the heartbeat loop.
                    None => {
                        tracing::info!(
                            "Service {}:{} is being unregistered, stopping heartbeat loop",
                            service,
                            instance,
                        );
                        receiver.close();
                        break;
                    }
                }

                // Waiti till either TTL update timeout triggers or request to stop
                // heartbeat loop is received.
                tokio::select! {
                    _ = tokio::time::sleep(std::time::Duration::from_secs(ttl as u64 / 2)) => {},

                    cmd = receiver.recv() => {
                        match cmd.unwrap() {
                            HeartbeatLoopCommand::StopHeartbeat(tx) => {
                                tracing::info!(
                                    "Service {}:{} stopped heartbeat loop",
                                    service, instance,
                                );
                                tx.send(HeartbeatLoopCommand::Ok).expect("Command sender disappeared");
                                break;
                            },
                            c => panic!("Unexpected command for heartbeat loop control: {:?}", c),
                        }
                    }
                }
            }
            Ok(())
        });

        // Wait till heartbeat loop is explicitly stopped or error occurs during heartbeat update.
        h.await.unwrap()
    }
/// Start heartbeat loop for the service.
/// In case this function returns error, it's always an indication of lost heartbet
/// and the caller must take an appropriate action in response to heartbeat loss.
Start heartbeat loop for the service.
In case this function returns error, it's always an indication of lost heartbet
and the caller must take an appropriate action in response to heartbeat loss.
Start heartbeat loop for the service . In case this function returns error it ' s always an indication of lost heartbet and the caller must take an appropriate action in response to heartbeat loss .
async fn start_heartbeat(&mut self) -> Result<(), ServiceError> {
        assert!(self.hb_channel.is_none(), "Heartbeat loop already started");
        let (sender, mut receiver) = channel::<HeartbeatLoopCommand>(1);
        self.hb_channel = Some(sender);
        let ttl: u64 = self.ttl as u64;
        let service = self.service.clone();
        let instance = self.instance.clone();
        let descr = Arc::clone(&self.descriptor);
        tracing::info!(
            "Service {}:{} starting heartbeat loop with interval {} secs.",
            self.service,
            self.instance,
            ttl
        );
        let mut heartbeat_lost_time = None;
        let h = tokio::spawn(async move {
            loop {
                match descr.lock().await.as_mut() {
                    Some(d) => {
                        // We care only about TTL loss reported explicitly, as it's
                        // the only reliable trigger for heartbeat-loss related actions.
                        // Direct I/O errors upon heartbeat update don't count as a valid
                        // reason to judge about heartbeat loss unless all attempts to report
                        // heartbeat failed within TTL time (measured since the moment of
                        // the first heartbeat update failure).
                        if let Err(e) = d.send_heartbeat().await {
                            match e {
                                ServiceError::HeartbeatLost { .. } => {
                                    tracing::error!(
                                        "Service {}:{} lost heartbeat",
                                        service,
                                        instance,
                                    );
                                    return Err(e);
                                }
                                _ => {
                                    if heartbeat_lost_time.is_none() {
                                        heartbeat_lost_time = Some(std::time::Instant::now());
                                        tracing::error!(
                                            "Service {}:{} failed to send heartbeat, error={}",
                                            service,
                                            instance,
                                            e,
                                        );
                                    } else {
                                        // Check whether TTL is elapsed and report error in case
                                        // all heartbeat update attemps failed within TTL interval.
                                        let elapsed =
                                            heartbeat_lost_time.as_ref().unwrap().elapsed();
                                        if elapsed.as_secs() > ttl {
                                            tracing::error!(
                                                "Service {}:{} lost heartbeat due to exceeding heartbeat TTL ({} sec)",
                                                service,
                                                instance,
                                                ttl,
                                            );
                                            return Err(ServiceError::HeartbeatLost {
                                                service,
                                                instance,
                                            });
                                        }
                                    }
                                }
                            }
                        } else {
                            // Heartbeat succeeded, reset the failure time.
                            if heartbeat_lost_time.is_some() {
                                heartbeat_lost_time.take();
                                tracing::info!(
                                    "Service {}:{} successfully recovered heartbeat within TTL interval",
                                    service,
                                    instance,
                                );
                            }
                        }
                    }
                    // Service is being unregistered, break the heartbeat loop.
                    None => {
                        tracing::info!(
                            "Service {}:{} is being unregistered, stopping heartbeat loop",
                            service,
                            instance,
                        );
                        receiver.close();
                        break;
                    }
                }
                // Waiti till either TTL update timeout triggers or request to stop
                // heartbeat loop is received.
                tokio::select! {
                    _ = tokio::time::sleep(std::time::Duration::from_secs(ttl as u64 / 2)) => {},
                    cmd = receiver.recv() => {
                        match cmd.unwrap() {
                            HeartbeatLoopCommand::StopHeartbeat(tx) => {
                                tracing::info!(
                                    "Service {}:{} stopped heartbeat loop",
                                    service, instance,
                                );
                                tx.send(HeartbeatLoopCommand::Ok).expect("Command sender disappeared");
                                break;
                            },
                            c => panic!("Unexpected command for heartbeat loop control: {:?}", c),
                        }
                    }
                }
            }
            Ok(())
        });
        // Wait till heartbeat loop is explicitly stopped or error occurs during heartbeat update.
        h.await.unwrap()
    }
async fn start_heartbeat ( & mut self ) -> Result < ( ) , ServiceError > { assert ! ( self hb_channel is_none ( ) " " ) ; let ( sender , mut receiver ) = channel :: < HeartbeatLoopCommand > ( 1 ) ; self . hb_channel = Some ( sender ) ; let ttl : u64 = self . ttl as u64 ; let service = self . service . clone ( ) ; let instance = self . instance . clone ( ) ; let descr = Arc :: clone ( & self . descriptor ) ; tracing :: info ! ( " " self service self instance ttl ) ; let mut heartbeat_lost_time = None ; let h = tokio :: spawn ( async move { loop { match descr . lock ( ) . await . as_mut ( ) { Some ( d ) => { // We care only about TTL loss reported explicitly, as it's // the only reliable trigger for heartbeat-loss related actions. // Direct I/O errors upon heartbeat update don't count as a valid // reason to judge about heartbeat loss unless all attempts to report // heartbeat failed within TTL time (measured since the moment of // the first heartbeat update failure). if let Err ( e ) = d . send_heartbeat ( ) . await { match e { ServiceError :: HeartbeatLost { .. } => { tracing :: error ! ( " " service instance ) ; return Err ( e ) ; } _ => { if heartbeat_lost_time . is_none ( ) { heartbeat_lost_time = Some ( std :: time :: Instant :: now ( ) ) ; tracing :: error ! ( " " service instance e ) ; } else { // Check whether TTL is elapsed and report error in case // all heartbeat update attemps failed within TTL interval. let elapsed = heartbeat_lost_time . as_ref ( ) . unwrap ( ) . elapsed ( ) ; if elapsed . as_secs ( ) > ttl { tracing :: error ! ( " " service instance ttl ) ; return Err ( ServiceError :: HeartbeatLost { service , instance , } ) ; } } } } } else { // Heartbeat succeeded, reset the failure time. if heartbeat_lost_time . is_some ( ) { heartbeat_lost_time . take ( ) ; tracing :: info ! ( " " service instance ) ; } } } // Service is being unregistered, break the heartbeat loop. None => { tracing :: info ! ( " " service instance ) ; receiver . close ( ) ; break ; } } // Waiti till either TTL update timeout triggers or request to stop // heartbeat loop is received. tokio :: select ! { tokio time sleep ( std time Duration from_secs ( ttl as u64 2 ) ) { } cmd receiver recv ( ) { match cmd unwrap ( ) { HeartbeatLoopCommand StopHeartbeat ( tx ) { tracing info ( " " service instance ) tx send ( HeartbeatLoopCommand Ok ) expect ( " " ) break } c panic ( " " c ) } } } } Ok ( ( ) ) } ) ; // Wait till heartbeat loop is explicitly stopped or error occurs during heartbeat update. h . await . unwrap ( ) }
Start heartbeat loop for the service.
Start heartbeat loop for the service .
&mut self
&mut self