pub(self) fn spawn(
        id: NodeId,
        target: NodeId,
        term: u64,
        config: Arc<Config>,
        last_log: LogId,
        commit_index: u64,
        network: Arc<N>,
        storage: Arc<S>,
        raft_core_tx: mpsc::UnboundedSender<(ReplicaEvent<S::SnapshotData>, Span)>,
    ) -> ReplicationStream<D> {
        // other component to ReplicationStream
        let (repl_tx, repl_rx) = mpsc::unbounded_channel();
        let heartbeat_timeout = Duration::from_millis(config.heartbeat_interval);
        let install_snapshot_timeout = Duration::from_millis(config.install_snapshot_timeout);

        let max_payload_entries = config.max_payload_entries as usize;
        let this = Self {
            id,
            target,
            term,
            network,
            storage,
            config,
            max_payload_entries,
            marker_r: std::marker::PhantomData,
            target_state: TargetReplState::Lagging,
            last_log_index: last_log.index,
            commit_index,
            next_index: last_log.index + 1,
            matched: last_log,
            raft_core_tx,
            repl_rx,
            heartbeat: interval(heartbeat_timeout),
            heartbeat_timeout,
            install_snapshot_timeout,
            replication_buffer: Vec::new(),
            outbound_buffer: Vec::new(),
        };

        let _handle = tokio::spawn(this.main().instrument(tracing::debug_span!("spawn")));

        ReplicationStream {
            // handle,
            repl_tx,
        }
    }
/// Spawn a new replication task for the target node.
Spawn a new replication task for the target node.
Spawn a new replication task for the target node .
pub(self) fn spawn(
        id: NodeId,
        target: NodeId,
        term: u64,
        config: Arc<Config>,
        last_log: LogId,
        commit_index: u64,
        network: Arc<N>,
        storage: Arc<S>,
        raft_core_tx: mpsc::UnboundedSender<(ReplicaEvent<S::SnapshotData>, Span)>,
    ) -> ReplicationStream<D> {
        // other component to ReplicationStream
        let (repl_tx, repl_rx) = mpsc::unbounded_channel();
        let heartbeat_timeout = Duration::from_millis(config.heartbeat_interval);
        let install_snapshot_timeout = Duration::from_millis(config.install_snapshot_timeout);
        let max_payload_entries = config.max_payload_entries as usize;
        let this = Self {
            id,
            target,
            term,
            network,
            storage,
            config,
            max_payload_entries,
            marker_r: std::marker::PhantomData,
            target_state: TargetReplState::Lagging,
            last_log_index: last_log.index,
            commit_index,
            next_index: last_log.index + 1,
            matched: last_log,
            raft_core_tx,
            repl_rx,
            heartbeat: interval(heartbeat_timeout),
            heartbeat_timeout,
            install_snapshot_timeout,
            replication_buffer: Vec::new(),
            outbound_buffer: Vec::new(),
        };
        let _handle = tokio::spawn(this.main().instrument(tracing::debug_span!("spawn")));
        ReplicationStream {
            // handle,
            repl_tx,
        }
    }
pub ( self ) fn spawn ( id : NodeId , target : NodeId , term : u64 , config : Arc < Config > , last_log : LogId , commit_index : u64 , network : Arc < N > , storage : Arc < S > , raft_core_tx : mpsc :: UnboundedSender < ( ReplicaEvent < S :: SnapshotData > , Span ) > , ) -> ReplicationStream < D > { // other component to ReplicationStream let ( repl_tx , repl_rx ) = mpsc :: unbounded_channel ( ) ; let heartbeat_timeout = Duration :: from_millis ( config . heartbeat_interval ) ; let install_snapshot_timeout = Duration :: from_millis ( config . install_snapshot_timeout ) ; let max_payload_entries = config . max_payload_entries as usize ; let this = Self { id , target , term , network , storage , config , max_payload_entries , marker_r : std :: marker :: PhantomData , target_state : TargetReplState :: Lagging , last_log_index : last_log . index , commit_index , next_index : last_log . index + 1 , matched : last_log , raft_core_tx , repl_rx , heartbeat : interval ( heartbeat_timeout ) , heartbeat_timeout , install_snapshot_timeout , replication_buffer : Vec :: new ( ) , outbound_buffer : Vec :: new ( ) , } ; let _handle = tokio :: spawn ( this . main ( ) . instrument ( tracing :: debug_span ! ( " " ) ) ) ; ReplicationStream { // handle, repl_tx , } }
Spawn a new replication task for the target node.
Spawn a new replication task for the target node .
id
NodeId
target
NodeId
term
u64
config
Arc<Config>
last_log
LogId
commit_index
u64
network
Arc<N>
storage
Arc<S>
raft_core_tx
mpsc::UnboundedSender<(ReplicaEvent<S::SnapshotData>, Span)>
id
NodeId
target
NodeId
term
u64
config
Arc<Config>
last_log
LogId
commit_index
u64
network
Arc<N>
storage
Arc<S>
raft_core_tx
mpsc::UnboundedSender<(ReplicaEvent<S::SnapshotData>, Span)>