/// svc has been changed, update spu
svc has been changed, update spu
svc has been changed update spu
async fn sync_spgs_to_statefulset(
        &mut self,
        listener: &mut K8ChangeListener<SpuGroupSpec>,
    ) -> Result<(), ClientError> {
        if !listener.has_change() {
            debug!("no spg change, skipping");
            return Ok(());
        }
        debug!("start syncing spg");
        let changes = listener.sync_changes().await;
        let epoch = changes.epoch;
        let (updates, deletes) = changes.parts();
        debug!(
            "received spg changes updates: {},deletes: {},epoch: {}",
            updates.len(),
            deletes.len(),
            epoch,
        );
        if let Some(config_obj) = self.configs.store().value("fluvio").await {
            let config = config_obj.inner_owned().spec;
            for group_item in updates.into_iter() {
                let spu_group = SpuGroupObj::new(group_item);
                self.sync_spg_to_statefulset(spu_group, &config).await?
            }
        } else {
            error!("fluvio config map not found, skipping");
        }
        Ok(())
    }
async fn sync_spgs_to_statefulset ( & mut self , listener : & mut K8ChangeListener < SpuGroupSpec > , ) -> Result < ( ) , ClientError > { if ! listener . has_change ( ) { debug ! ( " " ) ; return Ok ( ( ) ) ; } debug ! ( " " ) ; let changes = listener . sync_changes ( ) . await ; let epoch = changes . epoch ; let ( updates , deletes ) = changes . parts ( ) ; debug ! ( " " updates len ( ) deletes len ( ) epoch ) ; if let Some ( config_obj ) = self . configs . store ( ) . value ( " " ) . await { let config = config_obj . inner_owned ( ) . spec ; for group_item in updates . into_iter ( ) { let spu_group = SpuGroupObj :: new ( group_item ) ; self . sync_spg_to_statefulset ( spu_group , & config ) . await ? } } else { error ! ( " " ) ; } Ok ( ( ) ) }
svc has been changed, update spu
svc has been changed update spu
&mut self
listener
&mut K8ChangeListener<SpuGroupSpec>