don't return async::notready for messages without scheduling wakeup

substrate/polkadot/consensus/src/service.rs

@@ -108,22 +108,22 @@ impl Stream for Messages {
 		}
 
 		// check the network
-		match self.network_stream.poll() {
+		loop {
-			Err(_) => Err(bft::InputStreamConcluded.into()),
+			match self.network_stream.poll() {
-			Ok(Async::NotReady) => Ok(Async::NotReady),
+				Err(_) => return Err(bft::InputStreamConcluded.into()),
-			Ok(Async::Ready(None)) => Ok(Async::NotReady), // the input stream for agreements is never meant to logically end.
+				Ok(Async::NotReady) => return Ok(Async::NotReady),
-			Ok(Async::Ready(Some(message))) => {
+				Ok(Async::Ready(None)) => return Ok(Async::NotReady), // the input stream for agreements is never meant to logically end.
-				if message.parent_hash == self.parent_hash {
+				Ok(Async::Ready(Some(message))) => {
-					match process_message(message, &self.authorities) {
+					if message.parent_hash == self.parent_hash {
-						Ok(message) => Ok(Async::Ready(Some(message))),
+						match process_message(message, &self.authorities) {
-						Err(e) => {
+							Ok(message) => return Ok(Async::Ready(Some(message))),
-							debug!("Message validation failed: {:?}", e);
+							Err(e) => {
-							Ok(Async::NotReady)
+								debug!("Message validation failed: {:?}", e);
+							}
 						}
+					} else {
+						self.collection.push(message);
 					}
-				} else {
-					self.collection.push(message);
-					Ok(Async::NotReady)
 				}
 			}
 		}