jsonrpsee integration (#8783)

* Add tokio

* No need to map CallError to CallError

* jsonrpsee proc macros (#9673)

* port error types to `JsonRpseeError`

* migrate chain module to proc macro api

* make it compile with proc macros

* update branch

* update branch

* update to jsonrpsee master

* port system rpc

* port state rpc

* port childstate & offchain

* frame system rpc

* frame transaction payment

* bring back CORS hack to work with polkadot UI

* port babe rpc

* port manual seal rpc

* port frame mmr rpc

* port frame contracts rpc

* port finality grandpa rpc

* port sync state rpc

* resolve a few TODO + no jsonrpc deps

* Update bin/node/rpc-client/src/main.rs

* Update bin/node/rpc-client/src/main.rs

* Update bin/node/rpc-client/src/main.rs

* Update bin/node/rpc-client/src/main.rs

* Port over system_ rpc tests

* Make it compile

* Use prost 0.8

* Use prost 0.8

* Make it compile

* Ignore more failing tests

* Comment out WIP tests

* fix nit in frame system api

* Update lockfile

* No more juggling tokio versions

* No more wait_for_stop ?

* Remove browser-testing

* Arguments must be arrays

* Use same argument names

* Resolve todo: no wait_for_stop for WS server
Add todo: is parse_rpc_result used?
Cleanup imports

* fmt

* log

* One test passes

* update jsonrpsee

* update jsonrpsee

* cleanup rpc-servers crate

* jsonrpsee: add host and origin filtering (#9787)

* add access control in the jsonrpsee servers

* use master

* fix nits

* rpc runtime_version safe

* fix nits

* fix grumbles

* remove unused files

* resolve some todos

* jsonrpsee more cleanup (#9803)

* more cleanup

* resolve TODOs

* fix some unwraps

* remove type hints

* update jsonrpsee

* downgrade zeroize

* pin jsonrpsee rev

* remove unwrap nit

* Comment out more tests that aren't ported

* Comment out more tests

* Fix tests after merge

* Subscription test

* Invalid nonce test

* Pending exts

* WIP removeExtrinsic test

* Test remove_extrinsic

* Make state test: should_return_storage work

* Uncomment/fix the other non-subscription related state tests

* test: author_insertKey

* test: author_rotateKeys

* Get rest of state tests passing

* asyncify a little more

* Add todo to note #msg change

* Crashing test for has_session_keys

* Fix error conversion to avoid stack overflows
Port author_hasSessionKeys test
fmt

* test author_hasKey

* Add two missing tests
Add a check on the return type
Add todos for James's concerns

* RPC tests for state, author and system (#9859)

* Fix test runner

* Impl Default for SubscriptionTaskExecutor

* Keep the minimul amount of code needed to compile tests

* Re-instate `RpcSession` (for now)

* cleanup

* Port over RPC tests

* Add tokio

* No need to map CallError to CallError

* Port over system_ rpc tests

* Make it compile

* Use prost 0.8

* Use prost 0.8

* Make it compile

* Ignore more failing tests

* Comment out WIP tests

* Update lockfile

* No more juggling tokio versions

* No more wait_for_stop ?

* Remove browser-testing

* Arguments must be arrays

* Use same argument names

* Resolve todo: no wait_for_stop for WS server
Add todo: is parse_rpc_result used?
Cleanup imports

* fmt

* log

* One test passes

* Comment out more tests that aren't ported

* Comment out more tests

* Fix tests after merge

* Subscription test

* Invalid nonce test

* Pending exts

* WIP removeExtrinsic test

* Test remove_extrinsic

* Make state test: should_return_storage work

* Uncomment/fix the other non-subscription related state tests

* test: author_insertKey

* test: author_rotateKeys

* Get rest of state tests passing

* asyncify a little more

* Add todo to note #msg change

* Crashing test for has_session_keys

* Fix error conversion to avoid stack overflows
Port author_hasSessionKeys test
fmt

* test author_hasKey

* Add two missing tests
Add a check on the return type
Add todos for James's concerns

* offchain rpc tests

* Address todos

* fmt

Co-authored-by: James Wilson <james@jsdw.me>

* fix drop in state test

* update jsonrpsee

* fix ignored system test

* fix chain tests

* remove some boiler plate

* Port BEEFY RPC (#9883)

* Merge master

* Port beefy RPC (ty @niklas!)

* trivial changes left over from merge

* Remove unused code

* Update jsonrpsee

* fix build

* make tests compile again

* beefy update jsonrpsee

* fix: respect rpc methods policy

* update cargo.lock

* update jsonrpsee

* update jsonrpsee

* downgrade error logs

* update jsonrpsee

* Fix typo

* remove unused file

* Better name

* Port Babe RPC tests

* Put docs back

* Resolve todo

* Port tests for System RPCs

* Resolve todo

* fix build

* Updated jsonrpsee to current master

* fix: port finality grandpa rpc tests

* Move .into() outside of the match

* more review grumbles

* jsonrpsee: add `rpc handlers` back (#10245)

* add back RpcHandlers

* cargo fmt

* fix docs

* fix grumble: remove needless alloc

* resolve TODO

* fmt

* Fix typo

* grumble: Use constants based on BASE_ERROR

* grumble: DRY whitelisted listening addresses
grumble: s/JSONRPC/JSON-RPC/

* cleanup

* grumbles: Making readers aware of the possibility of gaps

* review grumbles

* grumbles

* remove notes from niklasad1

* Update `jsonrpsee`

* fix: jsonrpsee features

* jsonrpsee: fallback to random port in case the specified port failed (#10304)

* jsonrpsee: fallback to random port

* better comment

* Update client/rpc-servers/src/lib.rs

Co-authored-by: Maciej Hirsz <1096222+maciejhirsz@users.noreply.github.com>

* Update client/rpc-servers/src/lib.rs

Co-authored-by: Maciej Hirsz <1096222+maciejhirsz@users.noreply.github.com>

* address grumbles

* cargo fmt

* addrs already slice

Co-authored-by: Maciej Hirsz <1096222+maciejhirsz@users.noreply.github.com>

* Update jsonrpsee to 092081a0a2b8904c6ebd2cd99e16c7bc13ffc3ae

* lockfile

* update jsonrpsee

* fix warning

* Don't fetch jsonrpsee from crates

* make tests compile again

* fix rpc tests

* remove unused deps

* update tokio

* fix rpc tests again

* fix: test runner

`HttpServerBuilder::builder` fails unless it's called within tokio runtime

* cargo fmt

* grumbles: fix subscription aliases

* make clippy happy

* update remaining subscriptions alias

* cleanup

* cleanup

* fix chain subscription: less boiler plate (#10285)

* fix chain subscription: less boiler plate

* fix bad merge

* cargo fmt

* Switch to jsonrpsee 0.5

* fix build

* add missing features

* fix nit: remove needless Box::pin

* Integrate jsonrpsee metrics (#10395)

* draft metrics impl

* Use latest api

* Add missing file

* Http server metrics

* cleanup

* bump jsonrpsee

* Remove `ServerMetrics` and use a single middleware for both connection counting (aka sessions) and call metrics.

* fix build

* remove needless Arc::clone

* Update to jsonrpsee 0.6

* lolz

* fix metrics

* Revert "lolz"

This reverts commit eed6c6a56e78d8e307b4950f4c52a1c3a2322ba1.

* fix: in-memory rpc support subscriptions

* commit Cargo.lock

* Update tests to 0.7

* fix TODOs

* ws server: generate subscriptionIDs as Strings

Some libraries seems to expect the subscription IDs to be Strings, let's not break
this in this PR.

* Increase timeout

* Port over tests

* cleanup

* Using error codes from the spec

* fix clippy

* cargo fmt

* update jsonrpsee

* fix nits

* fix: rpc_query

* enable custom subid gen through spawn_tasks

* remove unsed deps

* unify tokio deps

* Revert "enable custom subid gen through spawn_tasks"

This reverts commit 5c5eb70328fe39d154fdb55c56e637b4548cf470.

* fix bad merge of `test-utils`

* fix more nits

* downgrade wasm-instrument to 0.1.0

* [jsonrpsee]: enable custom RPC subscription ID generatation (#10731)

* enable custom subid gen through spawn_tasks

* fix nits

* Update client/service/src/builder.rs

Co-authored-by: David <dvdplm@gmail.com>

* add Poc; needs jsonrpsee pr

* update jsonrpsee

* add re-exports

* add docs

Co-authored-by: David <dvdplm@gmail.com>

* cargo fmt

* fmt

* port RPC-API dev

* Remove unused file

* fix nit: remove async trait

* fix doc links

* fix merge nit: remove jsonrpc deps

* kill namespace on rpc apis

* companion for jsonrpsee v0.10 (#11158)

* companion for jsonrpsee v0.10

* update versions v0.10.0

* add some fixes

* spelling

* fix spaces

Co-authored-by: Niklas Adolfsson <niklasadolfsson1@gmail.com>

* send error before subs are closed

* fix unsubscribe method names: chain

* fix tests

* jsonrpc server: print binded local address

* grumbles: kill SubscriptionTaskExecutor

* Update client/sync-state-rpc/src/lib.rs

Co-authored-by: Bastian Köcher <bkchr@users.noreply.github.com>

* Update client/rpc/src/chain/chain_full.rs

Co-authored-by: Bastian Köcher <bkchr@users.noreply.github.com>

* Update client/rpc/src/chain/chain_full.rs

Co-authored-by: Bastian Köcher <bkchr@users.noreply.github.com>

* sync-state-rpc: kill anyhow

* no more anyhow

* remove todo

* jsonrpsee:  fix bad params in subscriptions. (#11251)

* update jsonrpsee

* fix error responses

* revert error codes

* dont do weird stuff in drop impl

* rpc servers: remove needless clone

* Remove silly constants

* chore: update jsonrpsee v0.12

* commit Cargo.lock

* deps: downgrade git2

* feat: CLI flag max subscriptions per connection

* metrics: use old logging format

* fix: read WS address from substrate output (#11379)

Co-authored-by: Niklas Adolfsson <niklasadolfsson1@gmail.com>
Co-authored-by: James Wilson <james@jsdw.me>
Co-authored-by: Maciej Hirsz <hello@maciej.codes>
Co-authored-by: Maciej Hirsz <1096222+maciejhirsz@users.noreply.github.com>
Co-authored-by: Bastian Köcher <bkchr@users.noreply.github.com>

substrate/client/rpc/src/state/mod.rs

@@ -23,11 +23,15 @@ mod state_full;
 #[cfg(test)]
 mod tests;
 
-use futures::FutureExt;
-use jsonrpc_pubsub::{manager::SubscriptionManager, typed::Subscriber, SubscriptionId};
-use rpc::Result as RpcResult;
 use std::sync::Arc;
 
+use crate::SubscriptionTaskExecutor;
+
+use jsonrpsee::{
+	core::{async_trait, Error as JsonRpseeError, RpcResult},
+	ws_server::PendingSubscription,
+};
+
 use sc_rpc_api::{state::ReadProof, DenyUnsafe};
 use sp_core::{
 	storage::{PrefixedStorageKey, StorageChangeSet, StorageData, StorageKey},
@@ -38,7 +42,7 @@ use sp_version::RuntimeVersion;
 
 use sp_api::{CallApiAt, Metadata, ProvideRuntimeApi};
 
-use self::error::{Error, FutureResult};
+use self::error::Error;
 
 use sc_client_api::{
 	Backend, BlockBackend, BlockchainEvents, ExecutorProvider, ProofProvider, StorageProvider,
@@ -49,144 +53,122 @@ use sp_blockchain::{HeaderBackend, HeaderMetadata};
 const STORAGE_KEYS_PAGED_MAX_COUNT: u32 = 1000;
 
 /// State backend API.
+#[async_trait]
 pub trait StateBackend<Block: BlockT, Client>: Send + Sync + 'static
 where
 	Block: BlockT + 'static,
 	Client: Send + Sync + 'static,
 {
 	/// Call runtime method at given block.
-	fn call(
+	async fn call(
 		&self,
 		block: Option<Block::Hash>,
 		method: String,
 		call_data: Bytes,
-	) -> FutureResult<Bytes>;
+	) -> Result<Bytes, Error>;
 
 	/// Returns the keys with prefix, leave empty to get all the keys.
-	fn storage_keys(
+	async fn storage_keys(
 		&self,
 		block: Option<Block::Hash>,
 		prefix: StorageKey,
-	) -> FutureResult<Vec<StorageKey>>;
+	) -> Result<Vec<StorageKey>, Error>;
 
 	/// Returns the keys with prefix along with their values, leave empty to get all the pairs.
-	fn storage_pairs(
+	async fn storage_pairs(
 		&self,
 		block: Option<Block::Hash>,
 		prefix: StorageKey,
-	) -> FutureResult<Vec<(StorageKey, StorageData)>>;
+	) -> Result<Vec<(StorageKey, StorageData)>, Error>;
 
 	/// Returns the keys with prefix with pagination support.
-	fn storage_keys_paged(
+	async fn storage_keys_paged(
 		&self,
 		block: Option<Block::Hash>,
 		prefix: Option<StorageKey>,
 		count: u32,
 		start_key: Option<StorageKey>,
-	) -> FutureResult<Vec<StorageKey>>;
+	) -> Result<Vec<StorageKey>, Error>;
 
 	/// Returns a storage entry at a specific block's state.
-	fn storage(
+	async fn storage(
 		&self,
 		block: Option<Block::Hash>,
 		key: StorageKey,
-	) -> FutureResult<Option<StorageData>>;
+	) -> Result<Option<StorageData>, Error>;
 
 	/// Returns the hash of a storage entry at a block's state.
-	fn storage_hash(
+	async fn storage_hash(
 		&self,
 		block: Option<Block::Hash>,
 		key: StorageKey,
-	) -> FutureResult<Option<Block::Hash>>;
+	) -> Result<Option<Block::Hash>, Error>;
 
 	/// Returns the size of a storage entry at a block's state.
 	///
 	/// If data is available at `key`, it is returned. Else, the sum of values who's key has `key`
 	/// prefix is returned, i.e. all the storage (double) maps that have this prefix.
-	fn storage_size(
+	async fn storage_size(
 		&self,
 		block: Option<Block::Hash>,
 		key: StorageKey,
-	) -> FutureResult<Option<u64>>;
+	) -> Result<Option<u64>, Error>;
 
 	/// Returns the runtime metadata as an opaque blob.
-	fn metadata(&self, block: Option<Block::Hash>) -> FutureResult<Bytes>;
+	async fn metadata(&self, block: Option<Block::Hash>) -> Result<Bytes, Error>;
 
 	/// Get the runtime version.
-	fn runtime_version(&self, block: Option<Block::Hash>) -> FutureResult<RuntimeVersion>;
+	async fn runtime_version(&self, block: Option<Block::Hash>) -> Result<RuntimeVersion, Error>;
 
 	/// Query historical storage entries (by key) starting from a block given as the second
 	/// parameter.
 	///
 	/// NOTE This first returned result contains the initial state of storage for all keys.
 	/// Subsequent values in the vector represent changes to the previous state (diffs).
-	fn query_storage(
+	async fn query_storage(
 		&self,
 		from: Block::Hash,
 		to: Option<Block::Hash>,
 		keys: Vec<StorageKey>,
-	) -> FutureResult<Vec<StorageChangeSet<Block::Hash>>>;
+	) -> Result<Vec<StorageChangeSet<Block::Hash>>, Error>;
 
 	/// Query storage entries (by key) starting at block hash given as the second parameter.
-	fn query_storage_at(
+	async fn query_storage_at(
 		&self,
 		keys: Vec<StorageKey>,
 		at: Option<Block::Hash>,
-	) -> FutureResult<Vec<StorageChangeSet<Block::Hash>>>;
+	) -> Result<Vec<StorageChangeSet<Block::Hash>>, Error>;
 
 	/// Returns proof of storage entries at a specific block's state.
-	fn read_proof(
+	async fn read_proof(
 		&self,
 		block: Option<Block::Hash>,
 		keys: Vec<StorageKey>,
-	) -> FutureResult<ReadProof<Block::Hash>>;
-
-	/// New runtime version subscription
-	fn subscribe_runtime_version(
-		&self,
-		_meta: crate::Metadata,
-		subscriber: Subscriber<RuntimeVersion>,
-	);
-
-	/// Unsubscribe from runtime version subscription
-	fn unsubscribe_runtime_version(
-		&self,
-		_meta: Option<crate::Metadata>,
-		id: SubscriptionId,
-	) -> RpcResult<bool>;
-
-	/// New storage subscription
-	fn subscribe_storage(
-		&self,
-		_meta: crate::Metadata,
-		subscriber: Subscriber<StorageChangeSet<Block::Hash>>,
-		keys: Option<Vec<StorageKey>>,
-	);
-
-	/// Unsubscribe from storage subscription
-	fn unsubscribe_storage(
-		&self,
-		_meta: Option<crate::Metadata>,
-		id: SubscriptionId,
-	) -> RpcResult<bool>;
+	) -> Result<ReadProof<Block::Hash>, Error>;
 
 	/// Trace storage changes for block
-	fn trace_block(
+	async fn trace_block(
 		&self,
 		block: Block::Hash,
 		targets: Option<String>,
 		storage_keys: Option<String>,
 		methods: Option<String>,
-	) -> FutureResult<sp_rpc::tracing::TraceBlockResponse>;
+	) -> Result<sp_rpc::tracing::TraceBlockResponse, Error>;
+
+	/// New runtime version subscription
+	fn subscribe_runtime_version(&self, sink: PendingSubscription);
+
+	/// New storage subscription
+	fn subscribe_storage(&self, sink: PendingSubscription, keys: Option<Vec<StorageKey>>);
 }
 
 /// Create new state API that works on full node.
 pub fn new_full<BE, Block: BlockT, Client>(
 	client: Arc<Client>,
-	subscriptions: SubscriptionManager,
+	executor: SubscriptionTaskExecutor,
 	deny_unsafe: DenyUnsafe,
 	rpc_max_payload: Option<usize>,
-) -> (State<Block, Client>, ChildState<Block, Client>)
+) -> (StateApi<Block, Client>, ChildState<Block, Client>)
 where
 	Block: BlockT + 'static,
 	Block::Hash: Unpin,
@@ -207,168 +189,127 @@ where
 {
 	let child_backend = Box::new(self::state_full::FullState::new(
 		client.clone(),
-		subscriptions.clone(),
+		executor.clone(),
 		rpc_max_payload,
 	));
-	let backend =
-		Box::new(self::state_full::FullState::new(client, subscriptions, rpc_max_payload));
-	(State { backend, deny_unsafe }, ChildState { backend: child_backend })
+	let backend = Box::new(self::state_full::FullState::new(client, executor, rpc_max_payload));
+	(StateApi { backend, deny_unsafe }, ChildState { backend: child_backend })
 }
 
 /// State API with subscriptions support.
-pub struct State<Block, Client> {
+pub struct StateApi<Block, Client> {
 	backend: Box<dyn StateBackend<Block, Client>>,
 	/// Whether to deny unsafe calls
 	deny_unsafe: DenyUnsafe,
 }
 
-impl<Block, Client> StateApi<Block::Hash> for State<Block, Client>
+#[async_trait]
+impl<Block, Client> StateApiServer<Block::Hash> for StateApi<Block, Client>
 where
 	Block: BlockT + 'static,
 	Client: Send + Sync + 'static,
 {
-	type Metadata = crate::Metadata;
-
-	fn call(&self, method: String, data: Bytes, block: Option<Block::Hash>) -> FutureResult<Bytes> {
-		self.backend.call(block, method, data)
+	async fn call(
+		&self,
+		method: String,
+		data: Bytes,
+		block: Option<Block::Hash>,
+	) -> RpcResult<Bytes> {
+		self.backend.call(block, method, data).await.map_err(Into::into)
 	}
 
-	fn storage_keys(
+	async fn storage_keys(
 		&self,
 		key_prefix: StorageKey,
 		block: Option<Block::Hash>,
-	) -> FutureResult<Vec<StorageKey>> {
-		self.backend.storage_keys(block, key_prefix)
+	) -> RpcResult<Vec<StorageKey>> {
+		self.backend.storage_keys(block, key_prefix).await.map_err(Into::into)
 	}
 
-	fn storage_pairs(
+	async fn storage_pairs(
 		&self,
 		key_prefix: StorageKey,
 		block: Option<Block::Hash>,
-	) -> FutureResult<Vec<(StorageKey, StorageData)>> {
-		if let Err(err) = self.deny_unsafe.check_if_safe() {
-			return async move { Err(err.into()) }.boxed()
-		}
-
-		self.backend.storage_pairs(block, key_prefix)
+	) -> RpcResult<Vec<(StorageKey, StorageData)>> {
+		self.deny_unsafe.check_if_safe()?;
+		self.backend.storage_pairs(block, key_prefix).await.map_err(Into::into)
 	}
 
-	fn storage_keys_paged(
+	async fn storage_keys_paged(
 		&self,
 		prefix: Option<StorageKey>,
 		count: u32,
 		start_key: Option<StorageKey>,
 		block: Option<Block::Hash>,
-	) -> FutureResult<Vec<StorageKey>> {
+	) -> RpcResult<Vec<StorageKey>> {
 		if count > STORAGE_KEYS_PAGED_MAX_COUNT {
-			return async move {
-				Err(Error::InvalidCount { value: count, max: STORAGE_KEYS_PAGED_MAX_COUNT })
-			}
-			.boxed()
+			return Err(JsonRpseeError::from(Error::InvalidCount {
+				value: count,
+				max: STORAGE_KEYS_PAGED_MAX_COUNT,
+			}))
 		}
-		self.backend.storage_keys_paged(block, prefix, count, start_key)
+		self.backend
+			.storage_keys_paged(block, prefix, count, start_key)
+			.await
+			.map_err(Into::into)
 	}
 
-	fn storage(
+	async fn storage(
 		&self,
 		key: StorageKey,
 		block: Option<Block::Hash>,
-	) -> FutureResult<Option<StorageData>> {
-		self.backend.storage(block, key)
+	) -> RpcResult<Option<StorageData>> {
+		self.backend.storage(block, key).await.map_err(Into::into)
 	}
 
-	fn storage_hash(
+	async fn storage_hash(
 		&self,
 		key: StorageKey,
 		block: Option<Block::Hash>,
-	) -> FutureResult<Option<Block::Hash>> {
-		self.backend.storage_hash(block, key)
+	) -> RpcResult<Option<Block::Hash>> {
+		self.backend.storage_hash(block, key).await.map_err(Into::into)
 	}
 
-	fn storage_size(
+	async fn storage_size(
 		&self,
 		key: StorageKey,
 		block: Option<Block::Hash>,
-	) -> FutureResult<Option<u64>> {
-		self.backend.storage_size(block, key)
+	) -> RpcResult<Option<u64>> {
+		self.backend.storage_size(block, key).await.map_err(Into::into)
 	}
 
-	fn metadata(&self, block: Option<Block::Hash>) -> FutureResult<Bytes> {
-		self.backend.metadata(block)
+	async fn metadata(&self, block: Option<Block::Hash>) -> RpcResult<Bytes> {
+		self.backend.metadata(block).await.map_err(Into::into)
 	}
 
-	fn query_storage(
+	async fn runtime_version(&self, at: Option<Block::Hash>) -> RpcResult<RuntimeVersion> {
+		self.backend.runtime_version(at).await.map_err(Into::into)
+	}
+
+	async fn query_storage(
 		&self,
 		keys: Vec<StorageKey>,
 		from: Block::Hash,
 		to: Option<Block::Hash>,
-	) -> FutureResult<Vec<StorageChangeSet<Block::Hash>>> {
-		if let Err(err) = self.deny_unsafe.check_if_safe() {
-			return async move { Err(err.into()) }.boxed()
-		}
-
-		self.backend.query_storage(from, to, keys)
+	) -> RpcResult<Vec<StorageChangeSet<Block::Hash>>> {
+		self.deny_unsafe.check_if_safe()?;
+		self.backend.query_storage(from, to, keys).await.map_err(Into::into)
 	}
 
-	fn query_storage_at(
+	async fn query_storage_at(
 		&self,
 		keys: Vec<StorageKey>,
 		at: Option<Block::Hash>,
-	) -> FutureResult<Vec<StorageChangeSet<Block::Hash>>> {
-		self.backend.query_storage_at(keys, at)
+	) -> RpcResult<Vec<StorageChangeSet<Block::Hash>>> {
+		self.backend.query_storage_at(keys, at).await.map_err(Into::into)
 	}
 
-	fn read_proof(
+	async fn read_proof(
 		&self,
 		keys: Vec<StorageKey>,
 		block: Option<Block::Hash>,
-	) -> FutureResult<ReadProof<Block::Hash>> {
-		self.backend.read_proof(block, keys)
-	}
-
-	fn subscribe_storage(
-		&self,
-		meta: Self::Metadata,
-		subscriber: Subscriber<StorageChangeSet<Block::Hash>>,
-		keys: Option<Vec<StorageKey>>,
-	) {
-		if keys.is_none() {
-			if let Err(err) = self.deny_unsafe.check_if_safe() {
-				subscriber.reject(err.into())
-				    .expect("subscription rejection can only fail if it's been already rejected, and we're rejecting it for the first time; qed");
-				return
-			}
-		}
-
-		self.backend.subscribe_storage(meta, subscriber, keys)
-	}
-
-	fn unsubscribe_storage(
-		&self,
-		meta: Option<Self::Metadata>,
-		id: SubscriptionId,
-	) -> RpcResult<bool> {
-		self.backend.unsubscribe_storage(meta, id)
-	}
-
-	fn runtime_version(&self, at: Option<Block::Hash>) -> FutureResult<RuntimeVersion> {
-		self.backend.runtime_version(at)
-	}
-
-	fn subscribe_runtime_version(
-		&self,
-		meta: Self::Metadata,
-		subscriber: Subscriber<RuntimeVersion>,
-	) {
-		self.backend.subscribe_runtime_version(meta, subscriber);
-	}
-
-	fn unsubscribe_runtime_version(
-		&self,
-		meta: Option<Self::Metadata>,
-		id: SubscriptionId,
-	) -> RpcResult<bool> {
-		self.backend.unsubscribe_runtime_version(meta, id)
+	) -> RpcResult<ReadProof<Block::Hash>> {
+		self.backend.read_proof(block, keys).await.map_err(Into::into)
 	}
 
 	/// Re-execute the given block with the tracing targets given in `targets`
@@ -376,88 +317,102 @@ where
 	///
 	/// Note: requires the node to run with `--rpc-methods=Unsafe`.
 	/// Note: requires runtimes compiled with wasm tracing support, `--features with-tracing`.
-	fn trace_block(
+	async fn trace_block(
 		&self,
 		block: Block::Hash,
 		targets: Option<String>,
 		storage_keys: Option<String>,
 		methods: Option<String>,
-	) -> FutureResult<sp_rpc::tracing::TraceBlockResponse> {
-		if let Err(err) = self.deny_unsafe.check_if_safe() {
-			return async move { Err(err.into()) }.boxed()
+	) -> RpcResult<sp_rpc::tracing::TraceBlockResponse> {
+		self.deny_unsafe.check_if_safe()?;
+		self.backend
+			.trace_block(block, targets, storage_keys, methods)
+			.await
+			.map_err(Into::into)
+	}
+
+	fn subscribe_runtime_version(&self, sink: PendingSubscription) {
+		self.backend.subscribe_runtime_version(sink)
+	}
+
+	fn subscribe_storage(&self, sink: PendingSubscription, keys: Option<Vec<StorageKey>>) {
+		if keys.is_none() {
+			if let Err(err) = self.deny_unsafe.check_if_safe() {
+				let _ = sink.reject(JsonRpseeError::from(err));
+				return
+			}
 		}
 
-		self.backend.trace_block(block, targets, storage_keys, methods)
+		self.backend.subscribe_storage(sink, keys)
 	}
 }
 
 /// Child state backend API.
+#[async_trait]
 pub trait ChildStateBackend<Block: BlockT, Client>: Send + Sync + 'static
 where
 	Block: BlockT + 'static,
 	Client: Send + Sync + 'static,
 {
 	/// Returns proof of storage for a child key entries at a specific block's state.
-	fn read_child_proof(
+	async fn read_child_proof(
 		&self,
 		block: Option<Block::Hash>,
 		storage_key: PrefixedStorageKey,
 		keys: Vec<StorageKey>,
-	) -> FutureResult<ReadProof<Block::Hash>>;
+	) -> Result<ReadProof<Block::Hash>, Error>;
 
 	/// Returns the keys with prefix from a child storage,
 	/// leave prefix empty to get all the keys.
-	fn storage_keys(
+	async fn storage_keys(
 		&self,
 		block: Option<Block::Hash>,
 		storage_key: PrefixedStorageKey,
 		prefix: StorageKey,
-	) -> FutureResult<Vec<StorageKey>>;
+	) -> Result<Vec<StorageKey>, Error>;
 
 	/// Returns the keys with prefix from a child storage with pagination support.
-	fn storage_keys_paged(
+	async fn storage_keys_paged(
 		&self,
 		block: Option<Block::Hash>,
 		storage_key: PrefixedStorageKey,
 		prefix: Option<StorageKey>,
 		count: u32,
 		start_key: Option<StorageKey>,
-	) -> FutureResult<Vec<StorageKey>>;
+	) -> Result<Vec<StorageKey>, Error>;
 
 	/// Returns a child storage entry at a specific block's state.
-	fn storage(
+	async fn storage(
 		&self,
 		block: Option<Block::Hash>,
 		storage_key: PrefixedStorageKey,
 		key: StorageKey,
-	) -> FutureResult<Option<StorageData>>;
+	) -> Result<Option<StorageData>, Error>;
 
 	/// Returns child storage entries at a specific block's state.
-	fn storage_entries(
+	async fn storage_entries(
 		&self,
 		block: Option<Block::Hash>,
 		storage_key: PrefixedStorageKey,
 		keys: Vec<StorageKey>,
-	) -> FutureResult<Vec<Option<StorageData>>>;
+	) -> Result<Vec<Option<StorageData>>, Error>;
 
 	/// Returns the hash of a child storage entry at a block's state.
-	fn storage_hash(
+	async fn storage_hash(
 		&self,
 		block: Option<Block::Hash>,
 		storage_key: PrefixedStorageKey,
 		key: StorageKey,
-	) -> FutureResult<Option<Block::Hash>>;
+	) -> Result<Option<Block::Hash>, Error>;
 
 	/// Returns the size of a child storage entry at a block's state.
-	fn storage_size(
+	async fn storage_size(
 		&self,
 		block: Option<Block::Hash>,
 		storage_key: PrefixedStorageKey,
 		key: StorageKey,
-	) -> FutureResult<Option<u64>> {
-		self.storage(block, storage_key, key)
-			.map(|x| x.map(|r| r.map(|v| v.0.len() as u64)))
-			.boxed()
+	) -> Result<Option<u64>, Error> {
+		self.storage(block, storage_key, key).await.map(|x| x.map(|x| x.0.len() as u64))
 	}
 }
 
@@ -466,76 +421,84 @@ pub struct ChildState<Block, Client> {
 	backend: Box<dyn ChildStateBackend<Block, Client>>,
 }
 
-impl<Block, Client> ChildStateApi<Block::Hash> for ChildState<Block, Client>
+#[async_trait]
+impl<Block, Client> ChildStateApiServer<Block::Hash> for ChildState<Block, Client>
 where
 	Block: BlockT + 'static,
 	Client: Send + Sync + 'static,
 {
-	type Metadata = crate::Metadata;
-
-	fn read_child_proof(
-		&self,
-		child_storage_key: PrefixedStorageKey,
-		keys: Vec<StorageKey>,
-		block: Option<Block::Hash>,
-	) -> FutureResult<ReadProof<Block::Hash>> {
-		self.backend.read_child_proof(block, child_storage_key, keys)
-	}
-
-	fn storage(
-		&self,
-		storage_key: PrefixedStorageKey,
-		key: StorageKey,
-		block: Option<Block::Hash>,
-	) -> FutureResult<Option<StorageData>> {
-		self.backend.storage(block, storage_key, key)
-	}
-
-	fn storage_entries(
-		&self,
-		storage_key: PrefixedStorageKey,
-		keys: Vec<StorageKey>,
-		block: Option<Block::Hash>,
-	) -> FutureResult<Vec<Option<StorageData>>> {
-		self.backend.storage_entries(block, storage_key, keys)
-	}
-
-	fn storage_keys(
+	async fn storage_keys(
 		&self,
 		storage_key: PrefixedStorageKey,
 		key_prefix: StorageKey,
 		block: Option<Block::Hash>,
-	) -> FutureResult<Vec<StorageKey>> {
-		self.backend.storage_keys(block, storage_key, key_prefix)
+	) -> RpcResult<Vec<StorageKey>> {
+		self.backend
+			.storage_keys(block, storage_key, key_prefix)
+			.await
+			.map_err(Into::into)
 	}
 
-	fn storage_keys_paged(
+	async fn storage_keys_paged(
 		&self,
 		storage_key: PrefixedStorageKey,
 		prefix: Option<StorageKey>,
 		count: u32,
 		start_key: Option<StorageKey>,
 		block: Option<Block::Hash>,
-	) -> FutureResult<Vec<StorageKey>> {
-		self.backend.storage_keys_paged(block, storage_key, prefix, count, start_key)
+	) -> RpcResult<Vec<StorageKey>> {
+		self.backend
+			.storage_keys_paged(block, storage_key, prefix, count, start_key)
+			.await
+			.map_err(Into::into)
 	}
 
-	fn storage_hash(
+	async fn storage(
 		&self,
 		storage_key: PrefixedStorageKey,
 		key: StorageKey,
 		block: Option<Block::Hash>,
-	) -> FutureResult<Option<Block::Hash>> {
-		self.backend.storage_hash(block, storage_key, key)
+	) -> RpcResult<Option<StorageData>> {
+		self.backend.storage(block, storage_key, key).await.map_err(Into::into)
 	}
 
-	fn storage_size(
+	async fn storage_entries(
+		&self,
+		storage_key: PrefixedStorageKey,
+		keys: Vec<StorageKey>,
+		block: Option<Block::Hash>,
+	) -> RpcResult<Vec<Option<StorageData>>> {
+		self.backend.storage_entries(block, storage_key, keys).await.map_err(Into::into)
+	}
+
+	async fn storage_hash(
 		&self,
 		storage_key: PrefixedStorageKey,
 		key: StorageKey,
 		block: Option<Block::Hash>,
-	) -> FutureResult<Option<u64>> {
-		self.backend.storage_size(block, storage_key, key)
+	) -> RpcResult<Option<Block::Hash>> {
+		self.backend.storage_hash(block, storage_key, key).await.map_err(Into::into)
+	}
+
+	async fn storage_size(
+		&self,
+		storage_key: PrefixedStorageKey,
+		key: StorageKey,
+		block: Option<Block::Hash>,
+	) -> RpcResult<Option<u64>> {
+		self.backend.storage_size(block, storage_key, key).await.map_err(Into::into)
+	}
+
+	async fn read_child_proof(
+		&self,
+		child_storage_key: PrefixedStorageKey,
+		keys: Vec<StorageKey>,
+		block: Option<Block::Hash>,
+	) -> RpcResult<ReadProof<Block::Hash>> {
+		self.backend
+			.read_child_proof(block, child_storage_key, keys)
+			.await
+			.map_err(Into::into)
 	}
 }
 

substrate/client/rpc/src/state/state_full.rs

@@ -18,17 +18,26 @@
 
 //! State API backend for full nodes.
 
-use futures::{
-	future,
-	future::{err, try_join_all},
-	stream, FutureExt, SinkExt, StreamExt,
-};
-use jsonrpc_pubsub::{manager::SubscriptionManager, typed::Subscriber, SubscriptionId};
-use log::warn;
-use rpc::Result as RpcResult;
-use std::{collections::HashMap, sync::Arc};
+use std::{collections::HashMap, marker::PhantomData, sync::Arc};
 
+use super::{
+	client_err,
+	error::{Error, Result},
+	ChildStateBackend, StateBackend,
+};
+use crate::SubscriptionTaskExecutor;
+
+use futures::{future, stream, FutureExt, StreamExt};
+use jsonrpsee::{
+	core::{async_trait, Error as JsonRpseeError},
+	PendingSubscription,
+};
+use sc_client_api::{
+	Backend, BlockBackend, BlockchainEvents, CallExecutor, ExecutorProvider, ProofProvider,
+	StorageProvider,
+};
 use sc_rpc_api::state::ReadProof;
+use sp_api::{CallApiAt, Metadata, ProvideRuntimeApi};
 use sp_blockchain::{
 	CachedHeaderMetadata, Error as ClientError, HeaderBackend, HeaderMetadata,
 	Result as ClientResult,
@@ -42,19 +51,6 @@ use sp_core::{
 use sp_runtime::{generic::BlockId, traits::Block as BlockT};
 use sp_version::RuntimeVersion;
 
-use sp_api::{CallApiAt, Metadata, ProvideRuntimeApi};
-
-use super::{
-	client_err,
-	error::{Error, FutureResult, Result},
-	ChildStateBackend, StateBackend,
-};
-use sc_client_api::{
-	Backend, BlockBackend, BlockchainEvents, CallExecutor, ExecutorProvider, ProofProvider,
-	StorageNotification, StorageProvider,
-};
-use std::marker::PhantomData;
-
 /// Ranges to query in state_queryStorage.
 struct QueryStorageRange<Block: BlockT> {
 	/// Hashes of all the blocks in the range.
@@ -64,7 +60,7 @@ struct QueryStorageRange<Block: BlockT> {
 /// State API backend for full nodes.
 pub struct FullState<BE, Block: BlockT, Client> {
 	client: Arc<Client>,
-	subscriptions: SubscriptionManager,
+	executor: SubscriptionTaskExecutor,
 	_phantom: PhantomData<(BE, Block)>,
 	rpc_max_payload: Option<usize>,
 }
@@ -81,10 +77,10 @@ where
 	/// Create new state API backend for full nodes.
 	pub fn new(
 		client: Arc<Client>,
-		subscriptions: SubscriptionManager,
+		executor: SubscriptionTaskExecutor,
 		rpc_max_payload: Option<usize>,
 	) -> Self {
-		Self { client, subscriptions, _phantom: PhantomData, rpc_max_payload }
+		Self { client, executor, _phantom: PhantomData, rpc_max_payload }
 	}
 
 	/// Returns given block hash or best block hash if None is passed.
@@ -174,6 +170,7 @@ where
 	}
 }
 
+#[async_trait]
 impl<BE, Block, Client> StateBackend<Block, Client> for FullState<BE, Block, Client>
 where
 	Block: BlockT + 'static,
@@ -193,14 +190,13 @@ where
 		+ 'static,
 	Client::Api: Metadata<Block>,
 {
-	fn call(
+	async fn call(
 		&self,
 		block: Option<Block::Hash>,
 		method: String,
 		call_data: Bytes,
-	) -> FutureResult<Bytes> {
-		let r = self
-			.block_or_best(block)
+	) -> std::result::Result<Bytes, Error> {
+		self.block_or_best(block)
 			.and_then(|block| {
 				self.client
 					.executor()
@@ -213,43 +209,37 @@ where
 					)
 					.map(Into::into)
 			})
-			.map_err(client_err);
-		async move { r }.boxed()
+			.map_err(client_err)
 	}
 
-	fn storage_keys(
+	async fn storage_keys(
 		&self,
 		block: Option<Block::Hash>,
 		prefix: StorageKey,
-	) -> FutureResult<Vec<StorageKey>> {
-		let r = self
-			.block_or_best(block)
+	) -> std::result::Result<Vec<StorageKey>, Error> {
+		self.block_or_best(block)
 			.and_then(|block| self.client.storage_keys(&BlockId::Hash(block), &prefix))
-			.map_err(client_err);
-		async move { r }.boxed()
+			.map_err(client_err)
 	}
 
-	fn storage_pairs(
+	async fn storage_pairs(
 		&self,
 		block: Option<Block::Hash>,
 		prefix: StorageKey,
-	) -> FutureResult<Vec<(StorageKey, StorageData)>> {
-		let r = self
-			.block_or_best(block)
+	) -> std::result::Result<Vec<(StorageKey, StorageData)>, Error> {
+		self.block_or_best(block)
 			.and_then(|block| self.client.storage_pairs(&BlockId::Hash(block), &prefix))
-			.map_err(client_err);
-		async move { r }.boxed()
+			.map_err(client_err)
 	}
 
-	fn storage_keys_paged(
+	async fn storage_keys_paged(
 		&self,
 		block: Option<Block::Hash>,
 		prefix: Option<StorageKey>,
 		count: u32,
 		start_key: Option<StorageKey>,
-	) -> FutureResult<Vec<StorageKey>> {
-		let r = self
-			.block_or_best(block)
+	) -> std::result::Result<Vec<StorageKey>, Error> {
+		self.block_or_best(block)
 			.and_then(|block| {
 				self.client.storage_keys_iter(
 					&BlockId::Hash(block),
@@ -258,40 +248,36 @@ where
 				)
 			})
 			.map(|iter| iter.take(count as usize).collect())
-			.map_err(client_err);
-		async move { r }.boxed()
+			.map_err(client_err)
 	}
 
-	fn storage(
+	async fn storage(
 		&self,
 		block: Option<Block::Hash>,
 		key: StorageKey,
-	) -> FutureResult<Option<StorageData>> {
-		let r = self
-			.block_or_best(block)
+	) -> std::result::Result<Option<StorageData>, Error> {
+		self.block_or_best(block)
 			.and_then(|block| self.client.storage(&BlockId::Hash(block), &key))
-			.map_err(client_err);
-		async move { r }.boxed()
+			.map_err(client_err)
 	}
 
-	fn storage_size(
+	async fn storage_size(
 		&self,
 		block: Option<Block::Hash>,
 		key: StorageKey,
-	) -> FutureResult<Option<u64>> {
+	) -> std::result::Result<Option<u64>, Error> {
 		let block = match self.block_or_best(block) {
 			Ok(b) => b,
-			Err(e) => return async move { Err(client_err(e)) }.boxed(),
+			Err(e) => return Err(client_err(e)),
 		};
 
 		match self.client.storage(&BlockId::Hash(block), &key) {
-			Ok(Some(d)) => return async move { Ok(Some(d.0.len() as u64)) }.boxed(),
-			Err(e) => return async move { Err(client_err(e)) }.boxed(),
+			Ok(Some(d)) => return Ok(Some(d.0.len() as u64)),
+			Err(e) => return Err(client_err(e)),
 			Ok(None) => {},
 		}
 
-		let r = self
-			.client
+		self.client
 			.storage_pairs(&BlockId::Hash(block), &key)
 			.map(|kv| {
 				let item_sum = kv.iter().map(|(_, v)| v.0.len() as u64).sum::<u64>();
@@ -301,48 +287,46 @@ where
 					None
 				}
 			})
-			.map_err(client_err);
-		async move { r }.boxed()
+			.map_err(client_err)
 	}
 
-	fn storage_hash(
+	async fn storage_hash(
 		&self,
 		block: Option<Block::Hash>,
 		key: StorageKey,
-	) -> FutureResult<Option<Block::Hash>> {
-		let r = self
-			.block_or_best(block)
+	) -> std::result::Result<Option<Block::Hash>, Error> {
+		self.block_or_best(block)
 			.and_then(|block| self.client.storage_hash(&BlockId::Hash(block), &key))
-			.map_err(client_err);
-		async move { r }.boxed()
+			.map_err(client_err)
 	}
 
-	fn metadata(&self, block: Option<Block::Hash>) -> FutureResult<Bytes> {
-		let r = self.block_or_best(block).map_err(client_err).and_then(|block| {
+	async fn metadata(&self, block: Option<Block::Hash>) -> std::result::Result<Bytes, Error> {
+		self.block_or_best(block).map_err(client_err).and_then(|block| {
 			self.client
 				.runtime_api()
 				.metadata(&BlockId::Hash(block))
 				.map(Into::into)
 				.map_err(|e| Error::Client(Box::new(e)))
-		});
-		async move { r }.boxed()
+		})
 	}
 
-	fn runtime_version(&self, block: Option<Block::Hash>) -> FutureResult<RuntimeVersion> {
-		let r = self.block_or_best(block).map_err(client_err).and_then(|block| {
+	async fn runtime_version(
+		&self,
+		block: Option<Block::Hash>,
+	) -> std::result::Result<RuntimeVersion, Error> {
+		self.block_or_best(block).map_err(client_err).and_then(|block| {
 			self.client
 				.runtime_version_at(&BlockId::Hash(block))
 				.map_err(|e| Error::Client(Box::new(e)))
-		});
-		async move { r }.boxed()
+		})
 	}
 
-	fn query_storage(
+	async fn query_storage(
 		&self,
 		from: Block::Hash,
 		to: Option<Block::Hash>,
 		keys: Vec<StorageKey>,
-	) -> FutureResult<Vec<StorageChangeSet<Block::Hash>>> {
+	) -> std::result::Result<Vec<StorageChangeSet<Block::Hash>>, Error> {
 		let call_fn = move || {
 			let range = self.query_storage_range(from, to)?;
 			let mut changes = Vec::new();
@@ -350,168 +334,151 @@ where
 			self.query_storage_unfiltered(&range, &keys, &mut last_values, &mut changes)?;
 			Ok(changes)
 		};
-
-		let r = call_fn();
-		async move { r }.boxed()
+		call_fn()
 	}
 
-	fn query_storage_at(
+	async fn query_storage_at(
 		&self,
 		keys: Vec<StorageKey>,
 		at: Option<Block::Hash>,
-	) -> FutureResult<Vec<StorageChangeSet<Block::Hash>>> {
+	) -> std::result::Result<Vec<StorageChangeSet<Block::Hash>>, Error> {
 		let at = at.unwrap_or_else(|| self.client.info().best_hash);
-		self.query_storage(at, Some(at), keys)
+		self.query_storage(at, Some(at), keys).await
 	}
 
-	fn read_proof(
+	async fn read_proof(
 		&self,
 		block: Option<Block::Hash>,
 		keys: Vec<StorageKey>,
-	) -> FutureResult<ReadProof<Block::Hash>> {
-		let r = self
-			.block_or_best(block)
+	) -> std::result::Result<ReadProof<Block::Hash>, Error> {
+		self.block_or_best(block)
 			.and_then(|block| {
 				self.client
 					.read_proof(&BlockId::Hash(block), &mut keys.iter().map(|key| key.0.as_ref()))
 					.map(|proof| proof.iter_nodes().map(|node| node.into()).collect())
 					.map(|proof| ReadProof { at: block, proof })
 			})
-			.map_err(client_err);
-		async move { r }.boxed()
+			.map_err(client_err)
 	}
 
-	fn subscribe_runtime_version(
-		&self,
-		_meta: crate::Metadata,
-		subscriber: Subscriber<RuntimeVersion>,
-	) {
-		self.subscriptions.add(subscriber, |sink| {
-			let version = self
-				.block_or_best(None)
-				.and_then(|block| {
-					self.client.runtime_version_at(&BlockId::Hash(block)).map_err(Into::into)
-				})
-				.map_err(client_err)
-				.map_err(Into::into);
+	fn subscribe_runtime_version(&self, pending: PendingSubscription) {
+		let client = self.client.clone();
 
-			let client = self.client.clone();
-			let mut previous_version = version.clone();
+		let initial = match self
+			.block_or_best(None)
+			.and_then(|block| {
+				self.client.runtime_version_at(&BlockId::Hash(block)).map_err(Into::into)
+			})
+			.map_err(|e| Error::Client(Box::new(e)))
+		{
+			Ok(initial) => initial,
+			Err(e) => {
+				pending.reject(JsonRpseeError::from(e));
+				return
+			},
+		};
 
-			// A stream of all best blocks.
-			let stream =
-				client.import_notification_stream().filter(|n| future::ready(n.is_new_best));
+		let mut previous_version = initial.clone();
 
-			let stream = stream.filter_map(move |n| {
+		// A stream of new versions
+		let version_stream = client
+			.import_notification_stream()
+			.filter(|n| future::ready(n.is_new_best))
+			.filter_map(move |n| {
 				let version = client
 					.runtime_version_at(&BlockId::hash(n.hash))
-					.map_err(|e| Error::Client(Box::new(e)))
-					.map_err(Into::into);
+					.map_err(|e| Error::Client(Box::new(e)));
 
-				if previous_version != version {
-					previous_version = version.clone();
-					future::ready(Some(Ok::<_, ()>(version)))
-				} else {
-					future::ready(None)
+				match version {
+					Ok(version) if version != previous_version => {
+						previous_version = version.clone();
+						future::ready(Some(version))
+					},
+					_ => future::ready(None),
 				}
 			});
 
-			stream::iter(vec![Ok(version)])
-				.chain(stream)
-				.forward(sink.sink_map_err(|e| warn!("Error sending notifications: {:?}", e)))
-				// we ignore the resulting Stream (if the first stream is over we are unsubscribed)
-				.map(|_| ())
-		});
+		let stream = futures::stream::once(future::ready(initial)).chain(version_stream);
+
+		let fut = async move {
+			if let Some(mut sink) = pending.accept() {
+				sink.pipe_from_stream(stream).await;
+			}
+		}
+		.boxed();
+
+		self.executor
+			.spawn("substrate-rpc-subscription", Some("rpc"), fut.map(drop).boxed());
 	}
 
-	fn unsubscribe_runtime_version(
-		&self,
-		_meta: Option<crate::Metadata>,
-		id: SubscriptionId,
-	) -> RpcResult<bool> {
-		Ok(self.subscriptions.cancel(id))
-	}
-
-	fn subscribe_storage(
-		&self,
-		_meta: crate::Metadata,
-		subscriber: Subscriber<StorageChangeSet<Block::Hash>>,
-		keys: Option<Vec<StorageKey>>,
-	) {
-		let keys = Into::<Option<Vec<_>>>::into(keys);
+	fn subscribe_storage(&self, pending: PendingSubscription, keys: Option<Vec<StorageKey>>) {
 		let stream = match self.client.storage_changes_notification_stream(keys.as_deref(), None) {
 			Ok(stream) => stream,
-			Err(err) => {
-				let _ = subscriber.reject(client_err(err).into());
+			Err(blockchain_err) => {
+				pending.reject(JsonRpseeError::from(Error::Client(Box::new(blockchain_err))));
 				return
 			},
 		};
 
 		// initial values
-		let initial = stream::iter(
-			keys.map(|keys| {
-				let block = self.client.info().best_hash;
-				let changes = keys
-					.into_iter()
-					.map(|key| {
-						let v = self.client.storage(&BlockId::Hash(block), &key).ok().flatten();
-						(key, v)
-					})
-					.collect();
-				vec![Ok(Ok(StorageChangeSet { block, changes }))]
-			})
-			.unwrap_or_default(),
-		);
+		let initial = stream::iter(keys.map(|keys| {
+			let block = self.client.info().best_hash;
+			let changes = keys
+				.into_iter()
+				.map(|key| {
+					let v = self.client.storage(&BlockId::Hash(block), &key).ok().flatten();
+					(key, v)
+				})
+				.collect();
+			StorageChangeSet { block, changes }
+		}));
 
-		self.subscriptions.add(subscriber, |sink| {
-			let stream = stream.map(|StorageNotification { block, changes }| {
-				Ok(Ok::<_, rpc::Error>(StorageChangeSet {
-					block,
-					changes: changes
-						.iter()
-						.filter_map(|(o_sk, k, v)| o_sk.is_none().then(|| (k.clone(), v.cloned())))
-						.collect(),
-				}))
-			});
-
-			initial
-				.chain(stream)
-				.forward(sink.sink_map_err(|e| warn!("Error sending notifications: {:?}", e)))
-				// we ignore the resulting Stream (if the first stream is over we are unsubscribed)
-				.map(|_| ())
+		// let storage_stream = stream.map(|(block, changes)| StorageChangeSet {
+		let storage_stream = stream.map(|storage_notif| StorageChangeSet {
+			block: storage_notif.block,
+			changes: storage_notif
+				.changes
+				.iter()
+				.filter_map(|(o_sk, k, v)| o_sk.is_none().then(|| (k.clone(), v.cloned())))
+				.collect(),
 		});
+
+		let stream = initial
+			.chain(storage_stream)
+			.filter(|storage| future::ready(!storage.changes.is_empty()));
+
+		let fut = async move {
+			if let Some(mut sink) = pending.accept() {
+				sink.pipe_from_stream(stream).await;
+			}
+		}
+		.boxed();
+
+		self.executor
+			.spawn("substrate-rpc-subscription", Some("rpc"), fut.map(drop).boxed());
 	}
 
-	fn unsubscribe_storage(
-		&self,
-		_meta: Option<crate::Metadata>,
-		id: SubscriptionId,
-	) -> RpcResult<bool> {
-		Ok(self.subscriptions.cancel(id))
-	}
-
-	fn trace_block(
+	async fn trace_block(
 		&self,
 		block: Block::Hash,
 		targets: Option<String>,
 		storage_keys: Option<String>,
 		methods: Option<String>,
-	) -> FutureResult<sp_rpc::tracing::TraceBlockResponse> {
-		let block_executor = sc_tracing::block::BlockExecutor::new(
+	) -> std::result::Result<sp_rpc::tracing::TraceBlockResponse, Error> {
+		sc_tracing::block::BlockExecutor::new(
 			self.client.clone(),
 			block,
 			targets,
 			storage_keys,
 			methods,
 			self.rpc_max_payload,
-		);
-		let r = block_executor
-			.trace_block()
-			.map_err(|e| invalid_block::<Block>(block, None, e.to_string()));
-		async move { r }.boxed()
+		)
+		.trace_block()
+		.map_err(|e| invalid_block::<Block>(block, None, e.to_string()))
 	}
 }
 
+#[async_trait]
 impl<BE, Block, Client> ChildStateBackend<Block, Client> for FullState<BE, Block, Client>
 where
 	Block: BlockT + 'static,
@@ -530,14 +497,13 @@ where
 		+ 'static,
 	Client::Api: Metadata<Block>,
 {
-	fn read_child_proof(
+	async fn read_child_proof(
 		&self,
 		block: Option<Block::Hash>,
 		storage_key: PrefixedStorageKey,
 		keys: Vec<StorageKey>,
-	) -> FutureResult<ReadProof<Block::Hash>> {
-		let r = self
-			.block_or_best(block)
+	) -> std::result::Result<ReadProof<Block::Hash>, Error> {
+		self.block_or_best(block)
 			.and_then(|block| {
 				let child_info = match ChildType::from_prefixed_key(&storage_key) {
 					Some((ChildType::ParentKeyId, storage_key)) =>
@@ -553,19 +519,16 @@ where
 					.map(|proof| proof.iter_nodes().map(|node| node.into()).collect())
 					.map(|proof| ReadProof { at: block, proof })
 			})
-			.map_err(client_err);
-
-		async move { r }.boxed()
+			.map_err(client_err)
 	}
 
-	fn storage_keys(
+	async fn storage_keys(
 		&self,
 		block: Option<Block::Hash>,
 		storage_key: PrefixedStorageKey,
 		prefix: StorageKey,
-	) -> FutureResult<Vec<StorageKey>> {
-		let r = self
-			.block_or_best(block)
+	) -> std::result::Result<Vec<StorageKey>, Error> {
+		self.block_or_best(block)
 			.and_then(|block| {
 				let child_info = match ChildType::from_prefixed_key(&storage_key) {
 					Some((ChildType::ParentKeyId, storage_key)) =>
@@ -574,21 +537,18 @@ where
 				};
 				self.client.child_storage_keys(&BlockId::Hash(block), &child_info, &prefix)
 			})
-			.map_err(client_err);
-
-		async move { r }.boxed()
+			.map_err(client_err)
 	}
 
-	fn storage_keys_paged(
+	async fn storage_keys_paged(
 		&self,
 		block: Option<Block::Hash>,
 		storage_key: PrefixedStorageKey,
 		prefix: Option<StorageKey>,
 		count: u32,
 		start_key: Option<StorageKey>,
-	) -> FutureResult<Vec<StorageKey>> {
-		let r = self
-			.block_or_best(block)
+	) -> std::result::Result<Vec<StorageKey>, Error> {
+		self.block_or_best(block)
 			.and_then(|block| {
 				let child_info = match ChildType::from_prefixed_key(&storage_key) {
 					Some((ChildType::ParentKeyId, storage_key)) =>
@@ -603,19 +563,16 @@ where
 				)
 			})
 			.map(|iter| iter.take(count as usize).collect())
-			.map_err(client_err);
-
-		async move { r }.boxed()
+			.map_err(client_err)
 	}
 
-	fn storage(
+	async fn storage(
 		&self,
 		block: Option<Block::Hash>,
 		storage_key: PrefixedStorageKey,
 		key: StorageKey,
-	) -> FutureResult<Option<StorageData>> {
-		let r = self
-			.block_or_best(block)
+	) -> std::result::Result<Option<StorageData>, Error> {
+		self.block_or_best(block)
 			.and_then(|block| {
 				let child_info = match ChildType::from_prefixed_key(&storage_key) {
 					Some((ChildType::ParentKeyId, storage_key)) =>
@@ -624,28 +581,25 @@ where
 				};
 				self.client.child_storage(&BlockId::Hash(block), &child_info, &key)
 			})
-			.map_err(client_err);
-
-		async move { r }.boxed()
+			.map_err(client_err)
 	}
 
-	fn storage_entries(
+	async fn storage_entries(
 		&self,
 		block: Option<Block::Hash>,
 		storage_key: PrefixedStorageKey,
 		keys: Vec<StorageKey>,
-	) -> FutureResult<Vec<Option<StorageData>>> {
-		let child_info = match ChildType::from_prefixed_key(&storage_key) {
-			Some((ChildType::ParentKeyId, storage_key)) =>
-				Arc::new(ChildInfo::new_default(storage_key)),
-			None => return err(client_err(sp_blockchain::Error::InvalidChildStorageKey)).boxed(),
-		};
-		let block = match self.block_or_best(block) {
-			Ok(b) => b,
-			Err(e) => return err(client_err(e)).boxed(),
+	) -> std::result::Result<Vec<Option<StorageData>>, Error> {
+		let child_info = if let Some((ChildType::ParentKeyId, storage_key)) =
+			ChildType::from_prefixed_key(&storage_key)
+		{
+			Arc::new(ChildInfo::new_default(storage_key))
+		} else {
+			return Err(client_err(sp_blockchain::Error::InvalidChildStorageKey))
 		};
+		let block = self.block_or_best(block).map_err(client_err)?;
 		let client = self.client.clone();
-		try_join_all(keys.into_iter().map(move |key| {
+		future::try_join_all(keys.into_iter().map(move |key| {
 			let res = client
 				.clone()
 				.child_storage(&BlockId::Hash(block), &child_info, &key)
@@ -653,17 +607,16 @@ where
 
 			async move { res }
 		}))
-		.boxed()
+		.await
 	}
 
-	fn storage_hash(
+	async fn storage_hash(
 		&self,
 		block: Option<Block::Hash>,
 		storage_key: PrefixedStorageKey,
 		key: StorageKey,
-	) -> FutureResult<Option<Block::Hash>> {
-		let r = self
-			.block_or_best(block)
+	) -> std::result::Result<Option<Block::Hash>, Error> {
+		self.block_or_best(block)
 			.and_then(|block| {
 				let child_info = match ChildType::from_prefixed_key(&storage_key) {
 					Some((ChildType::ParentKeyId, storage_key)) =>
@@ -672,9 +625,7 @@ where
 				};
 				self.client.child_storage_hash(&BlockId::Hash(block), &child_info, &key)
 			})
-			.map_err(client_err);
-
-		async move { r }.boxed()
+			.map_err(client_err)
 	}
 }
 

substrate/client/rpc/src/state/tests.rs

@@ -18,9 +18,13 @@
 
 use self::error::Error;
 use super::*;
-use crate::testing::TaskExecutor;
+use crate::testing::{test_executor, timeout_secs};
 use assert_matches::assert_matches;
-use futures::{executor, StreamExt};
+use futures::executor;
+use jsonrpsee::{
+	core::Error as RpcError,
+	types::{error::CallError as RpcCallError, EmptyParams, ErrorObject},
+};
 use sc_block_builder::BlockBuilderProvider;
 use sc_rpc_api::DenyUnsafe;
 use sp_consensus::BlockOrigin;
@@ -36,8 +40,8 @@ fn prefixed_storage_key() -> PrefixedStorageKey {
 	child_info.prefixed_storage_key()
 }
 
-#[test]
-fn should_return_storage() {
+#[tokio::test]
+async fn should_return_storage() {
 	const KEY: &[u8] = b":mock";
 	const VALUE: &[u8] = b"hello world";
 	const CHILD_VALUE: &[u8] = b"hello world !";
@@ -51,49 +55,46 @@ fn should_return_storage() {
 		.add_extra_storage(b":map:acc2".to_vec(), vec![1, 2, 3])
 		.build();
 	let genesis_hash = client.genesis_hash();
-	let (client, child) = new_full(
-		Arc::new(client),
-		SubscriptionManager::new(Arc::new(TaskExecutor)),
-		DenyUnsafe::No,
-		None,
-	);
+	let (client, child) = new_full(Arc::new(client), test_executor(), DenyUnsafe::No, None);
 	let key = StorageKey(KEY.to_vec());
 
 	assert_eq!(
-		executor::block_on(client.storage(key.clone(), Some(genesis_hash).into()))
+		client
+			.storage(key.clone(), Some(genesis_hash).into())
+			.await
 			.map(|x| x.map(|x| x.0.len()))
 			.unwrap()
 			.unwrap() as usize,
 		VALUE.len(),
 	);
 	assert_matches!(
-		executor::block_on(client.storage_hash(key.clone(), Some(genesis_hash).into()))
+		client
+			.storage_hash(key.clone(), Some(genesis_hash).into())
+			.await
 			.map(|x| x.is_some()),
 		Ok(true)
 	);
 	assert_eq!(
-		executor::block_on(client.storage_size(key.clone(), None)).unwrap().unwrap() as usize,
+		client.storage_size(key.clone(), None).await.unwrap().unwrap() as usize,
 		VALUE.len(),
 	);
 	assert_eq!(
-		executor::block_on(client.storage_size(StorageKey(b":map".to_vec()), None))
-			.unwrap()
-			.unwrap() as usize,
+		client.storage_size(StorageKey(b":map".to_vec()), None).await.unwrap().unwrap() as usize,
 		2 + 3,
 	);
 	assert_eq!(
-		executor::block_on(
-			child
-				.storage(prefixed_storage_key(), key, Some(genesis_hash).into())
-				.map(|x| x.map(|x| x.unwrap().0.len()))
-		)
-		.unwrap() as usize,
+		child
+			.storage(prefixed_storage_key(), key, Some(genesis_hash).into())
+			.await
+			.map(|x| x.map(|x| x.0.len()))
+			.unwrap()
+			.unwrap() as usize,
 		CHILD_VALUE.len(),
 	);
 }
 
-#[test]
-fn should_return_storage_entries() {
+#[tokio::test]
+async fn should_return_storage_entries() {
 	const KEY1: &[u8] = b":mock";
 	const KEY2: &[u8] = b":turtle";
 	const VALUE: &[u8] = b"hello world";
@@ -107,22 +108,15 @@ fn should_return_storage_entries() {
 		.add_extra_child_storage(&child_info, KEY2.to_vec(), CHILD_VALUE2.to_vec())
 		.build();
 	let genesis_hash = client.genesis_hash();
-	let (_client, child) = new_full(
-		Arc::new(client),
-		SubscriptionManager::new(Arc::new(TaskExecutor)),
-		DenyUnsafe::No,
-		None,
-	);
+	let (_client, child) = new_full(Arc::new(client), test_executor(), DenyUnsafe::No, None);
 
 	let keys = &[StorageKey(KEY1.to_vec()), StorageKey(KEY2.to_vec())];
 	assert_eq!(
-		executor::block_on(child.storage_entries(
-			prefixed_storage_key(),
-			keys.to_vec(),
-			Some(genesis_hash).into()
-		))
-		.map(|x| x.into_iter().map(|x| x.map(|x| x.0.len()).unwrap()).sum::<usize>())
-		.unwrap(),
+		child
+			.storage_entries(prefixed_storage_key(), keys.to_vec(), Some(genesis_hash).into())
+			.await
+			.map(|x| x.into_iter().map(|x| x.map(|x| x.0.len()).unwrap()).sum::<usize>())
+			.unwrap(),
 		CHILD_VALUE1.len() + CHILD_VALUE2.len()
 	);
 
@@ -130,18 +124,16 @@ fn should_return_storage_entries() {
 	let mut failing_keys = vec![StorageKey(b":soup".to_vec())];
 	failing_keys.extend_from_slice(keys);
 	assert_matches!(
-		executor::block_on(child.storage_entries(
-			prefixed_storage_key(),
-			failing_keys,
-			Some(genesis_hash).into()
-		))
-		.map(|x| x.iter().all(|x| x.is_some())),
+		child
+			.storage_entries(prefixed_storage_key(), failing_keys, Some(genesis_hash).into())
+			.await
+			.map(|x| x.iter().all(|x| x.is_some())),
 		Ok(false)
 	);
 }
 
-#[test]
-fn should_return_child_storage() {
+#[tokio::test]
+async fn should_return_child_storage() {
 	let child_info = ChildInfo::new_default(STORAGE_KEY);
 	let client = Arc::new(
 		substrate_test_runtime_client::TestClientBuilder::new()
@@ -149,49 +141,30 @@ fn should_return_child_storage() {
 			.build(),
 	);
 	let genesis_hash = client.genesis_hash();
-	let (_client, child) =
-		new_full(client, SubscriptionManager::new(Arc::new(TaskExecutor)), DenyUnsafe::No, None);
+	let (_client, child) = new_full(client, test_executor(), DenyUnsafe::No, None);
 	let child_key = prefixed_storage_key();
 	let key = StorageKey(b"key".to_vec());
 
 	assert_matches!(
-		executor::block_on(child.storage(
+		child.storage(
 			child_key.clone(),
 			key.clone(),
 			Some(genesis_hash).into(),
-		)),
+		).await,
 		Ok(Some(StorageData(ref d))) if d[0] == 42 && d.len() == 1
 	);
-
-	// should fail if key does not exist.
-	let failing_key = StorageKey(b":soup".to_vec());
 	assert_matches!(
-		executor::block_on(child.storage(
-			prefixed_storage_key(),
-			failing_key,
-			Some(genesis_hash).into()
-		))
-		.map(|x| x.is_some()),
-		Ok(false)
-	);
-
-	assert_matches!(
-		executor::block_on(child.storage_hash(
-			child_key.clone(),
-			key.clone(),
-			Some(genesis_hash).into(),
-		))
-		.map(|x| x.is_some()),
+		child
+			.storage_hash(child_key.clone(), key.clone(), Some(genesis_hash).into(),)
+			.await
+			.map(|x| x.is_some()),
 		Ok(true)
 	);
-	assert_matches!(
-		executor::block_on(child.storage_size(child_key.clone(), key.clone(), None)),
-		Ok(Some(1))
-	);
+	assert_matches!(child.storage_size(child_key.clone(), key.clone(), None).await, Ok(Some(1)));
 }
 
-#[test]
-fn should_return_child_storage_entries() {
+#[tokio::test]
+async fn should_return_child_storage_entries() {
 	let child_info = ChildInfo::new_default(STORAGE_KEY);
 	let client = Arc::new(
 		substrate_test_runtime_client::TestClientBuilder::new()
@@ -200,17 +173,14 @@ fn should_return_child_storage_entries() {
 			.build(),
 	);
 	let genesis_hash = client.genesis_hash();
-	let (_client, child) =
-		new_full(client, SubscriptionManager::new(Arc::new(TaskExecutor)), DenyUnsafe::No, None);
+	let (_client, child) = new_full(client, test_executor(), DenyUnsafe::No, None);
 	let child_key = prefixed_storage_key();
 	let keys = vec![StorageKey(b"key1".to_vec()), StorageKey(b"key2".to_vec())];
 
-	let res = executor::block_on(child.storage_entries(
-		child_key.clone(),
-		keys.clone(),
-		Some(genesis_hash).into(),
-	))
-	.unwrap();
+	let res = child
+		.storage_entries(child_key.clone(), keys.clone(), Some(genesis_hash).into())
+		.await
+		.unwrap();
 
 	assert_matches!(
 		res[0],
@@ -232,46 +202,37 @@ fn should_return_child_storage_entries() {
 		Ok(true)
 	);
 	assert_matches!(
-		executor::block_on(child.storage_size(child_key.clone(), keys[0].clone(), None)),
+		child.storage_size(child_key.clone(), keys[0].clone(), None).await,
 		Ok(Some(1))
 	);
 }
 
-#[test]
-fn should_call_contract() {
+#[tokio::test]
+async fn should_call_contract() {
 	let client = Arc::new(substrate_test_runtime_client::new());
 	let genesis_hash = client.genesis_hash();
-	let (client, _child) =
-		new_full(client, SubscriptionManager::new(Arc::new(TaskExecutor)), DenyUnsafe::No, None);
+	let (client, _child) = new_full(client, test_executor(), DenyUnsafe::No, None);
+
+	use jsonrpsee::{core::Error, types::error::CallError};
 
 	assert_matches!(
-		executor::block_on(client.call(
-			"balanceOf".into(),
-			Bytes(vec![1, 2, 3]),
-			Some(genesis_hash).into()
-		)),
-		Err(Error::Client(_))
+		client
+			.call("balanceOf".into(), Bytes(vec![1, 2, 3]), Some(genesis_hash).into())
+			.await,
+		Err(Error::Call(CallError::Failed(_)))
 	)
 }
 
-#[test]
-fn should_notify_about_storage_changes() {
-	let (subscriber, id, mut transport) = Subscriber::new_test("test");
-
-	{
+#[tokio::test]
+async fn should_notify_about_storage_changes() {
+	let mut sub = {
 		let mut client = Arc::new(substrate_test_runtime_client::new());
-		let (api, _child) = new_full(
-			client.clone(),
-			SubscriptionManager::new(Arc::new(TaskExecutor)),
-			DenyUnsafe::No,
-			None,
-		);
+		let (api, _child) = new_full(client.clone(), test_executor(), DenyUnsafe::No, None);
 
-		api.subscribe_storage(Default::default(), subscriber, None.into());
-
-		// assert id assigned
-		assert!(matches!(executor::block_on(id), Ok(Ok(SubscriptionId::String(_)))));
+		let api_rpc = api.into_rpc();
+		let sub = api_rpc.subscribe("state_subscribeStorage", EmptyParams::new()).await.unwrap();
 
+		// Cause a change:
 		let mut builder = client.new_block(Default::default()).unwrap();
 		builder
 			.push_transfer(runtime::Transfer {
@@ -282,38 +243,32 @@ fn should_notify_about_storage_changes() {
 			})
 			.unwrap();
 		let block = builder.build().unwrap().block;
-		executor::block_on(client.import(BlockOrigin::Own, block)).unwrap();
-	}
+		client.import(BlockOrigin::Own, block).await.unwrap();
 
-	// Check notification sent to transport
-	executor::block_on((&mut transport).take(2).collect::<Vec<_>>());
-	assert!(executor::block_on(transport.next()).is_none());
+		sub
+	};
+
+	// We should get a message back on our subscription about the storage change:
+	// NOTE: previous versions of the subscription code used to return an empty value for the
+	// "initial" storage change here
+	assert_matches!(timeout_secs(1, sub.next::<StorageChangeSet<H256>>()).await, Ok(Some(_)));
+	assert_matches!(timeout_secs(1, sub.next::<StorageChangeSet<H256>>()).await, Ok(None));
 }
 
-#[test]
-fn should_send_initial_storage_changes_and_notifications() {
-	let (subscriber, id, mut transport) = Subscriber::new_test("test");
-
-	{
+#[tokio::test]
+async fn should_send_initial_storage_changes_and_notifications() {
+	let mut sub = {
 		let mut client = Arc::new(substrate_test_runtime_client::new());
-		let (api, _child) = new_full(
-			client.clone(),
-			SubscriptionManager::new(Arc::new(TaskExecutor)),
-			DenyUnsafe::No,
-			None,
-		);
+		let (api, _child) = new_full(client.clone(), test_executor(), DenyUnsafe::No, None);
 
 		let alice_balance_key =
 			blake2_256(&runtime::system::balance_of_key(AccountKeyring::Alice.into()));
 
-		api.subscribe_storage(
-			Default::default(),
-			subscriber,
-			Some(vec![StorageKey(alice_balance_key.to_vec())]).into(),
-		);
-
-		// assert id assigned
-		assert!(matches!(executor::block_on(id), Ok(Ok(SubscriptionId::String(_)))));
+		let api_rpc = api.into_rpc();
+		let sub = api_rpc
+			.subscribe("state_subscribeStorage", [[StorageKey(alice_balance_key.to_vec())]])
+			.await
+			.unwrap();
 
 		let mut builder = client.new_block(Default::default()).unwrap();
 		builder
@@ -325,23 +280,22 @@ fn should_send_initial_storage_changes_and_notifications() {
 			})
 			.unwrap();
 		let block = builder.build().unwrap().block;
-		executor::block_on(client.import(BlockOrigin::Own, block)).unwrap();
-	}
+		client.import(BlockOrigin::Own, block).await.unwrap();
 
-	// Check for the correct number of notifications
-	executor::block_on((&mut transport).take(2).collect::<Vec<_>>());
-	assert!(executor::block_on(transport.next()).is_none());
+		sub
+	};
+
+	assert_matches!(timeout_secs(1, sub.next::<StorageChangeSet<H256>>()).await, Ok(Some(_)));
+	assert_matches!(timeout_secs(1, sub.next::<StorageChangeSet<H256>>()).await, Ok(Some(_)));
+
+	// No more messages to follow
+	assert_matches!(timeout_secs(1, sub.next::<StorageChangeSet<H256>>()).await, Ok(None));
 }
 
-#[test]
-fn should_query_storage() {
-	fn run_tests(mut client: Arc<TestClient>) {
-		let (api, _child) = new_full(
-			client.clone(),
-			SubscriptionManager::new(Arc::new(TaskExecutor)),
-			DenyUnsafe::No,
-			None,
-		);
+#[tokio::test]
+async fn should_query_storage() {
+	async fn run_tests(mut client: Arc<TestClient>) {
+		let (api, _child) = new_full(client.clone(), test_executor(), DenyUnsafe::No, None);
 
 		let mut add_block = |nonce| {
 			let mut builder = client.new_block(Default::default()).unwrap();
@@ -393,7 +347,7 @@ fn should_query_storage() {
 		let keys = (1..6).map(|k| StorageKey(vec![k])).collect::<Vec<_>>();
 		let result = api.query_storage(keys.clone(), genesis_hash, Some(block1_hash).into());
 
-		assert_eq!(executor::block_on(result).unwrap(), expected);
+		assert_eq!(result.await.unwrap(), expected);
 
 		// Query all changes
 		let result = api.query_storage(keys.clone(), genesis_hash, None.into());
@@ -406,23 +360,28 @@ fn should_query_storage() {
 				(StorageKey(vec![5]), Some(StorageData(vec![1]))),
 			],
 		});
-		assert_eq!(executor::block_on(result).unwrap(), expected);
+		assert_eq!(result.await.unwrap(), expected);
 
 		// Query changes up to block2.
 		let result = api.query_storage(keys.clone(), genesis_hash, Some(block2_hash));
 
-		assert_eq!(executor::block_on(result).unwrap(), expected);
+		assert_eq!(result.await.unwrap(), expected);
 
 		// Inverted range.
 		let result = api.query_storage(keys.clone(), block1_hash, Some(genesis_hash));
 
 		assert_eq!(
-			executor::block_on(result).map_err(|e| e.to_string()),
-			Err(Error::InvalidBlockRange {
-				from: format!("1 ({:?})", block1_hash),
-				to: format!("0 ({:?})", genesis_hash),
-				details: "from number > to number".to_owned(),
-			})
+			result.await.map_err(|e| e.to_string()),
+			Err(RpcError::Call(RpcCallError::Custom(ErrorObject::owned(
+				4001,
+				Error::InvalidBlockRange {
+					from: format!("1 ({:?})", block1_hash),
+					to: format!("0 ({:?})", genesis_hash),
+					details: "from number > to number".to_owned(),
+				}
+				.to_string(),
+				None::<()>,
+			))))
 			.map_err(|e| e.to_string())
 		);
 
@@ -433,15 +392,20 @@ fn should_query_storage() {
 		let result = api.query_storage(keys.clone(), genesis_hash, Some(random_hash1));
 
 		assert_eq!(
-			executor::block_on(result).map_err(|e| e.to_string()),
-			Err(Error::InvalidBlockRange {
-				from: format!("{:?}", genesis_hash),
-				to: format!("{:?}", Some(random_hash1)),
-				details: format!(
-					"UnknownBlock: Header was not found in the database: {:?}",
-					random_hash1
-				),
-			})
+			result.await.map_err(|e| e.to_string()),
+			Err(RpcError::Call(RpcCallError::Custom(ErrorObject::owned(
+				4001,
+				Error::InvalidBlockRange {
+					from: format!("{:?}", genesis_hash),
+					to: format!("{:?}", Some(random_hash1)),
+					details: format!(
+						"UnknownBlock: Header was not found in the database: {:?}",
+						random_hash1
+					),
+				}
+				.to_string(),
+				None::<()>,
+			))))
 			.map_err(|e| e.to_string())
 		);
 
@@ -449,15 +413,20 @@ fn should_query_storage() {
 		let result = api.query_storage(keys.clone(), random_hash1, Some(genesis_hash));
 
 		assert_eq!(
-			executor::block_on(result).map_err(|e| e.to_string()),
-			Err(Error::InvalidBlockRange {
-				from: format!("{:?}", random_hash1),
-				to: format!("{:?}", Some(genesis_hash)),
-				details: format!(
-					"UnknownBlock: Header was not found in the database: {:?}",
-					random_hash1
-				),
-			})
+			result.await.map_err(|e| e.to_string()),
+			Err(RpcError::Call(RpcCallError::Custom(ErrorObject::owned(
+				4001,
+				Error::InvalidBlockRange {
+					from: format!("{:?}", random_hash1),
+					to: format!("{:?}", Some(genesis_hash)),
+					details: format!(
+						"UnknownBlock: Header was not found in the database: {:?}",
+						random_hash1
+					),
+				}
+				.to_string(),
+				None::<()>,
+			))))
 			.map_err(|e| e.to_string()),
 		);
 
@@ -465,15 +434,20 @@ fn should_query_storage() {
 		let result = api.query_storage(keys.clone(), random_hash1, None);
 
 		assert_eq!(
-			executor::block_on(result).map_err(|e| e.to_string()),
-			Err(Error::InvalidBlockRange {
-				from: format!("{:?}", random_hash1),
-				to: format!("{:?}", Some(block2_hash)), // Best block hash.
-				details: format!(
-					"UnknownBlock: Header was not found in the database: {:?}",
-					random_hash1
-				),
-			})
+			result.await.map_err(|e| e.to_string()),
+			Err(RpcError::Call(RpcCallError::Custom(ErrorObject::owned(
+				4001,
+				Error::InvalidBlockRange {
+					from: format!("{:?}", random_hash1),
+					to: format!("{:?}", Some(block2_hash)), // Best block hash.
+					details: format!(
+						"UnknownBlock: Header was not found in the database: {:?}",
+						random_hash1
+					),
+				}
+				.to_string(),
+				None::<()>,
+			))))
 			.map_err(|e| e.to_string()),
 		);
 
@@ -481,15 +455,20 @@ fn should_query_storage() {
 		let result = api.query_storage(keys.clone(), random_hash1, Some(random_hash2));
 
 		assert_eq!(
-			executor::block_on(result).map_err(|e| e.to_string()),
-			Err(Error::InvalidBlockRange {
-				from: format!("{:?}", random_hash1), // First hash not found.
-				to: format!("{:?}", Some(random_hash2)),
-				details: format!(
-					"UnknownBlock: Header was not found in the database: {:?}",
-					random_hash1
-				),
-			})
+			result.await.map_err(|e| e.to_string()),
+			Err(RpcError::Call(RpcCallError::Custom(ErrorObject::owned(
+				4001,
+				Error::InvalidBlockRange {
+					from: format!("{:?}", random_hash1), // First hash not found.
+					to: format!("{:?}", Some(random_hash2)),
+					details: format!(
+						"UnknownBlock: Header was not found in the database: {:?}",
+						random_hash1
+					),
+				}
+				.to_string(),
+				None::<()>
+			))))
 			.map_err(|e| e.to_string()),
 		);
 
@@ -497,7 +476,7 @@ fn should_query_storage() {
 		let result = api.query_storage_at(keys.clone(), Some(block1_hash));
 
 		assert_eq!(
-			executor::block_on(result).unwrap(),
+			result.await.unwrap(),
 			vec![StorageChangeSet {
 				block: block1_hash,
 				changes: vec![
@@ -511,19 +490,14 @@ fn should_query_storage() {
 		);
 	}
 
-	run_tests(Arc::new(substrate_test_runtime_client::new()));
-	run_tests(Arc::new(TestClientBuilder::new().build()));
+	run_tests(Arc::new(substrate_test_runtime_client::new())).await;
+	run_tests(Arc::new(TestClientBuilder::new().build())).await;
 }
 
-#[test]
-fn should_return_runtime_version() {
+#[tokio::test]
+async fn should_return_runtime_version() {
 	let client = Arc::new(substrate_test_runtime_client::new());
-	let (api, _child) = new_full(
-		client.clone(),
-		SubscriptionManager::new(Arc::new(TaskExecutor)),
-		DenyUnsafe::No,
-		None,
-	);
+	let (api, _child) = new_full(client.clone(), test_executor(), DenyUnsafe::No, None);
 
 	let result = "{\"specName\":\"test\",\"implName\":\"parity-test\",\"authoringVersion\":1,\
 		\"specVersion\":2,\"implVersion\":2,\"apis\":[[\"0xdf6acb689907609b\",4],\
@@ -532,7 +506,7 @@ fn should_return_runtime_version() {
 		[\"0xf78b278be53f454c\",2],[\"0xab3c0572291feb8b\",1],[\"0xbc9d89904f5b923f\",1]],\
 		\"transactionVersion\":1,\"stateVersion\":1}";
 
-	let runtime_version = executor::block_on(api.runtime_version(None.into())).unwrap();
+	let runtime_version = api.runtime_version(None.into()).await.unwrap();
 	let serialized = serde_json::to_string(&runtime_version).unwrap();
 	assert_eq!(serialized, result);
 
@@ -540,28 +514,26 @@ fn should_return_runtime_version() {
 	assert_eq!(deserialized, runtime_version);
 }
 
-#[test]
-fn should_notify_on_runtime_version_initially() {
-	let (subscriber, id, mut transport) = Subscriber::new_test("test");
-
-	{
+#[tokio::test]
+async fn should_notify_on_runtime_version_initially() {
+	let mut sub = {
 		let client = Arc::new(substrate_test_runtime_client::new());
-		let (api, _child) = new_full(
-			client.clone(),
-			SubscriptionManager::new(Arc::new(TaskExecutor)),
-			DenyUnsafe::No,
-			None,
-		);
+		let (api, _child) = new_full(client, test_executor(), DenyUnsafe::No, None);
 
-		api.subscribe_runtime_version(Default::default(), subscriber);
+		let api_rpc = api.into_rpc();
+		let sub = api_rpc
+			.subscribe("state_subscribeRuntimeVersion", EmptyParams::new())
+			.await
+			.unwrap();
 
-		// assert id assigned
-		assert!(matches!(executor::block_on(id), Ok(Ok(SubscriptionId::String(_)))));
-	}
+		sub
+	};
 
 	// assert initial version sent.
-	executor::block_on((&mut transport).take(1).collect::<Vec<_>>());
-	assert!(executor::block_on(transport.next()).is_none());
+	assert_matches!(timeout_secs(10, sub.next::<RuntimeVersion>()).await, Ok(Some(_)));
+
+	sub.close();
+	assert_matches!(timeout_secs(10, sub.next::<RuntimeVersion>()).await, Ok(None));
 }
 
 #[test]
@@ -572,38 +544,24 @@ fn should_deserialize_storage_key() {
 	assert_eq!(k.0.len(), 32);
 }
 
-#[test]
-fn wildcard_storage_subscriptions_are_rpc_unsafe() {
-	let (subscriber, id, _) = Subscriber::new_test("test");
-
+#[tokio::test]
+async fn wildcard_storage_subscriptions_are_rpc_unsafe() {
 	let client = Arc::new(substrate_test_runtime_client::new());
-	let (api, _child) = new_full(
-		client.clone(),
-		SubscriptionManager::new(Arc::new(TaskExecutor)),
-		DenyUnsafe::Yes,
-		None,
-	);
+	let (api, _child) = new_full(client, test_executor(), DenyUnsafe::Yes, None);
 
-	api.subscribe_storage(Default::default(), subscriber, None.into());
-
-	let error = executor::block_on(id).unwrap().unwrap_err();
-	assert_eq!(error.to_string(), "Method not found: RPC call is unsafe to be called externally");
+	let api_rpc = api.into_rpc();
+	let err = api_rpc.subscribe("state_subscribeStorage", EmptyParams::new()).await;
+	assert_matches!(err, Err(RpcError::Call(RpcCallError::Custom(e))) if e.message() == "RPC call is unsafe to be called externally");
 }
 
-#[test]
-fn concrete_storage_subscriptions_are_rpc_safe() {
-	let (subscriber, id, _) = Subscriber::new_test("test");
-
+#[tokio::test]
+async fn concrete_storage_subscriptions_are_rpc_safe() {
 	let client = Arc::new(substrate_test_runtime_client::new());
-	let (api, _child) = new_full(
-		client.clone(),
-		SubscriptionManager::new(Arc::new(TaskExecutor)),
-		DenyUnsafe::Yes,
-		None,
-	);
+	let (api, _child) = new_full(client, test_executor(), DenyUnsafe::Yes, None);
+	let api_rpc = api.into_rpc();
 
 	let key = StorageKey(STORAGE_KEY.to_vec());
-	api.subscribe_storage(Default::default(), subscriber, Some(vec![key]));
+	let sub = api_rpc.subscribe("state_subscribeStorage", [[key]]).await;
 
-	assert!(matches!(executor::block_on(id), Ok(Ok(SubscriptionId::String(_)))));
+	assert!(sub.is_ok());
 }