Revert "Use Subscription Manager from jsonrpc-pubsub (#6208)" (#6252)

This reverts commit ea1eb4e57f.

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;
-use jsonrpc_pubsub::{typed::Subscriber, SubscriptionId, manager::SubscriptionManager};
+use sc_rpc_api::{DenyUnsafe, Subscriptions};
+use jsonrpc_pubsub::{typed::Subscriber, SubscriptionId};
 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: SubscriptionManager,
+	subscriptions: Subscriptions,
 	/// 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: SubscriptionManager,
+		subscriptions: Subscriptions,
 		keystore: BareCryptoStorePtr,
 		deny_unsafe: DenyUnsafe,
 	) -> Self {
@@ -81,6 +81,7 @@ 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: SubscriptionManager::new(Arc::new(crate::testing::TaskExecutor)),
+			subscriptions: Subscriptions::new(Arc::new(crate::testing::TaskExecutor)),
 			keystore: self.keystore.clone(),
 			deny_unsafe: DenyUnsafe::No,
 		}
@@ -133,14 +133,8 @@ fn should_watch_extrinsic() {
 		uxt(AccountKeyring::Alice, 0).encode().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!(),
-	};
-
+	// then
+	assert_eq!(executor::block_on(id_rx.compat()), Ok(Ok(1.into())));
 	// check notifications
 	let replacement = {
 		let tx = Transfer {
@@ -153,22 +147,15 @@ fn should_watch_extrinsic() {
 	};
 	AuthorApi::submit_extrinsic(&p, replacement.encode().into()).wait().unwrap();
 	let (res, data) = executor::block_on(data.into_future().compat()).unwrap();
-
-	let expected = Some(format!(
-		r#"{{"jsonrpc":"2.0","method":"test","params":{{"result":"ready","subscription":"{}"}}}}"#,
-		id,
-	));
-	assert_eq!(res, expected);
-
+	assert_eq!(
+		res,
+		Some(r#"{"jsonrpc":"2.0","method":"test","params":{"result":"ready","subscription":1}}"#.into())
+	);
 	let h = blake2_256(&replacement.encode());
-	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);
+	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)))
+	);
 }
 
 #[test]