fn process_changes(
        &self,
        _time_source: &dyn TimeSource,
        artifacts: Vec<UnvalidatedArtifact<CertificationMessage>>,
    ) -> (Vec<Advert<CertificationArtifact>>, ProcessingResult) {
        {
            let mut certification_pool = self.certification_pool.write().unwrap();
            for artifact in artifacts {
                certification_pool.insert(artifact.message)
            }
        }
        let mut adverts = Vec::new();
        let change_set = self.client.on_state_change(
            self.consensus_pool_cache.as_ref(),
            self.certification_pool.clone(),
        );
        let changed = if !change_set.is_empty() {
            ProcessingResult::StateChanged
        } else {
            ProcessingResult::StateUnchanged
        };

        for action in change_set.iter() {
            match action {
                certification::ChangeAction::AddToValidated(msg) => {
                    adverts.push(CertificationArtifact::message_to_advert(msg))
                }
                certification::ChangeAction::MoveToValidated(msg) => {
                    adverts.push(CertificationArtifact::message_to_advert(msg))
                }
                certification::ChangeAction::HandleInvalid(msg, reason) => {
                    self.invalidated_artifacts.inc();
                    warn!(
                        self.log,
                        "Invalid certification message ({:?}): {:?}", reason, msg
                    );
                }
                _ => {}
            }
        }
        self.certification_pool
            .write()
            .unwrap()
            .apply_changes(change_set);
        (adverts, changed)
    }
/// The method processes changes in the certification pool.
The method processes changes in the certification pool.
The method processes changes in the certification pool .
fn process_changes(
        &self,
        _time_source: &dyn TimeSource,
        artifacts: Vec<UnvalidatedArtifact<CertificationMessage>>,
    ) -> (Vec<Advert<CertificationArtifact>>, ProcessingResult) {
        {
            let mut certification_pool = self.certification_pool.write().unwrap();
            for artifact in artifacts {
                certification_pool.insert(artifact.message)
            }
        }
        let mut adverts = Vec::new();
        let change_set = self.client.on_state_change(
            self.consensus_pool_cache.as_ref(),
            self.certification_pool.clone(),
        );
        let changed = if !change_set.is_empty() {
            ProcessingResult::StateChanged
        } else {
            ProcessingResult::StateUnchanged
        };
        for action in change_set.iter() {
            match action {
                certification::ChangeAction::AddToValidated(msg) => {
                    adverts.push(CertificationArtifact::message_to_advert(msg))
                }
                certification::ChangeAction::MoveToValidated(msg) => {
                    adverts.push(CertificationArtifact::message_to_advert(msg))
                }
                certification::ChangeAction::HandleInvalid(msg, reason) => {
                    self.invalidated_artifacts.inc();
                    warn!(
                        self.log,
                        "Invalid certification message ({:?}): {:?}", reason, msg
                    );
                }
                _ => {}
            }
        }
        self.certification_pool
            .write()
            .unwrap()
            .apply_changes(change_set);
        (adverts, changed)
    }
fn process_changes ( & self , _time_source : & dyn TimeSource , artifacts : Vec < UnvalidatedArtifact < CertificationMessage > > , ) -> ( Vec < Advert < CertificationArtifact > > , ProcessingResult ) { { let mut certification_pool = self . certification_pool . write ( ) . unwrap ( ) ; for artifact in artifacts { certification_pool . insert ( artifact . message ) } } let mut adverts = Vec :: new ( ) ; let change_set = self . client . on_state_change ( self . consensus_pool_cache . as_ref ( ) , self . certification_pool . clone ( ) , ) ; let changed = if ! change_set . is_empty ( ) { ProcessingResult :: StateChanged } else { ProcessingResult :: StateUnchanged } ; for action in change_set . iter ( ) { match action { certification :: ChangeAction :: AddToValidated ( msg ) => { adverts . push ( CertificationArtifact :: message_to_advert ( msg ) ) } certification :: ChangeAction :: MoveToValidated ( msg ) => { adverts . push ( CertificationArtifact :: message_to_advert ( msg ) ) } certification :: ChangeAction :: HandleInvalid ( msg , reason ) => { self . invalidated_artifacts . inc ( ) ; warn ! ( self log " " reason msg ) ; } _ => { } } } self . certification_pool . write ( ) . unwrap ( ) . apply_changes ( change_set ) ; ( adverts , changed ) }
The method processes changes in the certification pool.
The method processes changes in the certification pool .
&self
_time_source
&dyn TimeSource
artifacts
Vec<UnvalidatedArtifact<CertificationMessage>>
&self
_time_source
&dyn TimeSource
artifacts
Vec<UnvalidatedArtifact<CertificationMessage>>