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: mod.rs
    path of file: ./repos/vector/src/sources/mongodb_metrics
    
    the code of the file until where you have to start completion: use std::time::{Duration, Instant};

use chrono::Utc;
use futures::{
    future::{join_all, try_join_all},
    StreamExt,
};
use mongodb::{
    bson::{self, doc, from_document, Bson, Document},
    error::Error as MongoError,
    options::ClientOptions,
    Client,
};
use serde_with::serde_as;
use snafu::{ResultExt, Snafu};
use tokio::time;
use tokio_stream::wrappers::IntervalStream;
use vector_lib::configurable::configurable_component;
use vector_lib::{metric_tags, ByteSizeOf, EstimatedJsonEncodedSizeOf};

use crate::{
    config::{SourceConfig, SourceContext, SourceOutput},
    event::metric::{Metric, MetricKind, MetricTags, MetricValue},
    internal_events::{
        CollectionCompleted, EndpointBytesReceived, MongoDbMetricsBsonParseError,
        MongoDbMetricsEventsReceived, MongoDbMetricsRequestError, StreamClosedError,
    },
};

mod types;
use types::{CommandBuildInfo, CommandIsMaster, CommandServerStatus, NodeType};
use vector_lib::config::LogNamespace;

macro_rules! tags {
    ($tags:expr) => { $tags.clone() };
    ($tags:expr, $($key:expr => $value:expr),*) => {
        {
            let mut tags = $tags.clone();
            $(
                tags.replace($key.into(), $val
    
    async fn build(&self, mut cx: SourceContext) -> crate::Result<super::Source> {
        let namespace = Some(self.namespace.clone()).filter(|namespace| !namespace.is_empty());

        let sources = try_join_all(
            self.endpoints
                .iter()
                .map(|endpoint| MongoDbMetrics::new(endpoint, namespace.clone())),
        )
        .await?;

        let duration = self.scrape_interval_secs;
        let shutdown = cx.shutdown;
        Ok(Box::pin(async move {
            let mut interval = IntervalStream::new(time::interval(duration)).take_until(shutdown);
            while interval.next().await.is_some() {
                let start = Instant::now();
                let metrics = join_all(sources.iter().map(|mongodb| mongodb.collect())).await;
                emit!(CollectionCompleted {
                    start,
                    end: Instant::now()
                });

                let metrics: Vec<Metric> = metrics.into_iter().flatten().collect();
                let count = metrics.len();

                if (cx.out.send_batch(metrics).await).is_err() {
                    emit!(StreamClosedError { count });
                    return Err(());
                }
            }

            Ok(())
        }))
    }