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: monitor.rs
    path of file: ./repos/mongo-rust-driver/src/sdam
    
    the code of the file until where you have to start completion: use std::{
    sync::{
        atomic::{AtomicU32, Ordering},
        Arc,
    },
    time::{Duration, Instant},
};

use bson::doc;
use tokio::sync::watch;

use super::{
    description::server::{ServerDescription, TopologyVersion},
    topology::{SdamEventEmitter, TopologyCheckRequestReceiver},
    TopologyUpdater,
    TopologyWatcher,
};
use crate::{
    client::options::ServerMonitoringMode,
    cmap::{establish::ConnectionEstablisher, Connection},
    error::{Error, Result},
    event::sdam::{
        SdamEvent,
        ServerHeartbeatFailedEvent,
        ServerHeartbeatStartedEvent,
        ServerHeartbeatSucceededEvent,
    },
    hello::{hello_command, run_hello, AwaitableHelloOptions, HelloReply},
    options::{ClientOptions, ServerAddress},
    runtime::{self, stream::DEFAULT_CONNECT_TIMEOUT, WorkerHandle, WorkerHandleListener},
};

fn next_monitoring_connection_id() -> u32 {
    static MONITORING_CONNECTION_ID: AtomicU32 = AtomicU32::new(0);

    MONITORING_CONNECTION_ID.fetch_add(1, Ordering::SeqCst)
}

pub(crate) const DEFAULT_HEARTBEAT_FREQUENCY: Duration = Duration::from_secs(10);
pub(crate) const MIN_HEARTBEAT_FREQUENCY: Duration = Duration::from_millis(500);

/// Monitor that performs regular heartbeats to determine server status.
pub(crate) struct Monitor {
    address: ServerAddress,
    connection: Option<Connection>,
    connection_establisher: ConnectionEstablisher,
    topology_updater: TopologyUpdater,
    topology_watcher: TopologyWatcher,
    sdam_event_emitter: Option<SdamEventEmitter>,
    client_options: ClientOptions,

    /// Whether this monitor is allowed to use the streaming protocol.
    allow_streaming: bool,

    /// The most recent topology version returned by the server in a hello response.
    topology_version: Option<TopologyVersion>,

    /// The RTT monitor; once it's started this is None.
    pending_rtt_monitor: Option<RttMonitor>,

    /// Handle to the RTT monitor, used to get the latest known round trip time for a given server
    /// and to reset the RTT when the monitor disconnects from the server.
    rtt_monitor_handle: RttMonitorHandle,

    /// Handle to the `Server` instance in the `Topology`. This is used to detect when a server has
    /// been removed from the topology and no longer needs to be monitored and to receive
    /// cancellation requests.
    request_receiver: MonitorRequestReceiver,
}

impl Monitor {
    pub(crate) fn start(
        address: ServerAddress,
        topology_updater: TopologyUpdater,
        topology_watcher: TopologyWatcher,
        sdam_event_emitter: Option<SdamEventEmitter>,
        manager_receiver: MonitorRequestReceiver,
        client_options: ClientOptions,
        connection_establisher: ConnectionEstablisher,
    ) {
        let (rtt_monitor, rtt_monitor_handle) = RttMonitor::new(
            address.clone(),
            topology_watcher.clone(),
            connection_establisher.clone(),
            client_options.clone(),
        );
        let allow_streaming = match client_options
            .server_monitoring_mode
            .clone()
            .unwrap_or(ServerMonitoringMode::Auto)
        {
            ServerMonitoringMode::Stream => true,
            ServerMonitoringMode::Poll => false,
            ServerMonitoringMode::Auto => !crate::cmap::is_faas(),
        };
        let monitor = Self {
            address,
            client_options,
            connection_establisher,
            topology_updater,
            topology_watcher,
            sdam_event_emitter,
            pending_rtt_monitor: Some(rtt_monitor),
            rtt_monitor_handle,
            request_receiver: manager_receiver,
            connection: None,
            allow_streaming,
            topology_version: None,
        };

        runtime::spawn(monitor.execute());
    }

    async fn execute(mut self) {
        let heartbeat_frequency = self.heartbeat_frequency();

        while self.is_alive() {
            let check_succeeded = self.check_server().await;

            if self.topology_version.is_some() && self.allow_streaming {
                if let Some(rtt_monitor) = self.pending_rtt_monitor.take() {
                    runtime::spawn(rtt_monitor.execute());
                }
            }

            // In the streaming protocol, we read from the socket continuously
            // rather than polling at specific intervals, unless the most recent check
            // failed.
            //
            // We only go to sleep when using the polling protocol (i.e. server never returned a
            // topologyVersion) or when the most recent check failed.
            if self.topology_version.is_none() || !check_succeeded || !self.allow_streaming {
                self.request_receiver
                    .wait_for_check_request(
                        self.client_options.min_heartbeat_frequency(),
                        heartbeat_frequency,
                    )
                    .await;
            }
        }
    }

    fn is_alive(&self) -> bool {
        self.request_receiver.is_alive()
    }

