pub(crate) fn checked_message_stream<Block: BlockT, S>(
	inner: S,
	voters: Arc<VoterSet<AuthorityId>>,
)
	-> impl Stream<Item=SignedMessage<Block>,Error=Error> where
	S: Stream<Item=Vec<u8>,Error=()>
{
	inner
		.filter_map(|raw| {
			let decoded = GossipMessage::<Block>::decode(&mut &raw[..]);
			if decoded.is_none() {
				debug!(target: "afg", "Skipping malformed message {:?}", raw);
			}
			decoded
		})
		.and_then(move |msg| {
			match msg {
				GossipMessage::VoteOrPrecommit(msg) => {
					// check signature.
					if !voters.contains_key(&msg.message.id) {
						debug!(target: "afg", "Skipping message from unknown voter {}", msg.message.id);
						return Ok(None);
					}

					match &msg.message.message {
						Prevote(prevote) => {
							telemetry!(CONSENSUS_INFO; "afg.received_prevote";
								"voter" => ?format!("{}", msg.message.id),
								"target_number" => ?prevote.target_number,
								"target_hash" => ?prevote.target_hash,
							);
						},
						Precommit(precommit) => {
							telemetry!(CONSENSUS_INFO; "afg.received_precommit";
								"voter" => ?format!("{}", msg.message.id),
								"target_number" => ?precommit.target_number,
								"target_hash" => ?precommit.target_hash,
							);
						},
					};

					Ok(Some(msg.message))
				}
				_ => {
					debug!(target: "afg", "Skipping unknown message type");
					return Ok(None);
				}
			}
		})
		.filter_map(|x| x)
		.map_err(|()| Error::Network(format!("Failed to receive message on unbounded stream")))
}
/// converts a message stream into a stream of signed messages.
/// the output stream checks signatures also.
converts a message stream into a stream of signed messages.
the output stream checks signatures also.
converts a message stream into a stream of signed messages . the output stream checks signatures also .
pub(crate) fn checked_message_stream<Block: BlockT, S>(
	inner: S,
	voters: Arc<VoterSet<AuthorityId>>,
)
	-> impl Stream<Item=SignedMessage<Block>,Error=Error> where
	S: Stream<Item=Vec<u8>,Error=()>
{
	inner
		.filter_map(|raw| {
			let decoded = GossipMessage::<Block>::decode(&mut &raw[..]);
			if decoded.is_none() {
				debug!(target: "afg", "Skipping malformed message {:?}", raw);
			}
			decoded
		})
		.and_then(move |msg| {
			match msg {
				GossipMessage::VoteOrPrecommit(msg) => {
					// check signature.
					if !voters.contains_key(&msg.message.id) {
						debug!(target: "afg", "Skipping message from unknown voter {}", msg.message.id);
						return Ok(None);
					}
					match &msg.message.message {
						Prevote(prevote) => {
							telemetry!(CONSENSUS_INFO; "afg.received_prevote";
								"voter" => ?format!("{}", msg.message.id),
								"target_number" => ?prevote.target_number,
								"target_hash" => ?prevote.target_hash,
							);
						},
						Precommit(precommit) => {
							telemetry!(CONSENSUS_INFO; "afg.received_precommit";
								"voter" => ?format!("{}", msg.message.id),
								"target_number" => ?precommit.target_number,
								"target_hash" => ?precommit.target_hash,
							);
						},
					};
					Ok(Some(msg.message))
				}
				_ => {
					debug!(target: "afg", "Skipping unknown message type");
					return Ok(None);
				}
			}
		})
		.filter_map(|x| x)
		.map_err(|()| Error::Network(format!("Failed to receive message on unbounded stream")))
}
pub ( crate ) fn checked_message_stream < Block : BlockT , S > ( inner : S , voters : Arc < VoterSet < AuthorityId > > , ) -> impl Stream < Item = SignedMessage < Block > , Error = Error > where S : Stream < Item = Vec < u8 > , Error = ( ) > { inner . filter_map ( | raw | { let decoded = GossipMessage :: < Block > :: decode ( & mut & raw [ .. ] ) ; if decoded . is_none ( ) { debug ! ( target " " " " raw ) ; } decoded } ) . and_then ( move | msg | { match msg { GossipMessage :: VoteOrPrecommit ( msg ) => { // check signature. if ! voters . contains_key ( & msg . message . id ) { debug ! ( target " " " " msg message id ) ; return Ok ( None ) ; } match & msg . message . message { Prevote ( prevote ) => { telemetry ! ( CONSENSUS_INFO " " " " format ( " " msg message id ) " " prevote target_number " " prevote target_hash ) ; } , Precommit ( precommit ) => { telemetry ! ( CONSENSUS_INFO " " " " format ( " " msg message id ) " " precommit target_number " " precommit target_hash ) ; } , } ; Ok ( Some ( msg . message ) ) } _ => { debug ! ( target " " " " ) ; return Ok ( None ) ; } } } ) . filter_map ( | x | x ) . map_err ( | ( ) | Error :: Network ( format ! ( " " ) ) ) }
converts a message stream into a stream of signed messages.
converts a message stream into a stream of signed messages .
inner
S
voters
Arc<VoterSet<AuthorityId>>
inner
S
voters
Arc<VoterSet<AuthorityId>>