/// Process all incoming messages from other nodes.
///
/// Commit messages are handled with extra care. If a commit message references
/// a currently backgrounded round, we send it to that round so that when we commit
/// on that round, our commit message will be informed by those that we've seen.
///
/// Otherwise, we will simply handle the commit and issue a finalization command
/// to the environment.
Process all incoming messages from other nodes.
Commit messages are handled with extra care. If a commit message references
a currently backgrounded round, we send it to that round so that when we commit
on that round, our commit message will be informed by those that we've seen.

Otherwise, we will simply handle the commit and issue a finalization command
to the environment.
Process all incoming messages from other nodes . Commit messages are handled with extra care . If a commit message references a currently backgrounded round we send it to that round so that when we commit on that round our commit message will be informed by those that we ' ve seen . Otherwise we will simply handle the commit and issue a finalization command to the environment .
fn process_incoming(&mut self, cx: &mut Context) -> Result<(), E::Error> {
		while let Poll::Ready(Some(item)) = Stream::poll_next(Pin::new(&mut self.global_in), cx) {
			match item? {
				CommunicationIn::Commit(round_number, commit, mut process_commit_outcome) => {
					trace!(target: "afg", "Got commit for round_number {:?}: target_number: {:?}, target_hash: {:?}",
						round_number,
						commit.target_number,
						commit.target_hash,
					);
					let commit: Commit<_, _, _, _> = commit.into();
					let mut inner = self.inner.lock();
					// if the commit is for a background round dispatch to round committer.
					// that returns Some if there wasn't one.
					if let Some(commit) = inner.past_rounds.import_commit(round_number, commit) {
						// otherwise validate the commit and signal the finalized block
						// (if any) to the environment
						let validation_result = validate_commit(&commit, &self.voters, &*self.env)?;
						if let Some((finalized_hash, finalized_number)) = validation_result.ghost {
							// this can't be moved to a function because the compiler
							// will complain about getting two mutable borrows to self
							// (due to the call to `self.rounds.get_mut`).
							let last_finalized_number = &mut self.last_finalized_number;
							// clean up any background rounds
							inner.past_rounds.update_finalized(finalized_number);
							if finalized_number > *last_finalized_number {
								*last_finalized_number = finalized_number;
								self.env.finalize_block(
									finalized_hash,
									finalized_number,
									round_number,
									commit,
								)?;
							}
							process_commit_outcome
								.run(CommitProcessingOutcome::Good(GoodCommit::new()));
						} else {
							// Failing validation of a commit is bad.
							process_commit_outcome.run(CommitProcessingOutcome::Bad(
								BadCommit::from(validation_result),
							));
						}
					} else {
						// Import to backgrounded round is good.
						process_commit_outcome
							.run(CommitProcessingOutcome::Good(GoodCommit::new()));
					}
				},
				CommunicationIn::CatchUp(catch_up, mut process_catch_up_outcome) => {
					trace!(target: "afg", "Got catch-up message for round {}", catch_up.round_number);
					let mut inner = self.inner.lock();
					let round = if let Some(round) = validate_catch_up(
						catch_up,
						&*self.env,
						&self.voters,
						inner.best_round.round_number(),
					) {
						round
					} else {
						process_catch_up_outcome
							.run(CatchUpProcessingOutcome::Bad(BadCatchUp::new()));
						return Ok(())
					};
					let state = round.state();
					// beyond this point, we set this round to the past and
					// start voting in the next round.
					let mut just_completed = VotingRound::completed(
						round,
						inner.best_round.finalized_sender(),
						None,
						self.env.clone(),
					);
					let new_best = VotingRound::new(
						just_completed.round_number() + 1,
						self.voters.clone(),
						self.last_finalized_in_rounds.clone(),
						Some(just_completed.bridge_state()),
						inner.best_round.finalized_sender(),
						self.env.clone(),
					);
					// update last-finalized in rounds _after_ starting new round.
					// otherwise the base could be too eagerly set forward.
					if let Some((f_hash, f_num)) = state.finalized.clone() {
						if f_num > self.last_finalized_in_rounds.1 {
							self.last_finalized_in_rounds = (f_hash, f_num);
						}
					}
					self.env.completed(
						just_completed.round_number(),
						just_completed.round_state(),
						just_completed.dag_base(),
						just_completed.historical_votes(),
					)?;
					inner.past_rounds.push(&*self.env, just_completed);
					let old_best = std::mem::replace(&mut inner.best_round, new_best);
					inner.past_rounds.push(&*self.env, old_best);
					process_catch_up_outcome
						.run(CatchUpProcessingOutcome::Good(GoodCatchUp::new()));
				},
			}
		}
		Ok(())
	}
