[RPC] Move runtime version from chain to state (#1243)

* Move runtimeVersion to state, add rudimentary test for subscription.

* Bump to latest jsonrpc.

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

@@ -64,7 +64,7 @@ build_rpc_trait! {
 
 			/// Unsubscribe from extrinsic watching.
 			#[rpc(name = "author_unwatchExtrinsic")]
-			fn unwatch_extrinsic(&self, SubscriptionId) -> Result<bool>;
+			fn unwatch_extrinsic(&self, Self::Metadata, SubscriptionId) -> Result<bool>;
 		}
 
 	}
@@ -149,7 +149,7 @@ impl<B, E, P, RA> AuthorApi<ExHash<P>, BlockHash<P>> for Author<B, E, P, RA> whe
 		})
 	}
 
-	fn unwatch_extrinsic(&self, id: SubscriptionId) -> Result<bool> {
+	fn unwatch_extrinsic(&self, _metadata: Self::Metadata, id: SubscriptionId) -> Result<bool> {
 		Ok(self.subscriptions.cancel(id))
 	}
 }

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

@@ -21,13 +21,11 @@ use std::sync::Arc;
 use client::{self, Client, BlockchainEvents};
 use jsonrpc_macros::{pubsub, Trailing};
 use jsonrpc_pubsub::SubscriptionId;
+use primitives::{H256, Blake2Hasher};
 use rpc::Result as RpcResult;
 use rpc::futures::{stream, Future, Sink, Stream};
-use primitives::H256;
 use runtime_primitives::generic::{BlockId, SignedBlock};
 use runtime_primitives::traits::{Block as BlockT, Header, NumberFor};
-use runtime_version::RuntimeVersion;
-use primitives::{Blake2Hasher, storage};
 
 use subscriptions::Subscriptions;
 
@@ -60,10 +58,6 @@ build_rpc_trait! {
 		#[rpc(name = "chain_getFinalisedHead")]
 		fn finalised_head(&self) -> Result<Hash>;
 
-		/// Get the runtime version.
-		#[rpc(name = "chain_getRuntimeVersion")]
-		fn runtime_version(&self, Trailing<Hash>) -> Result<RuntimeVersion>;
-
 		#[pubsub(name = "chain_newHead")] {
 			/// New head subscription
 			#[rpc(name = "chain_subscribeNewHead", alias = ["subscribe_newHead", ])]
@@ -71,7 +65,7 @@ build_rpc_trait! {
 
 			/// Unsubscribe from new head subscription.
 			#[rpc(name = "chain_unsubscribeNewHead", alias = ["unsubscribe_newHead", ])]
-			fn unsubscribe_new_head(&self, SubscriptionId) -> RpcResult<bool>;
+			fn unsubscribe_new_head(&self, Self::Metadata, SubscriptionId) -> RpcResult<bool>;
 		}
 
 		#[pubsub(name = "chain_finalisedHead")] {
@@ -81,17 +75,7 @@ build_rpc_trait! {
 
 			/// Unsubscribe from new head subscription.
 			#[rpc(name = "chain_unsubscribeFinalisedHeads")]
-			fn unsubscribe_finalised_heads(&self, SubscriptionId) -> RpcResult<bool>;
-		}
-
-		#[pubsub(name = "chain_runtimeVersion")] {
-			/// New runtime version subscription
-			#[rpc(name = "chain_subscribeRuntimeVersion")]
-			fn subscribe_runtime_version(&self, Self::Metadata, pubsub::Subscriber<RuntimeVersion>);
-
-			/// Unsubscribe from runtime version subscription
-			#[rpc(name = "chain_unsubscribeRuntimeVersion")]
-			fn unsubscribe_runtime_version(&self, SubscriptionId) -> RpcResult<bool>;
+			fn unsubscribe_finalised_heads(&self, Self::Metadata, SubscriptionId) -> RpcResult<bool>;
 		}
 	}
 }
@@ -195,11 +179,6 @@ impl<B, E, Block, RA> ChainApi<Block::Hash, Block::Header, NumberFor<Block>, Sig
 		Ok(self.client.info()?.chain.finalized_hash)
 	}
 
-	fn runtime_version(&self, at: Trailing<Block::Hash>) -> Result<RuntimeVersion> {
-		let at = self.unwrap_or_best(at)?;
-		Ok(self.client.runtime_version_at(&BlockId::Hash(at))?)
-	}
-
 	fn subscribe_new_head(&self, _metadata: Self::Metadata, subscriber: pubsub::Subscriber<Block::Header>) {
 		self.subscribe_headers(
 			subscriber,
@@ -210,7 +189,7 @@ impl<B, E, Block, RA> ChainApi<Block::Hash, Block::Header, NumberFor<Block>, Sig
 		)
 	}
 
