fn process_changes(
        &self,
        time_source: &dyn TimeSource,
        artifacts: Vec<UnvalidatedArtifact<ConsensusMessage>>,
    ) -> (Vec<Advert<ConsensusArtifact>>, ProcessingResult) {
        {
            let mut consensus_pool = self.consensus_pool.write().unwrap();
            for artifact in artifacts {
                debug!(
                    tag => "consensus_trace",
                    self.log,
                    "process_change::artifact {}",
                    serde_json::to_string(&artifact).unwrap()
                );
                consensus_pool.insert(artifact)
            }
        }
        let mut adverts = Vec::new();
        let change_set = {
            let consensus_pool = self.consensus_pool.read().unwrap();
            let ingress_pool = Arc::clone(&self.ingress_pool) as Arc<_>;
            let ingress_pool = IngressPoolSelectWrapper::new(&ingress_pool);
            self.client.on_state_change(&*consensus_pool, &ingress_pool)
        };
        let changed = if !change_set.is_empty() {
            ProcessingResult::StateChanged
        } else {
            ProcessingResult::StateUnchanged
        };

        for change_action in change_set.iter() {
            debug!(
                tag => "consensus_trace",
                self.log,
                "process_change::change_action {}",
                serde_json::to_string(&change_action).unwrap()
            );
            match change_action {
                ConsensusAction::AddToValidated(to_add) => {
                    adverts.push(ConsensusArtifact::message_to_advert(to_add))
                }
                ConsensusAction::MoveToValidated(to_move) => {
                    adverts.push(ConsensusArtifact::message_to_advert(to_move))
                }
                ConsensusAction::RemoveFromValidated(_) => {}
                ConsensusAction::RemoveFromUnvalidated(_) => {}
                ConsensusAction::PurgeValidatedBelow(_) => {}
                ConsensusAction::PurgeUnvalidatedBelow(_) => {}
                ConsensusAction::HandleInvalid(artifact, s) => {
                    self.invalidated_artifacts.inc();
                    warn!(self.log, "Invalid artifact {} {:?}", s, artifact);
                }
            }
        }
        debug!(
            tag => "consensus_trace",
            self.log,
            "process_change::apply_changes {}",
            serde_json::to_string(&time_source.get_relative_time()).unwrap()
        );

        self.consensus_pool
            .write()
            .unwrap()
            .apply_changes(time_source, change_set);

        (adverts, changed)
    }
/// The method processes changes in the *Consensus* pool and ingress pool.
The method processes changes in the *Consensus* pool and ingress pool.
The method processes changes in the * Consensus * pool and ingress pool .
fn process_changes(
        &self,
        time_source: &dyn TimeSource,
        artifacts: Vec<UnvalidatedArtifact<ConsensusMessage>>,
    ) -> (Vec<Advert<ConsensusArtifact>>, ProcessingResult) {
        {
            let mut consensus_pool = self.consensus_pool.write().unwrap();
            for artifact in artifacts {
                debug!(
                    tag => "consensus_trace",
                    self.log,
                    "process_change::artifact {}",
                    serde_json::to_string(&artifact).unwrap()
                );
                consensus_pool.insert(artifact)
            }
        }
        let mut adverts = Vec::new();
        let change_set = {
            let consensus_pool = self.consensus_pool.read().unwrap();
            let ingress_pool = Arc::clone(&self.ingress_pool) as Arc<_>;
            let ingress_pool = IngressPoolSelectWrapper::new(&ingress_pool);
            self.client.on_state_change(&*consensus_pool, &ingress_pool)
        };
        let changed = if !change_set.is_empty() {
            ProcessingResult::StateChanged
        } else {
            ProcessingResult::StateUnchanged
        };
        for change_action in change_set.iter() {
            debug!(
                tag => "consensus_trace",
                self.log,
                "process_change::change_action {}",
                serde_json::to_string(&change_action).unwrap()
            );
            match change_action {
                ConsensusAction::AddToValidated(to_add) => {
                    adverts.push(ConsensusArtifact::message_to_advert(to_add))
                }
                ConsensusAction::MoveToValidated(to_move) => {
                    adverts.push(ConsensusArtifact::message_to_advert(to_move))
                }
                ConsensusAction::RemoveFromValidated(_) => {}
                ConsensusAction::RemoveFromUnvalidated(_) => {}
                ConsensusAction::PurgeValidatedBelow(_) => {}
                ConsensusAction::PurgeUnvalidatedBelow(_) => {}
                ConsensusAction::HandleInvalid(artifact, s) => {
                    self.invalidated_artifacts.inc();
                    warn!(self.log, "Invalid artifact {} {:?}", s, artifact);
                }
            }
        }
        debug!(
            tag => "consensus_trace",
            self.log,
            "process_change::apply_changes {}",
            serde_json::to_string(&time_source.get_relative_time()).unwrap()
        );
        self.consensus_pool
            .write()
            .unwrap()
            .apply_changes(time_source, change_set);
        (adverts, changed)
    }
fn process_changes ( & self , time_source : & dyn TimeSource , artifacts : Vec < UnvalidatedArtifact < ConsensusMessage > > , ) -> ( Vec < Advert < ConsensusArtifact > > , ProcessingResult ) { { let mut consensus_pool = self . consensus_pool . write ( ) . unwrap ( ) ; for artifact in artifacts { debug ! ( tag " " self log " " serde_json to_string ( artifact ) unwrap ( ) ) ; consensus_pool . insert ( artifact ) } } let mut adverts = Vec :: new ( ) ; let change_set = { let consensus_pool = self . consensus_pool . read ( ) . unwrap ( ) ; let ingress_pool = Arc :: clone ( & self . ingress_pool ) as Arc < _ > ; let ingress_pool = IngressPoolSelectWrapper :: new ( & ingress_pool ) ; self . client . on_state_change ( & * consensus_pool , & ingress_pool ) } ; let changed = if ! change_set . is_empty ( ) { ProcessingResult :: StateChanged } else { ProcessingResult :: StateUnchanged } ; for change_action in change_set . iter ( ) { debug ! ( tag " " self log " " serde_json to_string ( change_action ) unwrap ( ) ) ; match change_action { ConsensusAction :: AddToValidated ( to_add ) => { adverts . push ( ConsensusArtifact :: message_to_advert ( to_add ) ) } ConsensusAction :: MoveToValidated ( to_move ) => { adverts . push ( ConsensusArtifact :: message_to_advert ( to_move ) ) } ConsensusAction :: RemoveFromValidated ( _ ) => { } ConsensusAction :: RemoveFromUnvalidated ( _ ) => { } ConsensusAction :: PurgeValidatedBelow ( _ ) => { } ConsensusAction :: PurgeUnvalidatedBelow ( _ ) => { } ConsensusAction :: HandleInvalid ( artifact , s ) => { self . invalidated_artifacts . inc ( ) ; warn ! ( self log " " s artifact ) ; } } } debug ! ( tag " " self log " " serde_json to_string ( time_source get_relative_time ( ) ) unwrap ( ) ) ; self . consensus_pool . write ( ) . unwrap ( ) . apply_changes ( time_source , change_set ) ; ( adverts , changed ) }
The method processes changes in the *Consensus* pool and ingress pool.
The method processes changes in the * Consensus * pool and ingress pool .
&self
time_source
&dyn TimeSource
artifacts
Vec<UnvalidatedArtifact<ConsensusMessage>>
&self
time_source
&dyn TimeSource
artifacts
Vec<UnvalidatedArtifact<ConsensusMessage>>