fn process_incoming ( & mut self , cx : & mut Context ) -> Result < ( ) , E :: Error > { while let Poll :: Ready ( Some ( item ) ) = Stream :: poll_next ( Pin :: new ( & mut self . global_in ) , cx ) { match item ? { CommunicationIn :: Commit ( round_number , commit , mut process_commit_outcome ) => { trace ! ( target " " " " round_number commit target_number commit target_hash ) ; let commit : Commit < _ , _ , _ , _ > = commit . into ( ) ; let mut inner = self . inner . lock ( ) ; // if the commit is for a background round dispatch to round committer. // that returns Some if there wasn't one. if let Some ( commit ) = inner . past_rounds . import_commit ( round_number , commit ) { // otherwise validate the commit and signal the finalized block // (if any) to the environment let validation_result = validate_commit ( & commit , & self . voters , & * self . env ) ? ; if let Some ( ( finalized_hash , finalized_number ) ) = validation_result . ghost { // this can't be moved to a function because the compiler // will complain about getting two mutable borrows to self // (due to the call to `self.rounds.get_mut`). let last_finalized_number = & mut self . last_finalized_number ; // clean up any background rounds inner . past_rounds . update_finalized ( finalized_number ) ; if finalized_number > * last_finalized_number { * last_finalized_number = finalized_number ; self . env . finalize_block ( finalized_hash , finalized_number , round_number , commit , ) ? ; } process_commit_outcome . run ( CommitProcessingOutcome :: Good ( GoodCommit :: new ( ) ) ) ; } else { // Failing validation of a commit is bad. process_commit_outcome . run ( CommitProcessingOutcome :: Bad ( BadCommit :: from ( validation_result ) , ) ) ; } } else { // Import to backgrounded round is good. process_commit_outcome . run ( CommitProcessingOutcome :: Good ( GoodCommit :: new ( ) ) ) ; } } , CommunicationIn :: CatchUp ( catch_up , mut process_catch_up_outcome ) => { trace ! ( target " " " " catch_up round_number ) ; let mut inner = self . inner . lock ( ) ; let round = if let Some ( round ) = validate_catch_up ( catch_up , & * self . env , & self . voters , inner . best_round . round_number ( ) , ) { round } else { process_catch_up_outcome . run ( CatchUpProcessingOutcome :: Bad ( BadCatchUp :: new ( ) ) ) ; return Ok ( ( ) ) } ; let state = round . state ( ) ; // beyond this point, we set this round to the past and // start voting in the next round. let mut just_completed = VotingRound :: completed ( round , inner . best_round . finalized_sender ( ) , None , self . env . clone ( ) , ) ; let new_best = VotingRound :: new ( just_completed . round_number ( ) + 1 , self . voters . clone ( ) , self . last_finalized_in_rounds . clone ( ) , Some ( just_completed . bridge_state ( ) ) , inner . best_round . finalized_sender ( ) , self . env . clone ( ) , ) ; // update last-finalized in rounds _after_ starting new round. // otherwise the base could be too eagerly set forward. if let Some ( ( f_hash , f_num ) ) = state . finalized . clone ( ) { if f_num > self . last_finalized_in_rounds . 1 { self . last_finalized_in_rounds = ( f_hash , f_num ) ; } } self . env . completed ( just_completed . round_number ( ) , just_completed . round_state ( ) , just_completed . dag_base ( ) , just_completed . historical_votes ( ) , ) ? ; inner . past_rounds . push ( & * self . env , just_completed ) ; let old_best = std :: mem :: replace ( & mut inner . best_round , new_best ) ; inner . past_rounds . push ( & * self . env , old_best ) ; process_catch_up_outcome . run ( CatchUpProcessingOutcome :: Good ( GoodCatchUp :: new ( ) ) ) ; } , } } Ok ( ( ) ) }
Process all incoming messages from other nodes.
Process all incoming messages from other nodes .
&mut self
cx
&mut Context