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: milestone.rs
    path of file: ./repos/bee/bee-protocol/bee-protocol/src/block/payload
    
    the code of the file until where you have to start completion: // Copyright 2020-2022 IOTA Stiftung
// SPDX-License-Identifier: Apache-2.0

use std::{any::TypeId, convert::Infallible, ops::Deref};

use async_trait::async_trait;
use bee_block::{
    payload::{
        milestone::{MilestonePayload, MilestoneValidationError},
        Payload,
    },
    Block, BlockId,
};
use bee_runtime::{event::Bus, node::Node, shutdown_stream::ShutdownStream, worker::Worker};
use bee_tangle::{event::LatestMilestoneChanged, milestone_metadata::MilestoneMetadata, Tangle, TangleWorker};
use futures::{future::FutureExt, stream::StreamExt};
use log::{debug, error, info};
use tokio::sync::mpsc;
use tokio_stream::wrappers::UnboundedReceiverStream;

use crate::{
    config::ProtocolConfig,
    heartbeater::broadcast_heartbeat,
    peer::PeerManager,
    storage::StorageBackend,
    types::{metrics::NodeMetrics, milestone_key_manager::MilestoneKeyManager},
    MetricsWorker, MilestoneRequesterWorker, MilestoneSolidifierWorker, MilestoneSolidifierWorkerEvent,
    PeerManagerResWorker, RequestedMilestones,
};

#[derive(Debug)]
pub(crate) enum Error {
    BlockMilestoneParentsMismatch,
    InvalidMilestone(MilestoneValidationError),
}

pub(crate) struct MilestonePayloadWorkerEvent {
    pub(crate) block_id: BlockId,
    pub(crate) block: Block,
}

p
    
    async fn start(node: &mut N, config: Self::Config) -> Result<Self, Self::Error> {
        let milestone_solidifier = node.worker::<MilestoneSolidifierWorker>().unwrap().tx.clone();
        let tangle = node.resource::<Tangle<N::Backend>>();
        let requested_milestones = node.resource::<RequestedMilestones>();
        let peer_manager = node.resource::<PeerManager>();
        let metrics = node.resource::<NodeMetrics>();
        let key_manager = MilestoneKeyManager::new(
            config.coordinator.public_key_count,
            config.coordinator.public_key_ranges.into_boxed_slice(),
        );
        let bus = node.bus();
        let (tx, rx) = mpsc::unbounded_channel();

        node.spawn::<Self, _, _>(|shutdown| async move {
            info!("Running.");

            let mut receiver = ShutdownStream::new(shutdown, UnboundedReceiverStream::new(rx));

            while let Some(MilestonePayloadWorkerEvent { block_id, block }) = receiver.next().await {
                process(
                    &tangle,
                    block_id,
                    block,
                    &peer_manager,
                    &metrics,
                    &requested_milestones,
                    &milestone_solidifier,
                    &key_manager,
                    &bus,
                );
            }

            // Before the worker completely stops, the receiver needs to be drained for milestone payloads to be
            // analysed. Otherwise, information would be lost and not easily recoverable.

            let (_, mut receiver) = receiver.split();
            let mut count: usize = 0;

            while let Some(Some(MilestonePayloadWorkerEvent { block_id, block })) = receiver.next().now_or_never() {
                process(
                    &tangle,
                    block_id,
                    block,
                    &peer_manager,
                    &metrics,
                    &requested_milestones,
                    &milestone_solidifier,
                    &key_manager,
                    &bus,
                );
                count += 1;
            }

            debug!("Drained {} events.", count);

            info!("Stopped.");
        });

        Ok(Self { tx })
    }