From 7ef90a33b61f1314c168bc391344e4d2dff82028 Mon Sep 17 00:00:00 2001
From: Hernando Castano <castano.ha@gmail.com>
Date: Tue, 19 May 2020 17:23:51 -0400
Subject: [PATCH 01/16] WIP: Add a subscription manager.

The idea is to use the `Subscriptions` struct from Substrate, which
is used to drive subscription futures to completion, and modify
it for "general" use.
---
 pubsub/src/lib.rs     |   1 +
 pubsub/src/manager.rs | 115 ++++++++++++++++++++++++++++++++++++++++++
 pubsub/src/types.rs   |  44 ++++++++++++----
 3 files changed, 149 insertions(+), 11 deletions(-)
 create mode 100644 pubsub/src/manager.rs

diff --git a/pubsub/src/lib.rs b/pubsub/src/lib.rs
index a4f74eca5..145304486 100644
--- a/pubsub/src/lib.rs
+++ b/pubsub/src/lib.rs
@@ -9,6 +9,7 @@ extern crate log;
 
 mod delegates;
 mod handler;
+pub mod manager;
 pub mod oneshot;
 mod subscription;
 pub mod typed;
diff --git a/pubsub/src/manager.rs b/pubsub/src/manager.rs
new file mode 100644
index 000000000..3a1608fb6
--- /dev/null
+++ b/pubsub/src/manager.rs
@@ -0,0 +1,115 @@
+//! Provides an executor for subscription Futures.
+
+use std::collections::HashMap;
+use std::hash::Hash;
+use std::sync::{
+	atomic::{self, AtomicUsize},
+	Arc,
+};
+
+use crate::core::futures::sync::oneshot;
+use crate::core::futures::{future, Future};
+use crate::{
+	typed::{Sink, Subscriber},
+	SubscriptionId,
+};
+use log::{error, warn};
+use parking_lot::Mutex;
+
+/// Alias for an implementation of `futures::future::Executor`.
+pub type TaskExecutor = Arc<dyn future::Executor<Box<dyn Future<Item = (), Error = ()> + Send>> + Send + Sync>;
+
+/// Trait used to provide unique subscription ids.
+pub trait IdProvider {
+	// TODO: Maybe have this impl Into<u64>?
+	type Id: Clone + Default + Eq + Hash;
+
+	/// Returns next id for the subscription.
+	fn next_id(&self) -> Self::Id;
+}
+
+/// Trait used to drive subscription Futures to completion.
+pub trait SubscriptionManager {
+	/// Create a new `SubscriptionManager`.
+	fn new(&self) -> Self;
+	/// Borrows the internal task executor.
+	///
+	/// This can be used to spawn additional tasks on the underlying event loop.
+	fn executor(&self) -> &TaskExecutor;
+	/// Create new subscription for given subscriber.
+	///
+	/// Second parameter is a function that converts Subscriber sink into a future.
+	/// This future will be driven to completion by the underlying event loop
+	/// or will be cancelled in case #cancel is invoked.
+	fn add<T, E, G, R, F, N>(&self, subscriber: Subscriber<T, E>, into_future: G) -> SubscriptionId<N>
+	where
+		G: FnOnce(Sink<T, E>) -> R,
+		R: future::IntoFuture<Future = F, Item = (), Error = ()>,
+		F: future::Future<Item = (), Error = ()> + Send + 'static;
+	/// Cancel subscription.
+	///
+	/// Should true if subscription existed or false otherwise.
+	fn cancel<N>(&self, id: SubscriptionId<N>) -> bool;
+}
+
+/// Subscriptions manager.
+///
+/// Takes care of assigning unique subscription ids and
+/// driving the sinks into completion.
+#[derive(Clone)]
+pub struct Manager<I: Default + IdProvider> {
+	next_id: I,
+	active_subscriptions: Arc<Mutex<HashMap<I::Id, oneshot::Sender<()>>>>,
+	executor: TaskExecutor, // Make generic?
+}
+
+impl<I: Default + IdProvider> SubscriptionManager for Manager<I> {
+	fn new(&self) -> Self {
+		Self {
+			next_id: Default::default(),
+			active_subscriptions: Default::default(),
+			executor: self.executor,
+		}
+	}
+
+	fn executor(&self) -> &TaskExecutor {
+		&self.executor
+	}
+
+	fn add<T, E, G, R, F, N>(&self, subscriber: Subscriber<T, E>, into_future: G) -> SubscriptionId<N>
+	where
+		G: FnOnce(Sink<T, E>) -> R,
+		R: future::IntoFuture<Future = F, Item = (), Error = ()>,
+		F: future::Future<Item = (), Error = ()> + Send + 'static,
+	{
+		let id = self.next_id.next_id();
+		let subscription_id: SubscriptionId = id.into();
+		if let Ok(sink) = subscriber.assign_id(subscription_id.clone()) {
+			let (tx, rx) = oneshot::channel();
+			let future = into_future(sink)
+				.into_future()
+				.select(rx.map_err(|e| warn!("Error timing out: {:?}", e)))
+				.then(|_| Ok(()));
+
+			self.active_subscriptions.lock().insert(id, tx);
+			if self.executor.execute(Box::new(future)).is_err() {
+				error!("Failed to spawn RPC subscription task");
+			}
+		}
+
+		subscription_id
+	}
+
+	/// Cancel subscription.
+	///
+	/// Returns true if subscription existed or false otherwise.
+	fn cancel<N>(&self, id: SubscriptionId<N>) -> bool {
+		if let SubscriptionId::Number(id) = id {
+			if let Some(tx) = self.active_subscriptions.lock().remove(&id) {
+				let _ = tx.send(());
+				return true;
+			}
+		}
+		false
+	}
+}
diff --git a/pubsub/src/types.rs b/pubsub/src/types.rs
index be5fefc49..a00c12df6 100644
--- a/pubsub/src/types.rs
+++ b/pubsub/src/types.rs
@@ -3,6 +3,7 @@ use crate::core::futures::sync::mpsc;
 use std::sync::Arc;
 
 use crate::subscription::Session;
+use crate::manager::IdProvider;
 
 /// Raw transport sink for specific client.
 pub type TransportSender = mpsc::Sender<String>;
@@ -35,40 +36,41 @@ impl<T: PubSubMetadata> PubSubMetadata for Option<T> {
 }
 
 /// Unique subscription id.
+///
 /// NOTE Assigning same id to different requests will cause the previous request to be unsubscribed.
 #[derive(Debug, Clone, PartialEq, Eq, Hash)]
