pub async fn start(mut self) {
        self.known_version = self
            .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();

        loop {
            ::futures::select! {
                msg = self.consensus_events.select_next_some() => {
                    match msg {
                        CoordinatorMessage::Requested(target, subscription) => {
                            self.consensus_sync(target, subscription).await;
                        }
                        CoordinatorMessage::Commit(version) => {
                             self.handle_commit(version);
                        }
                        CoordinatorMessage::GetState(callback) => {
                            self.get_state(callback);
                        }
                    };
                },
                network_event = self.network_events.select_next_some() => {
                    match network_event {
                        Ok(event) => {
                            match event {
                                Event::NewPeer(peer_id) => {
                                    if self.autosync {
                                        self.peers.insert(peer_id, ());
                                    }
                                }
                                Event::LostPeer(peer_id) => {
                                    self.peers.remove(&peer_id);
                                }
                                Event::Message((peer_id, mut message)) => {
                                    if message.has_chunk_request() {
                                        self.process_chunk_request(peer_id, message.take_chunk_request()).await;
                                    }
                                    if message.has_chunk_response() {
                                        self.process_chunk_response(message.take_chunk_response()).await;
                                    }
                                }
                                _ => {}
                            }
                        },
                        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) {
        self.known_version = self
            .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();
        loop {
            ::futures::select! {
                msg = self.consensus_events.select_next_some() => {
                    match msg {
                        CoordinatorMessage::Requested(target, subscription) => {
                            self.consensus_sync(target, subscription).await;
                        }
                        CoordinatorMessage::Commit(version) => {
                             self.handle_commit(version);
                        }
                        CoordinatorMessage::GetState(callback) => {
                            self.get_state(callback);
                        }
                    };
                },
                network_event = self.network_events.select_next_some() => {
                    match network_event {
                        Ok(event) => {
                            match event {
                                Event::NewPeer(peer_id) => {
                                    if self.autosync {
                                        self.peers.insert(peer_id, ());
                                    }
                                }
                                Event::LostPeer(peer_id) => {
                                    self.peers.remove(&peer_id);
                                }
                                Event::Message((peer_id, mut message)) => {
                                    if message.has_chunk_request() {
                                        self.process_chunk_request(peer_id, message.take_chunk_request()).await;
                                    }
                                    if message.has_chunk_response() {
                                        self.process_chunk_response(message.take_chunk_response()).await;
                                    }
                                }
                                _ => {}
                            }
                        },
                        Err(err) => { error!("[state sync] network error {:?}", err); },
                    }
                },
                _ = interval.select_next_some() => {
                    self.check_progress().await;
                }
            }
        }
    }
pub async fn start ( mut self ) { self . known_version = self . get_latest_version ( ) . await . expect ( " " ) ; let mut interval = Interval :: new_interval ( Duration :: from_millis ( self . config . tick_interval_ms ) ) . compat ( ) . fuse ( ) ; loop { :: futures :: select ! { msg self consensus_events select_next_some ( ) { match msg { CoordinatorMessage Requested ( target subscription ) { self consensus_sync ( target subscription ) await } CoordinatorMessage Commit ( version ) { self handle_commit ( version ) } CoordinatorMessage GetState ( callback ) { self get_state ( callback ) } } } network_event self network_events select_next_some ( ) { match network_event { Ok ( event ) { match event { Event NewPeer ( peer_id ) { if self autosync { self peers insert ( peer_id ( ) ) } } Event LostPeer ( peer_id ) { self peers remove ( peer_id ) } Event Message ( ( peer_id mut message ) ) { if message has_chunk_request ( ) { self process_chunk_request ( peer_id message take_chunk_request ( ) ) await } if message has_chunk_response ( ) { self process_chunk_response ( message take_chunk_response ( ) ) await } } { } } } Err ( err ) { error ( " " err ) } } } interval select_next_some ( ) { self check_progress ( ) await } } } }
main routine.
main routine .
mut self
mut self