pub fn messages_for(&mut self, engine_id: ConsensusEngineId, topic: B::Hash)
		-> mpsc::UnboundedReceiver<Vec<u8>>
	{
		let (tx, rx) = mpsc::unbounded();

		let validator = match self.validators.get(&engine_id) {
			None => {
				self.live_message_sinks.entry((engine_id, topic)).or_default().push(tx);
				return rx;
			}
			Some(v) => v,
		};

		for entry in self.messages.iter_mut()
			.filter(|e| e.topic == topic && e.message.engine_id == engine_id)
		{
			let live = match entry.status {
				Status::Live => true,
				Status::Future => match validator.validate(&entry.message.data) {
					ValidationResult::Valid(_) => {
						entry.status = Status::Live;
						true
					}
					_ => {
						// don't send messages considered to be future still.
						// if messages are considered expired they'll be cleaned up when we
						// collect garbage.
						false
					}
				}
			};

			if live {
				entry.status = Status::Live;
				tx.unbounded_send(entry.message.data.clone())
					.expect("receiver known to be live; qed");
			}
		}

		self.live_message_sinks.entry((engine_id, topic)).or_default().push(tx);

		rx
	}
/// Get data of valid, incoming messages for a topic (but might have expired meanwhile)
Get data of valid, incoming messages for a topic (but might have expired meanwhile)
Get data of valid incoming messages for a topic ( but might have expired meanwhile )
pub fn messages_for(&mut self, engine_id: ConsensusEngineId, topic: B::Hash)
		-> mpsc::UnboundedReceiver<Vec<u8>>
	{
		let (tx, rx) = mpsc::unbounded();
		let validator = match self.validators.get(&engine_id) {
			None => {
				self.live_message_sinks.entry((engine_id, topic)).or_default().push(tx);
				return rx;
			}
			Some(v) => v,
		};
		for entry in self.messages.iter_mut()
			.filter(|e| e.topic == topic && e.message.engine_id == engine_id)
		{
			let live = match entry.status {
				Status::Live => true,
				Status::Future => match validator.validate(&entry.message.data) {
					ValidationResult::Valid(_) => {
						entry.status = Status::Live;
						true
					}
					_ => {
						// don't send messages considered to be future still.
						// if messages are considered expired they'll be cleaned up when we
						// collect garbage.
						false
					}
				}
			};
			if live {
				entry.status = Status::Live;
				tx.unbounded_send(entry.message.data.clone())
					.expect("receiver known to be live; qed");
			}
		}
		self.live_message_sinks.entry((engine_id, topic)).or_default().push(tx);
		rx
	}
pub fn messages_for ( & mut self , engine_id : ConsensusEngineId , topic : B :: Hash ) -> mpsc :: UnboundedReceiver < Vec < u8 > > { let ( tx , rx ) = mpsc :: unbounded ( ) ; let validator = match self . validators . get ( & engine_id ) { None => { self . live_message_sinks . entry ( ( engine_id , topic ) ) . or_default ( ) . push ( tx ) ; return rx ; } Some ( v ) => v , } ; for entry in self . messages . iter_mut ( ) . filter ( | e | e . topic == topic && e . message . engine_id == engine_id ) { let live = match entry . status { Status :: Live => true , Status :: Future => match validator . validate ( & entry . message . data ) { ValidationResult :: Valid ( _ ) => { entry . status = Status :: Live ; true } _ => { // don't send messages considered to be future still. // if messages are considered expired they'll be cleaned up when we // collect garbage. false } } } ; if live { entry . status = Status :: Live ; tx . unbounded_send ( entry . message . data . clone ( ) ) . expect ( " " ) ; } } self . live_message_sinks . entry ( ( engine_id , topic ) ) . or_default ( ) . push ( tx ) ; rx }
Get data of valid, incoming messages for a topic (but might have expired meanwhile)
Get data of valid incoming messages for a topic ( but might have expired meanwhile )
&mut self
engine_id
ConsensusEngineId
topic
B::Hash
&mut self
engine_id
ConsensusEngineId
topic
B::Hash