Use Subscription Manager from jsonrpc-pubsub

bin/node/browser-testing/Cargo.toml

@@ -9,7 +9,7 @@ license = "Apache-2.0"
 [dependencies]
 futures-timer = "3.0.2"
 libp2p = { version = "0.19.1", default-features = false }
-jsonrpc-core = "14.0.5"
+jsonrpc-core = "14.2"
 serde = "1.0.106"
 serde_json = "1.0.48"
 wasm-bindgen = { version = "=0.2.62", features = ["serde-serialize"] }

bin/node/cli/Cargo.toml

@@ -38,7 +38,7 @@ codec = { package = "parity-scale-codec", version = "1.3.0" }
 serde = { version = "1.0.102", features = ["derive"] }
 futures = { version = "0.3.1", features = ["compat"] }
 hex-literal = "0.2.1"
-jsonrpc-core = "14.0.3"
+jsonrpc-core = "14.2"
 log = "0.4.8"
 rand = "0.7.2"
 structopt = { version = "0.3.8", optional = true }

bin/node/rpc-client/Cargo.toml

@@ -14,7 +14,7 @@ targets = ["x86_64-unknown-linux-gnu"]
 env_logger = "0.7.0"
 futures = "0.1.29"
 hyper = "0.12.35"
-jsonrpc-core-client = { version = "14.0.5", default-features = false, features = ["http"] }
+jsonrpc-core-client = { version = "14.2", default-features = false, features = ["http"] }
 log = "0.4.8"
 node-primitives = { version = "2.0.0-rc2", path = "../primitives" }
 sc-rpc = { version = "2.0.0-rc2", path = "../../../client/rpc" }

bin/node/rpc/Cargo.toml

@@ -12,7 +12,7 @@ targets = ["x86_64-unknown-linux-gnu"]
 
 [dependencies]
 sc-client-api = { version = "2.0.0-rc2", path = "../../../client/api" }
-jsonrpc-core = "14.0.3"
+jsonrpc-core = "14.2"
 node-primitives = { version = "2.0.0-rc2", path = "../primitives" }
 node-runtime = { version = "2.0.0-rc2", path = "../runtime" }
 sp-runtime = { version = "2.0.0-rc2", path = "../../../primitives/runtime" }

bin/utils/subkey/Cargo.toml

@@ -31,7 +31,7 @@ rpassword = "4.0.1"
 itertools = "0.8.2"
 derive_more = { version = "0.99.2" }
 sc-rpc = { version = "2.0.0-rc2", path = "../../../client/rpc" }
-jsonrpc-core-client = { version = "14.0.3", features = ["http"] }
+jsonrpc-core-client = { version = "14.2", features = ["http"] }
 hyper = "0.12.35"
 libp2p = "0.19.1"
 serde_json = "1.0"

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" }

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"] }

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"

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" }

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;
 

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" }

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" }

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

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]

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
 	}