pub fn k8_event_stream_to_metadata_actions<S,E>(
    stream: TokenStreamResult<S::K8Spec,<S::K8Spec as K8Spec>::Status,E>,
    local_store: &LocalStore<S>
) -> Actions<LSChange<S>> 
    where 
        S: Spec + Debug + PartialEq + Debug,
        <S as Spec>::K8Spec: Debug,
        S::Key: Debug + Display + Clone,
        S::Status: Debug + PartialEq,
        E: MetadataClientError
{

    let (mut add_cnt, mut mod_cnt, mut del_cnt, mut skip_cnt) = (0, 0, 0, 0);
    let mut actions: Actions<LSChange<S>> = Actions::default();

    // loop through items and generate add/mod actions
    for token in stream.unwrap() {
        match token {
            Ok(watch_obj) => match watch_obj {
                K8Watch::ADDED(k8_obj) => {
                    let converted: Result<KVObject<S>,ScServerError> = k8_obj_to_kv_obj(k8_obj);  // help out compiler
                    match converted {
                        Ok(new_kv_value) =>  {
                            trace!("KV ({}): push ADD action", new_kv_value.key());
                            if let Some(old_value) = local_store.insert(new_kv_value.clone()) {
                                // some old value, check if same as new one, if they are same, no need for action
                                 warn!("detected exist value: {:#?} which sould not exists",old_value);
                                if old_value == new_kv_value {
                                    trace!("same value as old value, ignoring");
                                } else {
                                    trace!("generating update action: {:#?}",new_kv_value.key());
                                    actions.push(LSChange::update(new_kv_value,old_value));
                                    mod_cnt += 1;
                                }
                            } else {
                                // no existing value, which should be expected
                                debug!("adding {}:{} to local store",S::LABEL,new_kv_value.key());
                                actions.push(LSChange::add(new_kv_value));
                                
                                add_cnt += 1;
                            }
                        },
                        Err(err) => {
                            error!("{}", err);
                            skip_cnt += 1;
                        }
                    }
                },
                K8Watch::MODIFIED(k8_obj) => {
                    let converted: Result<KVObject<S>,ScServerError> = k8_obj_to_kv_obj(k8_obj);  // help out compiler
                    match converted {
                        Ok(new_kv_value) =>  {

                            if let Some(old_value) = local_store.insert(new_kv_value.clone()) {

                                if old_value == new_kv_value {
                                    // this is unexpected, 
                                    warn!("old and new value is same: {:#?}, ignoring",new_kv_value);
                                } else {
                                    // normal
                                    actions.push(LSChange::update(new_kv_value, old_value));
                                    mod_cnt += 1;
                                }
                            } else {
                                // doesn't exist, this is then new
                                warn!("KV ({}) - not found, generating add", new_kv_value.key());
                                actions.push(LSChange::add(new_kv_value));
                            }
                        },
                        Err(err) => {
                            error!("{}", err);
                            skip_cnt += 1;
                        }
                    }
                    
                }
                K8Watch::DELETED(k8_obj) => {
                    match k8_obj_to_kv_obj(k8_obj) {
                        Ok(kv_value) =>  {
                            trace!("KV ({}): push DEL action", kv_value.key());

                            // try to delete it
                            if let Some(_old_value) = local_store.remove(kv_value.key()) {
                                del_cnt += 1;
                                actions.push(LSChange::delete(kv_value));
                            } else {
                                skip_cnt += 1;
                                warn!("delete  should never fail since key exists: {}",kv_value.key());
                            }
                        },
                        Err(err) => {
                            error!("{}", err);
                            skip_cnt += 1;
                        }
                    }
                }
            },
            Err(err) => error!("Problem parsing {} event: {} ... (exiting)", S::LABEL,err),
        }
        
    }

    // log counters
    let all = add_cnt + mod_cnt + del_cnt + skip_cnt;
    trace!("K8 Streams {} [all:{}, add:{},mod:{},del:{},ski: {}", 
        S::LABEL,
        all, 
        add_cnt, 
        mod_cnt, 
        del_cnt, 
        skip_cnt);

    actions
}
///
/// Translates K8 events into metadata action.
///
Translates K8 events into metadata action.
Translates K8 events into metadata action .
pub fn k8_event_stream_to_metadata_actions<S,E>(
    stream: TokenStreamResult<S::K8Spec,<S::K8Spec as K8Spec>::Status,E>,
    local_store: &LocalStore<S>
) -> Actions<LSChange<S>> 
    where 
        S: Spec + Debug + PartialEq + Debug,
        <S as Spec>::K8Spec: Debug,
        S::Key: Debug + Display + Clone,
        S::Status: Debug + PartialEq,
        E: MetadataClientError
{
    let (mut add_cnt, mut mod_cnt, mut del_cnt, mut skip_cnt) = (0, 0, 0, 0);
    let mut actions: Actions<LSChange<S>> = Actions::default();
    // loop through items and generate add/mod actions
    for token in stream.unwrap() {
        match token {
            Ok(watch_obj) => match watch_obj {
                K8Watch::ADDED(k8_obj) => {
                    let converted: Result<KVObject<S>,ScServerError> = k8_obj_to_kv_obj(k8_obj);  // help out compiler
                    match converted {
                        Ok(new_kv_value) =>  {
                            trace!("KV ({}): push ADD action", new_kv_value.key());
                            if let Some(old_value) = local_store.insert(new_kv_value.clone()) {
                                // some old value, check if same as new one, if they are same, no need for action
                                 warn!("detected exist value: {:#?} which sould not exists",old_value);
                                if old_value == new_kv_value {
                                    trace!("same value as old value, ignoring");
                                } else {
                                    trace!("generating update action: {:#?}",new_kv_value.key());
                                    actions.push(LSChange::update(new_kv_value,old_value));
                                    mod_cnt += 1;
                                }
                            } else {
                                // no existing value, which should be expected
                                debug!("adding {}:{} to local store",S::LABEL,new_kv_value.key());
                                actions.push(LSChange::add(new_kv_value));
                                add_cnt += 1;
                            }
                        },
                        Err(err) => {
                            error!("{}", err);
                            skip_cnt += 1;
                        }
                    }
                },
                K8Watch::MODIFIED(k8_obj) => {
                    let converted: Result<KVObject<S>,ScServerError> = k8_obj_to_kv_obj(k8_obj);  // help out compiler
                    match converted {
                        Ok(new_kv_value) =>  {
                            if let Some(old_value) = local_store.insert(new_kv_value.clone()) {
                                if old_value == new_kv_value {
                                    // this is unexpected, 
                                    warn!("old and new value is same: {:#?}, ignoring",new_kv_value);
                                } else {
                                    // normal
                                    actions.push(LSChange::update(new_kv_value, old_value));
                                    mod_cnt += 1;
                                }
                            } else {
                                // doesn't exist, this is then new
                                warn!("KV ({}) - not found, generating add", new_kv_value.key());
                                actions.push(LSChange::add(new_kv_value));
                            }
                        },
                        Err(err) => {
                            error!("{}", err);
                            skip_cnt += 1;
                        }
                    }
                }
                K8Watch::DELETED(k8_obj) => {
                    match k8_obj_to_kv_obj(k8_obj) {
                        Ok(kv_value) =>  {
                            trace!("KV ({}): push DEL action", kv_value.key());
                            // try to delete it
                            if let Some(_old_value) = local_store.remove(kv_value.key()) {
                                del_cnt += 1;
                                actions.push(LSChange::delete(kv_value));
                            } else {
                                skip_cnt += 1;
                                warn!("delete  should never fail since key exists: {}",kv_value.key());
                            }
                        },
                        Err(err) => {
                            error!("{}", err);
                            skip_cnt += 1;
                        }
                    }
                }
            },
            Err(err) => error!("Problem parsing {} event: {} ... (exiting)", S::LABEL,err),
        }
    }
    // log counters
    let all = add_cnt + mod_cnt + del_cnt + skip_cnt;
    trace!("K8 Streams {} [all:{}, add:{},mod:{},del:{},ski: {}", 
        S::LABEL,
        all, 
        add_cnt, 
        mod_cnt, 
        del_cnt, 
        skip_cnt);
    actions
}
pub fn k8_event_stream_to_metadata_actions < S , E > ( stream : TokenStreamResult < S :: K8Spec , < S :: K8Spec as K8Spec > :: Status , E > , local_store : & LocalStore < S > ) -> Actions < LSChange < S > > where S : Spec + Debug + PartialEq + Debug , < S as Spec > :: K8Spec : Debug , S :: Key : Debug + Display + Clone , S :: Status : Debug + PartialEq , E : MetadataClientError { let ( mut add_cnt , mut mod_cnt , mut del_cnt , mut skip_cnt ) = ( 0 , 0 , 0 , 0 ) ; let mut actions : Actions < LSChange < S > > = Actions :: default ( ) ; // loop through items and generate add/mod actions for token in stream . unwrap ( ) { match token { Ok ( watch_obj ) => match watch_obj { K8Watch :: ADDED ( k8_obj ) => { let converted : Result < KVObject < S > , ScServerError > = k8_obj_to_kv_obj ( k8_obj ) ; // help out compiler match converted { Ok ( new_kv_value ) => { trace ! ( " " new_kv_value key ( ) ) ; if let Some ( old_value ) = local_store . insert ( new_kv_value . clone ( ) ) { // some old value, check if same as new one, if they are same, no need for action warn ! ( " " old_value ) ; if old_value == new_kv_value { trace ! ( " " ) ; } else { trace ! ( " " new_kv_value key ( ) ) ; actions . push ( LSChange :: update ( new_kv_value , old_value ) ) ; mod_cnt += 1 ; } } else { // no existing value, which should be expected debug ! ( " " S LABEL new_kv_value key ( ) ) ; actions . push ( LSChange :: add ( new_kv_value ) ) ; add_cnt += 1 ; } } , Err ( err ) => { error ! ( " " err ) ; skip_cnt += 1 ; } } } , K8Watch :: MODIFIED ( k8_obj ) => { let converted : Result < KVObject < S > , ScServerError > = k8_obj_to_kv_obj ( k8_obj ) ; // help out compiler match converted { Ok ( new_kv_value ) => { if let Some ( old_value ) = local_store . insert ( new_kv_value . clone ( ) ) { if old_value == new_kv_value { // this is unexpected,  warn ! ( " " new_kv_value ) ; } else { // normal actions . push ( LSChange :: update ( new_kv_value , old_value ) ) ; mod_cnt += 1 ; } } else { // doesn't exist, this is then new warn ! ( " " new_kv_value key ( ) ) ; actions . push ( LSChange :: add ( new_kv_value ) ) ; } } , Err ( err ) => { error ! ( " " err ) ; skip_cnt += 1 ; } } } K8Watch :: DELETED ( k8_obj ) => { match k8_obj_to_kv_obj ( k8_obj ) { Ok ( kv_value ) => { trace ! ( " " kv_value key ( ) ) ; // try to delete it if let Some ( _old_value ) = local_store . remove ( kv_value . key ( ) ) { del_cnt += 1 ; actions . push ( LSChange :: delete ( kv_value ) ) ; } else { skip_cnt += 1 ; warn ! ( " " kv_value key ( ) ) ; } } , Err ( err ) => { error ! ( " " err ) ; skip_cnt += 1 ; } } } } , Err ( err ) => error ! ( " " S LABEL err ) , } } // log counters let all = add_cnt + mod_cnt + del_cnt + skip_cnt ; trace ! ( " " S LABEL all add_cnt mod_cnt del_cnt skip_cnt ) ; actions }
Translates K8 events into metadata action.
Translates K8 events into metadata action .
stream
TokenStreamResult<S::K8Spec,<S::K8Spec as K8Spec>::Status,E>
local_store
&LocalStore<S>
stream
TokenStreamResult<S::K8Spec,<S::K8Spec as K8Spec>::Status,E>
local_store
&LocalStore<S>