pub async fn sync_from_spus(&self, spus: &SpuLocalStore, local_spu: SpuId) {
        let mut writer = self.write().await;
        // remove non existent spu
        let keys: Vec<SpuId> = writer.keys().copied().collect();
        for spu in keys {
            if !spus.contains_key(&spu) {
                debug!(spu, "spu no longer valid,removing");
                writer.remove(&spu);
            }
        }
        // add new spu that doesn't exists
        for spu in spus.all_keys() {
            if spu != local_spu && !writer.contains_key(&spu) {
                debug!(spu, "spu pending doesn't exists, creating");
                let pending = FollowerSpuPendingUpdates {
                    event: Arc::new(OffsetPublisher::new(0)),
                    replicas: Arc::new(RwLock::new(HashSet::new())),
                };
                writer.insert(spu, Arc::new(pending));
            }
        }
    }
/// update our self from current spu
update our self from current spu
update our self from current spu
pub async fn sync_from_spus(&self, spus: &SpuLocalStore, local_spu: SpuId) {
        let mut writer = self.write().await;
        // remove non existent spu
        let keys: Vec<SpuId> = writer.keys().copied().collect();
        for spu in keys {
            if !spus.contains_key(&spu) {
                debug!(spu, "spu no longer valid,removing");
                writer.remove(&spu);
            }
        }
        // add new spu that doesn't exists
        for spu in spus.all_keys() {
            if spu != local_spu && !writer.contains_key(&spu) {
                debug!(spu, "spu pending doesn't exists, creating");
                let pending = FollowerSpuPendingUpdates {
                    event: Arc::new(OffsetPublisher::new(0)),
                    replicas: Arc::new(RwLock::new(HashSet::new())),
                };
                writer.insert(spu, Arc::new(pending));
            }
        }
    }
pub async fn sync_from_spus ( & self , spus : & SpuLocalStore , local_spu : SpuId ) { let mut writer = self . write ( ) . await ; // remove non existent spu let keys : Vec < SpuId > = writer . keys ( ) . copied ( ) . collect ( ) ; for spu in keys { if ! spus . contains_key ( & spu ) { debug ! ( spu " " ) ; writer . remove ( & spu ) ; } } // add new spu that doesn't exists for spu in spus . all_keys ( ) { if spu != local_spu && ! writer . contains_key ( & spu ) { debug ! ( spu " " ) ; let pending = FollowerSpuPendingUpdates { event : Arc :: new ( OffsetPublisher :: new ( 0 ) ) , replicas : Arc :: new ( RwLock :: new ( HashSet :: new ( ) ) ) , } ; writer . insert ( spu , Arc :: new ( pending ) ) ; } } }
update our self from current spu
update our self from current spu
&self
spus
&SpuLocalStore
local_spu
SpuId
&self
spus
&SpuLocalStore
local_spu
SpuId