Chain head subscription (#126)

* Start WebSockets server.

* Expose non-working subscription.

* Dummy subscription for testing.

* Proper implementation with event loop.

* Finalized pubsub.

* Bump clap.

* Fix yml.

* Disable WS logs.

* Remove stale TransactionHash mention

* Fix build from nightly API change.

* Don't panic on invalid port.

* Bind server to random port.

* Send only best blocks.

substrate/substrate/client/src/client.rs

@@ -33,7 +33,7 @@ use {error, in_mem, block_builder, runtime_io, bft};
 pub type BlockchainEventStream = mpsc::UnboundedReceiver<BlockImportNotification>;
 
 /// Polkadot Client
-pub struct Client<B, E> where B: backend::Backend {
+pub struct Client<B, E> {
 	backend: B,
 	executor: E,
 	import_notification_sinks: Mutex<Vec<mpsc::UnboundedSender<BlockImportNotification>>>,

substrate/substrate/rpc-servers/Cargo.toml

@@ -4,6 +4,9 @@ version = "0.1.0"
 authors = ["Parity Technologies <admin@parity.io>"]
 
 [dependencies]
+jsonrpc-core = { git = "https://github.com/paritytech/jsonrpc.git" }
+jsonrpc-http-server = { git = "https://github.com/paritytech/jsonrpc.git" }
+jsonrpc-pubsub = { git = "https://github.com/paritytech/jsonrpc.git" }
+jsonrpc-ws-server = { git = "https://github.com/paritytech/jsonrpc.git" }
+log = "0.3"
 substrate-rpc = { path = "../rpc", version = "0.1" }
-jsonrpc-core = { git="https://github.com/paritytech/jsonrpc.git" }
-jsonrpc-http-server = { git="https://github.com/paritytech/jsonrpc.git" }

substrate/substrate/rpc-servers/src/lib.rs

@@ -18,30 +18,42 @@
 
 #[warn(missing_docs)]
 
-extern crate substrate_rpc as apis;
+pub extern crate substrate_rpc as apis;
 
 extern crate jsonrpc_core as rpc;
 extern crate jsonrpc_http_server as http;
+extern crate jsonrpc_pubsub as pubsub;
+extern crate jsonrpc_ws_server as ws;
+
+#[macro_use]
+extern crate log;
 
 use std::io;
 
+type Metadata = apis::metadata::Metadata;
+type RpcHandler = pubsub::PubSubHandler<Metadata>;
+
 /// Construct rpc `IoHandler`
-pub fn rpc_handler<S, T, C>(state: S, transaction_pool: T, chain: C) -> rpc::IoHandler where
+pub fn rpc_handler<S, C, A>(
+	state: S,
+	chain: C,
+	author: A,
+) -> RpcHandler where
 	S: apis::state::StateApi,
-	T: apis::author::AuthorApi,
-	C: apis::chain::ChainApi,
+	C: apis::chain::ChainApi<Metadata=Metadata>,
+	A: apis::author::AuthorApi,
 {
-	let mut io = rpc::IoHandler::new();
+	let mut io = pubsub::PubSubHandler::default();
 	io.extend_with(state.to_delegate());
-	io.extend_with(transaction_pool.to_delegate());
 	io.extend_with(chain.to_delegate());
+	io.extend_with(author.to_delegate());
 	io
 }
 
 /// Start HTTP server listening on given address.
 pub fn start_http(
 	addr: &std::net::SocketAddr,
-	io: rpc::IoHandler,
+	io: RpcHandler,
 ) -> io::Result<http::Server> {
 	http::ServerBuilder::new(io)
 		.threads(4)
@@ -49,3 +61,20 @@ pub fn start_http(
 		.cors(http::DomainsValidation::Disabled)
 		.start_http(addr)
 }
+
+/// Start WS server listening on given address.
+pub fn start_ws(
+	addr: &std::net::SocketAddr,
+	io: RpcHandler,
+) -> io::Result<ws::Server> {
+	ws::ServerBuilder::with_meta_extractor(io, |context: &ws::RequestContext| Metadata::new(context.sender()))
+		.start(addr)
+		.map_err(|err| match err {
+			ws::Error(ws::ErrorKind::Io(io), _) => io,
+			ws::Error(ws::ErrorKind::ConnectionClosed, _) => io::ErrorKind::BrokenPipe.into(),
+			ws::Error(e, _) => {
+				error!("{}", e);
+				io::ErrorKind::Other.into()
+			}
+		})
+}

substrate/substrate/rpc/Cargo.toml

@@ -4,15 +4,17 @@ version = "0.1.0"
 authors = ["Parity Technologies <admin@parity.io>"]
 
 [dependencies]
-parking_lot = "0.4"
-log = "0.3"
 error-chain = "0.11"
 jsonrpc-core = { git="https://github.com/paritytech/jsonrpc.git" }
 jsonrpc-macros = { git="https://github.com/paritytech/jsonrpc.git" }
+jsonrpc-pubsub = { git="https://github.com/paritytech/jsonrpc.git" }
+log = "0.3"
+parking_lot = "0.4"
 substrate-client = { path = "../client" }
+substrate-executor = { path = "../executor" }
 substrate-primitives = { path = "../primitives" }
 substrate-state-machine = { path = "../state-machine" }
-substrate-executor = { path = "../executor" }
+tokio-core = "0.1.12"
 
 [dev-dependencies]
 assert_matches = "1.1"

substrate/substrate/rpc/src/author/error.rs

@@ -14,6 +14,8 @@
 // You should have received a copy of the GNU General Public License
 // along with Substrate.  If not, see <http://www.gnu.org/licenses/>.
 
+//! Authoring RPC module errors.
+
 use client;
 use rpc;
 

substrate/substrate/rpc/src/author/tests.rs

@@ -15,7 +15,6 @@
 // along with Substrate.  If not, see <http://www.gnu.org/licenses/>.
 
 use primitives::block;
-use substrate_executor as executor;
 use super::*;
 use super::error::*;
 
@@ -38,7 +37,7 @@ impl AsyncAuthorApi for DummyTxPool {
 
 #[test]
 fn submit_transaction_should_not_cause_error() {
-	let mut p = Arc::new(Mutex::new(DummyTxPool::default()));
+	let p = Arc::new(Mutex::new(DummyTxPool::default()));
 
 	assert_matches!(
 		AuthorApi::submit_extrinsic(&p, block::Extrinsic(vec![])),

substrate/substrate/rpc/src/chain/mod.rs

@@ -17,12 +17,20 @@
 //! Substrate blockchain API.
 
 use std::sync::Arc;
+
 use primitives::block;
-use client::{self, Client};
+use client::{self, Client, BlockchainEvents};
 use state_machine;
 
-mod error;
+use jsonrpc_macros::pubsub;
+use jsonrpc_pubsub::SubscriptionId;
+use rpc::Result as RpcResult;
+use rpc::futures::{Future, Sink, Stream};
+use tokio_core::reactor::Remote;
 
+use subscriptions::Subscriptions;
+
+mod error;
 #[cfg(test)]
 mod tests;
 
@@ -31,6 +39,8 @@ use self::error::{Result, ResultExt};
 build_rpc_trait! {
 	/// Polkadot blockchain API
 	pub trait ChainApi {
+		type Metadata;
+
 		/// Get header of a relay chain block.
 		#[rpc(name = "chain_getHeader")]
 		fn header(&self, block::HeaderHash) -> Result<Option<block::Header>>;
@@ -38,19 +48,67 @@ build_rpc_trait! {
 		/// Get hash of the head.
 		#[rpc(name = "chain_getHead")]
 		fn head(&self) -> Result<block::HeaderHash>;
+
+		#[pubsub(name = "chain_newHead")] {
+			/// Hello subscription
+			#[rpc(name = "subscribe_newHead")]
+			fn subscribe_new_head(&self, Self::Metadata, pubsub::Subscriber<block::Header>);
+
+			/// Unsubscribe from hello subscription.
+			#[rpc(name = "unsubscribe_newHead")]
+			fn unsubscribe_new_head(&self, SubscriptionId) -> RpcResult<bool>;
+		}
 	}
 }
 
-impl<B, E> ChainApi for Arc<Client<B, E>> where
+/// Chain API with subscriptions support.
+pub struct Chain<B, E> {
+	/// Substrate client.
+	client: Arc<Client<B, E>>,
+	/// Current subscriptions.
+	subscriptions: Subscriptions,
+}
+
+impl<B, E> Chain<B, E> {
+	/// Create new Chain API RPC handler.
+	pub fn new(client: Arc<Client<B, E>>, remote: Remote) -> Self {
+		Chain {
+			client,
+			subscriptions: Subscriptions::new(remote),
+		}
+	}
+}
+
+impl<B, E> ChainApi for Chain<B, E> where
 	B: client::backend::Backend + Send + Sync + 'static,
 	E: state_machine::CodeExecutor + Send + Sync + 'static,
 	client::error::Error: From<<<B as client::backend::Backend>::State as state_machine::backend::Backend>::Error>,
 {
+	type Metadata = ::metadata::Metadata;
+
 	fn header(&self, hash: block::HeaderHash) -> Result<Option<block::Header>> {
-		client::Client::header(self, &block::Id::Hash(hash)).chain_err(|| "Blockchain error")
+		self.client.header(&block::Id::Hash(hash)).chain_err(|| "Blockchain error")
 	}
 
 	fn head(&self) -> Result<block::HeaderHash> {
-		Ok(client::Client::info(self).chain_err(|| "Blockchain error")?.chain.best_hash)
+		Ok(self.client.info().chain_err(|| "Blockchain error")?.chain.best_hash)
+	}
+
+	fn subscribe_new_head(&self, _metadata: Self::Metadata, subscriber: pubsub::Subscriber<block::Header>) {
+		self.subscriptions.add(subscriber, |sink| {
+			let stream = self.client.import_notification_stream()
+				.filter(|notification| notification.is_new_best)
+				.map(|notification| Ok(notification.header))
+				.map_err(|e| warn!("Block notification stream error: {:?}", e));
+			sink
+				.sink_map_err(|e| warn!("Error sending notifications: {:?}", e))
+				.send_all(stream)
+				// we ignore the resulting Stream (if the first stream is over we are unsubscribed)
+				.map(|_| ())
+		});
+	}
+
+	fn unsubscribe_new_head(&self, id: SubscriptionId) -> RpcResult<bool> {
+		Ok(self.subscriptions.cancel(id))
 	}
 }

substrate/substrate/rpc/src/chain/tests.rs

@@ -29,7 +29,13 @@ fn should_return_header() {
 		digest: Default::default(),
 	};
 
-	let client = Arc::new(client::new_in_mem(executor::WasmExecutor, || (test_genesis_block.clone(), vec![])).unwrap());
+	let core = ::tokio_core::reactor::Core::new().unwrap();
+	let remote = core.remote();
+
+	let client = Chain {
+		client: Arc::new(client::new_in_mem(executor::WasmExecutor, || (test_genesis_block.clone(), vec![])).unwrap()),
+		subscriptions: Subscriptions::new(remote),
+	};
 
 	assert_matches!(
 		ChainApi::header(&client, test_genesis_block.blake2_256().into()),

substrate/substrate/rpc/src/lib.rs

@@ -18,11 +18,13 @@
 
 #![warn(missing_docs)]
 
-extern crate parking_lot;
 extern crate jsonrpc_core as rpc;
+extern crate jsonrpc_pubsub;
+extern crate parking_lot;
 extern crate substrate_client as client;
 extern crate substrate_primitives as primitives;
 extern crate substrate_state_machine as state_machine;
+extern crate tokio_core;
 
 #[macro_use]
 extern crate error_chain;
@@ -39,6 +41,9 @@ extern crate assert_matches;
 #[cfg(test)]
 extern crate substrate_runtime_support as runtime_support;
 
-pub mod chain;
-pub mod state;
+mod subscriptions;
+
 pub mod author;
+pub mod chain;
+pub mod metadata;
+pub mod state;

substrate/substrate/rpc/src/metadata.rs

@@ -0,0 +1,46 @@
+// Copyright 2017-2018 Parity Technologies (UK) Ltd.
+// This file is part of Substrate.
+
+// Substrate is free software: you can redistribute it and/or modify
+// it under the terms of the GNU General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+
+// Substrate is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU General Public License for more details.
+
+// You should have received a copy of the GNU General Public License
+// along with Substrate.  If not, see <http://www.gnu.org/licenses/>.
+
+//! RPC Metadata
+use std::sync::Arc;
+
+use jsonrpc_pubsub::{Session, PubSubMetadata};
+
+/// RPC Metadata.
+///
+/// Manages peristent session for transports that support it
+/// and may contain some additional info extracted from specific transports
+/// (like remote client IP address, request headers, etc)
+#[derive(Default, Clone)]
+pub struct Metadata {
+	session: Option<Arc<Session>>,
+}
+
+impl ::rpc::Metadata for Metadata {}
+impl PubSubMetadata for Metadata {
+	fn session(&self) -> Option<Arc<Session>> {
+		self.session.clone()
+	}
+}
+
+impl Metadata {
+	/// Create new `Metadata` with session (Pub/Sub) support.
+	pub fn new(transport: ::rpc::futures::sync::mpsc::Sender<String>) -> Self {
+		Metadata {
+			session: Some(Arc::new(Session::new(transport))),
+		}
+	}
+}

substrate/substrate/rpc/src/subscriptions.rs

@@ -0,0 +1,86 @@
+// Copyright 2017-2018 Parity Technologies (UK) Ltd.
+// This file is part of Substrate.
+
+// Substrate is free software: you can redistribute it and/or modify
+// it under the terms of the GNU General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+
+// Substrate is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU General Public License for more details.
+
+// You should have received a copy of the GNU General Public License
+// along with Substrate.  If not, see <http://www.gnu.org/licenses/>.
+
+use std::collections::HashMap;
+use std::sync::atomic::{self, AtomicUsize};
+
+use jsonrpc_macros::pubsub;
+use jsonrpc_pubsub::SubscriptionId;
+use parking_lot::Mutex;
+use rpc::futures::sync::oneshot;
+use rpc::futures::{Future, future};
+use tokio_core::reactor::Remote;
+
+type Id = u64;
+
+/// Subscriptions manager.
+///
+/// Takes care of assigning unique subscription ids and
+/// driving the sinks into completion.
+#[derive(Debug)]
+pub struct Subscriptions {
+	next_id: AtomicUsize,
+	active_subscriptions: Mutex<HashMap<Id, oneshot::Sender<()>>>,
+	event_loop: Remote,
+}
+
+impl Subscriptions {
+	/// Creates new `Subscriptions` object.
+	pub fn new(event_loop: Remote) -> Self {
+		Subscriptions {
+			next_id: Default::default(),
+			active_subscriptions: Default::default(),
+			event_loop,
+		}
+	}
+
+	/// 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 bu underlying event loop
+	/// or will be cancelled in case #cancel is invoked.
+	pub fn add<T, E, G, R, F>(&self, subscriber: pubsub::Subscriber<T, E>, into_future: G) where
+		G: FnOnce(pubsub::Sink<T, E>) -> R,
+		R: future::IntoFuture<Future=F, Item=(), Error=()>,
+		F: future::Future<Item=(), Error=()> + Send + 'static,
+	{
+		let id = self.next_id.fetch_add(1, atomic::Ordering::AcqRel) as u64;
+		if let Ok(sink) = subscriber.assign_id(id.into()) {
+			let (tx, rx) = oneshot::channel();
+			let future = into_future(sink)
+				.into_future()
+				.select(rx.map_err(|e| warn!("Error timeing out: {:?}", e)))
+				.map(|_| ())
+				.map_err(|_| ());
+
+			self.active_subscriptions.lock().insert(id, tx);
+			self.event_loop.spawn(|_| future);
+		}
+	}
+
+	/// Cancel subscription.
+	///
+	/// Returns true if subscription existed or false otherwise.
+	pub fn cancel(&self, id: SubscriptionId) -> bool {
+		if let SubscriptionId::Number(id) = id {
+			if let Some(tx) = self.active_subscriptions.lock().remove(&id) {
+				let _ = tx.send(());
+				return true;
+			}
+		}
+		false
+	}
+}