grandpa-rpc: only share executor instead of sub manager

bin/node/cli/Cargo.toml

@@ -39,7 +39,6 @@ serde = { version = "1.0.102", features = ["derive"] }
 futures = { version = "0.3.1", features = ["compat"] }
 hex-literal = "0.3.1"
 jsonrpc-core = "14.2.0"
-jsonrpc-pubsub = "14.2.0"
 log = "0.4.8"
 rand = "0.7.2"
 structopt = { version = "0.3.8", optional = true }

bin/node/cli/src/service.rs

@@ -51,7 +51,7 @@ pub fn new_partial(config: &Configuration) -> Result<sc_service::PartialComponen
 	(
 		impl Fn(
 			node_rpc::DenyUnsafe,
-			jsonrpc_pubsub::manager::SubscriptionManager
+			sc_rpc::SubscriptionTaskExecutor
 		) -> node_rpc::IoHandler,
 		(
 			sc_consensus_babe::BabeBlockImport<Block, FullClient, FullGrandpaBlockImport>,

bin/node/rpc/Cargo.toml

@@ -12,7 +12,6 @@ targets = ["x86_64-unknown-linux-gnu"]
 
 [dependencies]
 jsonrpc-core = "14.2.0"
-jsonrpc-pubsub = "14.2.0"
 node-primitives = { version = "2.0.0-rc6", path = "../primitives" }
 node-runtime = { version = "2.0.0-rc6", path = "../runtime" }
 pallet-contracts-rpc = { version = "0.8.0-rc6", path = "../../../frame/contracts/rpc/" }

bin/node/rpc/src/lib.rs

@@ -32,7 +32,6 @@
 
 use std::sync::Arc;
 
-use jsonrpc_pubsub::manager::SubscriptionManager;
 use node_primitives::{Block, BlockNumber, AccountId, Index, Balance, Hash};
 use sc_consensus_babe::{Config, Epoch};
 use sc_consensus_babe_rpc::BabeRpcHandler;
@@ -46,6 +45,7 @@ use sp_block_builder::BlockBuilder;
 use sp_blockchain::{Error as BlockChainError, HeaderMetadata, HeaderBackend};
 use sp_consensus::SelectChain;
 use sp_consensus_babe::BabeApi;
+use sc_rpc::SubscriptionTaskExecutor;
 use sp_transaction_pool::TransactionPool;
 
 /// Light client extra dependencies.
@@ -78,8 +78,8 @@ pub struct GrandpaDeps {
 	pub shared_authority_set: SharedAuthoritySet<Hash, BlockNumber>,
 	/// Receives notifications about justification events from Grandpa.
 	pub justification_stream: GrandpaJustificationStream<Block>,
-	/// Subscription manager to keep track of pubsub subscribers.
-	pub subscriptions: SubscriptionManager,
+	/// Executor to drive the subscription manager in the Grandpa RPC handler.
+	pub sub_task_executor: SubscriptionTaskExecutor,
 }
 
 /// Full client dependencies.

client/finality-grandpa/rpc/src/lib.rs

@@ -19,6 +19,7 @@
 //! RPC API for GRANDPA.
 #![warn(missing_docs)]
 
+use std::sync::Arc;
 use futures::{FutureExt, TryFutureExt, TryStreamExt, StreamExt};
 use log::warn;
 use jsonrpc_derive::rpc;
@@ -27,6 +28,7 @@ use jsonrpc_core::futures::{
 	sink::Sink as Sink01,
 	stream::Stream as Stream01,
 	future::Future as Future01,
+	future::Executor as Executor01,
 };
 
 mod error;
@@ -92,12 +94,16 @@ pub struct GrandpaRpcHandler<AuthoritySet, VoterState, Block: BlockT> {
 
 impl<AuthoritySet, VoterState, Block: BlockT> GrandpaRpcHandler<AuthoritySet, VoterState, Block> {
 	/// Creates a new GrandpaRpcHandler instance.
-	pub fn new(
+	pub fn new<E>(
 		authority_set: AuthoritySet,
 		voter_state: VoterState,
 		justification_stream: GrandpaJustificationStream<Block>,
-		manager: SubscriptionManager,
-	) -> Self {
+		executor: E,
+	) -> Self
+	where
+		E: Executor01<Box<dyn Future01<Item = (), Error = ()> + Send>> + Send + Sync + 'static,
+	{
+		let manager = SubscriptionManager::new(Arc::new(executor));
 		Self {
 			authority_set,
 			voter_state,
@@ -232,13 +238,12 @@ mod tests {
 		VoterState: ReportVoterState + Send + Sync + 'static,
 	{
 		let (justification_sender, justification_stream) = GrandpaJustificationStream::channel();
-		let manager = SubscriptionManager::new(Arc::new(sc_rpc::testing::TaskExecutor));
 
 		let handler = GrandpaRpcHandler::new(
 			TestAuthoritySet,
 			voter_state,
 			justification_stream,
-			manager,
+			sc_rpc::testing::TaskExecutor,
 		);
 
 		let mut io = jsonrpc_core::MetaIoHandler::default();

client/service/src/builder.rs

@@ -73,17 +73,23 @@ pub trait RpcExtensionBuilder {
 
 	/// Returns an instance of the RPC extension for a particular `DenyUnsafe`
 	/// value, e.g. the RPC extension might not expose some unsafe methods.
-	fn build(&self, deny: sc_rpc::DenyUnsafe, subscriptions: SubscriptionManager) -> Self::Output;
+	fn build(&self,
+		deny: sc_rpc::DenyUnsafe,
+		sub_task_executor: sc_rpc::SubscriptionTaskExecutor
+	) -> Self::Output;
 }
 
 impl<F, R> RpcExtensionBuilder for F where
-	F: Fn(sc_rpc::DenyUnsafe, SubscriptionManager) -> R,
+	F: Fn(sc_rpc::DenyUnsafe, sc_rpc::SubscriptionTaskExecutor) -> R,
 	R: sc_rpc::RpcExtension<sc_rpc::Metadata>,
 {
 	type Output = R;
 
-	fn build(&self, deny: sc_rpc::DenyUnsafe, subscriptions: SubscriptionManager) -> Self::Output {
-		(*self)(deny, subscriptions)
+	fn build(&self,
+		deny: sc_rpc::DenyUnsafe,
+		sub_task_executor: sc_rpc::SubscriptionTaskExecutor
+	) -> Self::Output {
+		(*self)(deny, sub_task_executor)
 	}
 }
 
@@ -97,7 +103,10 @@ impl<R> RpcExtensionBuilder for NoopRpcExtensionBuilder<R> where
 {
 	type Output = R;
 
-	fn build(&self, _deny: sc_rpc::DenyUnsafe, _subscriptions: SubscriptionManager) -> Self::Output {
+	fn build(&self,
+		_deny: sc_rpc::DenyUnsafe,
+		_sub_task_executor: sc_rpc::SubscriptionTaskExecutor
+	) -> Self::Output {
 		self.0.clone()
 	}
 }
@@ -694,7 +703,7 @@ fn gen_handler<TBl, TBackend, TExPool, TRpc, TCl>(
 	};
 
 	let task_executor = sc_rpc::SubscriptionTaskExecutor::new(spawn_handle);
-	let subscriptions = SubscriptionManager::new(Arc::new(task_executor));
+	let subscriptions = SubscriptionManager::new(Arc::new(task_executor.clone()));
 
 	let (chain, state, child_state) = if let (Some(remote_blockchain), Some(on_demand)) =
 		(remote_blockchain, on_demand) {
@@ -723,7 +732,7 @@ fn gen_handler<TBl, TBackend, TExPool, TRpc, TCl>(
 	let author = sc_rpc::author::Author::new(
 		client,
 		transaction_pool,
-		subscriptions.clone(),
+		subscriptions,
 		keystore,
 		deny_unsafe,
 	);
@@ -741,7 +750,7 @@ fn gen_handler<TBl, TBackend, TExPool, TRpc, TCl>(
 		maybe_offchain_rpc,
 		author::AuthorApi::to_delegate(author),
 		system::SystemApi::to_delegate(system),
-		rpc_extensions_builder.build(deny_unsafe, subscriptions),
+		rpc_extensions_builder.build(deny_unsafe, task_executor),
 	))
 }