[servers]: subscriptionSink support params.

examples/Cargo.toml

@@ -25,6 +25,10 @@ path = "ws.rs"
 name = "ws_subscription"
 path = "ws_subscription.rs"
 
+[[example]]
+name = "ws_sub_with_params"
+path = "ws_sub_with_params.rs"
+
 [[example]]
 name = "proc_macro"
 path = "proc_macro.rs"

examples/ws_sub_with_params.rs

@@ -0,0 +1,73 @@
+// Copyright 2019 Parity Technologies (UK) Ltd.
+//
+// Permission is hereby granted, free of charge, to any
+// person obtaining a copy of this software and associated
+// documentation files (the "Software"), to deal in the
+// Software without restriction, including without
+// limitation the rights to use, copy, modify, merge,
+// publish, distribute, sublicense, and/or sell copies of
+// the Software, and to permit persons to whom the Software
+// is furnished to do so, subject to the following
+// conditions:
+//
+// The above copyright notice and this permission notice
+// shall be included in all copies or substantial portions
+// of the Software.
+//
+// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF
+// ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED
+// TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A
+// PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT
+// SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY
+// CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION
+// OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR
+// IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
+// DEALINGS IN THE SOFTWARE.
+
+use jsonrpsee::{
+	ws_client::{traits::SubscriptionClient, v2::params::JsonRpcParams, WsClientBuilder},
+	ws_server::WsServer,
+};
+use std::net::SocketAddr;
+
+#[tokio::main]
+async fn main() -> anyhow::Result<()> {
+	env_logger::init();
+	let addr = run_server().await?;
+	let url = format!("ws://{}", addr);
+
+	let client = WsClientBuilder::default().build(&url).await?;
+
+	// Subscription with a single parameter
+	let params = JsonRpcParams::Array(vec![3.into()]);
+	let mut sub_params_one = client.subscribe::<Option<char>>("sub_one_param", params, "unsub_one_param").await?;
+	println!("subscription with one param: {:?}", sub_params_one.next().await);
+
+	// Subscription with multiple parameters
+	let params = JsonRpcParams::Array(vec![2.into(), 5.into()]);
+	let mut sub_params_two = client.subscribe::<String>("sub_params_two", params, "unsub_params_two").await?;
+	println!("subscription with two params: {:?}", sub_params_two.next().await);
+
+	Ok(())
+}
+
+async fn run_server() -> anyhow::Result<SocketAddr> {
+	const LETTERS: &'static str = "abcdefghijklmnopqrstuvxyz";
+	let mut server = WsServer::new("127.0.0.1:0").await?;
+	let one_param = server.register_subscription("sub_one_param", "unsub_one_param").unwrap();
+	let two_params = server.register_subscription("sub_params_two", "unsub_params_two").unwrap();
+
+	std::thread::spawn(move || loop {
+		let _ = one_param.send_each(|idx| Ok(LETTERS.chars().nth(*idx)));
+		std::thread::sleep(std::time::Duration::from_millis(50));
+	});
+
+	std::thread::spawn(move || loop {
+		let _ = two_params.send_each(|params: &Vec<usize>| Ok(Some(LETTERS[params[0]..params[1]].to_string())));
+		std::thread::sleep(std::time::Duration::from_millis(100));
+	});
+
+	let addr = server.local_addr()?;
+	tokio::spawn(async move { server.start().await });
+	Ok(addr)
+}

examples/ws_subscription.rs

