/// The method processes changes in the ingress pool.
The method processes changes in the ingress pool.
The method processes changes in the ingress pool .
fn process_changes(
        &self,
        _time_source: &dyn TimeSource,
        artifacts: Vec<UnvalidatedArtifact<SignedIngress>>,
    ) -> (Vec<AdvertSendRequest<IngressArtifact>>, ProcessingResult) {
        {
            let mut ingress_pool = self.ingress_pool.write().unwrap();
            for artifact in artifacts {
                ingress_pool.insert(artifact)
            }
        }
        let change_set = {
            let pool = self.ingress_pool.read().unwrap();
            self.client.on_state_change(&*pool)
        };
        let mut adverts = Vec::new();
        for change_action in change_set.iter() {
            match change_action {
                IngressAction::MoveToValidated((
                    message_id,
                    source_node_id,
                    size,
                    attribute,
                    integrity_hash,
                )) => {
                    let advert_source = if *source_node_id == self.node_id {
                        AdvertSource::Produced
                    } else {
                        AdvertSource::Relayed
                    };
                    adverts.push(AdvertSendRequest {
                        advert: Advert {
                            size: *size,
                            id: message_id.clone(),
                            attribute: attribute.clone(),
                            integrity_hash: integrity_hash.clone(),
                        },
                        advert_class: self.advert_class(advert_source),
                    });
                }
                IngressAction::RemoveFromUnvalidated(_)
                | IngressAction::RemoveFromValidated(_)
                | IngressAction::PurgeBelowExpiry(_) => {}
            }
        }
        self.ingress_pool
            .write()
            .unwrap()
            .apply_changeset(change_set);
        (adverts, ProcessingResult::StateUnchanged)
    }
fn process_changes ( & self , _time_source : & dyn TimeSource , artifacts : Vec < UnvalidatedArtifact < SignedIngress > > , ) -> ( Vec < AdvertSendRequest < IngressArtifact > > , ProcessingResult ) { { let mut ingress_pool = self . ingress_pool . write ( ) . unwrap ( ) ; for artifact in artifacts { ingress_pool . insert ( artifact ) } } let change_set = { let pool = self . ingress_pool . read ( ) . unwrap ( ) ; self . client . on_state_change ( & * pool ) } ; let mut adverts = Vec :: new ( ) ; for change_action in change_set . iter ( ) { match change_action { IngressAction :: MoveToValidated ( ( message_id , source_node_id , size , attribute , integrity_hash , ) ) => { let advert_source = if * source_node_id == self . node_id { AdvertSource :: Produced } else { AdvertSource :: Relayed } ; adverts . push ( AdvertSendRequest { advert : Advert { size : * size , id : message_id . clone ( ) , attribute : attribute . clone ( ) , integrity_hash : integrity_hash . clone ( ) , } , advert_class : self . advert_class ( advert_source ) , } ) ; } IngressAction :: RemoveFromUnvalidated ( _ ) | IngressAction :: RemoveFromValidated ( _ ) | IngressAction :: PurgeBelowExpiry ( _ ) => { } } } self . ingress_pool . write ( ) . unwrap ( ) . apply_changeset ( change_set ) ; ( adverts , ProcessingResult :: StateUnchanged ) }
The method processes changes in the ingress pool.
The method processes changes in the ingress pool .
&self
_time_source
&dyn TimeSource
artifacts
Vec<UnvalidatedArtifact<SignedIngress>>