pub fn build(
        self,
        log: Logger,
        readers: Readers,
        channel_coordinator: Arc<ChannelCoordinator>,
        control_addr: SocketAddr,
        shutdown_valve: &Valve,
        state_size: Arc<AtomicUsize>,
    ) -> Domain {
        // initially, all nodes are not ready
        let not_ready = self
            .nodes
            .values()
            .map(|n| n.borrow().local_addr())
            .collect();

        let log = log.new(o!("domain" => self.index.index(), "shard" => self.shard.unwrap_or(0)));
        let control_reply_tx = TcpSender::connect(&control_addr).unwrap();
        let group_commit_queues = GroupCommitQueueSet::new(&self.persistence_parameters);

        Domain {
            index: self.index,
            shard: self.shard,
            _nshards: self.nshards,

            persistence_parameters: self.persistence_parameters,
            nodes: self.nodes,
            state: StateMap::default(),
            log,
            not_ready,
            mode: DomainMode::Forwarding,
            waiting: Default::default(),
            reader_triggered: Default::default(),
            replay_paths: Default::default(),
            replay_paths_by_dst: Default::default(),

            ingress_inject: Default::default(),

            shutdown_valve: shutdown_valve.clone(),
            readers,
            control_reply_tx,
            channel_coordinator,

            buffered_replay_requests: Default::default(),
            replay_batch_timeout: self.config.replay_batch_timeout,
            timed_purges: Default::default(),

            concurrent_replays: 0,
            max_concurrent_replays: self.config.concurrent_replays,
            replay_request_queue: Default::default(),
            delayed_for_self: Default::default(),

            group_commit_queues,

            state_size,
            total_time: Timer::new(),
            total_ptime: Timer::new(),
            wait_time: Timer::new(),
            process_times: TimerSet::new(),
            process_ptimes: TimerSet::new(),

            total_replay_time: Timer::new(),
            total_forward_time: Timer::new(),
        }
    }
/// Starts up the domain represented by this `DomainBuilder`.
Starts up the domain represented by this `DomainBuilder`.
Starts up the domain represented by this ` DomainBuilder ` .
pub fn build(
        self,
        log: Logger,
        readers: Readers,
        channel_coordinator: Arc<ChannelCoordinator>,
        control_addr: SocketAddr,
        shutdown_valve: &Valve,
        state_size: Arc<AtomicUsize>,
    ) -> Domain {
        // initially, all nodes are not ready
        let not_ready = self
            .nodes
            .values()
            .map(|n| n.borrow().local_addr())
            .collect();
        let log = log.new(o!("domain" => self.index.index(), "shard" => self.shard.unwrap_or(0)));
        let control_reply_tx = TcpSender::connect(&control_addr).unwrap();
        let group_commit_queues = GroupCommitQueueSet::new(&self.persistence_parameters);
        Domain {
            index: self.index,
            shard: self.shard,
            _nshards: self.nshards,
            persistence_parameters: self.persistence_parameters,
            nodes: self.nodes,
            state: StateMap::default(),
            log,
            not_ready,
            mode: DomainMode::Forwarding,
            waiting: Default::default(),
            reader_triggered: Default::default(),
            replay_paths: Default::default(),
            replay_paths_by_dst: Default::default(),
            ingress_inject: Default::default(),
            shutdown_valve: shutdown_valve.clone(),
            readers,
            control_reply_tx,
            channel_coordinator,
            buffered_replay_requests: Default::default(),
            replay_batch_timeout: self.config.replay_batch_timeout,
            timed_purges: Default::default(),
            concurrent_replays: 0,
            max_concurrent_replays: self.config.concurrent_replays,
            replay_request_queue: Default::default(),
            delayed_for_self: Default::default(),
            group_commit_queues,
            state_size,
            total_time: Timer::new(),
            total_ptime: Timer::new(),
            wait_time: Timer::new(),
            process_times: TimerSet::new(),
            process_ptimes: TimerSet::new(),
            total_replay_time: Timer::new(),
            total_forward_time: Timer::new(),
        }
    }
pub fn build ( self , log : Logger , readers : Readers , channel_coordinator : Arc < ChannelCoordinator > , control_addr : SocketAddr , shutdown_valve : & Valve , state_size : Arc < AtomicUsize > , ) -> Domain { // initially, all nodes are not ready let not_ready = self . nodes . values ( ) . map ( | n | n . borrow ( ) . local_addr ( ) ) . collect ( ) ; let log = log . new ( o ! ( " " self index index ( ) " " self shard unwrap_or ( 0 ) ) ) ; let control_reply_tx = TcpSender :: connect ( & control_addr ) . unwrap ( ) ; let group_commit_queues = GroupCommitQueueSet :: new ( & self . persistence_parameters ) ; Domain { index : self . index , shard : self . shard , _nshards : self . nshards , persistence_parameters : self . persistence_parameters , nodes : self . nodes , state : StateMap :: default ( ) , log , not_ready , mode : DomainMode :: Forwarding , waiting : Default :: default ( ) , reader_triggered : Default :: default ( ) , replay_paths : Default :: default ( ) , replay_paths_by_dst : Default :: default ( ) , ingress_inject : Default :: default ( ) , shutdown_valve : shutdown_valve . clone ( ) , readers , control_reply_tx , channel_coordinator , buffered_replay_requests : Default :: default ( ) , replay_batch_timeout : self . config . replay_batch_timeout , timed_purges : Default :: default ( ) , concurrent_replays : 0 , max_concurrent_replays : self . config . concurrent_replays , replay_request_queue : Default :: default ( ) , delayed_for_self : Default :: default ( ) , group_commit_queues , state_size , total_time : Timer :: new ( ) , total_ptime : Timer :: new ( ) , wait_time : Timer :: new ( ) , process_times : TimerSet :: new ( ) , process_ptimes : TimerSet :: new ( ) , total_replay_time : Timer :: new ( ) , total_forward_time : Timer :: new ( ) , } }
Starts up the domain represented by this `DomainBuilder`.
Starts up the domain represented by this ` DomainBuilder ` .
self
log
Logger
readers
Readers
channel_coordinator
Arc<ChannelCoordinator>
control_addr
SocketAddr
shutdown_valve
&Valve
state_size
Arc<AtomicUsize>
self
log
Logger
readers
Readers
channel_coordinator
Arc<ChannelCoordinator>
control_addr
SocketAddr
shutdown_valve
&Valve
state_size
Arc<AtomicUsize>