Struct coord::Timestamper

pub struct Timestamper {
    rt_sources: HashMap<GlobalId, RtTimestampConsumer>,
    tx: UnboundedSender<Message>,
    rx: Receiver<TimestampMessage>,
    timestamp_frequency: Duration,
    metrics: Metrics,
}

Fields

rt_sources: HashMap<GlobalId, RtTimestampConsumer>

Current list of up to date sources that use a real time consistency model

tx: UnboundedSender<Message>

Channel through which timestamp data updates are communicated through the coordinator

rx: Receiver<TimestampMessage>

Channel through which to timestamp metadata updates are received from the coordinator (to add or remove the timestamping of a source)

timestamp_frequency: Duration

Frequency at which thread should run

metrics: Metrics

Metrics that the timestamper reports.

Implementations

impl Timestamper

pub fn new(
    frequency: Duration,
    tx: UnboundedSender<Message>,
    rx: Receiver<TimestampMessage>,
    registry: &MetricsRegistry
) -> Self

pub fn run(&mut self)

Run the update function in a loop at the specified frequency. Acquires timestamps using either the Kafka topic ground truth

fn update_sources(&mut self) -> bool

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 drop_source(&mut self, id: GlobalId)

fn create_rt_connector(
    &self,
    id: GlobalId,
    sc: ExternalSourceConnector
) -> Option<RtTimestampConsumer>

Creates a RT connector TODO(rkhaitan): this function burns my eyes

fn create_rt_kinesis_connector(
    &self,
    _id: GlobalId,
    _kinc: KinesisSourceConnector
) -> Option<RtKinesisConnector>

fn create_rt_kafka_connector(
    &self,
    id: GlobalId,
    kc: KafkaSourceConnector
) -> Option<RtKafkaConnector>

fn create_rt_ocf_connector(
    &self,
    _id: GlobalId,
    _fc: FileSourceConnector
) -> Option<RtFileConnector>

fn create_rt_file_connector(
    &self,
    _id: GlobalId,
    _fc: FileSourceConnector
) -> Option<RtFileConnector>

fn create_rt_s3_connector(
    &self,
    _id: GlobalId,
    _fc: S3SourceConnector
) -> Option<RtS3Connector>