Use Subscription Manager from jsonrpc-pubsub (#6208)

* Bump jsonrpc pubsub, core, http, and ws

Right now these are the packages which _need_ to be updated
so I can just the latest `jsonrpc-pubsub` code. Once a release
it cut upstream the rest of the dependencies should be updated
as well.

* Use jsonrpc-pubsub's SubscriptionManager

This places sc-rpc-api::Subscriptions

* Bump jsonrpc-core outside of sc-rpc-*

* Update client/rpc tests

Right now one of the `author` tests is failing, I
need to think a bit about how best to fix it.

* Remove Subscriptions manager

There's no need for this implementation since we're
using the one from `jsonrpc-pubsub` now

* Fix author RPC test

This test used to check for a numerial subscription ID,
whereas now it uses a string based ID which is the default
provided by `jsonrpc-pubsub`'s subscription manager.

* Remove unused NumericIdProvider

* Add missing bracket

Removed one too many with that last one, lol

* Bump `jsonrpc` to v14.2

There's an exception though. `jsonrpc-derive` cannot be bumped
past v14.0.5 just yet since it has a dependency on `quote` pinned
to v1.0.1. This means that at the moment it won't build on Substrate
since it's using v1.0.3.

* Track `jsonrpc-derive` master branch

* Bump `quote` version to v1.0.6

* Bump `jsonrpc-derive` to v14.2.1

This includes support for `quote` v1.0.6

substrate/client/consensus/babe/rpc/Cargo.toml

@@ -14,9 +14,9 @@ targets = ["x86_64-unknown-linux-gnu"]
 [dependencies]
 sc-consensus-babe = { version = "0.8.0-rc2", path = "../" }
 sc-rpc-api = { version = "0.8.0-rc2", path = "../../../rpc-api" }
-jsonrpc-core = "14.0.3"
-jsonrpc-core-client = "14.0.5"
-jsonrpc-derive = "14.0.3"
+jsonrpc-core = "14.2"
+jsonrpc-core-client = "14.2"
+jsonrpc-derive = "14.2.1"
 sp-consensus-babe = { version = "0.8.0-rc2", path = "../../../../primitives/consensus/babe" }
 serde = { version = "1.0.104", features=["derive"] }
 sp-blockchain = { version = "2.0.0-rc2", path = "../../../../primitives/blockchain" }

substrate/client/consensus/manual-seal/Cargo.toml

@@ -14,9 +14,9 @@ targets = ["x86_64-unknown-linux-gnu"]
 [dependencies]
 derive_more = "0.99.2"
 futures = "0.3.4"
-jsonrpc-core = "14.0.5"
-jsonrpc-core-client = "14.0.5"
-jsonrpc-derive = "14.0.5"
+jsonrpc-core = "14.2"
+jsonrpc-core-client = "14.2"
+jsonrpc-derive = "14.2.1"
 log = "0.4.8"
 parking_lot = "0.10.0"
 serde = { version = "1.0", features=["derive"] }

substrate/client/finality-grandpa/rpc/Cargo.toml

@@ -10,9 +10,9 @@ license = "GPL-3.0-or-later WITH Classpath-exception-2.0"
 [dependencies]
 sc-finality-grandpa = { version = "0.8.0-rc2", path = "../" }
 finality-grandpa = { version = "0.12.3", features = ["derive-codec"] }
-jsonrpc-core = "14.0.3"
-jsonrpc-core-client = "14.0.3"
-jsonrpc-derive = "14.0.3"
+jsonrpc-core = "14.2"
+jsonrpc-core-client = "14.2"
+jsonrpc-derive = "14.2.1"
 futures = { version = "0.3.4", features = ["compat"] }
 serde = { version = "1.0.105", features = ["derive"] }
 serde_json = "1.0.50"

substrate/client/rpc-api/Cargo.toml

@@ -15,10 +15,10 @@ targets = ["x86_64-unknown-linux-gnu"]
 codec = { package = "parity-scale-codec", version = "1.3.0" }
 derive_more = "0.99.2"
 futures = { version = "0.3.1", features = ["compat"] }
-jsonrpc-core = "14.0.3"
-jsonrpc-core-client = "14.0.5"
-jsonrpc-derive = "14.0.3"
-jsonrpc-pubsub = "14.0.3"
+jsonrpc-core = "14.2"
+jsonrpc-core-client = "14.2"
+jsonrpc-derive = "14.2.1"
+jsonrpc-pubsub = "14.2"
 log = "0.4.8"
 parking_lot = "0.10.0"
 sp-core = { version = "2.0.0-rc2", path = "../../primitives/core" }

substrate/client/rpc-api/src/lib.rs

@@ -23,10 +23,8 @@
 mod errors;
 mod helpers;
 mod policy;
-mod subscriptions;
 
 pub use jsonrpc_core::IoHandlerExtension as RpcExtension;
-pub use subscriptions::{Subscriptions, TaskExecutor};
 pub use helpers::Receiver;
 pub use policy::DenyUnsafe;
 

substrate/client/rpc-api/src/subscriptions.rs

@@ -1,121 +0,0 @@
-// This file is part of Substrate.
-
-// Copyright (C) 2017-2020 Parity Technologies (UK) Ltd.
-// SPDX-License-Identifier: GPL-3.0-or-later WITH Classpath-exception-2.0
-
-// This program is free software: you can redistribute it and/or modify
-// it under the terms of the GNU General Public License as published by
-// the Free Software Foundation, either version 3 of the License, or
-// (at your option) any later version.
-
-// This program is distributed in the hope that it will be useful,
-// but WITHOUT ANY WARRANTY; without even the implied warranty of
-// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
-// GNU General Public License for more details.
-
-// You should have received a copy of the GNU General Public License
-// along with this program. If not, see <https://www.gnu.org/licenses/>.
-
-use std::collections::HashMap;
-use std::sync::{Arc, atomic::{self, AtomicUsize}};
-
-use log::{error, warn};
-use jsonrpc_pubsub::{SubscriptionId, typed::{Sink, Subscriber}};
-use parking_lot::Mutex;
-use jsonrpc_core::futures::sync::oneshot;
-use jsonrpc_core::futures::{Future, future};
-
-type Id = u64;
-
-/// Alias for a an implementation of `futures::future::Executor`.
-pub type TaskExecutor = Arc<dyn future::Executor<Box<dyn Future<Item = (), Error = ()> + Send>> + Send + Sync>;
-
-/// Generate unique ids for subscriptions.
-#[derive(Clone, Debug)]
-pub struct IdProvider {
-	next_id: Arc<AtomicUsize>,
-}
-impl Default for IdProvider {
-	fn default() -> Self {
-		IdProvider {
-			next_id: Arc::new(AtomicUsize::new(1)),
-		}
-	}
-}
-
-impl IdProvider {
-	/// Returns next id for the subscription.
-	pub fn next_id(&self) -> Id {
-		self.next_id.fetch_add(1, atomic::Ordering::AcqRel) as u64
-	}
-}
-
-/// Subscriptions manager.
-///
-/// Takes care of assigning unique subscription ids and
-/// driving the sinks into completion.
-#[derive(Clone)]
-pub struct Subscriptions {
-	next_id: IdProvider,
-	active_subscriptions: Arc<Mutex<HashMap<Id, oneshot::Sender<()>>>>,
-	executor: TaskExecutor,
-}
-
-impl Subscriptions {
-	/// Creates new `Subscriptions` object.
-	pub fn new(executor: TaskExecutor) -> Self {
-		Subscriptions {
-			next_id: Default::default(),
-			active_subscriptions: Default::default(),
-			executor,
-		}
-	}
-
-	/// Borrows the internal task executor.
-	///
-	/// This can be used to spawn additional tasks on the underlying event loop.
-	pub fn executor(&self) -> &TaskExecutor {
-		&self.executor
-	}
-
-	/// Creates new subscription for given subscriber.
-	///
-	/// Second parameter is a function that converts Subscriber sink into a future.
-	/// This future will be driven to completion by the underlying event loop
-	/// or will be cancelled in case #cancel is invoked.
-	pub fn add<T, E, G, R, F>(&self, subscriber: Subscriber<T, E>, into_future: G) -> SubscriptionId where
-		G: FnOnce(Sink<T, E>) -> R,
-		R: future::IntoFuture<Future=F, Item=(), Error=()>,
-		F: future::Future<Item=(), Error=()> + Send + 'static,
-	{
-		let id = self.next_id.next_id();
-		let subscription_id: SubscriptionId = id.into();
-		if let Ok(sink) = subscriber.assign_id(subscription_id.clone()) {
-			let (tx, rx) = oneshot::channel();
-			let future = into_future(sink)
-				.into_future()
-				.select(rx.map_err(|e| warn!("Error timeing out: {:?}", e)))
-				.then(|_| Ok(()));
-
-			self.active_subscriptions.lock().insert(id, tx);
-			if self.executor.execute(Box::new(future)).is_err() {
-				error!("Failed to spawn RPC subscription task");
-			}
-		}
-
-		subscription_id
-	}
-
-	/// Cancel subscription.
-	///
-	/// Returns true if subscription existed or false otherwise.
-	pub fn cancel(&self, id: SubscriptionId) -> bool {
-		if let SubscriptionId::Number(id) = id {
-			if let Some(tx) = self.active_subscriptions.lock().remove(&id) {
-				let _ = tx.send(());
-				return true;
-			}
-		}
-		false
-	}
-}

substrate/client/rpc-servers/Cargo.toml

@@ -12,13 +12,13 @@ description = "Substrate RPC servers."
 targets = ["x86_64-unknown-linux-gnu"]
 
 [dependencies]
-jsonrpc-core = "14.0.3"
-pubsub = { package = "jsonrpc-pubsub", version = "14.0.3" }
+jsonrpc-core = "14.2"
+pubsub = { package = "jsonrpc-pubsub", version = "14.2" }
 log = "0.4.8"
 serde = "1.0.101"
 serde_json = "1.0.41"
 sp-runtime = { version = "2.0.0-rc2", path = "../../primitives/runtime" }
 
 [target.'cfg(not(target_os = "unknown"))'.dependencies]
-http = { package = "jsonrpc-http-server", version = "14.0.3" }
-ws = { package = "jsonrpc-ws-server", version = "14.0.3" }
+http = { package = "jsonrpc-http-server", version = "14.2" }
+ws = { package = "jsonrpc-ws-server", version = "14.2" }

substrate/client/rpc/Cargo.toml

@@ -17,10 +17,10 @@ sc-client-api = { version = "2.0.0-rc2", path = "../api" }
 sp-api = { version = "2.0.0-rc2", path = "../../primitives/api" }
 codec = { package = "parity-scale-codec", version = "1.3.0" }
 futures = { version = "0.3.1", features = ["compat"] }
-jsonrpc-pubsub = "14.0.3"
+jsonrpc-pubsub = "14.2"
 log = "0.4.8"
 sp-core = { version = "2.0.0-rc2", path = "../../primitives/core" }
-rpc = { package = "jsonrpc-core", version = "14.0.3" }
+rpc = { package = "jsonrpc-core", version = "14.2" }
 sp-version = { version = "2.0.0-rc2", path = "../../primitives/version" }
 serde_json = "1.0.41"
 sp-session = { version = "2.0.0-rc2", path = "../../primitives/session" }

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

@@ -32,8 +32,8 @@ use rpc::futures::{
 };
 use futures::{StreamExt as _, compat::Compat};
 use futures::future::{ready, FutureExt, TryFutureExt};
-use sc_rpc_api::{DenyUnsafe, Subscriptions};
-use jsonrpc_pubsub::{typed::Subscriber, SubscriptionId};
+use sc_rpc_api::DenyUnsafe;
+use jsonrpc_pubsub::{typed::Subscriber, SubscriptionId, manager::SubscriptionManager};
 use codec::{Encode, Decode};
 use sp_core::{Bytes, traits::BareCryptoStorePtr};
 use sp_api::ProvideRuntimeApi;
@@ -55,7 +55,7 @@ pub struct Author<P, Client> {
 	/// Transactions pool
 	pool: Arc<P>,
 	/// Subscriptions manager
-	subscriptions: Subscriptions,
+	subscriptions: SubscriptionManager,
 	/// The key store.
 	keystore: BareCryptoStorePtr,
 	/// Whether to deny unsafe calls
@@ -67,7 +67,7 @@ impl<P, Client> Author<P, Client> {
 	pub fn new(
 		client: Arc<Client>,
 		pool: Arc<P>,
-		subscriptions: Subscriptions,
+		subscriptions: SubscriptionManager,
 		keystore: BareCryptoStorePtr,
 		deny_unsafe: DenyUnsafe,
 	) -> Self {
@@ -81,7 +81,6 @@ impl<P, Client> Author<P, Client> {
 	}
 }
 
-
 /// Currently we treat all RPC transactions as externals.
 ///
 /// Possibly in the future we could allow opt-in for special treatment

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

@@ -81,7 +81,7 @@ impl TestSetup {
 		Author {
 			client: self.client.clone(),
 			pool: self.pool.clone(),
-			subscriptions: Subscriptions::new(Arc::new(crate::testing::TaskExecutor)),
+			subscriptions: SubscriptionManager::new(Arc::new(crate::testing::TaskExecutor)),
 			keystore: self.keystore.clone(),
 			deny_unsafe: DenyUnsafe::No,
 		}
@@ -133,8 +133,14 @@ fn should_watch_extrinsic() {
 		uxt(AccountKeyring::Alice, 0).encode().into(),
 	);
 
-	// then
-	assert_eq!(executor::block_on(id_rx.compat()), Ok(Ok(1.into())));
+	let id = executor::block_on(id_rx.compat()).unwrap().unwrap();
+	assert_matches!(id, SubscriptionId::String(_));
+
+	let id = match id {
+		SubscriptionId::String(id) => id,
+		_ => unreachable!(),
+	};
+
 	// check notifications
 	let replacement = {
 		let tx = Transfer {
@@ -147,15 +153,22 @@ fn should_watch_extrinsic() {
 	};
 	AuthorApi::submit_extrinsic(&p, replacement.encode().into()).wait().unwrap();
 	let (res, data) = executor::block_on(data.into_future().compat()).unwrap();
-	assert_eq!(
-		res,
-		Some(r#"{"jsonrpc":"2.0","method":"test","params":{"result":"ready","subscription":1}}"#.into())
-	);
+
+	let expected = Some(format!(
+		r#"{{"jsonrpc":"2.0","method":"test","params":{{"result":"ready","subscription":"{}"}}}}"#,
+		id,
+	));
+	assert_eq!(res, expected);
+
 	let h = blake2_256(&replacement.encode());
-	assert_eq!(
-		executor::block_on(data.into_future().compat()).unwrap().0,
-		Some(format!(r#"{{"jsonrpc":"2.0","method":"test","params":{{"result":{{"usurped":"0x{}"}},"subscription":1}}}}"#, HexDisplay::from(&h)))
-	);
+	let expected = Some(format!(
+		r#"{{"jsonrpc":"2.0","method":"test","params":{{"result":{{"usurped":"0x{}"}},"subscription":"{}"}}}}"#,
+		HexDisplay::from(&h),
+		id,
+	));
+
+	let res = executor::block_on(data.into_future().compat()).unwrap().0;
+	assert_eq!(res, expected);
 }
 
 #[test]

substrate/client/rpc/src/chain/chain_full.rs

@@ -18,8 +18,8 @@
 
 use std::sync::Arc;
 use rpc::futures::future::result;
+use jsonrpc_pubsub::manager::SubscriptionManager;
 
-use sc_rpc_api::Subscriptions;
 use sc_client_api::{BlockchainEvents, BlockBackend};
 use sp_runtime::{generic::{BlockId, SignedBlock}, traits::{Block as BlockT}};
 
@@ -32,14 +32,14 @@ pub struct FullChain<Block: BlockT, Client> {
 	/// Substrate client.
 	client: Arc<Client>,
 	/// Current subscriptions.
-	subscriptions: Subscriptions,
+	subscriptions: SubscriptionManager,
 	/// phantom member to pin the block type
 	_phantom: PhantomData<Block>,
 }
 
 impl<Block: BlockT, Client> FullChain<Block, Client> {
 	/// Create new Chain API RPC handler.
-	pub fn new(client: Arc<Client>, subscriptions: Subscriptions) -> Self {
+	pub fn new(client: Arc<Client>, subscriptions: SubscriptionManager) -> Self {
 		Self {
 			client,
 			subscriptions,
@@ -56,7 +56,7 @@ impl<Block, Client> ChainBackend<Client, Block> for FullChain<Block, Client> whe
 		&self.client
 	}
 
-	fn subscriptions(&self) -> &Subscriptions {
+	fn subscriptions(&self) -> &SubscriptionManager {
 		&self.subscriptions
 	}
 

substrate/client/rpc/src/chain/chain_light.rs

@@ -19,8 +19,8 @@
 use std::sync::Arc;
 use futures::{future::ready, FutureExt, TryFutureExt};
 use rpc::futures::future::{result, Future, Either};
+use jsonrpc_pubsub::manager::SubscriptionManager;
 
-use sc_rpc_api::Subscriptions;
 use sc_client_api::light::{Fetcher, RemoteBodyRequest, RemoteBlockchain};
 use sp_runtime::{
 	generic::{BlockId, SignedBlock},
@@ -37,7 +37,7 @@ pub struct LightChain<Block: BlockT, Client, F> {
 	/// Substrate client.
 	client: Arc<Client>,
 	/// Current subscriptions.
-	subscriptions: Subscriptions,
+	subscriptions: SubscriptionManager,
 	/// Remote blockchain reference
 	remote_blockchain: Arc<dyn RemoteBlockchain<Block>>,
 	/// Remote fetcher reference.
@@ -48,7 +48,7 @@ impl<Block: BlockT, Client, F: Fetcher<Block>> LightChain<Block, Client, F> {
 	/// Create new Chain API RPC handler.
 	pub fn new(
 		client: Arc<Client>,
-		subscriptions: Subscriptions,
+		subscriptions: SubscriptionManager,
 		remote_blockchain: Arc<dyn RemoteBlockchain<Block>>,
 		fetcher: Arc<F>,
 	) -> Self {
@@ -70,7 +70,7 @@ impl<Block, Client, F> ChainBackend<Client, Block> for LightChain<Block, Client,
 		&self.client
 	}
 
-	fn subscriptions(&self) -> &Subscriptions {
+	fn subscriptions(&self) -> &SubscriptionManager {
 		&self.subscriptions
 	}
 

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

@@ -32,9 +32,8 @@ use rpc::{
 	futures::{stream, Future, Sink, Stream},
 };
 
-use sc_rpc_api::Subscriptions;
 use sc_client_api::{BlockchainEvents, light::{Fetcher, RemoteBlockchain}};
-use jsonrpc_pubsub::{typed::Subscriber, SubscriptionId};
+use jsonrpc_pubsub::{typed::Subscriber, SubscriptionId, manager::SubscriptionManager};
 use sp_rpc::{number::NumberOrHex, list::ListOrValue};
 use sp_runtime::{
 	generic::{BlockId, SignedBlock},
@@ -57,7 +56,7 @@ trait ChainBackend<Client, Block: BlockT>: Send + Sync + 'static
 	fn client(&self) -> &Arc<Client>;
 
 	/// Get subscriptions reference.
-	fn subscriptions(&self) -> &Subscriptions;
+	fn subscriptions(&self) -> &SubscriptionManager;
 
 	/// Tries to unwrap passed block hash, or uses best block hash otherwise.
 	fn unwrap_or_best(&self, hash: Option<Block::Hash>) -> Block::Hash {
@@ -177,7 +176,7 @@ trait ChainBackend<Client, Block: BlockT>: Send + Sync + 'static
 /// Create new state API that works on full node.
 pub fn new_full<Block: BlockT, Client>(
 	client: Arc<Client>,
-	subscriptions: Subscriptions,
+	subscriptions: SubscriptionManager,
 ) -> Chain<Block, Client>
 	where
 		Block: BlockT + 'static,
@@ -191,7 +190,7 @@ pub fn new_full<Block: BlockT, Client>(
 /// Create new state API that works on light node.
 pub fn new_light<Block: BlockT, Client, F: Fetcher<Block>>(
 	client: Arc<Client>,
-	subscriptions: Subscriptions,
+	subscriptions: SubscriptionManager,
 	remote_blockchain: Arc<dyn RemoteBlockchain<Block>>,
 	fetcher: Arc<F>,
 ) -> Chain<Block, Client>
@@ -279,7 +278,7 @@ impl<Block, Client> ChainApi<NumberFor<Block>, Block::Hash, Block::Header, Signe
 /// Subscribe to new headers.
 fn subscribe_headers<Block, Client, F, G, S, ERR>(
 	client: &Arc<Client>,
-	subscriptions: &Subscriptions,
+	subscriptions: &SubscriptionManager,
 	subscriber: Subscriber<Block::Header>,
 	best_block_hash: G,
 	stream: F,

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

@@ -31,7 +31,7 @@ use crate::testing::TaskExecutor;
 #[test]
 fn should_return_header() {
 	let client = Arc::new(substrate_test_runtime_client::new());
-	let api = new_full(client.clone(), Subscriptions::new(Arc::new(TaskExecutor)));
+	let api = new_full(client.clone(), SubscriptionManager::new(Arc::new(TaskExecutor)));
 
 	assert_matches!(
 		api.header(Some(client.genesis_hash()).into()).wait(),
@@ -63,7 +63,7 @@ fn should_return_header() {
 #[test]
 fn should_return_a_block() {
 	let mut client = Arc::new(substrate_test_runtime_client::new());
-	let api = new_full(client.clone(), Subscriptions::new(Arc::new(TaskExecutor)));
+	let api = new_full(client.clone(), SubscriptionManager::new(Arc::new(TaskExecutor)));
 
 	let block = client.new_block(Default::default()).unwrap().build().unwrap().block;
 	let block_hash = block.hash();
@@ -114,7 +114,7 @@ fn should_return_a_block() {
 #[test]
 fn should_return_block_hash() {
 	let mut client = Arc::new(substrate_test_runtime_client::new());
-	let api = new_full(client.clone(), Subscriptions::new(Arc::new(TaskExecutor)));
+	let api = new_full(client.clone(), SubscriptionManager::new(Arc::new(TaskExecutor)));
 
 	assert_matches!(
 		api.block_hash(None.into()),
@@ -158,7 +158,7 @@ fn should_return_block_hash() {
 #[test]
 fn should_return_finalized_hash() {
 	let mut client = Arc::new(substrate_test_runtime_client::new());
-	let api = new_full(client.clone(), Subscriptions::new(Arc::new(TaskExecutor)));
+	let api = new_full(client.clone(), SubscriptionManager::new(Arc::new(TaskExecutor)));
 
 	assert_matches!(
 		api.finalized_head(),
@@ -188,12 +188,15 @@ fn should_notify_about_latest_block() {
 
 	{
 		let mut client = Arc::new(substrate_test_runtime_client::new());
-		let api = new_full(client.clone(), Subscriptions::new(Arc::new(TaskExecutor)));
+		let api = new_full(client.clone(), SubscriptionManager::new(Arc::new(TaskExecutor)));
 
 		api.subscribe_all_heads(Default::default(), subscriber);
 
 		// assert id assigned
-		assert_eq!(executor::block_on(id.compat()), Ok(Ok(SubscriptionId::Number(1))));
+		assert!(matches!(
+			executor::block_on(id.compat()),
+			Ok(Ok(SubscriptionId::String(_)))
+		));
 
 		let block = client.new_block(Default::default()).unwrap().build().unwrap().block;
 		client.import(BlockOrigin::Own, block).unwrap();
@@ -215,12 +218,15 @@ fn should_notify_about_best_block() {
 
 	{
 		let mut client = Arc::new(substrate_test_runtime_client::new());
-		let api = new_full(client.clone(), Subscriptions::new(Arc::new(TaskExecutor)));
+		let api = new_full(client.clone(), SubscriptionManager::new(Arc::new(TaskExecutor)));
 
 		api.subscribe_new_heads(Default::default(), subscriber);
 
 		// assert id assigned
-		assert_eq!(executor::block_on(id.compat()), Ok(Ok(SubscriptionId::Number(1))));
+		assert!(matches!(
+			executor::block_on(id.compat()),
+			Ok(Ok(SubscriptionId::String(_)))
+		));
 
 		let block = client.new_block(Default::default()).unwrap().build().unwrap().block;
 		client.import(BlockOrigin::Own, block).unwrap();
@@ -242,12 +248,15 @@ fn should_notify_about_finalized_block() {
 
 	{
 		let mut client = Arc::new(substrate_test_runtime_client::new());
-		let api = new_full(client.clone(), Subscriptions::new(Arc::new(TaskExecutor)));
+		let api = new_full(client.clone(), SubscriptionManager::new(Arc::new(TaskExecutor)));
 
 		api.subscribe_finalized_heads(Default::default(), subscriber);
 
 		// assert id assigned
-		assert_eq!(executor::block_on(id.compat()), Ok(Ok(SubscriptionId::Number(1))));
+		assert!(matches!(
+			executor::block_on(id.compat()),
+			Ok(Ok(SubscriptionId::String(_)))
+		));
 
 		let block = client.new_block(Default::default()).unwrap().build().unwrap().block;
 		client.import(BlockOrigin::Own, block).unwrap();

substrate/client/rpc/src/lib.rs

@@ -24,7 +24,7 @@
 
 mod metadata;
 
-pub use sc_rpc_api::{DenyUnsafe, Subscriptions};
+pub use sc_rpc_api::DenyUnsafe;
 pub use self::metadata::Metadata;
 pub use rpc::IoHandlerExtension as RpcExtension;
 

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

@@ -25,10 +25,10 @@ mod state_light;
 mod tests;
 
 use std::sync::Arc;
-use jsonrpc_pubsub::{typed::Subscriber, SubscriptionId};
+use jsonrpc_pubsub::{typed::Subscriber, SubscriptionId, manager::SubscriptionManager};
 use rpc::{Result as RpcResult, futures::{Future, future::result}};
 
-use sc_rpc_api::{Subscriptions, state::ReadProof};
+use sc_rpc_api::state::ReadProof;
 use sc_client_api::light::{RemoteBlockchain, Fetcher};
 use sp_core::{Bytes, storage::{StorageKey, PrefixedStorageKey, StorageData, StorageChangeSet}};
 use sp_version::RuntimeVersion;
@@ -170,7 +170,7 @@ pub trait StateBackend<Block: BlockT, Client>: Send + Sync + 'static
 /// Create new state API that works on full node.
 pub fn new_full<BE, Block: BlockT, Client>(
 	client: Arc<Client>,
-	subscriptions: Subscriptions,
+	subscriptions: SubscriptionManager,
 ) -> (State<Block, Client>, ChildState<Block, Client>)
 	where
 		Block: BlockT + 'static,
@@ -191,7 +191,7 @@ pub fn new_full<BE, Block: BlockT, Client>(
 /// Create new state API that works on light node.
 pub fn new_light<BE, Block: BlockT, Client, F: Fetcher<Block>>(
 	client: Arc<Client>,
-	subscriptions: Subscriptions,
+	subscriptions: SubscriptionManager,
 	remote_blockchain: Arc<dyn RemoteBlockchain<Block>>,
 	fetcher: Arc<F>,
 ) -> (State<Block, Client>, ChildState<Block, Client>)

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

@@ -21,10 +21,10 @@ use std::sync::Arc;
 use std::ops::Range;
 use futures::{future, StreamExt as _, TryStreamExt as _};
 use log::warn;
-use jsonrpc_pubsub::{typed::Subscriber, SubscriptionId};
+use jsonrpc_pubsub::{typed::Subscriber, SubscriptionId, manager::SubscriptionManager};
 use rpc::{Result as RpcResult, futures::{stream, Future, Sink, Stream, future::result}};
 
-use sc_rpc_api::{Subscriptions, state::ReadProof};
+use sc_rpc_api::state::ReadProof;
 use sc_client_api::backend::Backend;
 use sp_blockchain::{Result as ClientResult, Error as ClientError, HeaderMetadata, CachedHeaderMetadata, HeaderBackend};
 use sc_client_api::BlockchainEvents;
@@ -60,7 +60,7 @@ struct QueryStorageRange<Block: BlockT> {
 /// State API backend for full nodes.
 pub struct FullState<BE, Block: BlockT, Client> {
 	client: Arc<Client>,
-	subscriptions: Subscriptions,
+	subscriptions: SubscriptionManager,
 	_phantom: PhantomData<(BE, Block)>
 }
 
@@ -72,7 +72,7 @@ impl<BE, Block: BlockT, Client> FullState<BE, Block, Client>
 		Block: BlockT + 'static,
 {
 	/// Create new state API backend for full nodes.
-	pub fn new(client: Arc<Client>, subscriptions: Subscriptions) -> Self {
+	pub fn new(client: Arc<Client>, subscriptions: SubscriptionManager) -> Self {
 		Self { client, subscriptions, _phantom: PhantomData }
 	}
 

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

@@ -28,7 +28,7 @@ use futures::{
 	StreamExt as _, TryStreamExt as _,
 };
 use hash_db::Hasher;
-use jsonrpc_pubsub::{typed::Subscriber, SubscriptionId};
+use jsonrpc_pubsub::{typed::Subscriber, SubscriptionId, manager::SubscriptionManager};
 use log::warn;
 use parking_lot::Mutex;
 use rpc::{
@@ -38,7 +38,7 @@ use rpc::{
 	futures::stream::Stream,
 };
 
-use sc_rpc_api::{Subscriptions, state::ReadProof};
+use sc_rpc_api::state::ReadProof;
 use sp_blockchain::{Error as ClientError, HeaderBackend};
 use sc_client_api::{
 	BlockchainEvents,
@@ -63,7 +63,7 @@ type StorageMap = HashMap<StorageKey, Option<StorageData>>;
 #[derive(Clone)]
 pub struct LightState<Block: BlockT, F: Fetcher<Block>, Client> {
 	client: Arc<Client>,
-	subscriptions: Subscriptions,
+	subscriptions: SubscriptionManager,
 	version_subscriptions: SimpleSubscriptions<Block::Hash, RuntimeVersion>,
 	storage_subscriptions: Arc<Mutex<StorageSubscriptions<Block>>>,
 	remote_blockchain: Arc<dyn RemoteBlockchain<Block>>,
@@ -143,7 +143,7 @@ impl<Block: BlockT, F: Fetcher<Block> + 'static, Client> LightState<Block, F, Cl
 	/// Create new state API backend for light nodes.
 	pub fn new(
 		client: Arc<Client>,
-		subscriptions: Subscriptions,
+		subscriptions: SubscriptionManager,
 		remote_blockchain: Arc<dyn RemoteBlockchain<Block>>,
 		fetcher: Arc<F>,
 	) -> Self {

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

@@ -55,7 +55,7 @@ fn should_return_storage() {
 		.add_extra_child_storage(&child_info, KEY.to_vec(), CHILD_VALUE.to_vec())
 		.build();
 	let genesis_hash = client.genesis_hash();
-	let (client, child) = new_full(Arc::new(client), Subscriptions::new(Arc::new(TaskExecutor)));
+	let (client, child) = new_full(Arc::new(client), SubscriptionManager::new(Arc::new(TaskExecutor)));
 	let key = StorageKey(KEY.to_vec());
 
 	assert_eq!(
@@ -90,7 +90,7 @@ fn should_return_child_storage() {
 		.add_child_storage(&child_info, "key", vec![42_u8])
 		.build());
 	let genesis_hash = client.genesis_hash();
-	let (_client, child) = new_full(client, Subscriptions::new(Arc::new(TaskExecutor)));
+	let (_client, child) = new_full(client, SubscriptionManager::new(Arc::new(TaskExecutor)));
 	let child_key = prefixed_storage_key();
 	let key = StorageKey(b"key".to_vec());
 
@@ -125,7 +125,7 @@ fn should_return_child_storage() {
 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, Subscriptions::new(Arc::new(TaskExecutor)));
+	let (client, _child) = new_full(client, SubscriptionManager::new(Arc::new(TaskExecutor)));
 
 	assert_matches!(
 		client.call("balanceOf".into(), Bytes(vec![1,2,3]), Some(genesis_hash).into()).wait(),
@@ -139,12 +139,15 @@ fn should_notify_about_storage_changes() {
 
 	{
 		let mut client = Arc::new(substrate_test_runtime_client::new());
-		let (api, _child) = new_full(client.clone(), Subscriptions::new(Arc::new(TaskExecutor)));
+		let (api, _child) = new_full(client.clone(), SubscriptionManager::new(Arc::new(TaskExecutor)));
 
 		api.subscribe_storage(Default::default(), subscriber, None.into());
 
 		// assert id assigned
-		assert_eq!(executor::block_on(id.compat()), Ok(Ok(SubscriptionId::Number(1))));
+		assert!(matches!(
+			executor::block_on(id.compat()),
+			Ok(Ok(SubscriptionId::String(_)))
+		));
 
 		let mut builder = client.new_block(Default::default()).unwrap();
 		builder.push_transfer(runtime::Transfer {
@@ -170,7 +173,7 @@ fn should_send_initial_storage_changes_and_notifications() {
 
 	{
 		let mut client = Arc::new(substrate_test_runtime_client::new());
-		let (api, _child) = new_full(client.clone(), Subscriptions::new(Arc::new(TaskExecutor)));
+		let (api, _child) = new_full(client.clone(), SubscriptionManager::new(Arc::new(TaskExecutor)));
 
 		let alice_balance_key = blake2_256(&runtime::system::balance_of_key(AccountKeyring::Alice.into()));
 
@@ -179,7 +182,10 @@ fn should_send_initial_storage_changes_and_notifications() {
 		]).into());
 
 		// assert id assigned
-		assert_eq!(executor::block_on(id.compat()), Ok(Ok(SubscriptionId::Number(1))));
+		assert!(matches!(
+			executor::block_on(id.compat()),
+			Ok(Ok(SubscriptionId::String(_)))
+		));
 
 		let mut builder = client.new_block(Default::default()).unwrap();
 		builder.push_transfer(runtime::Transfer {
@@ -205,7 +211,7 @@ fn should_send_initial_storage_changes_and_notifications() {
 #[test]
 fn should_query_storage() {
 	fn run_tests(mut client: Arc<TestClient>, has_changes_trie_config: bool) {
-		let (api, _child) = new_full(client.clone(), Subscriptions::new(Arc::new(TaskExecutor)));
+		let (api, _child) = new_full(client.clone(), SubscriptionManager::new(Arc::new(TaskExecutor)));
 
 		let mut add_block = |nonce| {
 			let mut builder = client.new_block(Default::default()).unwrap();
@@ -422,7 +428,7 @@ fn should_split_ranges() {
 #[test]
 fn should_return_runtime_version() {
 	let client = Arc::new(substrate_test_runtime_client::new());
-	let (api, _child) = new_full(client.clone(), Subscriptions::new(Arc::new(TaskExecutor)));
+	let (api, _child) = new_full(client.clone(), SubscriptionManager::new(Arc::new(TaskExecutor)));
 
 	let result = "{\"specName\":\"test\",\"implName\":\"parity-test\",\"authoringVersion\":1,\
 		\"specVersion\":2,\"implVersion\":2,\"apis\":[[\"0xdf6acb689907609b\",3],\
@@ -445,12 +451,16 @@ fn should_notify_on_runtime_version_initially() {
 
 	{
 		let client = Arc::new(substrate_test_runtime_client::new());
-		let (api, _child) = new_full(client.clone(), Subscriptions::new(Arc::new(TaskExecutor)));
+		let (api, _child) = new_full(client.clone(), SubscriptionManager::new(Arc::new(TaskExecutor)));
 
 		api.subscribe_runtime_version(Default::default(), subscriber);
 
 		// assert id assigned
-		assert_eq!(executor::block_on(id.compat()), Ok(Ok(SubscriptionId::Number(1))));
+		assert!(matches!(
+			executor::block_on(id.compat()),
+			Ok(Ok(SubscriptionId::String(_)))
+		));
+
 	}
 
 	// assert initial version sent.

substrate/client/service/Cargo.toml

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

substrate/client/service/src/builder.rs

@@ -36,6 +36,7 @@ use futures::{
 	Future, FutureExt, StreamExt,
 	future::ready,
 };
+use jsonrpc_pubsub::manager::SubscriptionManager;
 use sc_keystore::Store as Keystore;
 use log::{info, warn, error};
 use sc_network::config::{Role, FinalityProofProvider, OnDemand, BoxFinalityProofRequestBuilder};
@@ -1196,7 +1197,7 @@ ServiceBuilder<
 				chain_type: chain_spec.chain_type().clone(),
 			};
 
-			let subscriptions = sc_rpc::Subscriptions::new(Arc::new(task_manager.spawn_handle()));
+			let subscriptions = SubscriptionManager::new(Arc::new(task_manager.spawn_handle()));
 
 			let (chain, state, child_state) = if let (Some(remote_backend), Some(on_demand)) =
 				(remote_backend.as_ref(), on_demand.as_ref()) {