pub async fn start(mut self, network: Vec<(StateSynchronizerSender, StateSynchronizerEvents)>) {
        self.known_version = self
            .executor_proxy
            .get_latest_version()
            .await
            .expect("[start sync] failed to fetch latest version from storage");

        let mut interval =
            Interval::new_interval(Duration::from_millis(self.config.tick_interval_ms))
                .compat()
                .fuse();

        let network_senders: Vec<StateSynchronizerSender> =
            network.iter().map(|t| t.0.clone()).collect();
        let events: Vec<_> = network
            .into_iter()
            .enumerate()
            .map(|(idx, t)| t.1.map(move |e| (idx, e)))
            .collect();
        let mut network_events = select_all(events).fuse();

        loop {
            ::futures::select! {
                msg = self.client_events.select_next_some() => {
                    match msg {
                        CoordinatorMessage::Requested(target, subscription) => {
                            self.request_sync(target, subscription).await;
                        }
                        CoordinatorMessage::Commit(version) => {
                             self.commit(version).await;
                        }
                        CoordinatorMessage::GetState(callback) => {
                            self.get_state(callback);
                        }
                    };
                },
                (idx, network_event) = network_events.select_next_some() => {
                    match network_event {
                        Ok(event) => {
                            match event {
                                Event::NewPeer(peer_id) => {
                                    debug!("[state sync] new peer {}", peer_id);
                                    self.peer_manager.enable_peer(peer_id, network_senders[idx].clone());
                                    self.check_progress().await;
                                }
                                Event::LostPeer(peer_id) => {
                                    debug!("[state sync] lost peer {}", peer_id);
                                    self.peer_manager.disable_peer(&peer_id);
                                }
                                Event::Message((peer_id, mut message)) => {
                                    if message.has_chunk_request() {
                                        let known_version = message.get_chunk_request().get_known_version();
                                        if let Err(err) = self.process_chunk_request(peer_id, message.take_chunk_request()).await {
                                            error!("[state sync] failed to serve chunk request to {} with known version {}: {:?}", peer_id, known_version, err);
                                        }
                                    }
                                    if message.has_chunk_response() {
                                        if let Err(err) = self.process_chunk_response(&peer_id, message.take_chunk_response()).await {
                                            error!("[state sync] failed to process chunk response from {}: {:?}", peer_id, err);
                                        } else {
                                            self.peer_manager.update_score(&peer_id, PeerScoreUpdateType::Success);
                                        }
                                    }
                                }
                                _ => {}
                            }
                        },
                        Err(err) => { error!("[state sync] network error {:?}", err); },
                    }
                },
                _ = interval.select_next_some() => {
                    self.check_progress().await;
                }
            }
        }
    }
