Add asynchronous methods to assign_id or reject in pubsub. (paritytech#401)

* Add asynchronous methods to assign_id or reject in pubsub.

* Make the channel impl more clear.

* Fix typos.

Author: tomusdrw

macros/examples/generic-trait-bounds.rs renamed to macros/examples/generic-trait-bounds2.rs

@@ -70,6 +70,7 @@ macro_rules! metadata {
 	};
 }
 
+/// Build an RPC trait definition.
 #[macro_export]
 macro_rules! build_rpc_trait {
 	(

macros/examples/generic-trait.rs renamed to macros/examples/generic-trait2.rs

@@ -7,7 +7,7 @@ use jsonrpc_pubsub as pubsub;
 use serde;
 use util::to_value;
 
-use self::core::futures::{self, Sink as FuturesSink, sync};
+use self::core::futures::{self, Future, Sink as FuturesSink, sync};
 
 pub use self::pubsub::SubscriptionId;
 
@@ -30,7 +30,7 @@ impl<T, E> Subscriber<T, E> {
 	/// Create new subscriber for tests.
 	pub fn new_test<M: Into<String>>(method: M) -> (
 		Self,
-		sync::oneshot::Receiver<Result<SubscriptionId, core::Error>>,
+		pubsub::oneshot::Receiver<Result<SubscriptionId, core::Error>>,
 		sync::mpsc::Receiver<String>,
 	) {
 		let (subscriber, id, subscription) = pubsub::Subscriber::new_test(method);
@@ -42,18 +42,37 @@ impl<T, E> Subscriber<T, E> {
 		self.subscriber.reject(error)
 	}
 
+	/// Reject subscription with given error.
+	pub fn reject_async(self, error: core::Error) -> impl Future<Item = (), Error = ()> {
+		self.subscriber.reject_async(error)
+	}
+
 	/// Assign id to this subscriber.
 	/// This method consumes `Subscriber` and returns `Sink`
 	/// if the connection is still open or error otherwise.
 	pub fn assign_id(self, id: SubscriptionId) -> Result<Sink<T, E>, ()> {
-		let sink = self.subscriber.assign_id(id.clone())?;
-		Ok(Sink {
-			id,
-			sink,
-			buffered: None,
-			_data: PhantomData,
-		})
+		self.subscriber.assign_id(id.clone())
+			.map(|sink| Sink {
+				id,
+				sink,
+				buffered: None,
+				_data: PhantomData,
+			})
+	}
+
+	/// Assign id to this subscriber.
+	/// This method consumes `Subscriber` and returns `Sink`
+	/// if the connection is still open or error otherwise.
+	pub fn assign_id_async(self, id: SubscriptionId) -> impl Future<Item = Sink<T, E>, Error = ()> {
+		self.subscriber.assign_id_async(id.clone())
+			.map(|sink| Sink {
+				id,
+				sink,
+				buffered: None,
+				_data: PhantomData,
+			})
 	}
+
 }
 
 /// Subscriber sink.

macros/examples/meta-macros.rs renamed to macros/examples/meta-macros2.rs

@@ -34,11 +34,12 @@ fn main() {
 				return;
 			}
 
-			let sink = subscriber.assign_id(SubscriptionId::Number(5)).unwrap();
-			// or subscriber.reject(Error {} );
-			// or drop(subscriber)
 			let is_done = is_done.clone();
 			thread::spawn(move || {
+				let sink = subscriber.assign_id_async(SubscriptionId::Number(5)).wait().unwrap();
+				// or subscriber.reject(Error {} );
+				// or drop(subscriber)
+
 				loop {
 					if is_done.load(atomic::Ordering::AcqRel) {
 						return;

macros/examples/pubsub-macros.rs renamed to macros/examples/pubsub-macros2.rs

@@ -32,10 +32,11 @@ fn main() {
 				return;
 			}
 
-			let sink = subscriber.assign_id(SubscriptionId::Number(5)).unwrap();
-			// or subscriber.reject(Error {} );
-			// or drop(subscriber)
 			thread::spawn(move || {
+				let sink = subscriber.assign_id_async(SubscriptionId::Number(5)).wait().unwrap();
+				// or subscriber.reject(Error {} );
+				// or drop(subscriber)
+
 				loop {
 					thread::sleep(time::Duration::from_millis(100));
 					match sink.notify(Params::Array(vec![Value::Number(10.into())])).wait() {

macros/examples/std.rs renamed to macros/examples/std2.rs

@@ -36,10 +36,11 @@ fn main() {
 				return;
 			}
 
-			let sink = subscriber.assign_id(SubscriptionId::Number(5)).unwrap();
-			// or subscriber.reject(Error {} );
-			// or drop(subscriber)
 			thread::spawn(move || {
+				let sink = subscriber.assign_id_async(SubscriptionId::Number(5)).wait().unwrap();
+				// or subscriber.reject(Error {} );
+				// or drop(subscriber)
+
 				loop {
 					thread::sleep(time::Duration::from_millis(100));
 					match sink.notify(Params::Array(vec![Value::Number(10.into())])).wait() {