core/finality-grandpa: Minor refactorings (#3825)

* core/finality-grandpa: Improve code comments

* core/finality-grandpa: Rename VoteOrPrecommit to PrevoteOrPrecommit

According to the Grandpa paper [1]:

> A vote is a block hash, together with some metadata such as round
number and the type of vote, such as prevote or precommit, all signed
with a voter’s private key.

To reduce confusion this patch makes the code consistent with the
research paper.

[1] https://github.com/w3f/consensus/blob/master/pdf/grandpa.pdf

* core/finality-grandpa: Add comment for NetworkStream concept

* core/finality-grandpa: Improve round_communication doc comment

* core/finality-grandpa: Rename PrevoteOrPrecommit to Vote

* core/finality-grandpa: Represent NetworkStream state machine as enum

* core/finality-grandpa: Improve KeepTopics comment

substrate/core/finality-grandpa/src/communication/mod.rs

@@ -48,7 +48,7 @@ use crate::{
 };
 use crate::environment::HasVoted;
 use gossip::{
-	GossipMessage, FullCatchUpMessage, FullCommitMessage, VoteOrPrecommitMessage, GossipValidator
+	GossipMessage, FullCatchUpMessage, FullCommitMessage, VoteMessage, GossipValidator
 };
 use fg_primitives::{
 	AuthorityPair, AuthorityId, AuthoritySignature, SetId as SetIdNumber, RoundNumber,
@@ -148,12 +148,21 @@ impl<B, S, H> Network<B> for Arc<NetworkService<B, S, H>> where
 	type In = NetworkStream;
 
 	fn messages_for(&self, topic: B::Hash) -> Self::In {
+		// Given that one can only communicate with the Substrate network via the `NetworkService` via message-passing,
+		// and given that methods on the network consensus gossip are not exposed but only reachable by passing a
+		// closure into `with_gossip` on the `NetworkService` this function needs to make use of the `NetworkStream`
+		// construction.
+		//
+		// We create a oneshot channel and pass the sender within a closure to the network. At some point in the future
+		// the network passes the message channel back through the oneshot channel. But the consumer of this function
+		// expects a stream, not a stream within a oneshot. This complexity is abstracted within `NetworkStream`,
+		// waiting for the oneshot to resolve and from there on acting like a normal message channel.
 		let (tx, rx) = oneshot::channel();
 		self.with_gossip(move |gossip, _| {
 			let inner_rx = gossip.messages_for(GRANDPA_ENGINE_ID, topic);
 			let _ = tx.send(inner_rx);
 		});
-		NetworkStream { outer: rx, inner: None }
+		NetworkStream::PollingOneshot(rx)
 	}
 
 	fn register_validator(&self, validator: Arc<dyn network_gossip::Validator<B>>) {
@@ -202,10 +211,18 @@ impl<B, S, H> Network<B> for Arc<NetworkService<B, S, H>> where
 	}
 }
 
-/// A stream used by NetworkBridge in its implementation of Network.
-pub struct NetworkStream {
-	inner: Option<mpsc::UnboundedReceiver<network_gossip::TopicNotification>>,
-	outer: oneshot::Receiver<mpsc::UnboundedReceiver<network_gossip::TopicNotification>>
+/// A stream used by NetworkBridge in its implementation of Network. Given a oneshot that eventually returns a channel
+/// which eventually returns messages, instead of:
+///
+/// 1. polling the oneshot until it returns a message channel
+///
+/// 2. polling the message channel for messages
+///
+/// `NetworkStream` combines the two steps into one, requiring a consumer to only poll `NetworkStream` to retrieve
+/// messages directly.
+pub enum NetworkStream {
+	PollingOneshot(oneshot::Receiver<mpsc::UnboundedReceiver<network_gossip::TopicNotification>>),
+	PollingTopicNotifications(mpsc::UnboundedReceiver<network_gossip::TopicNotification>),
 }
 
 impl Stream for NetworkStream {
@@ -213,17 +230,21 @@ impl Stream for NetworkStream {
 	type Error = ();
 
 	fn poll(&mut self) -> Poll<Option<Self::Item>, Self::Error> {
-		if let Some(ref mut inner) = self.inner {
-			return inner.poll();
-		}
-		match self.outer.poll() {
-			Ok(futures::Async::Ready(mut inner)) => {
-				let poll_result = inner.poll();
-				self.inner = Some(inner);
-				poll_result
+		match self {
+			NetworkStream::PollingOneshot(oneshot) => {
+				match oneshot.poll() {
+					Ok(futures::Async::Ready(mut stream)) => {
+						let poll_result = stream.poll();
+						*self = NetworkStream::PollingTopicNotifications(stream);
+						poll_result
+					},
+					Ok(futures::Async::NotReady) => Ok(futures::Async::NotReady),
+					Err(_) => Err(())
+				}
+			},
+			NetworkStream::PollingTopicNotifications(stream) => {
+				stream.poll()
 			},
-			Ok(futures::Async::NotReady) => Ok(futures::Async::NotReady),
-			Err(_) => Err(())
 		}
 	}
 }
@@ -275,8 +296,8 @@ impl<B: BlockT, N: Network<B>> NetworkBridge<B, N> {
 				validator.note_round(Round(round.number), |_, _| {});
 
 				for signed in round.votes.iter() {
-					let message = gossip::GossipMessage::VoteOrPrecommit(
-						gossip::VoteOrPrecommitMessage::<B> {
+					let message = gossip::GossipMessage::Vote(
+						gossip::VoteMessage::<B> {
 							message: signed.clone(),
 							round: Round(round.number),
 							set_id: SetId(set_id),
@@ -341,7 +362,8 @@ impl<B: BlockT, N: Network<B>> NetworkBridge<B, N> {
 		);
 	}
 
-	/// Get the round messages for a round in the current set ID. These are signature-checked.
+	/// Get a stream of signature-checked round messages from the network as well as a sink for round messages to the
+	/// network all within the current set.
 	pub(crate) fn round_communication(
 		&self,
 		round: Round,
@@ -379,7 +401,7 @@ impl<B: BlockT, N: Network<B>> NetworkBridge<B, N> {
 			})
 			.and_then(move |msg| {
 				match msg {
-					GossipMessage::VoteOrPrecommit(msg) => {
+					GossipMessage::Vote(msg) => {
 						// check signature.
 						if !voters.contains_key(&msg.message.id) {
 							debug!(target: "afg", "Skipping message from unknown voter {}", msg.message.id);
@@ -707,7 +729,7 @@ impl<Block: BlockT, N: Network<Block>> Sink for OutgoingMessages<Block, N>
 				id: local_id.clone(),
 			};
 
-			let message = GossipMessage::VoteOrPrecommit(VoteOrPrecommitMessage::<Block> {
+			let message = GossipMessage::Vote(VoteMessage::<Block> {
 				message: signed.clone(),
 				round: Round(self.round),
 				set_id: SetId(self.set_id),