Split justification subscription into sender and receiver halves

client/finality-grandpa/src/environment.rs

@@ -51,7 +51,7 @@ use sp_consensus::SelectChain;
 use crate::authorities::{AuthoritySet, SharedAuthoritySet};
 use crate::communication::Network as NetworkT;
 use crate::consensus_changes::SharedConsensusChanges;
-use crate::finality_proof::{SharedFinalityNotifiers, JustificationNotification};
+use crate::finality_proof::{GrandpaJustificationSender, JustificationNotification};
 use crate::justification::GrandpaJustification;
 use crate::until_imported::UntilVoteTargetImported;
 use crate::voting_rule::VotingRule;
@@ -401,7 +401,7 @@ pub(crate) struct Environment<Backend, Block: BlockT, C, N: NetworkT<Block>, SC,
 	pub(crate) voter_set_state: SharedVoterSetState<Block>,
 	pub(crate) voting_rule: VR,
 	pub(crate) metrics: Option<Metrics>,
-	pub(crate) finality_notifiers: Option<SharedFinalityNotifiers<Block>>,
+	pub(crate) finality_subscription: Option<GrandpaJustificationSender<Block>>,
 	pub(crate) _phantom: PhantomData<Backend>,
 }
 
@@ -913,7 +913,7 @@ where
 			number,
 			(round, commit).into(),
 			false,
-			self.finality_notifiers.clone(),
+			&self.finality_subscription,
 		)
 	}
 
@@ -974,7 +974,7 @@ pub(crate) fn finalize_block<BE, Block, Client>(
 	number: NumberFor<Block>,
 	justification_or_commit: JustificationOrCommit<Block>,
 	initial_sync: bool,
-	finality_notifiers: Option<SharedFinalityNotifiers<Block>>,
+	finality_subscription: &Option<GrandpaJustificationSender<Block>>,
 ) -> Result<(), CommandOrError<Block::Hash, NumberFor<Block>>> where
 	Block:  BlockT,
 	BE: Backend<Block>,
@@ -1081,7 +1081,7 @@ pub(crate) fn finalize_block<BE, Block, Client>(
 			},
 		};
 
-		if let Some(notifiers) = finality_notifiers {
+		if let Some(subscription) = finality_subscription {
 			// Q: We `finalized()` this at L37, so can I be sure
 			// that it's fine to unwrap here?
 			if let Some(justification) = justification.clone() {
@@ -1092,10 +1092,7 @@ pub(crate) fn finalize_block<BE, Block, Client>(
 					justification,
 				};
 
-				for s in notifiers.read().iter() {
-					// TODO: Deal with Result
-					s.unbounded_send(notification.clone());
-				}
+                                subscription.notify(notification);
 			}
 		}
 

client/finality-grandpa/src/finality_proof.rs

@@ -147,7 +147,6 @@ impl<Block: BlockT> AuthoritySetForFinalityChecker<Block> for Arc<dyn FetchCheck
 	}
 }
 
