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: ping_pong.rs
    path of file: ./repos/mirrord/mirrord/intproxy/src
    
    the code of the file until where you have to start completion: //! Ping pong mechanism implementation on the internal proxy side.
//! This mechanism exists to keep the `proxy <-> agent` connection alive when there are no requests
//! from the layer.
//!
//! Realized using the [`DaemonMessage::Pong`](mirrord_protocol::codec::DaemonMessage::Pong) and
//! [`ClientMessage::Ping`] messages.

use std::time::Duration;

use mirrord_protocol::ClientMessage;
use thiserror::Error;
use tokio::time::{self, Interval, MissedTickBehavior};

use crate::{
    background_tasks::{BackgroundTask, MessageBus},
    ProxyMessage,
};

/// Errors that can occur when handling ping pong.
#[derive(Error, Debug)]
pub enum PingPongError {
    /// Agent sent pong but the proxy was not expecting one.
    #[error("received an unexpected pong from the agent")]
    UnmatchedPong,
    /// Agent did not send ping in time.
    #[error("agent did not respond to ping in time")]
    PongTimeout,
}

/// 
    
    async fn run(mut self, message_bus: &mut MessageBus<Self>) -> Result<(), Self::Error> {
        loop {
            tokio::select! {
                _ = self.ticker.tick() => {
                    if self.awaiting_pong {
                        tracing::error!("pong timeout");
                        break Err(PingPongError::PongTimeout);
                    } else {
                        tracing::trace!("sending ping");
                        let _ = message_bus.send(ProxyMessage::ToAgent(ClientMessage::Ping)).await;
                        self.ticker.reset();
                        self.awaiting_pong = true;
                    }
                },

                msg = message_bus.recv() => match (msg, self.awaiting_pong) {
                    (None, _) => {
                        tracing::trace!("message bus closed, exiting");
                        break Ok(())
                    },
                    (Some(AgentMessageNotification { pong: true }), true) => {
                        tracing::trace!("agent responded to ping");
                        self.awaiting_pong = false;
                        self.ticker.reset();
                    },
                    (Some(AgentMessageNotification { pong: false }), true) => {
                        tracing::trace!("agent sent message, still waiting for pong")
                    },
                    (Some(AgentMessageNotification { pong: true }), false) => {
                        tracing::error!("agent sent an unexpected pong");
                        break Err(PingPongError::UnmatchedPong)
                    },
                    (Some(AgentMessageNotification { pong: false }), false) => {
                        self.ticker.reset();
                    }
                },
            }
        }
    }