pub(crate) fn checked_commit_stream<Block: BlockT, S>(
	inner: S,
	voters: Arc<VoterSet<AuthorityId>>,
)
	-> impl Stream<Item=(u64, CompactCommit<Block>),Error=Error> where
	S: Stream<Item=Vec<u8>,Error=()>
{
	inner
		.filter_map(|raw| {
			// this could be optimized by decoding piecewise.
			let decoded = GossipMessage::<Block>::decode(&mut &raw[..]);
			if decoded.is_none() {
				trace!(target: "afg", "Skipping malformed commit message {:?}", raw);
			}
			decoded
		})
		.filter_map(move |msg| {
			match msg {
				GossipMessage::Commit(msg) => {
					let round = msg.round;
					let precommits_signed_by: Vec<String> =
						msg.message.auth_data.iter().map(move |(_, a)| {
							format!("{}", a)
						}).collect();
					telemetry!(CONSENSUS_INFO; "afg.received_commit";
						"contains_precommits_signed_by" => ?precommits_signed_by,
						"target_number" => ?msg.message.target_number,
						"target_hash" => ?msg.message.target_hash,
					);
					check_compact_commit::<Block>(msg.message, &*voters).map(move |c| (round, c))
				},
				_ => {
					debug!(target: "afg", "Skipping unknown message type");
					return None;
				}
			}
		})
		.map_err(|()| Error::Network(format!("Failed to receive message on unbounded stream")))
}
/// A stream for incoming commit messages. This checks all the signatures on the
/// messages.
A stream for incoming commit messages. This checks all the signatures on the
messages.
A stream for incoming commit messages . This checks all the signatures on the messages .
pub(crate) fn checked_commit_stream<Block: BlockT, S>(
	inner: S,
	voters: Arc<VoterSet<AuthorityId>>,
)
	-> impl Stream<Item=(u64, CompactCommit<Block>),Error=Error> where
	S: Stream<Item=Vec<u8>,Error=()>
{
	inner
		.filter_map(|raw| {
			// this could be optimized by decoding piecewise.
			let decoded = GossipMessage::<Block>::decode(&mut &raw[..]);
			if decoded.is_none() {
				trace!(target: "afg", "Skipping malformed commit message {:?}", raw);
			}
			decoded
		})
		.filter_map(move |msg| {
			match msg {
				GossipMessage::Commit(msg) => {
					let round = msg.round;
					let precommits_signed_by: Vec<String> =
						msg.message.auth_data.iter().map(move |(_, a)| {
							format!("{}", a)
						}).collect();
					telemetry!(CONSENSUS_INFO; "afg.received_commit";
						"contains_precommits_signed_by" => ?precommits_signed_by,
						"target_number" => ?msg.message.target_number,
						"target_hash" => ?msg.message.target_hash,
					);
					check_compact_commit::<Block>(msg.message, &*voters).map(move |c| (round, c))
				},
				_ => {
					debug!(target: "afg", "Skipping unknown message type");
					return None;
				}
			}
		})
		.map_err(|()| Error::Network(format!("Failed to receive message on unbounded stream")))
}
pub ( crate ) fn checked_commit_stream < Block : BlockT , S > ( inner : S , voters : Arc < VoterSet < AuthorityId > > , ) -> impl Stream < Item = ( u64 , CompactCommit < Block > ) , Error = Error > where S : Stream < Item = Vec < u8 > , Error = ( ) > { inner . filter_map ( | raw | { // this could be optimized by decoding piecewise. let decoded = GossipMessage :: < Block > :: decode ( & mut & raw [ .. ] ) ; if decoded . is_none ( ) { trace ! ( target " " " " raw ) ; } decoded } ) . filter_map ( move | msg | { match msg { GossipMessage :: Commit ( msg ) => { let round = msg . round ; let precommits_signed_by : Vec < String > = msg . message . auth_data . iter ( ) . map ( move | ( _ , a ) | { format ! ( " " a ) } ) . collect ( ) ; telemetry ! ( CONSENSUS_INFO " " " " precommits_signed_by " " msg message target_number " " msg message target_hash ) ; check_compact_commit :: < Block > ( msg . message , & * voters ) . map ( move | c | ( round , c ) ) } , _ => { debug ! ( target " " " " ) ; return None ; } } } ) . map_err ( | ( ) | Error :: Network ( format ! ( " " ) ) ) }
A stream for incoming commit messages.
A stream for incoming commit messages .
inner
S
voters
Arc<VoterSet<AuthorityId>>
inner
S
voters
Arc<VoterSet<AuthorityId>>