-	fn unsubscribe_new_head(&self, id: SubscriptionId) -> RpcResult<bool> {
+	fn unsubscribe_new_head(&self, _metadata: Self::Metadata, id: SubscriptionId) -> RpcResult<bool> {
 		Ok(self.subscriptions.cancel(id))
 	}
 
@@ -223,55 +202,7 @@ impl<B, E, Block, RA> ChainApi<Block::Hash, Block::Header, NumberFor<Block>, Sig
 		)
 	}
 
-	fn unsubscribe_finalised_heads(&self, id: SubscriptionId) -> RpcResult<bool> {
-		Ok(self.subscriptions.cancel(id))
-	}
-
-	fn subscribe_runtime_version(&self, _meta: Self::Metadata, subscriber: pubsub::Subscriber<RuntimeVersion>) {
-		let stream = match self.client.storage_changes_notification_stream(Some(&[storage::StorageKey(storage::well_known_keys::CODE.to_vec())])) {
-			Ok(stream) => stream,
-			Err(err) => {
-				let _ = subscriber.reject(error::Error::from(err).into());
-				return;
-			}
-		};
-
-		self.subscriptions.add(subscriber, |sink| {
-			let version = self.runtime_version(None.into())
-				.map_err(Into::into);
-
-			let client = self.client.clone();
-			let mut previous_version = version.clone();
-
-			let stream = stream
-				.map_err(|e| warn!("Error creating storage notification stream: {:?}", e))
-				.filter_map(move |_| {
-					let version = client.info().and_then(|info| {
-							client.runtime_version_at(&BlockId::hash(info.chain.best_hash))
-						})
-						.map_err(error::Error::from)
-						.map_err(Into::into);
-					if previous_version != version {
-						previous_version = version.clone();
-						Some(version)
-					} else {
-						None
-					}
-				});
-
-			sink
-				.sink_map_err(|e| warn!("Error sending notifications: {:?}", e))
-				.send_all(
-					stream::iter_result(vec![Ok(version)])
-					.chain(stream)
-				)
-				// we ignore the resulting Stream (if the first stream is over we are unsubscribed)
-				.map(|_| ())
-		});
-	}
-
-
-	fn unsubscribe_runtime_version(&self, id: SubscriptionId) -> RpcResult<bool> {
+	fn unsubscribe_finalised_heads(&self, _metadata: Self::Metadata, id: SubscriptionId) -> RpcResult<bool> {
 		Ok(self.subscriptions.cancel(id))
 	}
 }

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

@@ -17,7 +17,7 @@
 use super::*;
 use jsonrpc_macros::pubsub;
 use test_client::{self, TestClient};
-use test_client::runtime::{Block, Header, VERSION};
+use test_client::runtime::{Block, Header};
 use consensus::BlockOrigin;
 
 #[test]
@@ -246,19 +246,3 @@ fn should_notify_about_finalised_block() {
 	// no more notifications on this channel
 	assert_eq!(core.block_on(next.into_future()).unwrap().0, None);
 }
-
-#[test]
-fn should_return_runtime_version() {
-	let core = ::tokio::runtime::Runtime::new().unwrap();
-	let remote = core.executor();
-
-	let client = Chain {
-		client: Arc::new(test_client::new()),
-		subscriptions: Subscriptions::new(remote),
-	};
-
-	assert_matches!(
-		client.runtime_version(None.into()),
-		Ok(ref ver) if ver == &VERSION
-	);
-}

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

@@ -25,14 +25,14 @@ use client::{self, Client, CallExecutor, BlockchainEvents, runtime_api::Metadata
 use jsonrpc_macros::Trailing;
 use jsonrpc_macros::pubsub;
 use jsonrpc_pubsub::SubscriptionId;
-use primitives::H256;
+use primitives::{H256, Blake2Hasher, Bytes};
 use primitives::hexdisplay::HexDisplay;
-use primitives::storage::{StorageKey, StorageData, StorageChangeSet};
-use primitives::{Blake2Hasher, Bytes};
+use primitives::storage::{self, StorageKey, StorageData, StorageChangeSet};
 use rpc::Result as RpcResult;
 use rpc::futures::{stream, Future, Sink, Stream};
 use runtime_primitives::generic::BlockId;
 use runtime_primitives::traits::{Block as BlockT, Header, ProvideRuntimeApi};
+use runtime_version::RuntimeVersion;
 
 use subscriptions::Subscriptions;
 
@@ -67,6 +67,10 @@ build_rpc_trait! {
 		#[rpc(name = "state_getMetadata")]
 		fn metadata(&self, Trailing<Hash>) -> Result<Bytes>;
 
+		/// Get the runtime version.
+		#[rpc(name = "state_getRuntimeVersion", alias = ["chain_getRuntimeVersion", ])]
+		fn runtime_version(&self, Trailing<Hash>) -> Result<RuntimeVersion>;
+
 		/// 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.
@@ -74,6 +78,16 @@ build_rpc_trait! {
 		#[rpc(name = "state_queryStorage")]
 		fn query_storage(&self, Vec<StorageKey>, Hash, Trailing<Hash>) -> Result<Vec<StorageChangeSet<Hash>>>;
 
+		#[pubsub(name = "state_runtimeVersion")] {
+			/// New runtime version subscription
+			#[rpc(name = "state_subscribeRuntimeVersion", alias = ["chain_subscribeRuntimeVersion", ])]
+			fn subscribe_runtime_version(&self, Self::Metadata, pubsub::Subscriber<RuntimeVersion>);
+
+			/// Unsubscribe from runtime version subscription
+			#[rpc(name = "state_unsubscribeRuntimeVersion", alias = ["chain_unsubscribeRuntimeVersion", ])]
+			fn unsubscribe_runtime_version(&self, Self::Metadata, SubscriptionId) -> RpcResult<bool>;
+		}
+
 		#[pubsub(name = "state_storage")] {
 			/// New storage subscription
 			#[rpc(name = "state_subscribeStorage")]
@@ -81,7 +95,7 @@ build_rpc_trait! {
 
 			/// Unsubscribe from storage subscription
 			#[rpc(name = "state_unsubscribeStorage")]
-			fn unsubscribe_storage(&self, SubscriptionId) -> RpcResult<bool>;
+			fn unsubscribe_storage(&self, Self::Metadata, SubscriptionId) -> RpcResult<bool>;
 		}
 	}
 }
