Custom RPC implementation for node.

Cargo.toml

@@ -96,8 +96,10 @@ members = [
 	"node/cli",
 	"node/executor",
 	"node/primitives",
-	"node/runtime",
+	"node/rpc",
 	"node/rpc-client",
+	"node/runtime",
+	"node/testing",
 	"node-template",
 	"subkey",
 	"test-utils/chain-spec-builder",

core/rpc-servers/Cargo.toml

@@ -5,12 +5,12 @@ authors = ["Parity Technologies <[email protected]>"]
 edition = "2018"
 
 [dependencies]
-pubsub = { package = "jsonrpc-pubsub", version = "12.0.0" }
+jsonrpc-core = "13.0.0"
+pubsub = { package = "jsonrpc-pubsub", version = "13.0.0" }
 log = "0.4"
 serde = "1.0"
-substrate-rpc = { path = "../rpc" }
 sr-primitives = { path = "../sr-primitives" }
 
 [target.'cfg(not(target_os = "unknown"))'.dependencies]
-http = { package = "jsonrpc-http-server", version = "12.0.0" }
-ws = { package = "jsonrpc-ws-server", version = "12.0.0" }
+http = { package = "jsonrpc-http-server", version = "13.0.0" }
+ws = { package = "jsonrpc-ws-server", version = "13.0.0" }

core/rpc-servers/src/lib.rs

@@ -18,42 +18,28 @@
 
 #[warn(missing_docs)]
 
-pub use substrate_rpc as apis;
-
 use std::io;
+use jsonrpc_core::IoHandlerExtension;
 use log::error;
-use sr_primitives::{traits::{Block as BlockT, NumberFor}, generic::SignedBlock};
+use pubsub::PubSubMetadata;
 
 /// Maximal payload accepted by RPC servers.
 const MAX_PAYLOAD: usize = 15 * 1024 * 1024;
 
 /// Default maximum number of connections for WS RPC servers.
 const WS_MAX_CONNECTIONS: usize = 100;
 
-pub type Metadata = apis::metadata::Metadata;
-pub type RpcHandler = pubsub::PubSubHandler<Metadata>;
+/// The RPC IoHandler containing all requested APIs.
+pub type RpcHandler<T> = pubsub::PubSubHandler<T>;
 
 pub use self::inner::*;
 
 /// Construct rpc `IoHandler`
-pub fn rpc_handler<Block: BlockT, ExHash, S, C, A, Y>(
-	state: S,
-	chain: C,
-	author: A,
-	system: Y,
-) -> RpcHandler where
-	Block: BlockT + 'static,
-	ExHash: Send + Sync + 'static + sr_primitives::Serialize + sr_primitives::DeserializeOwned,
-	S: apis::state::StateApi<Block::Hash, Metadata=Metadata>,
-	C: apis::chain::ChainApi<NumberFor<Block>, Block::Hash, Block::Header, SignedBlock<Block>, Metadata=Metadata>,
-	A: apis::author::AuthorApi<ExHash, Block::Hash, Metadata=Metadata>,
-	Y: apis::system::SystemApi<Block::Hash, NumberFor<Block>>,
-{
+pub fn rpc_handler<M: PubSubMetadata>(
+	extension: impl IoHandlerExtension<M>
+) -> RpcHandler<M> {
 	let mut io = pubsub::PubSubHandler::default();
-	io.extend_with(state.to_delegate());
-	io.extend_with(chain.to_delegate());
-	io.extend_with(author.to_delegate());
-	io.extend_with(system.to_delegate());
+	extension.augment(&mut io);
 	io
 }
 
@@ -67,10 +53,10 @@ mod inner {
 	/// Start HTTP server listening on given address.
 	///
 	/// **Note**: Only available if `not(target_os = "unknown")`.
-	pub fn start_http(
+	pub fn start_http<M: pubsub::PubSubMetadata + Default>(
 		addr: &std::net::SocketAddr,
 		cors: Option<&Vec<String>>,
-		io: RpcHandler,
+		io: RpcHandler<M>,
 	) -> io::Result<http::Server> {
 		http::ServerBuilder::new(io)
 			.threads(4)
@@ -89,13 +75,13 @@ mod inner {
 	/// Start WS server listening on given address.
 	///
 	/// **Note**: Only available if `not(target_os = "unknown")`.
-	pub fn start_ws(
+	pub fn start_ws<M: pubsub::PubSubMetadata + From<jsonrpc_core::futures::sync::mpsc::Sender<String>>> (
 		addr: &std::net::SocketAddr,
 		max_connections: Option<usize>,
 		cors: Option<&Vec<String>>,
-		io: RpcHandler,
+		io: RpcHandler<M>,
 	) -> io::Result<ws::Server> {
-		ws::ServerBuilder::with_meta_extractor(io, |context: &ws::RequestContext| Metadata::new(context.sender()))
+		ws::ServerBuilder::with_meta_extractor(io, |context: &ws::RequestContext| context.sender().into())
 			.max_payload(MAX_PAYLOAD)
 			.max_connections(max_connections.unwrap_or(WS_MAX_CONNECTIONS))
 			.allowed_origins(map_cors(cors))

core/rpc/Cargo.toml

@@ -7,11 +7,11 @@ edition = "2018"
 [dependencies]
 derive_more = "0.14.0"
 futures = "0.1"
-futures03 = { package = "futures-preview", version = "=0.3.0-alpha.17", features = ["compat"] }
-jsonrpc-core = "12.0.0"
-jsonrpc-core-client = "12.0.0"
-jsonrpc-pubsub = "12.0.0"
-jsonrpc-derive = "12.0.0"
+futures03 = { package = "futures-preview", version = "0.3.0-alpha.17", features = ["compat"] }
+jsonrpc-core = "13.0.0"
+jsonrpc-core-client = "13.0.0"
+jsonrpc-pubsub = "13.0.0"
+jsonrpc-derive = "13.0.0"
 log = "0.4"
 parking_lot = "0.9.0"
 codec = { package = "parity-scale-codec", version = "1.0.0" }

core/rpc/src/lib.rs

@@ -15,19 +15,23 @@
 // along with Substrate.  If not, see <http://www.gnu.org/licenses/>.
 
 //! Substrate RPC interfaces.
+//!
+//! A collection of RPC methods and subscriptions supported by all substrate clients.
 
 #![warn(missing_docs)]
 
 mod errors;
 mod helpers;
+mod metadata;
 mod subscriptions;
 
+use jsonrpc_core as rpc;
+
+pub use metadata::Metadata;
+pub use rpc::IoHandlerExtension as RpcExtension;
 pub use subscriptions::Subscriptions;
 
 pub mod author;
 pub mod chain;
-pub mod metadata;
 pub mod state;
 pub mod system;
-
-use jsonrpc_core as rpc;

core/rpc/src/metadata.rs

@@ -52,3 +52,9 @@ impl Metadata {
 		(rx, Self::new(tx))
 	}
 }
+
+impl From<mpsc::Sender<String>> for Metadata {
+	fn from(sender: mpsc::Sender<String>) -> Self {
+		Self::new(sender)
+	}
+}

core/rpc/src/system/mod.rs

@@ -29,8 +29,8 @@ use network;
 use sr_primitives::traits::{self, Header as HeaderT};
 
 use self::error::Result;
-pub use self::helpers::{Properties, SystemInfo, Health, PeerInfo};
 
+pub use self::helpers::{Properties, SystemInfo, Health, PeerInfo};
 pub use self::gen_client::Client as SystemClient;
 
 /// Substrate system RPC API

core/service/Cargo.toml

@@ -32,7 +32,8 @@ client_db = { package = "substrate-client-db", path = "../../core/client/db", fe
 codec = { package = "parity-scale-codec", version = "1.0.0" }
 substrate-executor = { path = "../../core/executor" }
 transaction_pool = { package = "substrate-transaction-pool", path = "../../core/transaction-pool" }
-rpc = { package = "substrate-rpc-servers", path = "../../core/rpc-servers" }
+rpc-servers = { package = "substrate-rpc-servers", path = "../../core/rpc-servers" }
+rpc = { package = "substrate-rpc", path = "../../core/rpc" }
 tel = { package = "substrate-telemetry", path = "../../core/telemetry" }
 offchain = { package = "substrate-offchain", path = "../../core/offchain" }
 parity-multiaddr = { package = "parity-multiaddr", version = "0.5.0" }

core/service/src/components.rs

@@ -34,7 +34,7 @@ use sr_primitives::{
 };
 use crate::config::Configuration;
 use primitives::{Blake2Hasher, H256, traits::BareCryptoStorePtr};
-use rpc::{self, apis::system::SystemInfo};
+use rpc::{self, system::SystemInfo};
 use futures::{prelude::*, future::Executor};
 use futures03::{FutureExt as _, channel::mpsc, compat::Compat};
 
@@ -145,6 +145,9 @@ pub type PoolApi<C> = <C as Components>::TransactionPoolApi;
 pub trait RuntimeGenesis: Serialize + DeserializeOwned + BuildStorage {}
 impl<T: Serialize + DeserializeOwned + BuildStorage> RuntimeGenesis for T {}
 
+/// A transport-agnostic handler of the RPC queries.
+pub type RpcHandler = rpc_servers::RpcHandler<rpc::Metadata>;
+
 /// Something that can create and store initial session keys from given seeds.
 pub trait InitialSessionKeys<C: Components> {
 	/// Generate the initial session keys for the given seeds and store them in
@@ -168,46 +171,51 @@ impl<C: Components> InitialSessionKeys<Self> for C where
 }
 
 /// Something that can start the RPC service.
-pub trait StartRPC<C: Components> {
+pub trait StartRpc<C: Components> {
 	fn start_rpc(
 		client: Arc<ComponentClient<C>>,
-		system_send_back: mpsc::UnboundedSender<rpc::apis::system::Request<ComponentBlock<C>>>,
+		system_send_back: mpsc::UnboundedSender<rpc::system::Request<ComponentBlock<C>>>,
 		system_info: SystemInfo,
 		task_executor: TaskExecutor,
 		transaction_pool: Arc<TransactionPool<C::TransactionPoolApi>>,
+		rpc_extensions: impl rpc::RpcExtension<rpc::Metadata>,
 		keystore: KeyStorePtr,
-	) -> rpc::RpcHandler;
+	) -> RpcHandler;
 }
 
-impl<C: Components> StartRPC<Self> for C where
+impl<C: Components> StartRpc<C> for C where
 	ComponentClient<C>: ProvideRuntimeApi,
 	<ComponentClient<C> as ProvideRuntimeApi>::Api:
 		runtime_api::Metadata<ComponentBlock<C>> + session::SessionKeys<ComponentBlock<C>>,
 {
 	fn start_rpc(
 		client: Arc<ComponentClient<C>>,
-		system_send_back: mpsc::UnboundedSender<rpc::apis::system::Request<ComponentBlock<C>>>,
+		system_send_back: mpsc::UnboundedSender<rpc::system::Request<ComponentBlock<C>>>,
 		rpc_system_info: SystemInfo,
 		task_executor: TaskExecutor,
 		transaction_pool: Arc<TransactionPool<C::TransactionPoolApi>>,
+		rpc_extensions: impl rpc::RpcExtension<rpc::Metadata>,
 		keystore: KeyStorePtr,
-	) -> rpc::RpcHandler {
-		let subscriptions = rpc::apis::Subscriptions::new(task_executor.clone());
-		let chain = rpc::apis::chain::Chain::new(client.clone(), subscriptions.clone());
-		let state = rpc::apis::state::State::new(client.clone(), subscriptions.clone());
-		let author = rpc::apis::author::Author::new(
+	) -> RpcHandler {
+		use rpc::{chain, state, author, system};
+		let subscriptions = rpc::Subscriptions::new(task_executor.clone());
+		let chain = chain::Chain::new(client.clone(), subscriptions.clone());
+		let state = state::State::new(client.clone(), subscriptions.clone());
+		let author = rpc::author::Author::new(
 			client,
 			transaction_pool,
 			subscriptions,
 			keystore,
 		);
-		let system = rpc::apis::system::System::new(rpc_system_info, system_send_back);
-		rpc::rpc_handler::<ComponentBlock<C>, ComponentExHash<C>, _, _, _, _>(
-			state,
-			chain,
-			author,
-			system,
-		)
+		let system = system::System::new(rpc_system_info, system_send_back);
+
+		rpc_servers::rpc_handler((
+			state::StateApi::to_delegate(state),
+			chain::ChainApi::to_delegate(chain),
+			author::AuthorApi::to_delegate(author),
+			system::SystemApi::to_delegate(system),
+			rpc_extensions,
+		))
 	}
 }
 
@@ -299,7 +307,7 @@ pub trait ServiceTrait<C: Components>:
 	Deref<Target = Service<C>>
 	+ Send
 	+ 'static
-	+ StartRPC<C>
+	+ StartRpc<C>
 	+ MaintainTransactionPool<C>
 	+ OffchainWorker<C>
 	+ InitialSessionKeys<C>
@@ -308,7 +316,7 @@ impl<C: Components, T> ServiceTrait<C> for T where
 	T: Deref<Target = Service<C>>
 	+ Send
 	+ 'static
-	+ StartRPC<C>
+	+ StartRpc<C>
 	+ MaintainTransactionPool<C>
 	+ OffchainWorker<C>
 	+ InitialSessionKeys<C>
@@ -335,6 +343,8 @@ pub trait ServiceFactory: 'static + Sized {
 	type Genesis: RuntimeGenesis;
 	/// Other configuration for service members.
 	type Configuration: Default;
+	/// RPC initialisation.
+	type RpcExtensions: rpc::RpcExtension<rpc::Metadata>;
 	/// Extended full service type.
 	type FullService: ServiceTrait<FullComponents<Self>>;
 	/// Extended light service type.
@@ -407,6 +417,18 @@ pub trait ServiceFactory: 'static + Sized {
 			Err("Chain Specification doesn't contain any consensus_engine name".into())
 		}
 	}
+
+	/// Create custom RPC method handlers for full node.
+	fn build_full_rpc_extensions(
+		client: Arc<FullClient<Self>>,
+		transaction_pool: Arc<TransactionPool<Self::FullTransactionPoolApi>>,
+	) -> Self::RpcExtensions;
+
+	/// Create custom RPC method handlers for light node.
+	fn build_light_rpc_extensions(
+		client: Arc<LightClient<Self>>,
+		transaction_pool: Arc<TransactionPool<Self::LightTransactionPoolApi>>,
+	) -> Self::RpcExtensions;
 }
 
 /// A collection of types and function to generalize over full / light client type.
@@ -419,8 +441,10 @@ pub trait Components: Sized + 'static {
 	type Executor: 'static + client::CallExecutor<FactoryBlock<Self::Factory>, Blake2Hasher> + Send + Sync + Clone;
 	/// The type that implements the runtime API.
 	type RuntimeApi: Send + Sync;
-	/// A type that can start all runtime-dependent services.
+	/// The type that can start all runtime-dependent services.
 	type RuntimeServices: ServiceTrait<Self>;
+	/// The type that can extend the RPC methods.
+	type RpcExtensions: rpc::RpcExtension<rpc::Metadata>;
 	// TODO: Traitify transaction pool and allow people to implement their own. (#1242)
 	/// Extrinsic pool type.
 	type TransactionPoolApi: 'static + txpool::ChainApi<
@@ -468,6 +492,12 @@ pub trait Components: Sized + 'static {
 		config: &mut FactoryFullConfiguration<Self::Factory>,
 		client: Arc<ComponentClient<Self>>
 	) -> Result<Option<Self::SelectChain>, error::Error>;
+
+	/// Build RPC extensions
+	fn build_rpc_extensions(
+		client: Arc<ComponentClient<Self>>,
+		transaction_pool: Arc<TransactionPool<Self::TransactionPoolApi>>,
+	) -> Self::RpcExtensions;
 }
 
 /// A struct that implement `Components` for the full client.
@@ -529,6 +559,7 @@ impl<Factory: ServiceFactory> Components for FullComponents<Factory> {
 	type ImportQueue = Factory::FullImportQueue;
 	type RuntimeApi = Factory::RuntimeApi;
 	type RuntimeServices = Factory::FullService;
+	type RpcExtensions = Factory::RpcExtensions;
 	type SelectChain = Factory::SelectChain;
 
 	fn build_client(
@@ -594,6 +625,13 @@ impl<Factory: ServiceFactory> Components for FullComponents<Factory> {
 	) -> Result<Option<Arc<dyn FinalityProofProvider<<Self::Factory as ServiceFactory>::Block>>>, error::Error> {
 		Factory::build_finality_proof_provider(client)
 	}
+
+	fn build_rpc_extensions(
+		client: Arc<ComponentClient<Self>>,
+		transaction_pool: Arc<TransactionPool<Self::TransactionPoolApi>>,
+	) -> Self::RpcExtensions {
+		Factory::build_full_rpc_extensions(client, transaction_pool)
+	}
 }
 
 /// A struct that implement `Components` for the light client.
@@ -655,6 +693,7 @@ impl<Factory: ServiceFactory> Components for LightComponents<Factory> {
 	type ImportQueue = <Factory as ServiceFactory>::LightImportQueue;
 	type RuntimeApi = Factory::RuntimeApi;
 	type RuntimeServices = Factory::LightService;
+	type RpcExtensions = Factory::RpcExtensions;
 	type SelectChain = Factory::SelectChain;
 
 	fn build_client(
@@ -709,12 +748,20 @@ impl<Factory: ServiceFactory> Components for LightComponents<Factory> {
 	) -> Result<Option<Arc<dyn FinalityProofProvider<<Self::Factory as ServiceFactory>::Block>>>, error::Error> {
 		Ok(None)
 	}
+
 	fn build_select_chain(
 		_config: &mut FactoryFullConfiguration<Self::Factory>,
 		_client: Arc<ComponentClient<Self>>
 	) -> Result<Option<Self::SelectChain>, error::Error> {
 		Ok(None)
 	}
+
+	fn build_rpc_extensions(
+		client: Arc<ComponentClient<Self>>,
+		transaction_pool: Arc<TransactionPool<Self::TransactionPoolApi>>,
+	) -> Self::RpcExtensions {
+		Factory::build_light_rpc_extensions(client, transaction_pool)
+	}
 }
 
 #[cfg(test)]