-
 /// Justification for a finalized block.
 #[derive(Clone)]
 pub struct JustificationNotification<Block: BlockT> {
@@ -158,46 +157,76 @@ pub struct JustificationNotification<Block: BlockT> {
 }
 
 type JustificationStream<Block> = TracingUnboundedReceiver<JustificationNotification<Block>>;
-pub type SharedFinalitySubscribers<T> = Arc<RwLock<Vec<JustificationStream<T>>>>;
-
 type FinalityNotifier<T> = TracingUnboundedSender<JustificationNotification<T>>;
-pub type SharedFinalityNotifiers<T> = Arc<RwLock<Vec<FinalityNotifier<T>>>>;
+type SharedFinalityNotifiers<T> = Arc<RwLock<Vec<FinalityNotifier<T>>>>;
 
-pub struct FinalityProofSubscription<Block: BlockT> {
-	notifiers: SharedFinalityNotifiers<Block>,
+/// The sending half of the Grandpa justification channel.
+///
+/// Used to send notifictions about justifications generated
+/// at the end of a Grandpa round.
+#[derive(Clone)]
+pub struct GrandpaJustificationSender<Block: BlockT> {
+	notifiers: SharedFinalityNotifiers<Block>
 }
 
-impl<Block: BlockT> FinalityProofSubscription<Block> {
-	pub fn new() -> Self {
+impl<Block: BlockT> GrandpaJustificationSender<Block> {
+	/// The `notifiers` should be shared with a corresponding
+	/// `GrandpaJustificationReceiver`.
+	pub fn new(notifiers: SharedFinalityNotifiers<Block>) -> Self {
 		Self {
-			notifiers: Arc::new(RwLock::new(vec![])),
+			notifiers,
 		}
 	}
 
-	pub fn notifiers(&self) -> SharedFinalityNotifiers<Block> {
-		self.notifiers.clone()
-	}
+	/// Send out a notification to all subscribers that a new justification
+	/// is available for a block.
+	pub fn notify(&self, notification: JustificationNotification<Block>) -> Result<(), ()> {
+		{
+			// Clean up any closed sinks
+			self.notifiers.write().retain(|n| !n.is_closed());
+		}
 
-	// Will notify subsribers (receivers)
-	pub fn notify(&self, notification: JustificationNotification<Block>) {
-		// TODO: Look at `notify_justified()` in `client`
 		for s in self.notifiers.read().iter() {
 			s.unbounded_send(notification.clone());
 		}
 
-		todo!()
+		Ok(())
+	}
+}
+
+/// The receiving half of the Grandpa justification channel.
+///
+/// Used to recieve notifictions about justifications generated
+/// at the end of a Grandpa round.
+#[derive(Clone)]
+pub struct GrandpaJustificationReceiver<Block: BlockT> {
+	notifiers: SharedFinalityNotifiers<Block>
+}
+
+impl<Block: BlockT> GrandpaJustificationReceiver<Block> {
+	/// Crate a new receiver of justification notifications.
+	///
+	/// The `notifiers` should be shared with a corresponding
+	/// `GrandpaJustificationSender`.
+	pub fn new(notifiers: SharedFinalityNotifiers<Block>) -> Self {
+		Self {
+			notifiers,
+		}
 	}
 
-	pub fn stream(&self) -> JustificationStream<Block> {
-		// TODO: I only want one channel to be created per instance
-		let (sink, stream) = tracing_unbounded("mpsc_justification_notification_stream");
-		self.notifiers.write().push(sink);
-		stream
+	/// Subscribe to a channel through which justifications are sent
+	/// at the end of each Grandpa voting round.
+	pub fn subscribe(&self) -> JustificationStream<Block> {
+		let (sender, receiver) = tracing_unbounded("mpsc_justification_notification_stream");
+		self.notifiers.write().push(sender);
+		receiver
 	}
 }
 
+
+
 /// Finality proof provider for serving network requests.
-pub struct FinalityProofProvider<B,  Block: BlockT> {
+pub struct FinalityProofProvider<B, Block: BlockT> {
 	backend: Arc<B>,
 	authority_provider: Arc<dyn AuthoritySetForFinalityProver<Block>>,
 }

client/finality-grandpa/src/import.rs

@@ -42,7 +42,7 @@ use crate::authorities::{AuthoritySet, SharedAuthoritySet, DelayKind, PendingCha
 use crate::consensus_changes::SharedConsensusChanges;
 use crate::environment::finalize_block;
 use crate::justification::GrandpaJustification;
-use crate::finality_proof::SharedFinalityNotifiers;
+use crate::finality_proof::GrandpaJustificationSender;
 use std::marker::PhantomData;
 
 /// A block-import handler for GRANDPA.
@@ -61,7 +61,7 @@ pub struct GrandpaBlockImport<Backend, Block: BlockT, Client, SC> {
 	send_voter_commands: TracingUnboundedSender<VoterCommand<Block::Hash, NumberFor<Block>>>,
 	consensus_changes: SharedConsensusChanges<Block::Hash, NumberFor<Block>>,
 	authority_set_hard_forks: HashMap<Block::Hash, PendingChange<Block::Hash, NumberFor<Block>>>,
-	finality_notifiers: Option<SharedFinalityNotifiers<Block>>,
+	finality_subscription: Option<GrandpaJustificationSender<Block>>,
 	_phantom: PhantomData<Backend>,
 }
 
@@ -76,7 +76,7 @@ impl<Backend, Block: BlockT, Client, SC: Clone> Clone for
 			send_voter_commands: self.send_voter_commands.clone(),
 			consensus_changes: self.consensus_changes.clone(),
 			authority_set_hard_forks: self.authority_set_hard_forks.clone(),
-			finality_notifiers: self.finality_notifiers.clone(),
+			finality_subscription: self.finality_subscription.clone(),
 			_phantom: PhantomData,
 		}
 	}
@@ -561,7 +561,7 @@ impl<Backend, Block: BlockT, Client, SC> GrandpaBlockImport<Backend, Block, Clie
 		send_voter_commands: TracingUnboundedSender<VoterCommand<Block::Hash, NumberFor<Block>>>,
 		consensus_changes: SharedConsensusChanges<Block::Hash, NumberFor<Block>>,
 		authority_set_hard_forks: Vec<(SetId, PendingChange<Block::Hash, NumberFor<Block>>)>,
-		finality_notifiers: Option<SharedFinalityNotifiers<Block>>,
+		finality_subscription: Option<GrandpaJustificationSender<Block>>,
 	) -> GrandpaBlockImport<Backend, Block, Client, SC> {
 		// check for and apply any forced authority set hard fork that applies
 		// to the *current* authority set.
@@ -605,7 +605,7 @@ impl<Backend, Block: BlockT, Client, SC> GrandpaBlockImport<Backend, Block, Clie
 			send_voter_commands,
 			consensus_changes,
 			authority_set_hard_forks,
-			finality_notifiers,
+			finality_subscription,
 			_phantom: PhantomData,
 		}
 	}
@@ -651,7 +651,7 @@ where
 			number,
 			justification.into(),
 			initial_sync,
-			self.finality_notifiers.clone(),
+			&self.finality_subscription,
 		);
 
 		match result {

client/finality-grandpa/src/lib.rs

@@ -69,7 +69,7 @@ use sc_keystore::KeyStorePtr;
 use sp_inherents::InherentDataProviders;
 use sp_consensus::{SelectChain, BlockImport};
 use sp_core::Pair;
-use sp_utils::mpsc::{tracing_unbounded, TracingUnboundedReceiver, TracingUnboundedSender};
+use sp_utils::mpsc::{tracing_unbounded, TracingUnboundedReceiver};
 use sc_telemetry::{telemetry, CONSENSUS_INFO, CONSENSUS_DEBUG};
 use serde_json;
 
@@ -127,7 +127,7 @@ use import::GrandpaBlockImport;
 use until_imported::UntilGlobalMessageBlocksImported;
 use communication::{NetworkBridge, Network as NetworkT};
 use sp_finality_grandpa::{AuthorityList, AuthorityPair, AuthoritySignature, SetId};
-use crate::finality_proof::{FinalityProofSubscription, SharedFinalityNotifiers};
+use crate::finality_proof::GrandpaJustificationSender;
 
 // Re-export these two because it's just so damn convenient.
 pub use sp_finality_grandpa::{AuthorityId, ScheduledChange};
@@ -623,7 +623,7 @@ pub struct GrandpaParams<Block: BlockT, C, N, SC, VR> {
 	/// The prometheus metrics registry.
 	pub prometheus_registry: Option<prometheus_endpoint::Registry>,
 	/// A subscription to new block justifications
-	pub finality_subscription: Option<FinalityProofSubscription<Block>>,
+	pub finality_subscription: Option<GrandpaJustificationSender<Block>>,
 }
 
 /// Run a GRANDPA voter as a task. Provide configuration and a link to a
@@ -700,12 +700,6 @@ pub fn run_grandpa_voter<Block: BlockT, BE: 'static, C, N, SC, VR>(
 		future::Either::Right(future::pending())
 	};
 
-	let finality_notifiers = if let Some(s) = finality_subscription {
-		Some(s.notifiers())
-	} else {
-		None
-	};
-
 	let voter_work = VoterWork::new(
 		client,
 		config,
@@ -715,7 +709,7 @@ pub fn run_grandpa_voter<Block: BlockT, BE: 'static, C, N, SC, VR>(
 		persistent_data,
 		voter_commands_rx,
 		prometheus_registry,
-		finality_notifiers,
+		finality_subscription,
 	);
 
 	let voter_work = voter_work
@@ -773,7 +767,7 @@ where
 		persistent_data: PersistentData<Block>,
 		voter_commands_rx: TracingUnboundedReceiver<VoterCommand<Block::Hash, NumberFor<Block>>>,
 		prometheus_registry: Option<prometheus_endpoint::Registry>,
-		finality_notifiers: Option<SharedFinalityNotifiers<Block>>,
+		finality_subscription: Option<GrandpaJustificationSender<Block>>,
 	) -> Self {
 		let metrics = match prometheus_registry.as_ref().map(Metrics::register) {
 			Some(Ok(metrics)) => Some(metrics),
@@ -797,7 +791,7 @@ where
 			consensus_changes: persistent_data.consensus_changes.clone(),
 			voter_set_state: persistent_data.set_state.clone(),
 			metrics: metrics.as_ref().map(|m| m.environment.clone()),
-			finality_notifiers,
+			finality_subscription,
 			_phantom: PhantomData,
 		});
 
@@ -921,7 +915,7 @@ where
 					network: self.env.network.clone(),
 					voting_rule: self.env.voting_rule.clone(),
 					metrics: self.env.metrics.clone(),
-					finality_notifiers: self.env.finality_notifiers.clone(),
+					finality_subscription: self.env.finality_subscription.clone(),
 					_phantom: PhantomData,
 				});
 

client/finality-grandpa/src/observer.rs

@@ -125,7 +125,7 @@ fn grandpa_observer<BE, Block: BlockT, Client, S, F>(
 				finalized_number,
 				(round, commit).into(),
 				false,
-				None, // TODO: Should I include the finality_subscribers here?
+				&None, // TODO: Should I include the finality_subscribers here?
 			) {
 				Ok(_) => {},
 				Err(e) => return future::err(e),