Watch existing extrinsics RPC (#3873)

* Transaction pool watch intristics.

* Track extrinsic rpc methods.

* Test for pool watcher.

* Track extrinsic rpc test.

* Fix rpc naming.

* review fixes

* Update jsonrpc and use une subcription.

* Naming and dependencies.

substrate/core/rpc/Cargo.toml

@@ -9,10 +9,10 @@ api = { package = "substrate-rpc-api", path = "./api" }
 client = { package = "substrate-client", path = "../client" }
 codec = { package = "parity-scale-codec", version = "1.0.0" }
 futures03 = { package = "futures-preview", version = "0.3.0-alpha.19", features = ["compat"] }
-jsonrpc-pubsub = "13.1.0"
+jsonrpc-pubsub = "14.0"
 log = "0.4.8"
 primitives = { package = "substrate-primitives", path = "../primitives" }
-rpc = { package = "jsonrpc-core", version = "13.0.0" }
+rpc = { package = "jsonrpc-core", version = "14.0" }
 runtime_version = { package = "sr-version", path = "../sr-version" }
 serde_json = "1.0.41"
 session = { package = "substrate-session", path = "../session" }

substrate/core/rpc/api/Cargo.toml

@@ -8,10 +8,10 @@ edition = "2018"
 codec = { package = "parity-scale-codec", version = "1.0.0" }
 derive_more = "0.15.0"
 futures03 = { package = "futures-preview", version = "0.3.0-alpha.19", features = ["compat"] }
-jsonrpc-core = "13.2.0"
-jsonrpc-core-client = "13.2.0"
-jsonrpc-derive = "13.2.0"
-jsonrpc-pubsub = "13.2.0"
+jsonrpc-core = "14.0"
+jsonrpc-core-client = "14.0"
+jsonrpc-derive = "14.0"
+jsonrpc-pubsub = "14.0"
 log = "0.4.8"
 parking_lot = "0.9.0"
 primitives = { package = "substrate-primitives", path = "../../primitives" }

substrate/core/rpc/api/src/author/mod.rs

@@ -67,7 +67,7 @@ pub trait AuthorApi<Hash, BlockHash> {
 		subscribe,
 		name = "author_submitAndWatchExtrinsic"
 	)]
-	fn watch_extrinsic(&self,
+	fn submit_and_watch_extrinsic(&self,
 		metadata: Self::Metadata,
 		subscriber: Subscriber<Status<Hash, BlockHash>>,
 		bytes: Bytes
@@ -83,4 +83,16 @@ pub trait AuthorApi<Hash, BlockHash> {
 		metadata: Option<Self::Metadata>,
 		id: SubscriptionId
 	) -> Result<bool>;
+
+	/// Watch multiple extrinsics (own or from network)
+	#[pubsub(
+		subscription = "author_extrinsicUpdate",
+		subscribe,
+		name = "author_watchExtrinsic"
+	)]
+	fn watch_extrinsic(&self,
+    	metadata: Self::Metadata,
+		subscriber: Subscriber<Status<Hash, BlockHash>>,
+	    hash: Hash,
+	);
 }

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

@@ -153,6 +153,23 @@ impl<B, E, P, RA> AuthorApi<ExHash<P>, BlockHash<P>> for Author<B, E, P, RA> whe
 	}
 
 	fn watch_extrinsic(&self,
+    	_metadata: Self::Metadata,
+	    subscriber:  Subscriber<Status<ExHash<P>, BlockHash<P>>>,
+	    hash: ExHash<P>,
+	) {
+		let watcher = self.pool.watch(hash).into_stream().map(|v| Ok::<_, ()>(Ok(v)));
+		let subscriptions = self.subscriptions.clone();
+
+		subscriptions.add(subscriber,
+			move |sink| {
+				sink.sink_map_err(|_| unimplemented!())
+					.send_all(Compat::new(watcher))
+					.map(|_| ())
+			}
+		);
+	}
+
+	fn submit_and_watch_extrinsic(&self,
 		_metadata: Self::Metadata,
 		subscriber: Subscriber<Status<ExHash<P>, BlockHash<P>>>,
 		xt: Bytes

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

@@ -115,7 +115,7 @@ fn should_watch_extrinsic() {
 	let (subscriber, id_rx, data) = jsonrpc_pubsub::typed::Subscriber::new_test("test");
 
 	// when
-	p.watch_extrinsic(Default::default(), subscriber, uxt(AccountKeyring::Alice, 0).encode().into());
+	p.submit_and_watch_extrinsic(Default::default(), subscriber, uxt(AccountKeyring::Alice, 0).encode().into());
 
 	// then
 	assert_eq!(setup.runtime.block_on(id_rx), Ok(Ok(1.into())));
@@ -142,6 +142,53 @@ fn should_watch_extrinsic() {
 	);
 }
 
+
+#[test]
+fn should_watch_existing_extrinsic() {
+	// Initial setup is 1 submitted extrinsic
+	let mut runtime = runtime::Runtime::new().unwrap();
+	let client = Arc::new(test_client::new());
+	let pool = Arc::new(Pool::new(Default::default(), FullChainApi::new(client.clone())));
+	let keystore = KeyStore::new();
+	let p = Author {
+		client,
+		pool: pool.clone(),
+		subscriptions: Subscriptions::new(Arc::new(runtime.executor())),
+		keystore: keystore.clone(),
+	};
+	let (subscriber, id_rx, data) = jsonrpc_pubsub::typed::Subscriber::new_test("test");
+
+	let xt = uxt(AccountKeyring::Alice, 0).encode();
+	let xt_hash: H256 = blake2_256(&xt).into();
+	p.submit_extrinsic(xt.into()).wait().expect("Failed to submit extrinsic");
+
+	// Then we track it
+	p.watch_extrinsic(Default::default(), subscriber, xt_hash.into());
+	assert_eq!(runtime.block_on(id_rx), Ok(Ok(1.into())));
+
+	// Add replacement
+	let replacement = {
+		let tx = Transfer {
+			amount: 5,
+			nonce: 0,
+			from: AccountKeyring::Alice.into(),
+			to: Default::default(),
+		};
+		tx.into_signed_tx()
+	}.encode();
+	let replacement_hash = blake2_256(&replacement);
+	AuthorApi::submit_extrinsic(&p, replacement.into()).wait().unwrap();
+
+	// And check if the tracked one received usurped event
+	assert_eq!(
+		runtime.block_on(data.into_future()).unwrap().0,
+		Some(format!(
+			r#"{{"jsonrpc":"2.0","method":"test","params":{{"result":{{"usurped":"0x{}"}},"subscription":1}}}}"#,
+			HexDisplay::from(&replacement_hash))
+		)
+	);
+}
+
 #[test]
 fn should_return_watch_validation_error() {
 	//given
@@ -151,7 +198,7 @@ fn should_return_watch_validation_error() {
 	let (subscriber, id_rx, _data) = jsonrpc_pubsub::typed::Subscriber::new_test("test");
 
 	// when
-	p.watch_extrinsic(Default::default(), subscriber, uxt(AccountKeyring::Alice, 179).encode().into());
+	p.submit_and_watch_extrinsic(Default::default(), subscriber, uxt(AccountKeyring::Alice, 179).encode().into());
 
 	// then
 	let res = setup.runtime.block_on(id_rx).unwrap();