    /// Checks the the server by running a hello command. If an I/O error occurs, the
    /// connection will replaced with a new one.
    ///
    /// Returns whether the check succeeded or not.
    async fn check_server(&mut self) -> bool {
        let check_result = match self.perform_hello().await {
            HelloResult::Err(e) => {
                let previous_description = self.topology_watcher.server_description(&self.address);
                if e.is_network_error()
         
    
    async fn perform_hello(&mut self) -> HelloResult {
        let driver_connection_id = self
            .connection
            .as_ref()
            .map(|c| c.id)
            .unwrap_or(next_monitoring_connection_id());

        self.emit_event(|| {
            SdamEvent::ServerHeartbeatStarted(ServerHeartbeatStartedEvent {
                server_address: self.address.clone(),
                awaited: self.topology_version.is_some() && self.allow_streaming,
                driver_connection_id,
                server_connection_id: self.connection.as_ref().and_then(|c| c.server_id),
            })
        });

        let heartbeat_frequency = self.heartbeat_frequency();
        let timeout = if self.connect_timeout().is_zero() {
            // If connectTimeoutMS = 0, then the socket timeout for monitoring is unlimited.
            Duration::MAX
        } else if self.topology_version.is_some() {
            // For streaming responses, use connectTimeoutMS + heartbeatFrequencyMS for socket
            // timeout.
            heartbeat_frequency
                .checked_add(self.connect_timeout())
                .unwrap_or(Duration::MAX)
        } else {
            // Otherwise, just use connectTimeoutMS.
            self.connect_timeout()
        };

        let execute_hello = async {
            match self.connection {
                Some(ref mut conn) => {
                    // If the server indicated there was moreToCome, just read from the socket.
                    if conn.is_streaming() {
                        conn.receive_message()
                            .await
                            .and_then(|r| r.into_hello_reply())
                    // Otherwise, send a regular hello command.
                    } else {
                        // If the initial handshake returned a topology version, send it back to the
                        // server to begin streaming responses.
                        let opts = if self.allow_streaming {
                            self.topology_version.map(|tv| AwaitableHelloOptions {
                                topology_version: tv,
                                max_await_time: heartbeat_frequency,
                            })
                        } else {
                            None
                        };

                        let command = hello_command(
                            self.client_options.server_api.as_ref(),
                            self.client_options.load_balanced,
                            Some(conn.stream_description()?.hello_ok),
                            opts,
                        );

                        run_hello(conn, command).await
                    }
                }
                None => {
                    let start = Instant::now();
                    let res = self
                        .connection_establisher
                        .establish_monitoring_connection(self.address.clone(), driver_connection_id)
                        .await;
                    match res {
                        Ok((conn, hello_reply)) => {
                            self.rtt_monitor_handle.add_sample(start.elapsed());
                            self.connection = Some(conn);
                            Ok(hello_reply)
                        }
                        Err(e) => Err(e),
                    }
                }
            }
        };

        // Execute the hello while also listening for cancellation and keeping track of the timeout.
        let start = Instant::now();
        let result = tokio::select! {
            result = execute_hello => match result {
                Ok(reply) => HelloResult::Ok(reply),
                Err(e) => HelloResult::Err(e)
            },
            r = self.request_receiver.wait_for_cancellation() => {
                let reason_error = match r {
                    CancellationReason::Error(e) => e,
                    CancellationReason::ServerClosed => Error::internal("server closed")
                };
                HelloResult::Cancelled { reason: reason_error }
            }
            _ = tokio::time::sleep(timeout) => {
                HelloResult::Err(Error::network_timeout())
            }
        };
        let duration = start.elapsed();

        let awaited = self.topology_version.is_some() && self.allow_streaming;
        match result {
            HelloResult::Ok(ref r) => {
                if !awaited {
                    self.rtt_monitor_handle.add_sample(duration);
                }
                self.emit_event(|| {
                    let mut reply = r
                        .raw_command_response
                        .to_document()
                        .unwrap_or_else(|e| doc! { "deserialization error": e.to_string() });
                    // if this hello call is part of a handshake, remove speculative authentication
                    // information before publishing an event
                    reply.remove("speculativeAuthenticate");
                    SdamEvent::ServerHeartbeatSucceeded(ServerHeartbeatSucceededEvent {
                        duration,
                        reply,
                        server_address: self.address.clone(),
                        awaited,
                        driver_connection_id,
                        server_connection_id: self.connection.as_ref().and_then(|c| c.server_id),
                    })
                });

                // If the response included a topology version, cache it so that we can return it in
                // the next hello.
                self.topology_version = r.command_response.topology_version;
            }
            HelloResult::Err(ref e) | HelloResult::Cancelled { reason: ref e } => {
                self.emit_event(|| {
                    SdamEvent::ServerHeartbeatFailed(ServerHeartbeatFailedEvent {
                        duration,
                        failure: e.clone(),
                        server_address: self.address.clone(),
                        awaited,
                        driver_connection_id,
                        server_connection_id: self.connection.as_ref().and_then(|c| c.server_id),
                    })
                });

                // Per the spec, cancelled requests and errors both require the monitoring
                // connection to be closed.
                self.connection = None;
                self.rtt_monitor_handle.reset_average_rtt();
                self.topology_version.take();
            }
        }

        result
    }