/// main routine. starts sync coordinator that listens for CoordinatorMsg
main routine. starts sync coordinator that listens for CoordinatorMsg
main routine . starts sync coordinator that listens for CoordinatorMsg
pub async fn start(mut self, network: Vec<(StateSynchronizerSender, StateSynchronizerEvents)>) {
        self.known_version = self
            .executor_proxy
            .get_latest_version()
            .await
            .expect("[start sync] failed to fetch latest version from storage");
        let mut interval =
            Interval::new_interval(Duration::from_millis(self.config.tick_interval_ms))
                .compat()
                .fuse();
        let network_senders: Vec<StateSynchronizerSender> =
            network.iter().map(|t| t.0.clone()).collect();
        let events: Vec<_> = network
            .into_iter()
            .enumerate()
            .map(|(idx, t)| t.1.map(move |e| (idx, e)))
            .collect();
        let mut network_events = select_all(events).fuse();
        loop {
            ::futures::select! {
                msg = self.client_events.select_next_some() => {
                    match msg {
                        CoordinatorMessage::Requested(target, subscription) => {
                            self.request_sync(target, subscription).await;
                        }
                        CoordinatorMessage::Commit(version) => {
                             self.commit(version).await;
                        }
                        CoordinatorMessage::GetState(callback) => {
                            self.get_state(callback);
                        }
                    };
                },
                (idx, network_event) = network_events.select_next_some() => {
                    match network_event {
                        Ok(event) => {
                            match event {
                                Event::NewPeer(peer_id) => {
                                    debug!("[state sync] new peer {}", peer_id);
                                    self.peer_manager.enable_peer(peer_id, network_senders[idx].clone());
                                    self.check_progress().await;
                                }
                                Event::LostPeer(peer_id) => {
                                    debug!("[state sync] lost peer {}", peer_id);
                                    self.peer_manager.disable_peer(&peer_id);
                                }
                                Event::Message((peer_id, mut message)) => {
                                    if message.has_chunk_request() {
                                        let known_version = message.get_chunk_request().get_known_version();
                                        if let Err(err) = self.process_chunk_request(peer_id, message.take_chunk_request()).await {
                                            error!("[state sync] failed to serve chunk request to {} with known version {}: {:?}", peer_id, known_version, err);
                                        }
                                    }
                                    if message.has_chunk_response() {
                                        if let Err(err) = self.process_chunk_response(&peer_id, message.take_chunk_response()).await {
                                            error!("[state sync] failed to process chunk response from {}: {:?}", peer_id, err);
                                        } else {
                                            self.peer_manager.update_score(&peer_id, PeerScoreUpdateType::Success);
                                        }
                                    }
                                }
                                _ => {}
                            }
                        },
                        Err(err) => { error!("[state sync] network error {:?}", err); },
                    }
                },
                _ = interval.select_next_some() => {
                    self.check_progress().await;
                }
            }
        }
    }
pub async fn start ( mut self , network : Vec < ( StateSynchronizerSender , StateSynchronizerEvents ) > ) { self . known_version = self . executor_proxy . get_latest_version ( ) . await . expect ( " " ) ; let mut interval = Interval :: new_interval ( Duration :: from_millis ( self . config . tick_interval_ms ) ) . compat ( ) . fuse ( ) ; let network_senders : Vec < StateSynchronizerSender > = network . iter ( ) . map ( | t | t . 0 . clone ( ) ) . collect ( ) ; let events : Vec < _ > = network . into_iter ( ) . enumerate ( ) . map ( | ( idx , t ) | t . 1 . map ( move | e | ( idx , e ) ) ) . collect ( ) ; let mut network_events = select_all ( events ) . fuse ( ) ; loop { :: futures :: select ! { msg self client_events select_next_some ( ) { match msg { CoordinatorMessage Requested ( target subscription ) { self request_sync ( target subscription ) await } CoordinatorMessage Commit ( version ) { self commit ( version ) await } CoordinatorMessage GetState ( callback ) { self get_state ( callback ) } } } ( idx network_event ) network_events select_next_some ( ) { match network_event { Ok ( event ) { match event { Event NewPeer ( peer_id ) { debug ( " " peer_id ) self peer_manager enable_peer ( peer_id network_senders [ idx ] clone ( ) ) self check_progress ( ) await } Event LostPeer ( peer_id ) { debug ( " " peer_id ) self peer_manager disable_peer ( peer_id ) } Event Message ( ( peer_id mut message ) ) { if message has_chunk_request ( ) { let known_version message get_chunk_request ( ) get_known_version ( ) if let Err ( err ) self process_chunk_request ( peer_id message take_chunk_request ( ) ) await { error ( " " peer_id known_version err ) } } if message has_chunk_response ( ) { if let Err ( err ) self process_chunk_response ( peer_id message take_chunk_response ( ) ) await { error ( " " peer_id err ) } else { self peer_manager update_score ( peer_id PeerScoreUpdateType Success ) } } } { } } } Err ( err ) { error ( " " err ) } } } interval select_next_some ( ) { self check_progress ( ) await } } } }
main routine.
main routine .
mut self
network
Vec<(StateSynchronizerSender, StateSynchronizerEvents)>