@@ -54,10 +54,10 @@ async fn main() -> anyhow::Result<()> {
 
 async fn run_server() -> anyhow::Result<SocketAddr> {
 	let mut server = WsServer::new("127.0.0.1:0").await?;
-	let mut subscription = server.register_subscription("subscribe_hello", "unsubscribe_hello").unwrap();
+	let subscription = server.register_subscription::<()>("subscribe_hello", "unsubscribe_hello").unwrap();
 
 	std::thread::spawn(move || loop {
-		subscription.send(&"hello my friend").unwrap();
+		subscription.broadcast(&"hello my friend").unwrap();
 		std::thread::sleep(std::time::Duration::from_secs(1));
 	});
 

tests/tests/helpers.rs

@@ -25,7 +25,10 @@
 // DEALINGS IN THE SOFTWARE.
 
 use futures_channel::oneshot;
-use jsonrpsee::{http_server::HttpServerBuilder, ws_server::WsServer};
+use jsonrpsee::{
+	http_server::HttpServerBuilder,
+	ws_server::{SubscriptionSink, WsServer},
+};
 use std::net::SocketAddr;
 use std::time::Duration;
 
@@ -36,9 +39,11 @@ pub async fn websocket_server_with_subscription() -> SocketAddr {
 		let rt = tokio::runtime::Runtime::new().unwrap();
 
 		let mut server = rt.block_on(WsServer::new("127.0.0.1:0")).unwrap();
-		let mut sub_hello = server.register_subscription("subscribe_hello", "unsubscribe_hello").unwrap();
-		let mut sub_foo = server.register_subscription("subscribe_foo", "unsubscribe_foo").unwrap();
-
+		let sub_hello: SubscriptionSink<()> =
+			server.register_subscription("subscribe_hello", "unsubscribe_hello").unwrap();
+		let sub_foo: SubscriptionSink<()> = server.register_subscription("subscribe_foo", "unsubscribe_foo").unwrap();
+		let sub_add_one: SubscriptionSink<u64> =
+			server.register_subscription("subscribe_add_one", "unsubscribe_add_one").unwrap();
 		server.register_method("say_hello", |_| Ok("hello")).unwrap();
 
 		server_started_tx.send(server.local_addr().unwrap()).unwrap();
@@ -49,8 +54,9 @@ pub async fn websocket_server_with_subscription() -> SocketAddr {
 			loop {
 				tokio::time::sleep(Duration::from_millis(100)).await;
 
-				sub_hello.send(&"hello from subscription").unwrap();
-				sub_foo.send(&1337_u64).unwrap();
+				sub_hello.broadcast(&"hello from subscription").unwrap();
+				sub_foo.broadcast(&1337_u64).unwrap();
+				sub_add_one.send_each(|p| Ok(Some(*p + 1))).unwrap();
 			}
 		});
 	});

tests/tests/integration_tests.rs

@@ -54,6 +54,20 @@ async fn ws_subscription_works() {
 	}
 }
 
+#[tokio::test]
+async fn ws_subscription_with_input_works() {
+	let server_addr = websocket_server_with_subscription().await;
+	let server_url = format!("ws://{}", server_addr);
+	let client = WsClientBuilder::default().build(&server_url).await.unwrap();
+	let mut add_one: Subscription<u64> =
+		client.subscribe("subscribe_add_one", vec![1.into()].into(), "unsubscribe_add_one").await.unwrap();
+
+	for _ in 0..2 {
+		let two = add_one.next().await.unwrap();
+		assert_eq!(two, 2);
+	}
+}
+
 #[tokio::test]
 async fn ws_method_call_works() {
 	let server_addr = websocket_server().await;

types/src/error.rs

@@ -40,7 +40,7 @@ pub enum Error {
 	Request(String),
 	/// Frontend/backend channel error.
 	#[error("Frontend/backend channel error: {0}")]
-	Internal(#[source] futures_channel::mpsc::SendError),
+	Internal(#[from] futures_channel::mpsc::SendError),
 	/// Invalid response,
 	#[error("Invalid response: {0}")]
 	InvalidResponse(Mismatch<String>),

types/src/v2/params.rs

@@ -83,10 +83,8 @@ impl<'a> RpcParams<'a> {
 	where
 		T: Deserialize<'a>,
 	{
-		match self.0 {
-			None => Err(CallError::InvalidParams),
-			Some(params) => serde_json::from_str(params).map_err(|_| CallError::InvalidParams),
-		}
+		let params = self.0.unwrap_or("null");
+		serde_json::from_str(params).map_err(|_| CallError::InvalidParams)
 	}
 
 	/// Attempt to parse only the first parameter from an array into type T