pub async fn subscribe_metrics(mn: Arc<Self>, mut metrics_rx: watch::Receiver<RaftMetrics>) {
        // TODO: return a handle for join
        // TODO: every state change triggers add_non_voter!!!
        let mut running_rx = mn.running_rx.clone();
        let mut jh = mn.join_handles.lock().await;

        // TODO: reduce dependency: it does not need all of the fields in MetaNode
        let mn = mn.clone();

        let span = tracing::span!(tracing::Level::INFO, "watch-metrics");

        let h = tokio::task::spawn(
            {
                async move {
                    loop {
                        let changed = tokio::select! {
                            _ = running_rx.changed() => {
                               return Ok::<(), StoreError>(());
                            }
                            changed = metrics_rx.changed() => {
                                changed
                            }
                        };
                        if changed.is_ok() {
                            let mm = metrics_rx.borrow().clone();
                            if let Some(cur) = mm.current_leader {
                                if cur == mn.sto.id {
                                    // TODO: check result
                                    let _rst = mn.add_configured_non_voters().await;

                                    if _rst.is_err() {
                                        tracing::info!(
                                            "fail to add non-voter: my id={}, rst:{:?}",
                                            mn.sto.id,
                                            _rst
                                        );
                                    }
                                }
                            }
                        } else {
                            // shutting down
                            break;
                        }
                    }

                    Ok::<(), StoreError>(())
                }
            }
            .instrument(span),
        );
        jh.push(h);
    }
// spawn a monitor to watch raft state changes such as leader changes,
// and manually add non-voter to cluster so that non-voter receives raft logs.
spawn a monitor to watch raft state changes such as leader changes,
and manually add non-voter to cluster so that non-voter receives raft logs.
spawn a monitor to watch raft state changes such as leader changes and manually add non - voter to cluster so that non - voter receives raft logs .
pub async fn subscribe_metrics(mn: Arc<Self>, mut metrics_rx: watch::Receiver<RaftMetrics>) {
        // TODO: return a handle for join
        // TODO: every state change triggers add_non_voter!!!
        let mut running_rx = mn.running_rx.clone();
        let mut jh = mn.join_handles.lock().await;
        // TODO: reduce dependency: it does not need all of the fields in MetaNode
        let mn = mn.clone();
        let span = tracing::span!(tracing::Level::INFO, "watch-metrics");
        let h = tokio::task::spawn(
            {
                async move {
                    loop {
                        let changed = tokio::select! {
                            _ = running_rx.changed() => {
                               return Ok::<(), StoreError>(());
                            }
                            changed = metrics_rx.changed() => {
                                changed
                            }
                        };
                        if changed.is_ok() {
                            let mm = metrics_rx.borrow().clone();
                            if let Some(cur) = mm.current_leader {
                                if cur == mn.sto.id {
                                    // TODO: check result
                                    let _rst = mn.add_configured_non_voters().await;
                                    if _rst.is_err() {
                                        tracing::info!(
                                            "fail to add non-voter: my id={}, rst:{:?}",
                                            mn.sto.id,
                                            _rst
                                        );
                                    }
                                }
                            }
                        } else {
                            // shutting down
                            break;
                        }
                    }
                    Ok::<(), StoreError>(())
                }
            }
            .instrument(span),
        );
        jh.push(h);
    }
pub async fn subscribe_metrics ( mn : Arc < Self > , mut metrics_rx : watch :: Receiver < RaftMetrics > ) { // TODO: return a handle for join // TODO: every state change triggers add_non_voter!!! let mut running_rx = mn . running_rx . clone ( ) ; let mut jh = mn . join_handles . lock ( ) . await ; // TODO: reduce dependency: it does not need all of the fields in MetaNode let mn = mn . clone ( ) ; let span = tracing :: span ! ( tracing Level INFO " " ) ; let h = tokio :: task :: spawn ( { async move { loop { let changed = tokio :: select ! { running_rx changed ( ) { return Ok ( ) StoreError ( ( ) ) } changed metrics_rx changed ( ) { changed } } ; if changed . is_ok ( ) { let mm = metrics_rx . borrow ( ) . clone ( ) ; if let Some ( cur ) = mm . current_leader { if cur == mn . sto . id { // TODO: check result let _rst = mn . add_configured_non_voters ( ) . await ; if _rst . is_err ( ) { tracing :: info ! ( " " mn sto id _rst ) ; } } } } else { // shutting down break ; } } Ok :: < ( ) , StoreError > ( ( ) ) } } . instrument ( span ) , ) ; jh . push ( h ) ; }
spawn a monitor to watch raft state changes such as leader changes,
and manually add non-voter to cluster so that non-voter receives raft logs.
spawn a monitor to watch raft state changes such as leader changes and manually add non - voter to cluster so that non - voter receives raft logs .
mn
Arc<Self>
metrics_rx
watch::Receiver<RaftMetrics>
mn
Arc<Self>
metrics_rx
watch::Receiver<RaftMetrics>