From c32679a7bf9fb617c49d62228dd4c6cd8945db32 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Tomasz=20Drwi=C4=99ga?= <tomasz@parity.io>
Date: Mon, 10 Apr 2017 17:00:04 +0200
Subject: [PATCH 1/2] Improving the external API.

---
 core/src/io.rs             |  3 ++-
 core/src/types/request.rs  | 12 ++++++++++++
 macros/src/auto_args.rs    |  1 +
 pubsub/src/subscription.rs | 10 ++++++++++
 4 files changed, 25 insertions(+), 1 deletion(-)

diff --git a/core/src/io.rs b/core/src/io.rs
index 6830ae613..64274278f 100644
--- a/core/src/io.rs
+++ b/core/src/io.rs
@@ -192,7 +192,8 @@ impl<T: Metadata, S: Middleware<T>> MetaIoHandler<T, S> {
 		}).boxed()
 	}
 
-	fn handle_call(&self, call: Call, meta: T) -> BoxFuture<Option<Output>, ()> {
+	/// Handle single call asynchronously.
+	pub fn handle_call(&self, call: Call, meta: T) -> BoxFuture<Option<Output>, ()> {
 		match call {
 			Call::MethodCall(method) => {
 				let params = method.params.unwrap_or(Params::None);
diff --git a/core/src/types/request.rs b/core/src/types/request.rs
index 074f1eac0..213a1ab6c 100644
--- a/core/src/types/request.rs
+++ b/core/src/types/request.rs
@@ -47,6 +47,18 @@ pub enum Call {
 
 }
 
+impl From<MethodCall> for Call {
+	fn from(mc: MethodCall) -> Self {
+		Call::MethodCall(mc)
+	}
+}
+
+impl From<Notification> for Call {
+	fn from(n: Notification) -> Self {
+		Call::Notification(n)
+	}
+}
+
 impl Serialize for Call {
 	fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
 	where S: Serializer {
diff --git a/macros/src/auto_args.rs b/macros/src/auto_args.rs
index dd1b46725..1ddb2f050 100644
--- a/macros/src/auto_args.rs
+++ b/macros/src/auto_args.rs
@@ -233,6 +233,7 @@ macro_rules! build_rpc_trait {
 				)
 			}),
 			($unsubscribe, move |base, id| {
+				use futures::Future;
 				Self::$unsub_method(base, id).map($crate::to_value).boxed()
 			}),
 		);
diff --git a/pubsub/src/subscription.rs b/pubsub/src/subscription.rs
index 2c837e88a..48e7d3480 100644
--- a/pubsub/src/subscription.rs
+++ b/pubsub/src/subscription.rs
@@ -1,5 +1,6 @@
 //! Subscription primitives.
 
+use std::fmt;
 use std::collections::HashMap;
 use std::sync::Arc;
 use parking_lot::Mutex;
@@ -19,6 +20,15 @@ pub struct Session {
 	on_drop: Mutex<Vec<Box<Fn() + Send>>>,
 }
 
+impl fmt::Debug for Session {
+	fn fmt(&self, fmt: &mut fmt::Formatter) -> fmt::Result {
+		fmt.debug_struct("pubsub::Session")
+			.field("active_subscriptions", &self.active_subscriptions.lock().len())
+			.field("transport", &self.transport)
+			.finish()
+	}
+}
+
 impl Session {
 	/// Creates new session given transport raw send capabilities.
 	/// Session should be created as part of metadata, `sender` should be returned by transport.

From c7839abc18bc39f3bc53c649d2ff2f8fc0271da2 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Tomasz=20Drwi=C4=99ga?= <tomasz@parity.io>
Date: Mon, 10 Apr 2017 17:22:50 +0200
Subject: [PATCH 2/2] Fix output format for subscriptions.

---
 macros/src/auto_args.rs |  2 +-
 macros/src/lib.rs       |  2 +-
 macros/src/pubsub.rs    | 10 ++++++++--
 3 files changed, 10 insertions(+), 4 deletions(-)

diff --git a/macros/src/auto_args.rs b/macros/src/auto_args.rs
index 1ddb2f050..5296eec60 100644
--- a/macros/src/auto_args.rs
+++ b/macros/src/auto_args.rs
@@ -233,7 +233,7 @@ macro_rules! build_rpc_trait {
 				)
 			}),
 			($unsubscribe, move |base, id| {
-				use futures::Future;
+				use $crate::jsonrpc_core::futures::Future;
 				Self::$unsub_method(base, id).map($crate::to_value).boxed()
 			}),
 		);
diff --git a/macros/src/lib.rs b/macros/src/lib.rs
index b222eb295..2dd165827 100644
--- a/macros/src/lib.rs
+++ b/macros/src/lib.rs
@@ -1,4 +1,4 @@
-extern crate jsonrpc_core;
+pub extern crate jsonrpc_core;
 extern crate jsonrpc_pubsub;
 extern crate serde;
 
diff --git a/macros/src/pubsub.rs b/macros/src/pubsub.rs
index 33ea822a1..c4c1e62b2 100644
--- a/macros/src/pubsub.rs
+++ b/macros/src/pubsub.rs
@@ -25,8 +25,9 @@ impl<T> Subscriber<T> {
 	}
 
 	pub fn assign_id(self, id: SubscriptionId) -> Result<Sink<T>, ()> {
-		let sink = self.subscriber.assign_id(id)?;
+		let sink = self.subscriber.assign_id(id.clone())?;
 		Ok(Sink {
+			id: id,
 			sink: sink,
 			_data: PhantomData,
 		})
@@ -35,12 +36,17 @@ impl<T> Subscriber<T> {
 
 pub struct Sink<T> {
 	sink: pubsub::Sink,
+	id: SubscriptionId,
 	_data: PhantomData<T>,
 }
 
 impl<T: serde::Serialize> Sink<T> {
 	pub fn send(&self, val: T) -> pubsub::SinkResult {
+		let id = self.id.clone().into();
 		let val = to_value(val);
-		self.sink.send(core::Params::Array(vec![val]))
+		self.sink.send(core::Params::Map(vec![
+			("subscription".to_owned(), id),
+			("result".to_owned(), val),
+		].into_iter().collect()))
 	}
 }