fn update_sources(&mut self) -> bool {
        // First check if there are some new source that we should
        // start checking
        loop {
            match self.rx.try_recv() {
                Ok(TimestampMessage::Add(source_id, sc)) => {
                    let (sc, enc, env, cons) = if let SourceConnector::External {
                        connector,
                        encoding,
                        envelope,
                        consistency,
                        ts_frequency: _,
                        timeline: _,
                        persist: _,
                    } = sc
                    {
                        (connector, encoding, envelope, consistency)
                    } else {
                        log::debug!("Local source {} cannot be timestamped. Ignoring", source_id);
                        continue;
                    };

                    if !self.rt_sources.contains_key(&source_id)
                        && !self.byo_sources.contains_key(&source_id)
                    {
                        // Did not know about source, must update
                        match cons {
                            Consistency::RealTime => {
                                info!(
                                    "Timestamping Source {} with Real Time Consistency.",
                                    source_id
                                );
                                let consumer = self.create_rt_connector(source_id, sc);
                                if let Some(consumer) = consumer {
                                    self.rt_sources.insert(source_id, consumer);
                                }
                            }
                            Consistency::BringYourOwn(consistency_topic) => {
                                info!("Timestamping Source {} with BYO Consistency. Consistency Source: {:?}.",
                                      source_id, consistency_topic);
                                let consumer = self.create_byo_connector(
                                    source_id,
                                    sc,
                                    enc.value(),
                                    env,
                                    consistency_topic.topic,
                                );
                                if let Some(consumer) = consumer {
                                    self.byo_sources.insert(source_id, consumer);
                                }
                            }
                        }
                    }
                }
                Ok(TimestampMessage::Drop(id)) => {
                    self.drop_source(id);
                }
                Err(TryRecvError::Empty) => return false,
                Err(TryRecvError::Disconnected) => {
                    // First, let's remove all of the threads consuming metadata
                    // from realtime Kafka sources
                    for (_, src) in self.rt_sources.iter_mut() {
                        if let RtTimestampConnector::Kafka(RtKafkaConnector {
                            coordination_state,
                            ..
                        }) = &src.connector
                        {
                            coordination_state.stop.store(true, Ordering::SeqCst);
                        }
                    }
                    return true;
                }
            }
        }
    }
/// Updates list of timestamp sources based on coordinator information. If using
/// using the real-time timestamping logic, then maintain a list of Kafka consumers
/// that poll topics to check how much data has been generated. If using the Kafka
/// source timestamping logic, then keep a mapping of (name,id) to translate user-
/// defined timestamps to GlobalIds
Updates list of timestamp sources based on coordinator information. If using
using the real-time timestamping logic, then maintain a list of Kafka consumers
that poll topics to check how much data has been generated. If using the Kafka
source timestamping logic, then keep a mapping of (name,id) to translate user
defined timestamps to GlobalIds
Updates list of timestamp sources based on coordinator information . If using using the real - time timestamping logic then maintain a list of Kafka consumers that poll topics to check how much data has been generated . If using the Kafka source timestamping logic then keep a mapping of ( name id ) to translate user defined timestamps to GlobalIds
fn update_sources(&mut self) -> bool {
        // First check if there are some new source that we should
        // start checking
        loop {
            match self.rx.try_recv() {
                Ok(TimestampMessage::Add(source_id, sc)) => {
                    let (sc, enc, env, cons) = if let SourceConnector::External {
                        connector,
                        encoding,
                        envelope,
                        consistency,
                        ts_frequency: _,
                        timeline: _,
                        persist: _,
                    } = sc
                    {
                        (connector, encoding, envelope, consistency)
                    } else {
                        log::debug!("Local source {} cannot be timestamped. Ignoring", source_id);
                        continue;
                    };
                    if !self.rt_sources.contains_key(&source_id)
                        && !self.byo_sources.contains_key(&source_id)
                    {
                        // Did not know about source, must update
                        match cons {
                            Consistency::RealTime => {
                                info!(
                                    "Timestamping Source {} with Real Time Consistency.",
                                    source_id
                                );
                                let consumer = self.create_rt_connector(source_id, sc);
                                if let Some(consumer) = consumer {
                                    self.rt_sources.insert(source_id, consumer);
                                }
                            }
                            Consistency::BringYourOwn(consistency_topic) => {
                                info!("Timestamping Source {} with BYO Consistency. Consistency Source: {:?}.",
                                      source_id, consistency_topic);
                                let consumer = self.create_byo_connector(
                                    source_id,
                                    sc,
                                    enc.value(),
                                    env,
                                    consistency_topic.topic,
                                );
                                if let Some(consumer) = consumer {
                                    self.byo_sources.insert(source_id, consumer);
                                }
                            }
                        }
                    }
                }
                Ok(TimestampMessage::Drop(id)) => {
                    self.drop_source(id);
                }
                Err(TryRecvError::Empty) => return false,
                Err(TryRecvError::Disconnected) => {
                    // First, let's remove all of the threads consuming metadata
                    // from realtime Kafka sources
                    for (_, src) in self.rt_sources.iter_mut() {
                        if let RtTimestampConnector::Kafka(RtKafkaConnector {
                            coordination_state,
                            ..
                        }) = &src.connector
                        {
                            coordination_state.stop.store(true, Ordering::SeqCst);
                        }
                    }
                    return true;
                }
            }
        }
    }
fn update_sources ( & mut self ) -> bool { // First check if there are some new source that we should // start checking loop { match self . rx . try_recv ( ) { Ok ( TimestampMessage :: Add ( source_id , sc ) ) => { let ( sc , enc , env , cons ) = if let SourceConnector :: External { connector , encoding , envelope , consistency , ts_frequency : _ , timeline : _ , persist : _ , } = sc { ( connector , encoding , envelope , consistency ) } else { log :: debug ! ( " " source_id ) ; continue ; } ; if ! self . rt_sources . contains_key ( & source_id ) && ! self . byo_sources . contains_key ( & source_id ) { // Did not know about source, must update match cons { Consistency :: RealTime => { info ! ( " " source_id ) ; let consumer = self . create_rt_connector ( source_id , sc ) ; if let Some ( consumer ) = consumer { self . rt_sources . insert ( source_id , consumer ) ; } } Consistency :: BringYourOwn ( consistency_topic ) => { info ! ( " " source_id consistency_topic ) ; let consumer = self . create_byo_connector ( source_id , sc , enc . value ( ) , env , consistency_topic . topic , ) ; if let Some ( consumer ) = consumer { self . byo_sources . insert ( source_id , consumer ) ; } } } } } Ok ( TimestampMessage :: Drop ( id ) ) => { self . drop_source ( id ) ; } Err ( TryRecvError :: Empty ) => return false , Err ( TryRecvError :: Disconnected ) => { // First, let's remove all of the threads consuming metadata // from realtime Kafka sources for ( _ , src ) in self . rt_sources . iter_mut ( ) { if let RtTimestampConnector :: Kafka ( RtKafkaConnector { coordination_state , .. } ) = & src . connector { coordination_state . stop . store ( true , Ordering :: SeqCst ) ; } } return true ; } } } }
Updates list of timestamp sources based on coordinator information.
Updates list of timestamp sources based on coordinator information .
&mut self
&mut self