jsonrpsee integration

bin/node-template/node/Cargo.toml

@@ -42,7 +42,7 @@ frame-system = { version = "4.0.0-dev", path = "../../../frame/system" }
 pallet-transaction-payment = { version = "4.0.0-dev", default-features = false, path = "../../../frame/transaction-payment" }
 
 # These dependencies are used for the node template's RPCs
-jsonrpc-core = "18.0.0"
+jsonrpsee = { version = "0.12.0", features = ["server"] }
 sc-rpc = { version = "4.0.0-dev", path = "../../../client/rpc" }
 sp-api = { version = "4.0.0-dev", path = "../../../primitives/api" }
 sc-rpc-api = { version = "0.10.0-dev", path = "../../../client/rpc-api" }

bin/node-template/node/src/rpc.rs

@@ -7,13 +7,15 @@
 
 use std::sync::Arc;
 
+use jsonrpsee::RpcModule;
 use node_template_runtime::{opaque::Block, AccountId, Balance, Index};
-pub use sc_rpc_api::DenyUnsafe;
 use sc_transaction_pool_api::TransactionPool;
 use sp_api::ProvideRuntimeApi;
 use sp_block_builder::BlockBuilder;
 use sp_blockchain::{Error as BlockChainError, HeaderBackend, HeaderMetadata};
 
+pub use sc_rpc_api::DenyUnsafe;
+
 /// Full client dependencies.
 pub struct FullDeps<C, P> {
 	/// The client instance to use.
@@ -25,7 +27,9 @@ pub struct FullDeps<C, P> {
 }
 
 /// Instantiate all full RPC extensions.
-pub fn create_full<C, P>(deps: FullDeps<C, P>) -> jsonrpc_core::IoHandler<sc_rpc::Metadata>
+pub fn create_full<C, P>(
+	deps: FullDeps<C, P>,
+) -> Result<RpcModule<()>, Box<dyn std::error::Error + Send + Sync>>
 where
 	C: ProvideRuntimeApi<Block>,
 	C: HeaderBackend<Block> + HeaderMetadata<Block, Error = BlockChainError> + 'static,
@@ -35,20 +39,19 @@ where
 	C::Api: BlockBuilder<Block>,
 	P: TransactionPool + 'static,
 {
-	use pallet_transaction_payment_rpc::{TransactionPayment, TransactionPaymentApi};
-	use substrate_frame_rpc_system::{FullSystem, SystemApi};
+	use pallet_transaction_payment_rpc::{TransactionPaymentApiServer, TransactionPaymentRpc};
+	use substrate_frame_rpc_system::{SystemApiServer, SystemRpc};
 
-	let mut io = jsonrpc_core::IoHandler::default();
+	let mut module = RpcModule::new(());
 	let FullDeps { client, pool, deny_unsafe } = deps;
 
-	io.extend_with(SystemApi::to_delegate(FullSystem::new(client.clone(), pool, deny_unsafe)));
-
-	io.extend_with(TransactionPaymentApi::to_delegate(TransactionPayment::new(client)));
+	module.merge(SystemRpc::new(client.clone(), pool.clone(), deny_unsafe).into_rpc())?;
+	module.merge(TransactionPaymentRpc::new(client).into_rpc())?;
 
 	// Extend this RPC with a custom API by using the following syntax.
 	// `YourRpcStruct` should have a reference to a client, which is needed
 	// to call into the runtime.
-	// `io.extend_with(YourRpcTrait::to_delegate(YourRpcStruct::new(ReferenceToClient, ...)));`
+	// `module.merge(YourRpcTrait::into_rpc(YourRpcStruct::new(ReferenceToClient, ...)))?;`
 
-	io
+	Ok(module)
 }

bin/node-template/node/src/service.rs

@@ -228,8 +228,7 @@ pub fn new_full(mut config: Configuration) -> Result<TaskManager, ServiceError>
 		Box::new(move |deny_unsafe, _| {
 			let deps =
 				crate::rpc::FullDeps { client: client.clone(), pool: pool.clone(), deny_unsafe };
-
-			Ok(crate::rpc::create_full(deps))
+			crate::rpc::create_full(deps).map_err(Into::into)
 		})
 	};
 