@@ -268,7 +282,59 @@ impl<B, E, Block, RA> StateApi<Block::Hash> for State<B, E, Block, RA> where
 		})
 	}
 
-	fn unsubscribe_storage(&self, id: SubscriptionId) -> RpcResult<bool> {
+	fn unsubscribe_storage(&self, _meta: Self::Metadata, id: SubscriptionId) -> RpcResult<bool> {
+		Ok(self.subscriptions.cancel(id))
+	}
+
+	fn runtime_version(&self, at: Trailing<Block::Hash>) -> Result<RuntimeVersion> {
+		let at = self.unwrap_or_best(at)?;
+		Ok(self.client.runtime_version_at(&BlockId::Hash(at))?)
+	}
+
+	fn subscribe_runtime_version(&self, _meta: Self::Metadata, subscriber: pubsub::Subscriber<RuntimeVersion>) {
+		let stream = match self.client.storage_changes_notification_stream(Some(&[StorageKey(storage::well_known_keys::CODE.to_vec())])) {
+			Ok(stream) => stream,
+			Err(err) => {
+				let _ = subscriber.reject(error::Error::from(err).into());
+				return;
+			}
+		};
+
+		self.subscriptions.add(subscriber, |sink| {
+			let version = self.runtime_version(None.into())
+				.map_err(Into::into);
+
+			let client = self.client.clone();
+			let mut previous_version = version.clone();
+
+			let stream = stream
+				.map_err(|e| warn!("Error creating storage notification stream: {:?}", e))
+				.filter_map(move |_| {
+					let version = client.info().and_then(|info| {
+							client.runtime_version_at(&BlockId::hash(info.chain.best_hash))
+						})
+						.map_err(error::Error::from)
+						.map_err(Into::into);
+					if previous_version != version {
+						previous_version = version.clone();
+						Some(version)
+					} else {
+						None
+					}
+				});
+
+			sink
+				.sink_map_err(|e| warn!("Error sending notifications: {:?}", e))
+				.send_all(
+					stream::iter_result(vec![Ok(version)])
+					.chain(stream)
+				)
+				// we ignore the resulting Stream (if the first stream is over we are unsubscribed)
+				.map(|_| ())
+		});
+	}
+
+	fn unsubscribe_runtime_version(&self, _meta: Self::Metadata, id: SubscriptionId) -> RpcResult<bool> {
 		Ok(self.subscriptions.cancel(id))
 	}
 }

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

@@ -178,3 +178,39 @@ fn should_query_storage() {
 	});
 	assert_eq!(result.unwrap(), expected);
 }
+
+
+#[test]
+fn should_return_runtime_version() {
+	let core = ::tokio::runtime::Runtime::new().unwrap();
+
+	let client = Arc::new(test_client::new());
+	let api = State::new(client.clone(), Subscriptions::new(core.executor()));
+
+	assert_matches!(
+		api.runtime_version(None.into()),
+		Ok(ref ver) if ver == &runtime::VERSION
+	);
+}
+
+#[test]
+fn should_notify_on_runtime_version_initially() {
+	let mut core = ::tokio::runtime::Runtime::new().unwrap();
+	let (subscriber, id, transport) = pubsub::Subscriber::new_test("test");
+
+	{
+		let client = Arc::new(test_client::new());
+		let api = State::new(client.clone(), Subscriptions::new(core.executor()));
+
+		api.subscribe_runtime_version(Default::default(), subscriber);
+
+		// assert id assigned
+		assert_eq!(core.block_on(id), Ok(Ok(SubscriptionId::Number(1))));
+	}
+
+	// assert initial version sent.
+	let (notification, next) = core.block_on(transport.into_future()).unwrap();
+	assert!(notification.is_some());
+		// no more notifications on this channel
+	assert_eq!(core.block_on(next.into_future()).unwrap().0, None);
+}