pub(self) fn drain_raft_rx(&mut self, first: RaftEvent<D>, span: Span) {
        let mut event_opt = Some((first, span));
        let mut iters = 0;
        loop {
            // Just ensure we don't get stuck draining a REALLY hot replication feed.
            if iters > self.config.max_payload_entries {
                return;
            }

            // Unpack the event opt, else return if we don't have one to process.
            let (event, span) = match event_opt.take() {
                Some(event) => event,
                None => return,
            };

            let _ent = span.enter();

            // Process the event.
            match event {
                RaftEvent::UpdateCommitIndex { commit_index } => {
                    self.commit_index = commit_index;
                }

                RaftEvent::Replicate { entry, commit_index } => {
                    self.commit_index = commit_index;
                    self.last_log_index = entry.log_id.index;
                    if self.target_state == TargetReplState::LineRate {
                        self.replication_buffer.push(entry);
                    }
                }

                RaftEvent::Terminate => {
                    self.target_state = TargetReplState::Shutdown;
                    return;
                }
            }

            // Attempt to unpack the next event for the next loop iteration.
            if let Some(event_span) = self.repl_rx.recv().now_or_never() {
                event_opt = event_span;
            }
            iters += 1;
        }
    }
/// Fully drain the channel coming in from the Raft node.
Fully drain the channel coming in from the Raft node.
Fully drain the channel coming in from the Raft node .
pub(self) fn drain_raft_rx(&mut self, first: RaftEvent<D>, span: Span) {
        let mut event_opt = Some((first, span));
        let mut iters = 0;
        loop {
            // Just ensure we don't get stuck draining a REALLY hot replication feed.
            if iters > self.config.max_payload_entries {
                return;
            }
            // Unpack the event opt, else return if we don't have one to process.
            let (event, span) = match event_opt.take() {
                Some(event) => event,
                None => return,
            };
            let _ent = span.enter();
            // Process the event.
            match event {
                RaftEvent::UpdateCommitIndex { commit_index } => {
                    self.commit_index = commit_index;
                }
                RaftEvent::Replicate { entry, commit_index } => {
                    self.commit_index = commit_index;
                    self.last_log_index = entry.log_id.index;
                    if self.target_state == TargetReplState::LineRate {
                        self.replication_buffer.push(entry);
                    }
                }
                RaftEvent::Terminate => {
                    self.target_state = TargetReplState::Shutdown;
                    return;
                }
            }
            // Attempt to unpack the next event for the next loop iteration.
            if let Some(event_span) = self.repl_rx.recv().now_or_never() {
                event_opt = event_span;
            }
            iters += 1;
        }
    }
pub ( self ) fn drain_raft_rx ( & mut self , first : RaftEvent < D > , span : Span ) { let mut event_opt = Some ( ( first , span ) ) ; let mut iters = 0 ; loop { // Just ensure we don't get stuck draining a REALLY hot replication feed. if iters > self . config . max_payload_entries { return ; } // Unpack the event opt, else return if we don't have one to process. let ( event , span ) = match event_opt . take ( ) { Some ( event ) => event , None => return , } ; let _ent = span . enter ( ) ; // Process the event. match event { RaftEvent :: UpdateCommitIndex { commit_index } => { self . commit_index = commit_index ; } RaftEvent :: Replicate { entry , commit_index } => { self . commit_index = commit_index ; self . last_log_index = entry . log_id . index ; if self . target_state == TargetReplState :: LineRate { self . replication_buffer . push ( entry ) ; } } RaftEvent :: Terminate => { self . target_state = TargetReplState :: Shutdown ; return ; } } // Attempt to unpack the next event for the next loop iteration. if let Some ( event_span ) = self . repl_rx . recv ( ) . now_or_never ( ) { event_opt = event_span ; } iters += 1 ; } }
Fully drain the channel coming in from the Raft node.
Fully drain the channel coming in from the Raft node .
&mut self
first
RaftEvent<D>
span
Span
&mut self
first
RaftEvent<D>
span
Span