@@ -239,7 +238,7 @@ pub fn new_full(mut config: Configuration) -> Result<TaskManager, ServiceError>
 		keystore: keystore_container.sync_keystore(),
 		task_manager: &mut task_manager,
 		transaction_pool: transaction_pool.clone(),
-		rpc_extensions_builder,
+		rpc_builder: rpc_extensions_builder,
 		backend,
 		system_rpc_tx,
 		config,

bin/node/cli/Cargo.toml

@@ -37,6 +37,7 @@ crate-type = ["cdylib", "rlib"]
 clap = { version = "3.1.6", features = ["derive"], optional = true }
 codec = { package = "parity-scale-codec", version = "3.0.0" }
 serde = { version = "1.0.136", features = ["derive"] }
+jsonrpsee = { version = "0.12.0", features = ["server"] }
 futures = "0.3.21"
 hex-literal = "0.3.4"
 log = "0.4.16"

bin/node/cli/benches/block_production.rs

@@ -92,6 +92,10 @@ fn new_node(tokio_handle: Handle) -> node_cli::service::NewFullBase {
 		rpc_cors: None,
 		rpc_methods: Default::default(),
 		rpc_max_payload: None,
+		rpc_max_request_size: None,
+		rpc_max_response_size: None,
+		rpc_id_provider: None,
+		rpc_max_subs_per_conn: None,
 		ws_max_out_buffer_capacity: None,
 		prometheus_config: None,
 		telemetry_endpoints: None,

bin/node/cli/benches/transaction_pool.rs

@@ -84,6 +84,10 @@ fn new_node(tokio_handle: Handle) -> node_cli::service::NewFullBase {
 		rpc_cors: None,
 		rpc_methods: Default::default(),
 		rpc_max_payload: None,
+		rpc_max_request_size: None,
+		rpc_max_response_size: None,
+		rpc_id_provider: None,
+		rpc_max_subs_per_conn: None,
 		ws_max_out_buffer_capacity: None,
 		prometheus_config: None,
 		telemetry_endpoints: None,

bin/node/cli/src/service.rs

@@ -134,7 +134,7 @@ pub fn new_partial(
 			impl Fn(
 				node_rpc::DenyUnsafe,
 				sc_rpc::SubscriptionTaskExecutor,
-			) -> Result<node_rpc::IoHandler, sc_service::Error>,
+			) -> Result<jsonrpsee::RpcModule<()>, sc_service::Error>,
 			(
 				sc_consensus_babe::BabeBlockImport<Block, FullClient, FullGrandpaBlockImport>,
 				grandpa::LinkHalf<Block, FullClient, FullSelectChain>,
@@ -236,7 +236,7 @@ pub fn new_partial(
 		let justification_stream = grandpa_link.justification_stream();
 		let shared_authority_set = grandpa_link.shared_authority_set().clone();
 		let shared_voter_state = grandpa::SharedVoterState::empty();
-		let rpc_setup = shared_voter_state.clone();
+		let shared_voter_state2 = shared_voter_state.clone();
 
 		let finality_proof_provider = grandpa::FinalityProofProvider::new_for_service(
 			backend.clone(),
@@ -277,7 +277,7 @@ pub fn new_partial(
 			node_rpc::create_full(deps, rpc_backend.clone()).map_err(Into::into)
 		};
 
-		(rpc_extensions_builder, rpc_setup)
+		(rpc_extensions_builder, shared_voter_state2)
 	};
 
 	Ok(sc_service::PartialComponents {
@@ -332,7 +332,7 @@ pub fn new_full_base(
 		keystore_container,
 		select_chain,
 		transaction_pool,
-		other: (rpc_extensions_builder, import_setup, rpc_setup, mut telemetry),
+		other: (rpc_builder, import_setup, rpc_setup, mut telemetry),
 	} = new_partial(&config)?;
 
 	let shared_voter_state = rpc_setup;
@@ -386,7 +386,7 @@ pub fn new_full_base(
 		client: client.clone(),
 		keystore: keystore_container.sync_keystore(),
 		network: network.clone(),
-		rpc_extensions_builder: Box::new(rpc_extensions_builder),
+		rpc_builder: Box::new(rpc_builder),
 		transaction_pool: transaction_pool.clone(),
 		task_manager: &mut task_manager,
 		system_rpc_tx,

bin/node/cli/tests/common.rs

@@ -26,15 +26,14 @@ use nix::{
 use node_primitives::Block;
 use remote_externalities::rpc_api;
 use std::{
+	io::{BufRead, BufReader, Read},
 	ops::{Deref, DerefMut},
 	path::Path,
-	process::{Child, Command, ExitStatus},
+	process::{self, Child, Command, ExitStatus},
 	time::Duration,
 };
 use tokio::time::timeout;
 
-static LOCALHOST_WS: &str = "ws://127.0.0.1:9944/";
-
 /// Wait for the given `child` the given number of `secs`.
 ///
 /// Returns the `Some(exit status)` or `None` if the process did not finish in the given time.
@@ -63,8 +62,9 @@ pub fn wait_for(child: &mut Child, secs: u64) -> Result<ExitStatus, ()> {
 pub async fn wait_n_finalized_blocks(
 	n: usize,
 	timeout_secs: u64,
+	url: &str,
 ) -> Result<(), tokio::time::error::Elapsed> {
-	timeout(Duration::from_secs(timeout_secs), wait_n_finalized_blocks_from(n, LOCALHOST_WS)).await
+	timeout(Duration::from_secs(timeout_secs), wait_n_finalized_blocks_from(n, url)).await
 }
 
 /// Wait for at least n blocks to be finalized from a specified node
@@ -85,12 +85,23 @@ pub async fn wait_n_finalized_blocks_from(n: usize, url: &str) {
 
 /// Run the node for a while (3 blocks)
 pub async fn run_node_for_a_while(base_path: &Path, args: &[&str]) {
-	let mut cmd = Command::new(cargo_bin("substrate"));
+	let mut cmd = Command::new(cargo_bin("substrate"))
+		.stdout(process::Stdio::piped())
+		.stderr(process::Stdio::piped())
+		.args(args)
+		.arg("-d")
+		.arg(base_path)
+		.spawn()
+		.unwrap();
 
-	let mut child = KillChildOnDrop(cmd.args(args).arg("-d").arg(base_path).spawn().unwrap());
+	let stderr = cmd.stderr.take().unwrap();
+
+	let mut child = KillChildOnDrop(cmd);
+
+	let (ws_url, _) = find_ws_url_from_output(stderr);
 
 	// Let it produce some blocks.
-	let _ = wait_n_finalized_blocks(3, 30).await;
+	let _ = wait_n_finalized_blocks(3, 30, &ws_url).await;
 
 	assert!(child.try_wait().unwrap().is_none(), "the process should still be running");
 
@@ -134,3 +145,30 @@ impl DerefMut for KillChildOnDrop {
 		&mut self.0
 	}
 }
+
+/// Read the WS address from the output.
+///
+/// This is hack to get the actual binded sockaddr because
+/// substrate assigns a random port if the specified port was already binded.
+pub fn find_ws_url_from_output(read: impl Read + Send) -> (String, String) {
+	let mut data = String::new();
+
+	let ws_url = BufReader::new(read)
+		.lines()
+		.find_map(|line| {
+			let line =
+				line.expect("failed to obtain next line from stdout for WS address discovery");
+			data.push_str(&line);
+
+			// does the line contain our port (we expect this specific output from substrate).
+			let sock_addr = match line.split_once("Running JSON-RPC WS server: addr=") {
+				None => return None,
+				Some((_, after)) => after.split_once(",").unwrap().0,
+			};
+
+			Some(format!("ws://{}", sock_addr))
+		})
+		.expect("We should get a WebSocket address");
+
+	(ws_url, data)
+}

bin/node/cli/tests/running_the_node_and_interrupt.rs

@@ -25,7 +25,7 @@ use nix::{
 	},
 	unistd::Pid,
 };
-use std::process::{Child, Command};
+use std::process::{self, Child, Command};
 use tempfile::tempdir;
 
 pub mod common;
@@ -36,14 +36,22 @@ async fn running_the_node_works_and_can_be_interrupted() {
 		let base_path = tempdir().expect("could not create a temp dir");
 		let mut cmd = common::KillChildOnDrop(
 			Command::new(cargo_bin("substrate"))
+				.stdout(process::Stdio::piped())
+				.stderr(process::Stdio::piped())
 				.args(&["--dev", "-d"])
 				.arg(base_path.path())
 				.arg("--no-hardware-benchmarks")
 				.spawn()
 				.unwrap(),
 		);
 
-		common::wait_n_finalized_blocks(3, 30).await.unwrap();
+		let stderr = cmd.stderr.take().unwrap();
+
+		let (ws_url, _) = common::find_ws_url_from_output(stderr);
+
+		common::wait_n_finalized_blocks(3, 30, &ws_url)
+			.await
+			.expect("Blocks are produced in time");
 		assert!(cmd.try_wait().unwrap().is_none(), "the process should still be running");
 		kill(Pid::from_raw(cmd.id().try_into().unwrap()), signal).unwrap();
 		assert_eq!(
@@ -62,6 +70,8 @@ async fn running_the_node_works_and_can_be_interrupted() {
 async fn running_two_nodes_with_the_same_ws_port_should_work() {
 	fn start_node() -> Child {
 		Command::new(cargo_bin("substrate"))
+			.stdout(process::Stdio::piped())
+			.stderr(process::Stdio::piped())
 			.args(&["--dev", "--tmp", "--ws-port=45789", "--no-hardware-benchmarks"])
 			.spawn()
 			.unwrap()
@@ -70,7 +80,10 @@ async fn running_two_nodes_with_the_same_ws_port_should_work() {
 	let mut first_node = common::KillChildOnDrop(start_node());
 	let mut second_node = common::KillChildOnDrop(start_node());
 
-	let _ = common::wait_n_finalized_blocks(3, 30).await;
+	let stderr = first_node.stderr.take().unwrap();
+	let (ws_url, _) = common::find_ws_url_from_output(stderr);
+
+	common::wait_n_finalized_blocks(3, 30, &ws_url).await.unwrap();
 
 	assert!(first_node.try_wait().unwrap().is_none(), "The first node should still be running");
 	assert!(second_node.try_wait().unwrap().is_none(), "The second node should still be running");

bin/node/cli/tests/temp_base_path_works.rs

@@ -43,19 +43,21 @@ async fn temp_base_path_works() {
 			.unwrap(),
 	);
 
+	let mut stderr = child.stderr.take().unwrap();
+	let (ws_url, mut data) = common::find_ws_url_from_output(&mut stderr);
+
 	// Let it produce some blocks.
-	common::wait_n_finalized_blocks(3, 30).await.unwrap();
+	common::wait_n_finalized_blocks(3, 30, &ws_url).await.unwrap();
 	assert!(child.try_wait().unwrap().is_none(), "the process should still be running");
 
 	// Stop the process
 	kill(Pid::from_raw(child.id().try_into().unwrap()), SIGINT).unwrap();
 	assert!(common::wait_for(&mut child, 40).map(|x| x.success()).unwrap_or_default());
 
 	// Ensure the database has been deleted
-	let mut stderr = String::new();
-	child.stderr.as_mut().unwrap().read_to_string(&mut stderr).unwrap();
+	stderr.read_to_string(&mut data).unwrap();
 	let re = Regex::new(r"Database: .+ at (\S+)").unwrap();
-	let db_path = PathBuf::from(re.captures(stderr.as_str()).unwrap().get(1).unwrap().as_str());
+	let db_path = PathBuf::from(re.captures(data.as_str()).unwrap().get(1).unwrap().as_str());
 
 	assert!(!db_path.exists());
 }

bin/node/rpc/Cargo.toml

@@ -11,7 +11,7 @@ repository = "https://github.com/paritytech/substrate/"
 targets = ["x86_64-unknown-linux-gnu"]
 
 [dependencies]
-jsonrpc-core = "18.0.0"
+jsonrpsee = { version = "0.12.0", features = ["server"] }
 node-primitives = { version = "2.0.0", path = "../primitives" }
 pallet-contracts-rpc = { version = "4.0.0-dev", path = "../../../frame/contracts/rpc/" }
 pallet-mmr-rpc = { version = "3.0.0", path = "../../../frame/merkle-mountain-range/rpc/" }