You are an expert in writing code in many different languages. Your goal is to perform code completion for the following code, keeping in mind the rest of the code and the file meta data.
    Metadata details:
    filename: dummy_sink.rs
    path of file: ./repos/dozer/dozer-cli/src/pipeline
    
    the code of the file until where you have to start completion: use std::{collections::HashMap, time::Instant};

use dozer_core::{
    epoch::Epoch,
    event::EventHub,
    node::{PortHandle, Sink, SinkFactory},
    DEFAULT_PORT_HANDLE,
};
use dozer_types::log::debug;
use dozer_types::{
    chrono::Local,
    errors::internal::BoxedError,
    
    
    fn process(&mut self, op: TableOperation) -> Result<(), BoxedError> {
        if self.count == 0 {
            self.first_received = Some(Instant::now());
        }
        let diff = self.count - self.previous_op_count;
        if diff > 1000 {
            if self.count > 0 {
                info!(
                    "Rate: {:.0} op/s, Processed {} records. Elapsed {:?}",
                    diff as f64 / self.previous_started.elapsed().as_secs_f64(),
                    self.count,
                    self.previous_started.elapsed(),
                );
            }
            self.previous_started = Instant::now();
            self.previous_op_count = self.count;
        }

        self.count += match op.op {
            Operation::BatchInsert { ref new } => new.len(),
            _ => 1,
        };

        if let Some(stop_after) = self.stop_after {
            if self.count >= stop_after as usize {
                if let Some(first_received) = self.first_received {
                    info!("Stopping after {} records", stop_after);

                    info!(
                        "Rate: {:.0} op/s, Processed {} records. Elapsed {:?}",
                        stop_after as f64 / first_received.elapsed().as_secs_f64(),
                        self.count,
                        first_received.elapsed(),
                    );

                    if self.total_latency > 0 {
                        info!(
                            "Average latency: {}ms",
                            self.total_latency / stop_after as u64
                        );
                    }
                    std::process::exit(0);
                }
            }
        }

        if let Some(inserted_at_index) = self.inserted_at_index {
            let records = match op.op {
                Operation::BatchInsert { ref new } => new,
                Operation::Insert { ref new } => std::slice::from_ref(new),
                _ => &[],
            };

            for new in records {
                debug!("Received record: {:?}", new);
                let value = &new.values[inserted_at_index];
                if let Some(inserted_at) = value.to_timestamp() {
                    let latency = Local::now().naive_utc() - inserted_at.naive_utc();
                    self.total_latency += latency.num_milliseconds() as u64;
                    info!("Latency: {}ms", latency.num_milliseconds());
                } else {
                    warn!("expecting timestamp, got {:?}", value);
                }
            }
        }
        Ok(())
    }