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/dozer/dozer-core/src/executor/source_node
    
    the code of the file until where you have to start completion: use std::{fmt::Debug, future::Future, pin::pin, sync::Arc, time::SystemTime};

use daggy::petgraph::visit::IntoNodeIdentifiers;
use dozer_types::{
    log::debug, models::ingestion_types::TransactionInfo, node::OpIdentifier, types::TableOperation,
};
use dozer_types::{models::ingestion_types::IngestionMessage, node::SourceState};
use futures::{future::Either, StreamExt};
use tokio::{
    runtime::Runtime,
    sync::mpsc::{channel, Receiver, Sender},
};

use crate::{
    builder_dag::NodeKind,
    epoch::Epoch,
    errors::ExecutionError,
    executor_operation::ExecutorOperation,
    forwarder::ChannelManager,
    node::{PortHandle, Source},
};

use super::{execution_dag::ExecutionDag, node::Node, ExecutorOptions};

/// The source operation collector.
#[derive(Debug)]
pub struct SourceNode<F> {
    /// To decide when to emit `Commit`, we keep track of source state.
    sources: Vec<RunningSource>,
    /// Structs for running a source.
    source_runners: Vec<SourceRunner>,
    /// Receivers from sources.
    receivers: Vec<Receiver<(PortHandle, IngestionMessage)>>,
    /// The current epoch id.
    epoch_id: u64,
    /// The shutdown future.
    shutdown: F,
    /// The runtime to run the source in.
    runtime: Arc<Runtime>,
}

impl<F: Future + Unpin> Node for SourceNode<F> {
    fn run(mut self) -> Result<(), ExecutionError> {
        let mut handles = vec![];
        for mut source_runner in self.source_runners {
            handles.push(Some(self.runtime.spawn(async move {
                source_runner
                    .source
                    .start(source_runner.sender, source_runner.last_checkpoint)
                    .await
            })));
        }
        let mut num_running_sources = handles.len();

        let mut stream = pin!(stream::receivers_stream(self.receivers));
        loop {
            let next = stream.next();
            let next = pin!(next);
            match self
                .runtime
                .block_on(futures::future::select(self.shutdown, next))
            {
                Either::Left((_, _)) => {
                    send_to_all_nodes(&self.sources, ExecutorOperation::Terminate)?;
                    return Ok(());
                }
                Either::Right((next, shutdown)) => {
                    let next = next.expect("We return just when the stream ends");
                    self.shutdown = shutdown;
                    let index = next.0;
                    let Some((port, message)) = next.1 else {
                        debug!("[{}] quit", self.sources[index].channel_manager.owner().id);
                        match self.runtime.block_on(
                            handles[index]
                                .take()
                                .expect("Shouldn't receive message from dropped receiver"),
                        ) {
                            Ok(Ok(())) => {
                                num_running_sources -= 1;
                                if num_running_sources == 0 {
                                    send_to_all_nodes(&self.sources, ExecutorOperation::Terminate)?;
                                    return Ok(());
                                }
                                continue;
                            }
                            Ok(Err(e)) => return Err(ExecutionError::Source(e)),
                            Err(e) => {
                                panic!("Source panicked: {e}");
                            }
                        }
                    };
                    let source = &mut self.sources[index];
                    match message {
                        IngestionMessage::OperationEvent { op, id, .. } => {
                            source.state = SourceState::NonRestartable;
                            source
                                .channel_manager
                                .send_op(TableOperation { op, id, port })?;
                        }
                        IngestionMessage::TransactionInfo(info) => match info {
                            TransactionInfo::Commit { id } => {
                                if let Some(id) = id {
                                    source.state = SourceState::Restartable(id);
                                } else {
                                    source.state = SourceState::NonRestartable;
                                }

                                let source_states = Arc::new(
                                    self.sources
                                        .iter()
                                        .map(|source| {
                                            (
                                                source.channel_manager.owne
    
    fn run(mut self) -> Result<(), ExecutionError> {
        let mut handles = vec![];
        for mut source_runner in self.source_runners {
            handles.push(Some(self.runtime.spawn(async move {
                source_runner
                    .source
                    .start(source_runner.sender, source_runner.last_checkpoint)
                    .await
            })));
        }
        let mut num_running_sources = handles.len();

        let mut stream = pin!(stream::receivers_stream(self.receivers));
        loop {
            let next = stream.next();
            let next = pin!(next);
            match self
                .runtime
                .block_on(futures::future::select(self.shutdown, next))
            {
                Either::Left((_, _)) => {
                    send_to_all_nodes(&self.sources, ExecutorOperation::Terminate)?;
                    return Ok(());
                }
                Either::Right((next, shutdown)) => {
                    let next = next.expect("We return just when the stream ends");
                    self.shutdown = shutdown;
                    let index = next.0;
                    let Some((port, message)) = next.1 else {
                        debug!("[{}] quit", self.sources[index].channel_manager.owner().id);
                        match self.runtime.block_on(
                            handles[index]
                                .take()
                                .expect("Shouldn't receive message from dropped receiver"),
                        ) {
                            Ok(Ok(())) => {
                                num_running_sources -= 1;
                                if num_running_sources == 0 {
                                    send_to_all_nodes(&self.sources, ExecutorOperation::Terminate)?;
                                    return Ok(());
                                }
                                continue;
                            }
                            Ok(Err(e)) => return Err(ExecutionError::Source(e)),
                            Err(e) => {
                                panic!("Source panicked: {e}");
                            }
                        }
                    };
                    let source = &mut self.sources[index];
                    match message {
                        IngestionMessage::OperationEvent { op, id, .. } => {
                            source.state = SourceState::NonRestartable;
                            source
                                .channel_manager
                                .send_op(TableOperation { op, id, port })?;
                        }
                        IngestionMessage::TransactionInfo(info) => match info {
                            TransactionInfo::Commit { id } => {
                                if let Some(id) = id {
                                    source.state = SourceState::Restartable(id);
                                } else {
                                    source.state = SourceState::NonRestartable;
                                }

                                let source_states = Arc::new(
                                    self.sources
                                        .iter()
                                        .map(|source| {
                                            (
                                                source.channel_manager.owner().clone(),
                                                source.state.clone(),
                                            )
                                        })
                                        .collect(),
                                );
                                let epoch =
                                    Epoch::new(self.epoch_id, source_states, SystemTime::now());
                                send_to_all_nodes(
                                    &self.sources,
                                    ExecutorOperation::Commit { epoch },
                                )?;
                                self.epoch_id += 1;
                            }
                            TransactionInfo::SnapshottingStarted => {
                                source.channel_manager.send_snapshotting_started(
                                    source.channel_manager.owner().id.clone(),
                                )?;
                            }
                            TransactionInfo::SnapshottingDone { id } => {
                                source.channel_manager.send_snapshotting_done(
                                    source.channel_manager.owner().id.clone(),
                                    id,
                                )?;
                            }
                        },
                    }
                }
            }
        }
    }