RpcHandlers Refactorings (#6846)

* allow access to the underlying Pubsub instance from RpcHandlers

* bump Cargo.lock

* no more Arc<RpcHandlers>

* bump Cargo.lock

* Debug,.

* Arc<RpcHandlers>

* RpcHandler

* RpcHandlers::io_handler

* remove chain spec from cli

* address pr comments

* remove stray newline

Co-authored-by: Ashley <ashley.ruglys@gmail.com>

Co-authored-by: Tomasz Drwięga <tomasz@parity.io>
Co-authored-by: Ashley <ashley.ruglys@gmail.com>

substrate/client/service/Cargo.toml

@@ -26,7 +26,8 @@ test-helpers = []
 derive_more = "0.99.2"
 futures01 = { package = "futures", version = "0.1.29" }
 futures = { version = "0.3.4", features = ["compat"] }
-jsonrpc-pubsub = "14.2.0"
+jsonrpc-pubsub = "14.2"
+jsonrpc-core = "14.2"
 rand = "0.7.3"
 parking_lot = "0.10.0"
 lazy_static = "1.4.0"

substrate/client/service/src/builder.rs

@@ -442,7 +442,7 @@ pub fn build_offchain_workers<TBl, TBackend, TCl>(
 /// Spawn the tasks that are required to run a node.
 pub fn spawn_tasks<TBl, TBackend, TExPool, TRpc, TCl>(
 	params: SpawnTasksParams<TBl, TCl, TExPool, TRpc, TBackend>,
-) -> Result<Arc<RpcHandlers>, Error>
+) -> Result<RpcHandlers, Error>
 	where
 		TCl: ProvideRuntimeApi<TBl> + HeaderMetadata<TBl, Error=sp_blockchain::Error> + Chain<TBl> +
 		BlockBackend<TBl> + BlockIdTo<TBl, Error=sp_blockchain::Error> + ProofProvider<TBl> +
@@ -540,7 +540,7 @@ pub fn spawn_tasks<TBl, TBackend, TExPool, TRpc, TCl>(
 	);
 	let rpc = start_rpc_servers(&config, gen_handler)?;
 	// This is used internally, so don't restrict access to unsafe RPC
-	let rpc_handlers = Arc::new(RpcHandlers(gen_handler(sc_rpc::DenyUnsafe::No)));
+	let rpc_handlers = RpcHandlers(Arc::new(gen_handler(sc_rpc::DenyUnsafe::No).into()));
 
 	// Telemetry
 	let telemetry = config.telemetry_endpoints.clone().and_then(|endpoints| {

substrate/client/service/src/lib.rs

@@ -96,7 +96,8 @@ impl<T: MallocSizeOf> MallocSizeOfWasm for T {}
 impl<T> MallocSizeOfWasm for T {}
 
 /// RPC handlers that can perform RPC queries.
-pub struct RpcHandlers(sc_rpc_server::RpcHandler<sc_rpc::Metadata>);
+#[derive(Clone)]
+pub struct RpcHandlers(Arc<jsonrpc_core::MetaIoHandler<sc_rpc::Metadata>>);
 
 impl RpcHandlers {
 	/// Starts an RPC query.
@@ -115,6 +116,11 @@ impl RpcHandlers {
 			.map(|res| res.expect("this should never fail"))
 			.boxed()
 	}
+
+	/// Provides access to the underlying `MetaIoHandler`
+	pub fn io_handler(&self) -> Arc<jsonrpc_core::MetaIoHandler<sc_rpc::Metadata>> {
+		self.0.clone()
+	}
 }
 
 /// Sinks to propagate network status updates.

substrate/client/service/src/task_manager/mod.rs

@@ -336,9 +336,12 @@ impl TaskManager {
 		}
 	}
 
-	/// Set what the task manager should keep alive.
-	pub(super) fn keep_alive<T: 'static + Send + Sync>(&mut self, to_keep_alive: T) {
-		self.keep_alive = Box::new(to_keep_alive);
+	/// Set what the task manager should keep alive, can be called multiple times.
+	pub fn keep_alive<T: 'static + Send + Sync>(&mut self, to_keep_alive: T) {
+		// allows this fn to safely called multiple times.
+		use std::mem;
+		let old = mem::replace(&mut self.keep_alive, Box::new(()));
+		self.keep_alive = Box::new((to_keep_alive, old));
 	}
 
 	/// Register another TaskManager to terminate and gracefully shutdown when the parent