Add subscription ID recognition to tungstenite and ws-rpc client (#662)

* fix error

* fix clippy warning

* seems to wrok

* check for subscirption id

* fix clippy

* fix typo

* remove error logs

* add info log

* we expect an immediate response

* add tests for rpc client

* fix test and clean up ws_rpc client

* fix test

* fix test

* fix clippy

* fix tests

* git commit

* add missing comma

* extract helper functions

* add missing helper file

* remove code from ws-client and use helpers

* fix compile errors

.github/workflows/ci.yml

@@ -153,6 +153,9 @@ jobs:
           pallet_balances_tests,
           pallet_transaction_payment_tests,
           state_tests,
+          tungstenite_client_test,
+          ws_client_test,
+          state_tests,
           runtime_update_sync,
           runtime_update_async,
         ]

examples/examples/transfer_with_tungstenite_client.rs

@@ -55,8 +55,22 @@ fn main() {
 	let bob_balance = api.get_account_data(&bob.into()).unwrap().unwrap_or_default().free;
 	println!("[+] Bob's Free Balance is {}\n", bob_balance);
 
-	// Generate extrinsic.
-	let xt = api.balance_transfer_allow_death(MultiAddress::Id(bob.into()), 1000000000000);
+	// We first generate an extrinsic that will fail to be executed due to missing funds.
+	let xt = api.balance_transfer_allow_death(MultiAddress::Id(bob.into()), bob_balance + 1);
+	println!(
+		"Sending an extrinsic from Alice (Key = {}),\n\nto Bob (Key = {})\n",
+		alice.public(),
+		bob
+	);
+	println!("[+] Composed extrinsic: {:?}\n", xt);
+
+	// Send and watch extrinsic until it fails onchain.
+	let result = api.submit_and_watch_extrinsic_until(xt, XtStatus::InBlock);
+	assert!(result.is_err());
+	println!("[+] Extrinsic did not get included due to: {:?}\n", result);
+
+	// This time, we generate an extrinsic that will succeed.
+	let xt = api.balance_transfer_allow_death(MultiAddress::Id(bob.into()), bob_balance / 2);
 	println!(
 		"Sending an extrinsic from Alice (Key = {}),\n\nto Bob (Key = {})\n",
 		alice.public(),
@@ -70,7 +84,7 @@ fn main() {
 		.unwrap()
 		.block_hash
 		.unwrap();
-	println!("[+] Extrinsic got included. Hash: {:?}\n", block_hash);
+	println!("[+] Extrinsic got included. Block Hash: {:?}\n", block_hash);
 
 	// Verify that Bob's free Balance increased.
 	let bob_new_balance = api.get_account_data(&bob.into()).unwrap().unwrap().free;

examples/examples/transfer_with_ws_client.rs

@@ -54,8 +54,22 @@ fn main() {
 	let bob_balance = api.get_account_data(&bob.into()).unwrap().unwrap_or_default().free;
 	println!("[+] Bob's Free Balance is {}\n", bob_balance);
 
-	// Generate extrinsic.
-	let xt = api.balance_transfer_allow_death(MultiAddress::Id(bob.into()), 1000000000000);
+	// We first generate an extrinsic that will fail to be executed due to missing funds.
+	let xt = api.balance_transfer_allow_death(MultiAddress::Id(bob.into()), bob_balance + 1);
+	println!(
+		"Sending an extrinsic from Alice (Key = {}),\n\nto Bob (Key = {})\n",
+		alice.public(),
+		bob
+	);
+	println!("[+] Composed extrinsic: {:?}\n", xt);
+
+	// Send and watch extrinsic until it fails onchain.
+	let result = api.submit_and_watch_extrinsic_until(xt, XtStatus::InBlock);
+	assert!(result.is_err());
+	println!("[+] Extrinsic did not get included due to: {:?}\n", result);
+
+	// This time, we generate an extrinsic that will succeed.
+	let xt = api.balance_transfer_allow_death(MultiAddress::Id(bob.into()), bob_balance / 2);
 	println!(
 		"Sending an extrinsic from Alice (Key = {}),\n\nto Bob (Key = {})\n",
 		alice.public(),
@@ -69,7 +83,7 @@ fn main() {
 		.unwrap()
 		.block_hash
 		.unwrap();
-	println!("[+] Extrinsic got included. Hash: {:?}\n", block_hash);
+	println!("[+] Extrinsic got included. Block Hash: {:?}\n", block_hash);
 
 	// Verify that Bob's free Balance increased.
 	let bob_new_balance = api.get_account_data(&bob.into()).unwrap().unwrap().free;

src/rpc/error.rs

@@ -22,6 +22,7 @@ pub type Result<T> = core::result::Result<T, Error>;
 #[derive(Debug)]
 pub enum Error {
 	SerdeJson(serde_json::error::Error),
+	ExtrinsicFailed(String),
 	MpscSend(String),
 	InvalidUrl(String),
 	RecvError(String),

src/rpc/helpers.rs

@@ -0,0 +1,148 @@
+/*
+   Copyright 2019 Supercomputing Systems AG
+
+   Licensed under the Apache License, Version 2.0 (the "License");
+   you may not use this file except in compliance with the License.
+   You may obtain a copy of the License at
+
+	   http://www.apache.org/licenses/LICENSE-2.0
+
+   Unless required by applicable law or agreed to in writing, software
+   distributed under the License is distributed on an "AS IS" BASIS,
+   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+   See the License for the specific language governing permissions and
+   limitations under the License.
+
+*/
+
+use alloc::{
+	format,
+	string::{String, ToString},
+};
+use serde_json::Value;
+
+pub fn read_subscription_id(value: &Value) -> Option<String> {
+	value["result"].as_str().map(|str| str.to_string())
+}
+
+pub fn read_error_message(value: &Value, msg: &str) -> String {
+	match value["error"].as_str() {
+		Some(error_message) => error_message.to_string(),
+		None => format!("Unexpected Response: {}", msg),
+	}
+}
+
+pub fn subscription_id_matches(value: &Value, subscription_id: &str) -> bool {
+	match value["params"]["subscription"].as_str() {
+		Some(retrieved_subscription_id) => subscription_id == retrieved_subscription_id,
+		None => false,
+	}
+}
+
+#[cfg(test)]
+mod tests {
+	use super::*;
+	use serde_json::json;
+
+	#[test]
+	fn read_valid_subscription_response() {
+		let subcription_id = "tejkataa12124a";
+		let value = json!({
+			"result": subcription_id,
+			"id": 43,
+			"and_so_on": "test",
+		});
+
+		let maybe_subcription_id = read_subscription_id(&value);
+		assert_eq!(maybe_subcription_id, Some(subcription_id.to_string()));
+	}
+
+	#[test]
+	fn read_invalid_subscription_response() {
+		let subcription_id = "tejkataa12124a";
+		let value = json!({
+			"error": subcription_id,
+			"id": 43,
+			"and_so_on": "test",
+		});
+
+		let maybe_subcription_id = read_subscription_id(&value);
+		assert!(maybe_subcription_id.is_none());
+	}
+
+	#[test]
+	fn read_error_message_returns_error_if_available() {
+		let error_message = "some_error_message";
+		let value = json!({
+			"error": error_message,
+			"id": 43,
+			"and_so_on": "test",
+		});
+
+		let msg = serde_json::to_string(&value).unwrap();
+
+		let message = read_error_message(&value, &msg);
+		assert!(message.contains(error_message));
+		assert!(message.contains("error"));
+	}
+
+	#[test]
+	fn read_error_message_returns_full_msg_if_error_is_not_available() {
+		let error_message = "some_error_message";
+		let value = json!({
+			"result": error_message,
+			"id": 43,
+			"and_so_on": "test",
+		});
+
+		let msg = serde_json::to_string(&value).unwrap();
+
+		let message = read_error_message(&value, &msg);
+		assert!(message.contains(&msg));
+	}
+
+	#[test]
+	fn subscription_id_matches_returns_true_for_equal_id() {
+		let subcription_id = "tejkataa12124a";
+		let value = json!({
+			"params": {
+				"subscription": subcription_id,
+				"message": "Test"
+			},
+			"id": 43,
+			"and_so_on": "test",
+		});
+
+		assert!(subscription_id_matches(&value, subcription_id));
+	}
+
+	#[test]
+	fn subscription_id_matches_returns_false_for_not_equal_id() {
+		let subcription_id = "tejkataa12124a";
+		let value = json!({
+			"params": {
+				"subscription": "something else",
+				"message": "Test"
+			},
+			"id": 43,
+			"and_so_on": "test",
+		});
+
+		assert!(!subscription_id_matches(&value, subcription_id));
+	}
+
+	#[test]
+	fn subscription_id_matches_returns_false_for_missing_subscription() {
+		let subcription_id = "tejkataa12124a";
+		let value = json!({
+			"params": {
+				"result": subcription_id,
+				"message": "Test"
+			},
+			"id": 43,
+			"and_so_on": "test",
+		});
+
+		assert!(!subscription_id_matches(&value, subcription_id));
+	}
+}

src/rpc/mod.rs

@@ -35,6 +35,8 @@ pub use jsonrpsee_client::JsonrpseeClient;
 pub mod jsonrpsee_client;
 
 pub mod error;
+#[cfg(any(feature = "ws-client", feature = "tungstenite-client"))]
+mod helpers;
 
 pub use error::{Error, Result};
 

src/rpc/tungstenite_client/client.rs

@@ -15,7 +15,7 @@
 
 */
 use crate::rpc::{
-	to_json_req, tungstenite_client::subscription::TungsteniteSubscriptionWrapper,
+	helpers, to_json_req, tungstenite_client::subscription::TungsteniteSubscriptionWrapper,
 	Error as RpcClientError, Request, Result, Subscribe,
 };
 use ac_primitives::RpcParams;
@@ -134,9 +134,22 @@ fn subscribe_to_server(
 	// Subscribe to server
 	socket.send(Message::Text(json_req))?;
 
+	// Read the first message response - must be the subscription id.
+	let msg = read_until_text_message(&mut socket)?;
+	let value: Value = serde_json::from_str(&msg)?;
+
+	let subcription_id = match helpers::read_subscription_id(&value) {
+		Some(id) => id,
+		None => {
+			let message = helpers::read_error_message(&value, &msg);
+			result_in.send(message)?;
+			return Ok(())
+		},
+	};
+
 	loop {
 		let msg = read_until_text_message(&mut socket)?;
-		send_message_to_client(result_in.clone(), msg.as_str())?;
+		send_message_to_client(result_in.clone(), &msg, &subcription_id)?;
 	}
 }
 
@@ -147,19 +160,18 @@ pub fn do_reconnect(error: &RpcClientError) -> bool {
 	)
 }
 
-fn send_message_to_client(result_in: ThreadOut<String>, message: &str) -> Result<()> {
-	debug!("got on_subscription_msg {}", message);
+fn send_message_to_client(
+	result_in: ThreadOut<String>,
+	message: &str,
+	subscription_id: &str,
+) -> Result<()> {
+	info!("got on_subscription_msg {}", message);
 	let value: Value = serde_json::from_str(message)?;
 
-	match value["id"].as_str() {
-		Some(_idstr) => {
-			warn!("Expected subscription, but received an id response instead: {:?}", value);
-		},
-		None => {
-			let message = serde_json::to_string(&value["params"]["result"])?;
-			result_in.send(message)?;
-		},
-	};
+	if helpers::subscription_id_matches(&value, subscription_id) {
+		result_in.send(serde_json::to_string(&value["params"]["result"])?)?;
+	}
+
 	Ok(())
 }
 

src/rpc/tungstenite_client/subscription.rs

@@ -15,7 +15,7 @@
 
 */
 
-use crate::rpc::{HandleSubscription, Result};
+use crate::rpc::{Error, HandleSubscription, Result};
 use core::marker::PhantomData;
 use serde::de::DeserializeOwned;
 use std::sync::mpsc::Receiver;
@@ -42,7 +42,7 @@ impl<Notification: DeserializeOwned> HandleSubscription<Notification>
 			// Sender was disconnected, therefore no further messages are to be expected.
 			Err(_e) => return None,
 		};
-		Some(serde_json::from_str(&notification).map_err(|e| e.into()))
+		Some(serde_json::from_str(&notification).map_err(|_| Error::ExtrinsicFailed(notification)))
 	}
 
 	async fn unsubscribe(self) -> Result<()> {

src/rpc/ws_client/client.rs

@@ -50,7 +50,7 @@ impl WsRpcClient {
 impl Request for WsRpcClient {
 	async fn request<R: DeserializeOwned>(&self, method: &str, params: RpcParams) -> Result<R> {
 		let json_req = to_json_req(method, params)?;
-		let response = self.direct_rpc_request(json_req, RequestHandler::default())??;
+		let response = self.direct_rpc_request(json_req, RequestHandler)??;
 		let deserialized_value: R = serde_json::from_str(&response)?;
 		Ok(deserialized_value)
 	}