-pub enum SubscriptionId {
-	/// U64 number
-	Number(u64),
+pub enum SubscriptionId<N> {
+	/// Number
+	Number(N),
 	/// String
 	String(String),
 }
 
-impl SubscriptionId {
+impl<N> SubscriptionId<N> {
 	/// Parses `core::Value` into unique subscription id.
-	pub fn parse_value(val: &core::Value) -> Option<SubscriptionId> {
+	pub fn parse_value(val: &core::Value) -> Option<SubscriptionId<N>> {
 		match *val {
 			core::Value::String(ref val) => Some(SubscriptionId::String(val.clone())),
-			core::Value::Number(ref val) => val.as_u64().map(SubscriptionId::Number),
+			core::Value::Number(ref val) => val.as_u64().map(SubscriptionId::Number), // TODO: Fix
 			_ => None,
 		}
 	}
 }
 
-impl From<String> for SubscriptionId {
+impl<N> From<String> for SubscriptionId<N> {
 	fn from(other: String) -> Self {
 		SubscriptionId::String(other)
 	}
 }
 
-impl From<u64> for SubscriptionId {
-	fn from(other: u64) -> Self {
+impl<N, I: IdProvider> From<I> for SubscriptionId<N> {
+	fn from(other: I::Id) -> Self {
 		SubscriptionId::Number(other)
 	}
 }
 
-impl From<SubscriptionId> for core::Value {
-	fn from(sub: SubscriptionId) -> Self {
+impl<N> From<SubscriptionId<N>> for core::Value {
+	fn from(sub: SubscriptionId<N>) -> Self {
 		match sub {
 			SubscriptionId::Number(val) => core::Value::Number(val.into()),
 			SubscriptionId::String(val) => core::Value::String(val),
@@ -76,6 +78,26 @@ impl From<SubscriptionId> for core::Value {
 	}
 }
 
+macro_rules! impl_from_num {
+	($num:ty) => {
+		impl<N> From<$num> for SubscriptionId<N> {
+			fn from(other: $num) -> Self {
+				SubscriptionId::Number(other)
+			}
+		}
+	}
+}
+
+impl_from_num!(u8);
+impl_from_num!(u16);
+impl_from_num!(u32);
+impl_from_num!(u64);
+
+impl_from_num!(i8);
+impl_from_num!(i16);
+impl_from_num!(i32);
+impl_from_num!(i64);
+
 #[cfg(test)]
 mod tests {
 	use super::SubscriptionId;

From 988372068bdef4a579262e7335eda0fb66ed12ed Mon Sep 17 00:00:00 2001
From: Hernando Castano <castano.ha@gmail.com>
Date: Wed, 20 May 2020 20:57:16 -0400
Subject: [PATCH 02/16] Allow IdProvider::Id and SubscriptionId to work
 together

Adds trait bounds that allow conversion between the two, removing
the need for generics in SubscriptionId.
---
 pubsub/src/manager.rs | 55 +++++++++++--------------------------------
 pubsub/src/types.rs   | 32 ++++++++-----------------
 2 files changed, 24 insertions(+), 63 deletions(-)

diff --git a/pubsub/src/manager.rs b/pubsub/src/manager.rs
index 3a1608fb6..1ac78627e 100644
--- a/pubsub/src/manager.rs
+++ b/pubsub/src/manager.rs
@@ -2,10 +2,7 @@
 
 use std::collections::HashMap;
 use std::hash::Hash;
-use std::sync::{
-	atomic::{self, AtomicUsize},
-	Arc,
-};
+use std::sync::Arc;
 
 use crate::core::futures::sync::oneshot;
 use crate::core::futures::{future, Future};
@@ -21,54 +18,31 @@ pub type TaskExecutor = Arc<dyn future::Executor<Box<dyn Future<Item = (), Error
 
 /// Trait used to provide unique subscription ids.
 pub trait IdProvider {
-	// TODO: Maybe have this impl Into<u64>?
-	type Id: Clone + Default + Eq + Hash;
+	/// A unique ID used to identify a subscription.
+	type Id: Copy + Clone + Default + Eq + Hash + Into<SubscriptionId> + From<SubscriptionId>;
 
 	/// Returns next id for the subscription.
 	fn next_id(&self) -> Self::Id;
 }
 
-/// Trait used to drive subscription Futures to completion.
-pub trait SubscriptionManager {
-	/// Create a new `SubscriptionManager`.
-	fn new(&self) -> Self;
-	/// Borrows the internal task executor.
-	///
-	/// This can be used to spawn additional tasks on the underlying event loop.
-	fn executor(&self) -> &TaskExecutor;
-	/// Create new subscription for given subscriber.
-	///
-	/// Second parameter is a function that converts Subscriber sink into a future.
-	/// This future will be driven to completion by the underlying event loop
-	/// or will be cancelled in case #cancel is invoked.
-	fn add<T, E, G, R, F, N>(&self, subscriber: Subscriber<T, E>, into_future: G) -> SubscriptionId<N>
-	where
-		G: FnOnce(Sink<T, E>) -> R,
-		R: future::IntoFuture<Future = F, Item = (), Error = ()>,
-		F: future::Future<Item = (), Error = ()> + Send + 'static;
-	/// Cancel subscription.
-	///
-	/// Should true if subscription existed or false otherwise.
-	fn cancel<N>(&self, id: SubscriptionId<N>) -> bool;
-}
-
 /// Subscriptions manager.
 ///
 /// Takes care of assigning unique subscription ids and
 /// driving the sinks into completion.
 #[derive(Clone)]
-pub struct Manager<I: Default + IdProvider> {
+pub struct SubscriptionManager<I: Default + IdProvider> {
 	next_id: I,
 	active_subscriptions: Arc<Mutex<HashMap<I::Id, oneshot::Sender<()>>>>,
 	executor: TaskExecutor, // Make generic?
 }
 
-impl<I: Default + IdProvider> SubscriptionManager for Manager<I> {
-	fn new(&self) -> Self {
+impl<I: Default + IdProvider> SubscriptionManager<I> {
+	/// Creates a new SubscriptionManager.
+	pub fn new(executor: TaskExecutor) -> Self {
 		Self {
 			next_id: Default::default(),
 			active_subscriptions: Default::default(),
-			executor: self.executor,
+			executor,
 		}
 	}
 
@@ -76,7 +50,7 @@ impl<I: Default + IdProvider> SubscriptionManager for Manager<I> {
 		&self.executor
 	}
 
-	fn add<T, E, G, R, F, N>(&self, subscriber: Subscriber<T, E>, into_future: G) -> SubscriptionId<N>
+	fn add<T, E, G, R, F>(&self, subscriber: Subscriber<T, E>, into_future: G) -> SubscriptionId
 	where
 		G: FnOnce(Sink<T, E>) -> R,
 		R: future::IntoFuture<Future = F, Item = (), Error = ()>,
@@ -103,13 +77,12 @@ impl<I: Default + IdProvider> SubscriptionManager for Manager<I> {
 	/// Cancel subscription.
 	///
 	/// Returns true if subscription existed or false otherwise.
-	fn cancel<N>(&self, id: SubscriptionId<N>) -> bool {
-		if let SubscriptionId::Number(id) = id {
-			if let Some(tx) = self.active_subscriptions.lock().remove(&id) {
-				let _ = tx.send(());
-				return true;
-			}
+	fn cancel(&self, id: SubscriptionId) -> bool {
+		if let Some(tx) = self.active_subscriptions.lock().remove(&id.into()) {
+			let _ = tx.send(());
+			return true;
 		}
+
 		false
 	}
 }
diff --git a/pubsub/src/types.rs b/pubsub/src/types.rs
index a00c12df6..0339e1d95 100644
--- a/pubsub/src/types.rs
+++ b/pubsub/src/types.rs
@@ -3,7 +3,6 @@ use crate::core::futures::sync::mpsc;
 use std::sync::Arc;
 
 use crate::subscription::Session;
-use crate::manager::IdProvider;
 
 /// Raw transport sink for specific client.
 pub type TransportSender = mpsc::Sender<String>;
@@ -39,38 +38,32 @@ impl<T: PubSubMetadata> PubSubMetadata for Option<T> {
 ///
 /// NOTE Assigning same id to different requests will cause the previous request to be unsubscribed.
 #[derive(Debug, Clone, PartialEq, Eq, Hash)]
-pub enum SubscriptionId<N> {
+pub enum SubscriptionId {
 	/// Number
-	Number(N),
+	Number(u64),
 	/// String
 	String(String),
 }
 
-impl<N> SubscriptionId<N> {
+impl SubscriptionId {
 	/// Parses `core::Value` into unique subscription id.
-	pub fn parse_value(val: &core::Value) -> Option<SubscriptionId<N>> {
+	pub fn parse_value(val: &core::Value) -> Option<SubscriptionId> {
 		match *val {
 			core::Value::String(ref val) => Some(SubscriptionId::String(val.clone())),
-			core::Value::Number(ref val) => val.as_u64().map(SubscriptionId::Number), // TODO: Fix
+			core::Value::Number(ref val) => val.as_u64().map(SubscriptionId::Number),
 			_ => None,
 		}
 	}
 }
 
-impl<N> From<String> for SubscriptionId<N> {
+impl From<String> for SubscriptionId {
 	fn from(other: String) -> Self {
 		SubscriptionId::String(other)
 	}
 }
 
-impl<N, I: IdProvider> From<I> for SubscriptionId<N> {
-	fn from(other: I::Id) -> Self {
-		SubscriptionId::Number(other)
-	}
-}
-
-impl<N> From<SubscriptionId<N>> for core::Value {
-	fn from(sub: SubscriptionId<N>) -> Self {
+impl From<SubscriptionId> for core::Value {
+	fn from(sub: SubscriptionId) -> Self {
 		match sub {
 			SubscriptionId::Number(val) => core::Value::Number(val.into()),
 			SubscriptionId::String(val) => core::Value::String(val),
@@ -80,9 +73,9 @@ impl<N> From<SubscriptionId<N>> for core::Value {
 
 macro_rules! impl_from_num {
 	($num:ty) => {
-		impl<N> From<$num> for SubscriptionId<N> {
+		impl From<$num> for SubscriptionId {
 			fn from(other: $num) -> Self {
-				SubscriptionId::Number(other)
+				SubscriptionId::Number(other.into())
 			}
 		}
 	}
@@ -93,11 +86,6 @@ impl_from_num!(u16);
 impl_from_num!(u32);
 impl_from_num!(u64);
 
-impl_from_num!(i8);
-impl_from_num!(i16);
-impl_from_num!(i32);
-impl_from_num!(i64);
-
 #[cfg(test)]
 mod tests {
 	use super::SubscriptionId;

From a41638e0f6b17155cb22f22cb9cbb2078ea03db4 Mon Sep 17 00:00:00 2001
From: Hernando Castano <castano.ha@gmail.com>
Date: Wed, 20 May 2020 21:21:20 -0400
Subject: [PATCH 03/16] Update SubscriptionId tests

---
 pubsub/src/types.rs | 80 +++++++++++++++++++++++++++++++++------------
 1 file changed, 59 insertions(+), 21 deletions(-)

diff --git a/pubsub/src/types.rs b/pubsub/src/types.rs
index 0339e1d95..68fd27a6f 100644
--- a/pubsub/src/types.rs
+++ b/pubsub/src/types.rs
@@ -39,9 +39,9 @@ impl<T: PubSubMetadata> PubSubMetadata for Option<T> {
 /// NOTE Assigning same id to different requests will cause the previous request to be unsubscribed.
 #[derive(Debug, Clone, PartialEq, Eq, Hash)]
 pub enum SubscriptionId {
-	/// Number
+	/// A numerical ID, represented by a `u64`.
 	Number(u64),
-	/// String
+	/// A non-numerical ID, for example a hash.
 	String(String),
 }
 
@@ -92,24 +92,62 @@ mod tests {
 	use crate::core::Value;
 
 	#[test]
-	fn should_convert_between_value_and_subscription_id() {
-		// given
-		let val1 = Value::Number(5.into());
-		let val2 = Value::String("asdf".into());
-		let val3 = Value::Null;
-
-		// when
-		let res1 = SubscriptionId::parse_value(&val1);
-		let res2 = SubscriptionId::parse_value(&val2);
-		let res3 = SubscriptionId::parse_value(&val3);
-
-		// then
-		assert_eq!(res1, Some(SubscriptionId::Number(5)));
-		assert_eq!(res2, Some(SubscriptionId::String("asdf".into())));
-		assert_eq!(res3, None);
-
-		// and back
-		assert_eq!(Value::from(res1.unwrap()), val1);
-		assert_eq!(Value::from(res2.unwrap()), val2);
+	fn should_convert_between_number_value_and_subscription_id() {
+		let val = Value::Number(5.into());
+		let res = SubscriptionId::parse_value(&val);
+
+		assert_eq!(res, Some(SubscriptionId::Number(5)));
+		assert_eq!(Value::from(res.unwrap()), val);
+	}
+
+	#[test]
+	fn should_convert_between_string_value_and_subscription_id() {
+		let val = Value::String("asdf".into());
+		let res = SubscriptionId::parse_value(&val);
+
+		assert_eq!(res, Some(SubscriptionId::String("asdf".into())));
+		assert_eq!(Value::from(res.unwrap()), val);
+	}
+
+	#[test]
+	fn should_convert_between_null_value_and_subscription_id() {
+		let val = Value::Null;
+		let res = SubscriptionId::parse_value(&val);
+		assert_eq!(res, None);
+	}
+
+	#[test]
+	fn should_convert_from_u8_to_subscription_id() {
+		let val = 5u8;
+		let res: SubscriptionId = val.into();
+		assert_eq!(res, SubscriptionId::Number(5));
+	}
+
+	#[test]
+	fn should_convert_from_u16_to_subscription_id() {
+		let val = 5u16;
+		let res: SubscriptionId = val.into();
+		assert_eq!(res, SubscriptionId::Number(5));
+	}
+
+	#[test]
+	fn should_convert_from_u32_to_subscription_id() {
+		let val = 5u32;
+		let res: SubscriptionId = val.into();
+		assert_eq!(res, SubscriptionId::Number(5));
+	}
+
+	#[test]
+	fn should_convert_from_u64_to_subscription_id() {
+		let val = 5u64;
+		let res: SubscriptionId = val.into();
+		assert_eq!(res, SubscriptionId::Number(5));
+	}
+
+	#[test]
+	fn should_convert_from_string_to_subscription_id() {
+		let val = "String".to_string();
+		let res: SubscriptionId = val.into();
+		assert_eq!(res, SubscriptionId::String("String".to_string()));
 	}
 }

From 05b1a92a9f2cfab9c516eecf3cb68afb5d559ef5 Mon Sep 17 00:00:00 2001
From: Hernando Castano <castano.ha@gmail.com>
Date: Wed, 20 May 2020 21:22:23 -0400
Subject: [PATCH 04/16] Rustfmt

---
 pubsub/src/types.rs | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/pubsub/src/types.rs b/pubsub/src/types.rs
index 68fd27a6f..9e1437e74 100644
--- a/pubsub/src/types.rs
+++ b/pubsub/src/types.rs
@@ -78,7 +78,7 @@ macro_rules! impl_from_num {
 				SubscriptionId::Number(other.into())
 			}
 		}
-	}
+	};
 }
 
 impl_from_num!(u8);

From 0149fd4c44f619bf31e1194a27877ba86f9bf4dd Mon Sep 17 00:00:00 2001
From: Hernando Castano <castano.ha@gmail.com>
Date: Thu, 21 May 2020 16:51:41 -0400
Subject: [PATCH 05/16] Use `SubscriptionId` as the key for
 `active_subscriptions`

---
 pubsub/src/manager.rs | 9 ++++-----
 1 file changed, 4 insertions(+), 5 deletions(-)

diff --git a/pubsub/src/manager.rs b/pubsub/src/manager.rs
index 1ac78627e..a6733a805 100644
--- a/pubsub/src/manager.rs
+++ b/pubsub/src/manager.rs
@@ -1,7 +1,6 @@
 //! Provides an executor for subscription Futures.
 
 use std::collections::HashMap;
-use std::hash::Hash;
 use std::sync::Arc;
 
 use crate::core::futures::sync::oneshot;
@@ -19,7 +18,7 @@ pub type TaskExecutor = Arc<dyn future::Executor<Box<dyn Future<Item = (), Error
 /// Trait used to provide unique subscription ids.
 pub trait IdProvider {
 	/// A unique ID used to identify a subscription.
-	type Id: Copy + Clone + Default + Eq + Hash + Into<SubscriptionId> + From<SubscriptionId>;
+	type Id: Copy + Clone + Default + Into<SubscriptionId>;
 
 	/// Returns next id for the subscription.
 	fn next_id(&self) -> Self::Id;
@@ -32,7 +31,7 @@ pub trait IdProvider {
 #[derive(Clone)]
 pub struct SubscriptionManager<I: Default + IdProvider> {
 	next_id: I,
-	active_subscriptions: Arc<Mutex<HashMap<I::Id, oneshot::Sender<()>>>>,
+	active_subscriptions: Arc<Mutex<HashMap<SubscriptionId, oneshot::Sender<()>>>>,
 	executor: TaskExecutor, // Make generic?
 }
 
@@ -65,7 +64,7 @@ impl<I: Default + IdProvider> SubscriptionManager<I> {
 				.select(rx.map_err(|e| warn!("Error timing out: {:?}", e)))
 				.then(|_| Ok(()));
 
-			self.active_subscriptions.lock().insert(id, tx);
+			self.active_subscriptions.lock().insert(subscription_id.clone(), tx);
 			if self.executor.execute(Box::new(future)).is_err() {
 				error!("Failed to spawn RPC subscription task");
 			}
@@ -78,7 +77,7 @@ impl<I: Default + IdProvider> SubscriptionManager<I> {
 	///
 	/// Returns true if subscription existed or false otherwise.
 	fn cancel(&self, id: SubscriptionId) -> bool {
-		if let Some(tx) = self.active_subscriptions.lock().remove(&id.into()) {
+		if let Some(tx) = self.active_subscriptions.lock().remove(&id) {
 			let _ = tx.send(());
 			return true;
 		}

From f0fabd98e0620cddfa57cc12ae0f24dbd057bbad Mon Sep 17 00:00:00 2001
From: Hernando Castano <castano.ha@gmail.com>
Date: Thu, 21 May 2020 16:52:56 -0400
Subject: [PATCH 06/16] Add subscription ID providers.

Adds two subscription ID providers which can be used
by the SubscriptionManager. One provides a simple numeric
ID, while the other provides a random string.
---
 pubsub/Cargo.toml     |   1 +
 pubsub/src/manager.rs | 117 ++++++++++++++++++++++++++++++++++++++----
 pubsub/src/types.rs   |   8 +++
 3 files changed, 115 insertions(+), 11 deletions(-)

diff --git a/pubsub/Cargo.toml b/pubsub/Cargo.toml
index 852371f22..96f73dce0 100644
--- a/pubsub/Cargo.toml
+++ b/pubsub/Cargo.toml
@@ -15,6 +15,7 @@ log = "0.4"
 parking_lot = "0.10.0"
 jsonrpc-core = { version = "14.1", path = "../core" }
 serde = "1.0"
+rand = "0.7"
 
 [dev-dependencies]
 jsonrpc-tcp-server = { version = "14.1", path = "../tcp" }
diff --git a/pubsub/src/manager.rs b/pubsub/src/manager.rs
index a6733a805..d1ba8f55f 100644
--- a/pubsub/src/manager.rs
+++ b/pubsub/src/manager.rs
@@ -1,7 +1,11 @@
 //! Provides an executor for subscription Futures.
 
 use std::collections::HashMap;
-use std::sync::Arc;
+use std::iter;
+use std::sync::{
+	atomic::{AtomicUsize, Ordering},
+	Arc,
+};
 
 use crate::core::futures::sync::oneshot;
 use crate::core::futures::{future, Future};
@@ -9,37 +13,117 @@ use crate::{
 	typed::{Sink, Subscriber},
 	SubscriptionId,
 };
+
 use log::{error, warn};
 use parking_lot::Mutex;
+use rand::distributions::Alphanumeric;
+use rand::{thread_rng, Rng};
 
 /// Alias for an implementation of `futures::future::Executor`.
 pub type TaskExecutor = Arc<dyn future::Executor<Box<dyn Future<Item = (), Error = ()> + Send>> + Send + Sync>;
 
-/// Trait used to provide unique subscription ids.
+type ActiveSubscriptions = Arc<Mutex<HashMap<SubscriptionId, oneshot::Sender<()>>>>;
+
+/// Trait used to provide unique subscription IDs.
 pub trait IdProvider {
 	/// A unique ID used to identify a subscription.
-	type Id: Copy + Clone + Default + Into<SubscriptionId>;
+	type Id: Default + Into<SubscriptionId>;
 
-	/// Returns next id for the subscription.
+	/// Returns the next ID for the subscription.
 	fn next_id(&self) -> Self::Id;
 }
 
+/// Provides a thread-safe incrementing integer which
+/// can be used as a subscription ID.
+#[derive(Debug)]
+pub struct NumericIdProvider {
+	current_id: AtomicUsize,
+}
+
+impl NumericIdProvider {
+	/// Create a new NumericIdProvider.
+	pub fn new() -> Self {
+		Default::default()
+	}
+
+	/// Create a new NumericIdProvider starting from
+	/// the given ID.
+	pub fn with_id(id: AtomicUsize) -> Self {
+		Self { current_id: id }
+	}
+}
+
+impl IdProvider for NumericIdProvider {
+	type Id = usize;
+
+	fn next_id(&self) -> Self::Id {
+		self.current_id.fetch_add(1, Ordering::AcqRel)
+	}
+}
+
+impl Default for NumericIdProvider {
+	fn default() -> Self {
+		NumericIdProvider {
+			current_id: AtomicUsize::new(1),
+		}
+	}
+}
+
+/// Used to generate random strings for use as
+/// subscription IDs.
+#[derive(Copy, Clone, Eq, PartialEq, Hash, Debug)]
+pub struct RandomStringIdProvider {
+	len: usize,
+}
+
+impl RandomStringIdProvider {
+	/// Create a new RandomStringIdProvider.
+	pub fn new() -> Self {
+		Default::default()
+	}
+
+	/// Create a new RandomStringIdProvider, which will generate
+	/// random id strings of the given length.
+	pub fn with_len(len: usize) -> Self {
+		Self { len }
+	}
+}
+
+impl IdProvider for RandomStringIdProvider {
+	type Id = String;
+
+	fn next_id(&self) -> Self::Id {
+		let mut rng = thread_rng();
+		let id: String = iter::repeat(())
+			.map(|()| rng.sample(Alphanumeric))
+			.take(self.len)
+			.collect();
+		id
+	}
+}
+
+impl Default for RandomStringIdProvider {
+	fn default() -> Self {
+		Self { len: 16 }
+	}
+}
+
 /// Subscriptions manager.
 ///
 /// Takes care of assigning unique subscription ids and
 /// driving the sinks into completion.
 #[derive(Clone)]
-pub struct SubscriptionManager<I: Default + IdProvider> {
-	next_id: I,
-	active_subscriptions: Arc<Mutex<HashMap<SubscriptionId, oneshot::Sender<()>>>>,
+pub struct SubscriptionManager<I: IdProvider = RandomStringIdProvider> {
+	id_provider: I,
+	active_subscriptions: ActiveSubscriptions,
 	executor: TaskExecutor, // Make generic?
 }
 
-impl<I: Default + IdProvider> SubscriptionManager<I> {
+impl<I: IdProvider> SubscriptionManager<I> {
 	/// Creates a new SubscriptionManager.
-	pub fn new(executor: TaskExecutor) -> Self {
+	pub fn new_with_id_provider(id_provider: I, executor: TaskExecutor) -> Self {
 		Self {
-			next_id: Default::default(),
+			id_provider,
 			active_subscriptions: Default::default(),
 			executor,
 		}
@@ -55,7 +139,7 @@ impl<I: Default + IdProvider> SubscriptionManager<I> {
 		R: future::IntoFuture<Future = F, Item = (), Error = ()>,
 		F: future::Future<Item = (), Error = ()> + Send + 'static,
 	{
-		let id = self.next_id.next_id();
+		let id = self.id_provider.next_id();
 		let subscription_id: SubscriptionId = id.into();
 		if let Ok(sink) = subscriber.assign_id(subscription_id.clone()) {
 			let (tx, rx) = oneshot::channel();
@@ -85,3 +169,14 @@ impl<I: Default + IdProvider> SubscriptionManager<I> {
 		false
 	}
 }
+
+impl<I: Default + IdProvider> SubscriptionManager<I> {
+	/// Creates a new SubscriptionManager.
+	pub fn new(executor: TaskExecutor) -> Self {
+		Self {
+			id_provider: Default::default(),
+			active_subscriptions: Default::default(),
+			executor,
+		}
+	}
+}
diff --git a/pubsub/src/types.rs b/pubsub/src/types.rs
index 9e1437e74..e2e38fd4e 100644
--- a/pubsub/src/types.rs
+++ b/pubsub/src/types.rs
@@ -62,6 +62,14 @@ impl From<String> for SubscriptionId {
 	}
 }
 
+// TODO: Don't unwrap, and probably implement TryFrom instead of From
+use std::convert::TryInto;
+impl From<usize> for SubscriptionId {
+	fn from(id: usize) -> Self {
+		SubscriptionId::Number(id.try_into().unwrap())
+	}
+}
+
 impl From<SubscriptionId> for core::Value {
 	fn from(sub: SubscriptionId) -> Self {
 		match sub {

From c89511dbd95db4cd2a8cdd4fd6ef7ef5bf2bc837 Mon Sep 17 00:00:00 2001
From: Hernando Castano <castano.ha@gmail.com>
Date: Thu, 21 May 2020 17:30:37 -0400
Subject: [PATCH 07/16] Add some documentation

---
 pubsub/src/manager.rs | 29 +++++++++++++++++++++++++----
 1 file changed, 25 insertions(+), 4 deletions(-)

diff --git a/pubsub/src/manager.rs b/pubsub/src/manager.rs
index d1ba8f55f..f57411867 100644
--- a/pubsub/src/manager.rs
+++ b/pubsub/src/manager.rs
@@ -1,4 +1,18 @@
-//! Provides an executor for subscription Futures.
+//! The SubscriptionManager used to manage subscription based RPCs.
+//!
+//! The manager provides four main things in terms of functionality:
+//!
+//! 1. The ability to create unique subscription IDs through the
+//! use of the `IdProvider` trait. Two implementations are availble
+//! out of the box, a `NumericIdProvider` and a `RandomStringIdProvider`.
+//!
+//! 2. An executor with which to drive `Future`s to completion.
+//!
+//! 3. A way to add new subscriptions. Subscriptions should come in the form
+//! of a `Stream`. These subscriptions will be transformed into notifications
+//! by the manager, which can be consumed by the client.
+//!
+//! 4. A way to cancel any currently active subscription.
 
 use std::collections::HashMap;
 use std::iter;
@@ -129,11 +143,18 @@ impl<I: IdProvider> SubscriptionManager<I> {
 		}
 	}
 
-	fn executor(&self) -> &TaskExecutor {
+	/// Borrows the internal task executor.
+	///
+	/// This can be used to spawn additional tasks on the underlying event loop.
+	pub fn executor(&self) -> &TaskExecutor {
 		&self.executor
 	}
 
-	fn add<T, E, G, R, F>(&self, subscriber: Subscriber<T, E>, into_future: G) -> SubscriptionId
+	/// Creates new subscription for given subscriber.
+	///
+	/// Second parameter is a function that converts Subscriber Sink into a Future.
+	/// This future will be driven to completion by the underlying event loop
+	pub fn add<T, E, G, R, F>(&self, subscriber: Subscriber<T, E>, into_future: G) -> SubscriptionId
 	where
 		G: FnOnce(Sink<T, E>) -> R,
 		R: future::IntoFuture<Future = F, Item = (), Error = ()>,
@@ -160,7 +181,7 @@ impl<I: IdProvider> SubscriptionManager<I> {
 	/// Cancel subscription.
 	///
 	/// Returns true if subscription existed or false otherwise.
-	fn cancel(&self, id: SubscriptionId) -> bool {
+	pub fn cancel(&self, id: SubscriptionId) -> bool {
 		if let Some(tx) = self.active_subscriptions.lock().remove(&id) {
 			let _ = tx.send(());
 			return true;

From cd31378481e73dfcabcf94a1ffc9121f23bace97 Mon Sep 17 00:00:00 2001
From: Hernando Castano <castano.ha@gmail.com>
Date: Fri, 22 May 2020 10:53:04 -0400
Subject: [PATCH 08/16] Clean up comment and naming

---
 pubsub/src/manager.rs | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/pubsub/src/manager.rs b/pubsub/src/manager.rs
index f57411867..43218db89 100644
--- a/pubsub/src/manager.rs
+++ b/pubsub/src/manager.rs
@@ -130,12 +130,12 @@ impl Default for RandomStringIdProvider {
 pub struct SubscriptionManager<I: IdProvider = RandomStringIdProvider> {
 	id_provider: I,
 	active_subscriptions: ActiveSubscriptions,
-	executor: TaskExecutor, // Make generic?
+	executor: TaskExecutor,
 }
 
 impl<I: IdProvider> SubscriptionManager<I> {
 	/// Creates a new SubscriptionManager.
-	pub fn new_with_id_provider(id_provider: I, executor: TaskExecutor) -> Self {
+	pub fn with_id_provider(id_provider: I, executor: TaskExecutor) -> Self {
 		Self {
 			id_provider,
 			active_subscriptions: Default::default(),

From 3cf311306937742a25744e0e94759956f981355d Mon Sep 17 00:00:00 2001
From: Hernando Castano <castano.ha@gmail.com>
Date: Fri, 22 May 2020 11:05:17 -0400
Subject: [PATCH 09/16] Change the NumericIdProvider to use `u64` IDs

Instead of providing a guarantee that we can convert between
`usize` and `u64` we make the assumptions that it's unlikely
that we're running on an architecture larger than 64-bits and
we use a `u64` directly.
---
 pubsub/src/manager.rs | 4 ++--
 pubsub/src/types.rs   | 8 --------
 2 files changed, 2 insertions(+), 10 deletions(-)

diff --git a/pubsub/src/manager.rs b/pubsub/src/manager.rs
index 43218db89..9a0216832 100644
--- a/pubsub/src/manager.rs
+++ b/pubsub/src/manager.rs
@@ -68,10 +68,10 @@ impl NumericIdProvider {
 }
 
 impl IdProvider for NumericIdProvider {
-	type Id = usize;
+	type Id = u64;
 
 	fn next_id(&self) -> Self::Id {
-		self.current_id.fetch_add(1, Ordering::AcqRel)
+		self.current_id.fetch_add(1, Ordering::AcqRel) as u64
 	}
 }
 
diff --git a/pubsub/src/types.rs b/pubsub/src/types.rs
index e2e38fd4e..9e1437e74 100644
--- a/pubsub/src/types.rs
+++ b/pubsub/src/types.rs
@@ -62,14 +62,6 @@ impl From<String> for SubscriptionId {
 	}
 }
 
-// TODO: Don't unwrap, and probably implement TryFrom instead of From
-use std::convert::TryInto;
-impl From<usize> for SubscriptionId {
-	fn from(id: usize) -> Self {
-		SubscriptionId::Number(id.try_into().unwrap())
-	}
-}
-
 impl From<SubscriptionId> for core::Value {
 	fn from(sub: SubscriptionId) -> Self {
 		match sub {

From 7e435a0d9ccf2e282ead29c3c5871e5d6dcf1c19 Mon Sep 17 00:00:00 2001
From: Hernando Castano <castano.ha@gmail.com>
Date: Mon, 25 May 2020 18:40:08 -0400
Subject: [PATCH 10/16] Add tests for IdProvider and SubscriptionManager

Note: There's one test that doesn't pass yet which has to do
with the `cancel()` function of the SubscriptionManager.
---
 pubsub/Cargo.toml     |   1 +
 pubsub/src/manager.rs | 143 ++++++++++++++++++++++++++++++++++++++++--
 2 files changed, 139 insertions(+), 5 deletions(-)

diff --git a/pubsub/Cargo.toml b/pubsub/Cargo.toml
index 96f73dce0..79ecda57a 100644
--- a/pubsub/Cargo.toml
+++ b/pubsub/Cargo.toml
@@ -19,6 +19,7 @@ rand = "0.7"
 
 [dev-dependencies]
 jsonrpc-tcp-server = { version = "14.1", path = "../tcp" }
+futures = { version = "0.3", features = ["compat", "thread-pool"] }
 
 [badges]
 travis-ci = { repository = "paritytech/jsonrpc", branch = "master"}
diff --git a/pubsub/src/manager.rs b/pubsub/src/manager.rs
index 9a0216832..1b00669a4 100644
--- a/pubsub/src/manager.rs
+++ b/pubsub/src/manager.rs
@@ -22,7 +22,7 @@ use std::sync::{
 };
 
 use crate::core::futures::sync::oneshot;
-use crate::core::futures::{future, Future};
+use crate::core::futures::{future as future01, Future as Future01};
 use crate::{
 	typed::{Sink, Subscriber},
 	SubscriptionId,
@@ -34,7 +34,7 @@ use rand::distributions::Alphanumeric;
 use rand::{thread_rng, Rng};
 
 /// Alias for an implementation of `futures::future::Executor`.
-pub type TaskExecutor = Arc<dyn future::Executor<Box<dyn Future<Item = (), Error = ()> + Send>> + Send + Sync>;
+pub type TaskExecutor = Arc<dyn future01::Executor<Box<dyn Future01<Item = (), Error = ()> + Send>> + Send + Sync>;
 
 type ActiveSubscriptions = Arc<Mutex<HashMap<SubscriptionId, oneshot::Sender<()>>>>;
 
@@ -127,7 +127,7 @@ impl Default for RandomStringIdProvider {
 /// Takes care of assigning unique subscription ids and
 /// driving the sinks into completion.
 #[derive(Clone)]
-pub struct SubscriptionManager<I: IdProvider = RandomStringIdProvider> {
+pub struct SubscriptionManager<I: IdProvider> {
 	id_provider: I,
 	active_subscriptions: ActiveSubscriptions,
 	executor: TaskExecutor,
@@ -157,8 +157,8 @@ impl<I: IdProvider> SubscriptionManager<I> {
 	pub fn add<T, E, G, R, F>(&self, subscriber: Subscriber<T, E>, into_future: G) -> SubscriptionId
 	where
 		G: FnOnce(Sink<T, E>) -> R,
-		R: future::IntoFuture<Future = F, Item = (), Error = ()>,
-		F: future::Future<Item = (), Error = ()> + Send + 'static,
+		R: future01::IntoFuture<Future = F, Item = (), Error = ()>,
+		F: future01::Future<Item = (), Error = ()> + Send + 'static,
 	{
 		let id = self.id_provider.next_id();
 		let subscription_id: SubscriptionId = id.into();
@@ -170,6 +170,7 @@ impl<I: IdProvider> SubscriptionManager<I> {
 				.then(|_| Ok(()));
 
 			self.active_subscriptions.lock().insert(subscription_id.clone(), tx);
+			dbg!(&self.active_subscriptions.lock().get(&subscription_id));
 			if self.executor.execute(Box::new(future)).is_err() {
 				error!("Failed to spawn RPC subscription task");
 			}
@@ -182,6 +183,7 @@ impl<I: IdProvider> SubscriptionManager<I> {
 	///
 	/// Returns true if subscription existed or false otherwise.
 	pub fn cancel(&self, id: SubscriptionId) -> bool {
+		dbg!(&self.active_subscriptions.lock().get(&id));
 		if let Some(tx) = self.active_subscriptions.lock().remove(&id) {
 			let _ = tx.send(());
 			return true;
@@ -201,3 +203,134 @@ impl<I: Default + IdProvider> SubscriptionManager<I> {
 		}
 	}
 }
+
+#[cfg(test)]
+mod tests {
+	use super::*;
+	use crate::typed::Subscriber;
+	use futures::{compat::Future01CompatExt, executor, FutureExt};
+	use futures::{stream, StreamExt, TryStreamExt};
+
+	use crate::core::futures::sink::Sink as Sink01;
+	use crate::core::futures::stream::Stream as Stream01;
+
+	pub struct TestTaskExecutor;
+	type Boxed01Future01 = Box<dyn future01::Future<Item = (), Error = ()> + Send + 'static>;
+
+	impl future01::Executor<Boxed01Future01> for TestTaskExecutor {
+		fn execute(&self, future: Boxed01Future01) -> std::result::Result<(), future01::ExecuteError<Boxed01Future01>> {
+			let executor = executor::ThreadPool::new().expect("Failed to create thread pool executor for tests");
+			executor.spawn_ok(future.compat().map(drop));
+			Ok(())
+		}
+	}
+
+	#[test]
+	fn making_a_numeric_id_provider_works() {
+		let provider = NumericIdProvider::new();
+		let expected_id = 1;
+		let actual_id = provider.next_id();
+
+		assert_eq!(actual_id, expected_id);
+	}
+
+	#[test]
+	fn default_numeric_id_provider_works() {
+		let provider: NumericIdProvider = Default::default();
+		let expected_id = 1;
+		let actual_id = provider.next_id();
+
+		assert_eq!(actual_id, expected_id);
+	}
+
+	#[test]
+	fn numeric_id_provider_with_id_works() {
+		let provider = NumericIdProvider::with_id(AtomicUsize::new(5));
+		let expected_id = 5;
+		let actual_id = provider.next_id();
+
+		assert_eq!(actual_id, expected_id);
+	}
+
+	#[test]
+	fn random_string_provider_returns_id_with_correct_default_len() {
+		let provider = RandomStringIdProvider::new();
+		let expected_len = 16;
+		let actual_len = provider.next_id().len();
+
+		assert_eq!(actual_len, expected_len);
+	}
+
+	#[test]
+	fn random_string_provider_returns_id_with_correct_user_given_len() {
+		let expected_len = 10;
+		let provider = RandomStringIdProvider::with_len(expected_len);
+		let actual_len = provider.next_id().len();
+
+		assert_eq!(actual_len, expected_len);
+	}
+
+	#[test]
+	fn new_subscription_manager_works_with_numeric_id_provider() {
+		let manager = SubscriptionManager::<NumericIdProvider>::new(Arc::new(TestTaskExecutor));
+		let subscriber = Subscriber::<u64>::new_test("test_subTest").0;
+		let stream = stream::iter(vec![Ok(1)]).compat();
+
+		let id = manager.add(subscriber, |sink| {
+			let stream = stream.map(|res| Ok(res));
+
+			sink.sink_map_err(|_| ()).send_all(stream).map(|_| ())
+		});
+
+		assert!(matches!(id, SubscriptionId::Number(_)))
+	}
+
+	#[test]
+	fn new_subscription_manager_works_with_random_string_provider() {
+		let id_provider = RandomStringIdProvider::default();
+		let manager = SubscriptionManager::with_id_provider(id_provider, Arc::new(TestTaskExecutor));
+		let subscriber = Subscriber::<u64>::new_test("test_subTest").0;
+		let stream = stream::iter(vec![Ok(1)]).compat();
+
+		let id = manager.add(subscriber, |sink| {
+			let stream = stream.map(|res| Ok(res));
+
+			sink.sink_map_err(|_| ()).send_all(stream).map(|_| ())
+		});
+
+		assert!(matches!(id, SubscriptionId::String(_)))
+	}
+
+	#[test]
+	fn subscription_is_canceled_if_it_existed() {
+		let id_provider = RandomStringIdProvider::default();
+		let manager = SubscriptionManager::with_id_provider(id_provider, Arc::new(TestTaskExecutor));
+		let subscriber = Subscriber::<u64>::new_test("test_subTest").0;
+
+		let (mut tx, rx) = futures::channel::mpsc::channel(8);
+		tx.start_send(1).unwrap();
+		let stream = rx.map(|v| Ok::<_, ()>(v)).compat();
+
+		let id = manager.add(subscriber, |sink| {
+			let stream = stream.map(|res| Ok(res));
+
+			sink.sink_map_err(|_| ()).send_all(stream).map(|_| ())
+		});
+
+		let is_cancelled = manager.cancel(id);
+		dbg!(is_cancelled);
+
+		assert!(is_cancelled);
+	}
+
+	#[test]
+	fn subscription_is_not_canceled_because_it_didnt_exist() {
+		let manager = SubscriptionManager::<NumericIdProvider>::new(Arc::new(TestTaskExecutor));
+
+		let id: SubscriptionId = 23u32.into();
+		let is_cancelled = manager.cancel(id);
+		let is_not_cancelled = !is_cancelled;
+
+		assert!(is_not_cancelled);
+	}
+}

From 1e5846fa4ee09e3f1594ea69ba63ad6b47dc8c11 Mon Sep 17 00:00:00 2001
From: Hernando Castano <castano.ha@gmail.com>
Date: Tue, 26 May 2020 19:04:48 -0400
Subject: [PATCH 11/16] Restore RandomStringIdProvider as the default provider

---
 pubsub/src/manager.rs | 45 ++++++++++++++++++++++++++++++++++++-------
 1 file changed, 38 insertions(+), 7 deletions(-)

diff --git a/pubsub/src/manager.rs b/pubsub/src/manager.rs
index 1b00669a4..b129a0e2b 100644
--- a/pubsub/src/manager.rs
+++ b/pubsub/src/manager.rs
@@ -127,14 +127,28 @@ impl Default for RandomStringIdProvider {
 /// Takes care of assigning unique subscription ids and
 /// driving the sinks into completion.
 #[derive(Clone)]
-pub struct SubscriptionManager<I: IdProvider> {
+pub struct SubscriptionManager<I: IdProvider=RandomStringIdProvider> {
 	id_provider: I,
 	active_subscriptions: ActiveSubscriptions,
 	executor: TaskExecutor,
 }
 
-impl<I: IdProvider> SubscriptionManager<I> {
+impl SubscriptionManager {
 	/// Creates a new SubscriptionManager.
+	///
+	/// Uses `RandomStringIdProvider` as the ID provider.
+	pub fn new(executor: TaskExecutor) -> Self {
+		Self {
+			id_provider: RandomStringIdProvider::default(),
+			active_subscriptions: Default::default(),
+			executor,
+		}
+	}
+}
+
+impl<I: IdProvider> SubscriptionManager<I> {
+	/// Creates a new SubscriptionManager with the specified
+	/// ID provider.
 	pub fn with_id_provider(id_provider: I, executor: TaskExecutor) -> Self {
 		Self {
 			id_provider,
@@ -195,7 +209,7 @@ impl<I: IdProvider> SubscriptionManager<I> {
 
 impl<I: Default + IdProvider> SubscriptionManager<I> {
 	/// Creates a new SubscriptionManager.
-	pub fn new(executor: TaskExecutor) -> Self {
+	pub fn with_executor(executor: TaskExecutor) -> Self {
 		Self {
 			id_provider: Default::default(),
 			active_subscriptions: Default::default(),
@@ -270,9 +284,26 @@ mod tests {
 		assert_eq!(actual_len, expected_len);
 	}
 
+	#[test]
+	fn new_subscription_manager_defaults_to_random_string_provider() {
+		let manager = SubscriptionManager::new(Arc::new(TestTaskExecutor));
+		let subscriber = Subscriber::<u64>::new_test("test_subTest").0;
+		let stream = stream::iter(vec![Ok(1)]).compat();
+
+		let id = manager.add(subscriber, |sink| {
+			let stream = stream.map(|res| Ok(res));
+
+			sink.sink_map_err(|_| ()).send_all(stream).map(|_| ())
+		});
+
+		assert!(matches!(id, SubscriptionId::String(_)))
+	}
+
 	#[test]
 	fn new_subscription_manager_works_with_numeric_id_provider() {
-		let manager = SubscriptionManager::<NumericIdProvider>::new(Arc::new(TestTaskExecutor));
+		let id_provider = NumericIdProvider::default();
+		let manager = SubscriptionManager::with_id_provider(id_provider, Arc::new(TestTaskExecutor));
+
 		let subscriber = Subscriber::<u64>::new_test("test_subTest").0;
 		let stream = stream::iter(vec![Ok(1)]).compat();
 
@@ -289,6 +320,7 @@ mod tests {
 	fn new_subscription_manager_works_with_random_string_provider() {
 		let id_provider = RandomStringIdProvider::default();
 		let manager = SubscriptionManager::with_id_provider(id_provider, Arc::new(TestTaskExecutor));
+
 		let subscriber = Subscriber::<u64>::new_test("test_subTest").0;
 		let stream = stream::iter(vec![Ok(1)]).compat();
 
@@ -303,8 +335,7 @@ mod tests {
 
 	#[test]
 	fn subscription_is_canceled_if_it_existed() {
-		let id_provider = RandomStringIdProvider::default();
-		let manager = SubscriptionManager::with_id_provider(id_provider, Arc::new(TestTaskExecutor));
+		let manager = SubscriptionManager::<NumericIdProvider>::with_executor(Arc::new(TestTaskExecutor));
 		let subscriber = Subscriber::<u64>::new_test("test_subTest").0;
 
 		let (mut tx, rx) = futures::channel::mpsc::channel(8);
@@ -325,7 +356,7 @@ mod tests {
 
 	#[test]
 	fn subscription_is_not_canceled_because_it_didnt_exist() {
-		let manager = SubscriptionManager::<NumericIdProvider>::new(Arc::new(TestTaskExecutor));
+		let manager = SubscriptionManager::new(Arc::new(TestTaskExecutor));
 
 		let id: SubscriptionId = 23u32.into();
 		let is_cancelled = manager.cancel(id);

From 13347601de056f757194166d29b1090001288e33 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Tomasz=20Drwi=C4=99ga?= <tomasz@parity.io>
Date: Wed, 27 May 2020 16:49:28 +0200
Subject: [PATCH 12/16] Retain receiver.:

---
 pubsub/src/manager.rs | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/pubsub/src/manager.rs b/pubsub/src/manager.rs
index b129a0e2b..e8278850e 100644
--- a/pubsub/src/manager.rs
+++ b/pubsub/src/manager.rs
@@ -336,7 +336,7 @@ mod tests {
 	#[test]
 	fn subscription_is_canceled_if_it_existed() {
 		let manager = SubscriptionManager::<NumericIdProvider>::with_executor(Arc::new(TestTaskExecutor));
-		let subscriber = Subscriber::<u64>::new_test("test_subTest").0;
+		let (subscriber, _recv, _) = Subscriber::<u64>::new_test("test_subTest");
 
 		let (mut tx, rx) = futures::channel::mpsc::channel(8);
 		tx.start_send(1).unwrap();

From 476edee35b0595c2f14887bfbfdc0cb196d5a2f8 Mon Sep 17 00:00:00 2001
From: Hernando Castano <castano.ha@gmail.com>
Date: Wed, 27 May 2020 11:38:13 -0400
Subject: [PATCH 13/16] Make test executor a lazy static

---
 pubsub/Cargo.toml     |  1 +
 pubsub/src/manager.rs | 16 ++++++++++------
 2 files changed, 11 insertions(+), 6 deletions(-)

diff --git a/pubsub/Cargo.toml b/pubsub/Cargo.toml
index 79ecda57a..f7ccbc046 100644
--- a/pubsub/Cargo.toml
+++ b/pubsub/Cargo.toml
@@ -20,6 +20,7 @@ rand = "0.7"
 [dev-dependencies]
 jsonrpc-tcp-server = { version = "14.1", path = "../tcp" }
 futures = { version = "0.3", features = ["compat", "thread-pool"] }
+lazy_static = "1.4"
 
 [badges]
 travis-ci = { repository = "paritytech/jsonrpc", branch = "master"}
diff --git a/pubsub/src/manager.rs b/pubsub/src/manager.rs
index e8278850e..aef3e9fe6 100644
--- a/pubsub/src/manager.rs
+++ b/pubsub/src/manager.rs
@@ -184,7 +184,6 @@ impl<I: IdProvider> SubscriptionManager<I> {
 				.then(|_| Ok(()));
 
 			self.active_subscriptions.lock().insert(subscription_id.clone(), tx);
-			dbg!(&self.active_subscriptions.lock().get(&subscription_id));
 			if self.executor.execute(Box::new(future)).is_err() {
 				error!("Failed to spawn RPC subscription task");
 			}
@@ -197,7 +196,6 @@ impl<I: IdProvider> SubscriptionManager<I> {
 	///
 	/// Returns true if subscription existed or false otherwise.
 	pub fn cancel(&self, id: SubscriptionId) -> bool {
-		dbg!(&self.active_subscriptions.lock().get(&id));
 		if let Some(tx) = self.active_subscriptions.lock().remove(&id) {
 			let _ = tx.send(());
 			return true;
@@ -228,13 +226,21 @@ mod tests {
 	use crate::core::futures::sink::Sink as Sink01;
 	use crate::core::futures::stream::Stream as Stream01;
 
+	// Executor shared by all tests.
+	//
+	// This shared executor is used to prevent `Too many open files` errors
+	// on systems with a lot of cores.
+	lazy_static::lazy_static! {
+		static ref EXECUTOR: executor::ThreadPool = executor::ThreadPool::new()
+			.expect("Failed to create thread pool executor for tests");
+	}
+
 	pub struct TestTaskExecutor;
 	type Boxed01Future01 = Box<dyn future01::Future<Item = (), Error = ()> + Send + 'static>;
 
 	impl future01::Executor<Boxed01Future01> for TestTaskExecutor {
 		fn execute(&self, future: Boxed01Future01) -> std::result::Result<(), future01::ExecuteError<Boxed01Future01>> {
-			let executor = executor::ThreadPool::new().expect("Failed to create thread pool executor for tests");
-			executor.spawn_ok(future.compat().map(drop));
+			EXECUTOR.spawn_ok(future.compat().map(drop));
 			Ok(())
 		}
 	}
@@ -349,8 +355,6 @@ mod tests {
 		});
 
 		let is_cancelled = manager.cancel(id);
-		dbg!(is_cancelled);
-
 		assert!(is_cancelled);
 	}
 

From c3dafbc5cdad05cf61aebce05193c13095dbfd99 Mon Sep 17 00:00:00 2001
From: Hernando Castano <castano.ha@gmail.com>
Date: Wed, 27 May 2020 11:39:13 -0400
Subject: [PATCH 14/16] Rustfmt

---
 pubsub/src/manager.rs | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/pubsub/src/manager.rs b/pubsub/src/manager.rs
index aef3e9fe6..308e04dae 100644
--- a/pubsub/src/manager.rs
+++ b/pubsub/src/manager.rs
@@ -127,7 +127,7 @@ impl Default for RandomStringIdProvider {
 /// Takes care of assigning unique subscription ids and
 /// driving the sinks into completion.
 #[derive(Clone)]
-pub struct SubscriptionManager<I: IdProvider=RandomStringIdProvider> {
+pub struct SubscriptionManager<I: IdProvider = RandomStringIdProvider> {
 	id_provider: I,
 	active_subscriptions: ActiveSubscriptions,
 	executor: TaskExecutor,

From 52caa9afdf65d53667d983a7c02e0b0a861052e9 Mon Sep 17 00:00:00 2001
From: Hernando Castano <castano.ha@gmail.com>
Date: Wed, 27 May 2020 11:56:25 -0400
Subject: [PATCH 15/16] Add a comment to test

---
 pubsub/src/manager.rs | 2 ++
 1 file changed, 2 insertions(+)

diff --git a/pubsub/src/manager.rs b/pubsub/src/manager.rs
index 308e04dae..c0dae987f 100644
--- a/pubsub/src/manager.rs
+++ b/pubsub/src/manager.rs
@@ -342,6 +342,8 @@ mod tests {
 	#[test]
 	fn subscription_is_canceled_if_it_existed() {
 		let manager = SubscriptionManager::<NumericIdProvider>::with_executor(Arc::new(TestTaskExecutor));
+		// Need to bind receiver here (unlike the other tests) or else the subscriber
+		// will think the client has disconnected and not update `active_subscriptions`
 		let (subscriber, _recv, _) = Subscriber::<u64>::new_test("test_subTest");
 
 		let (mut tx, rx) = futures::channel::mpsc::channel(8);

From 2c27a2d8c44c9ea255f3eb76435db222dfe8bf83 Mon Sep 17 00:00:00 2001
From: Hernando Castano <castano.ha@gmail.com>
Date: Wed, 27 May 2020 12:18:34 -0400
Subject: [PATCH 16/16] Remove `matches!` macro

Our Windows CI runner isn't up to date and thinks this
is still a nightly feature
---
 pubsub/src/manager.rs | 18 +++++++++++++++---
 1 file changed, 15 insertions(+), 3 deletions(-)

diff --git a/pubsub/src/manager.rs b/pubsub/src/manager.rs
index c0dae987f..e8e725949 100644
--- a/pubsub/src/manager.rs
+++ b/pubsub/src/manager.rs
@@ -302,7 +302,11 @@ mod tests {
 			sink.sink_map_err(|_| ()).send_all(stream).map(|_| ())
 		});
 
-		assert!(matches!(id, SubscriptionId::String(_)))
+		if let SubscriptionId::String(_) = id {
+			assert!(true)
+		} else {
+			assert!(false, "Expected SubscriptionId::String");
+		}
 	}
 
 	#[test]
@@ -319,7 +323,11 @@ mod tests {
 			sink.sink_map_err(|_| ()).send_all(stream).map(|_| ())
 		});
 
-		assert!(matches!(id, SubscriptionId::Number(_)))
+		if let SubscriptionId::Number(_) = id {
+			assert!(true)
+		} else {
+			assert!(false, "Expected SubscriptionId::Number");
+		}
 	}
 
 	#[test]
@@ -336,7 +344,11 @@ mod tests {
 			sink.sink_map_err(|_| ()).send_all(stream).map(|_| ())
 		});
 
-		assert!(matches!(id, SubscriptionId::String(_)))
+		if let SubscriptionId::String(_) = id {
+			assert!(true)
+		} else {
+			assert!(false, "Expected SubscriptionId::String");
+		}
 	}
 
 	#[test]