pub fn on_incoming(
		&mut self,
		protocol: &mut Context<B>,
		who: PeerId,
		message: ConsensusMessage,
	) -> Option<(B::Hash, ConsensusMessage)> {
		let message_hash = HashFor::<B>::hash(&message.data[..]);

		if self.known_messages.contains_key(&message_hash) {
			trace!(target:"gossip", "Ignored already known message from {}", who);
			return None;
		}

		if let Some(ref mut peer) = self.peers.get_mut(&who) {
			use std::collections::hash_map::Entry;

			let engine_id = message.engine_id;
			// validate the message
			let (topic, status) = match self.validators.get(&engine_id)
				.map(|v| v.validate(&message.data))
			{
				Some(ValidationResult::Valid(topic)) => (topic, Status::Live),
				Some(ValidationResult::Future(topic)) => (topic, Status::Future),
				Some(ValidationResult::Invalid) => {
					trace!(target:"gossip", "Invalid message from {}", who);
					protocol.report_peer(
						who,
						Severity::Bad(format!("Sent invalid consensus message")),
					);
					return None;
				},
				Some(ValidationResult::Expired) => {
					trace!(target:"gossip", "Ignored expired message from {}", who);
					return None;
				},
				None => {
					protocol.report_peer(
						who.clone(),
						Severity::Useless(format!("Sent unknown consensus engine id")),
					);
					trace!(target:"gossip", "Unknown message engine id {:?} from {}",
						engine_id, who);
					return None;
				},
			};

			peer.known_messages.insert(message_hash);
			if let Entry::Occupied(mut entry) = self.live_message_sinks.entry((engine_id, topic)) {
				debug!(target: "gossip", "Pushing consensus message to sinks for {}.", topic);
				entry.get_mut().retain(|sink| {
					if let Err(e) = sink.unbounded_send(message.data.clone()) {
						trace!(target:"gossip", "Error broadcasting message notification: {:?}", e);
					}
					!sink.is_closed()
				});
				if entry.get().is_empty() {
					entry.remove_entry();
				}
			}
			self.multicast_inner(protocol, message_hash, topic, status, || message.clone(), false);
			Some((topic, message))
		} else {
			trace!(target:"gossip", "Ignored statement from unregistered peer {}", who);
			None
		}
	}
/// Handle an incoming ConsensusMessage for topic by who via protocol. Discard message if topic
/// already known, the message is old, its source peers isn't a registered peer or the connection
/// to them is broken. Return `Some(topic, message)` if it was added to the internal queue, `None`
/// in all other cases.
Handle an incoming ConsensusMessage for topic by who via protocol. Discard message if topic
already known, the message is old, its source peers isn't a registered peer or the connection
to them is broken. Return `Some(topic, message)` if it was added to the internal queue, `None`
in all other cases.
Handle an incoming ConsensusMessage for topic by who via protocol . Discard message if topic already known the message is old its source peers isn ' t a registered peer or the connection to them is broken . Return ` Some ( topic message ) ` if it was added to the internal queue ` None ` in all other cases .
pub fn on_incoming(
		&mut self,
		protocol: &mut Context<B>,
		who: PeerId,
		message: ConsensusMessage,
	) -> Option<(B::Hash, ConsensusMessage)> {
		let message_hash = HashFor::<B>::hash(&message.data[..]);
		if self.known_messages.contains_key(&message_hash) {
			trace!(target:"gossip", "Ignored already known message from {}", who);
			return None;
		}
		if let Some(ref mut peer) = self.peers.get_mut(&who) {
			use std::collections::hash_map::Entry;
			let engine_id = message.engine_id;
			// validate the message
			let (topic, status) = match self.validators.get(&engine_id)
				.map(|v| v.validate(&message.data))
			{
				Some(ValidationResult::Valid(topic)) => (topic, Status::Live),
				Some(ValidationResult::Future(topic)) => (topic, Status::Future),
				Some(ValidationResult::Invalid) => {
					trace!(target:"gossip", "Invalid message from {}", who);
					protocol.report_peer(
						who,
						Severity::Bad(format!("Sent invalid consensus message")),
					);
					return None;
				},
				Some(ValidationResult::Expired) => {
					trace!(target:"gossip", "Ignored expired message from {}", who);
					return None;
				},
				None => {
					protocol.report_peer(
						who.clone(),
						Severity::Useless(format!("Sent unknown consensus engine id")),
					);
					trace!(target:"gossip", "Unknown message engine id {:?} from {}",
						engine_id, who);
					return None;
				},
			};
			peer.known_messages.insert(message_hash);
			if let Entry::Occupied(mut entry) = self.live_message_sinks.entry((engine_id, topic)) {
				debug!(target: "gossip", "Pushing consensus message to sinks for {}.", topic);
				entry.get_mut().retain(|sink| {
					if let Err(e) = sink.unbounded_send(message.data.clone()) {
						trace!(target:"gossip", "Error broadcasting message notification: {:?}", e);
					}
					!sink.is_closed()
				});
				if entry.get().is_empty() {
					entry.remove_entry();
				}
			}
			self.multicast_inner(protocol, message_hash, topic, status, || message.clone(), false);
			Some((topic, message))
		} else {
			trace!(target:"gossip", "Ignored statement from unregistered peer {}", who);
			None
		}
	}
pub fn on_incoming ( & mut self , protocol : & mut Context < B > , who : PeerId , message : ConsensusMessage , ) -> Option < ( B :: Hash , ConsensusMessage ) > { let message_hash = HashFor :: < B > :: hash ( & message . data [ .. ] ) ; if self . known_messages . contains_key ( & message_hash ) { trace ! ( target " " " " who ) ; return None ; } if let Some ( ref mut peer ) = self . peers . get_mut ( & who ) { use std :: collections :: hash_map :: Entry ; let engine_id = message . engine_id ; // validate the message let ( topic , status ) = match self . validators . get ( & engine_id ) . map ( | v | v . validate ( & message . data ) ) { Some ( ValidationResult :: Valid ( topic ) ) => ( topic , Status :: Live ) , Some ( ValidationResult :: Future ( topic ) ) => ( topic , Status :: Future ) , Some ( ValidationResult :: Invalid ) => { trace ! ( target " " " " who ) ; protocol . report_peer ( who , Severity :: Bad ( format ! ( " " ) ) , ) ; return None ; } , Some ( ValidationResult :: Expired ) => { trace ! ( target " " " " who ) ; return None ; } , None => { protocol . report_peer ( who . clone ( ) , Severity :: Useless ( format ! ( " " ) ) , ) ; trace ! ( target " " " " engine_id who ) ; return None ; } , } ; peer . known_messages . insert ( message_hash ) ; if let Entry :: Occupied ( mut entry ) = self . live_message_sinks . entry ( ( engine_id , topic ) ) { debug ! ( target " " " " topic ) ; entry . get_mut ( ) . retain ( | sink | { if let Err ( e ) = sink . unbounded_send ( message . data . clone ( ) ) { trace ! ( target " " " " e ) ; } ! sink . is_closed ( ) } ) ; if entry . get ( ) . is_empty ( ) { entry . remove_entry ( ) ; } } self . multicast_inner ( protocol , message_hash , topic , status , | | message . clone ( ) , false ) ; Some ( ( topic , message ) ) } else { trace ! ( target " " " " who ) ; None } }
Handle an incoming ConsensusMessage for topic by who via protocol.
Handle an incoming ConsensusMessage for topic by who via protocol .
&mut self
protocol
&mut Context<B>
who
PeerId
message
ConsensusMessage
&mut self
protocol
&mut Context<B>
who
PeerId
message
ConsensusMessage