Finalized pubsub.

build.sh

@@ -1,5 +1,10 @@
 #!/bin/sh
 
+# NOTE `cargo install wasm-gc` before running this script.
+
+set -e
+export CARGO_INCREMENTAL=0
+
 cd demo/runtime/wasm && ./build.sh && cd ../../..
 cd substrate/executor/wasm && ./build.sh && cd ../../..
 cd substrate/test-runtime/wasm && ./build.sh && cd ../../..

demo/cli/Cargo.toml

@@ -6,12 +6,15 @@ description = "Substrate Demo node implementation in Rust."
 
 [dependencies]
 clap = { version = "2.27", features = ["yaml"] }
+ctrlc = { git = "https://github.com/paritytech/rust-ctrlc.git" }
+ed25519 = { path = "../../substrate/ed25519" }
 env_logger = "0.4"
+futures = "0.1.17"
 error-chain = "0.11"
-log = "0.3"
 hex-literal = "0.1"
+log = "0.3"
+tokio-core = "0.1.12"
 triehash = "0.1"
-ed25519 = { path = "../../substrate/ed25519" }
 substrate-client = { path = "../../substrate/client" }
 substrate-codec = { path = "../../substrate/codec" }
 substrate-runtime-io = { path = "../../substrate/runtime-io" }

demo/cli/src/lib.rs

@@ -18,18 +18,21 @@
 
 #![warn(missing_docs)]
 
-extern crate env_logger;
+extern crate ctrlc;
 extern crate ed25519;
+extern crate env_logger;
+extern crate futures;
+extern crate tokio_core;
 extern crate triehash;
-extern crate substrate_codec as codec;
-extern crate substrate_runtime_io as runtime_io;
-extern crate substrate_state_machine as state_machine;
 extern crate substrate_client as client;
+extern crate substrate_codec as codec;
 extern crate substrate_primitives as primitives;
 extern crate substrate_rpc;
 extern crate substrate_rpc_servers as rpc;
-extern crate demo_primitives;
+extern crate substrate_runtime_io as runtime_io;
+extern crate substrate_state_machine as state_machine;
 extern crate demo_executor;
+extern crate demo_primitives;
 extern crate demo_runtime;
 
 #[macro_use]
@@ -44,10 +47,12 @@ extern crate log;
 pub mod error;
 
 use std::sync::Arc;
+use client::genesis;
 use codec::Slicable;
 use demo_runtime::{GenesisConfig, ConsensusConfig, CouncilConfig, DemocracyConfig,
 	SessionConfig, StakingConfig, BuildExternalities};
-use client::genesis;
+use futures::{Future, Sink, Stream};
+
 
 struct DummyPool;
 impl substrate_rpc::author::AuthorApi for DummyPool {
@@ -127,17 +132,30 @@ pub fn run<I, T>(args: I) -> error::Result<()> where
 		(primitives::block::Header::decode(&mut block.header.encode().as_ref()).expect("to_vec() always gives a valid serialisation; qed"), storage.into_iter().collect())
 	};
 	let client = Arc::new(client::new_in_mem(executor, prepare_genesis)?);
-
-	let handler = || rpc::rpc_handler(client.clone(), client.clone(), DummyPool);
-	let http_address = "127.0.0.1:9933".parse().unwrap();
-	let http_server = rpc::start_http(&http_address, handler())?;
-	let ws_address = "127.0.0.1:9944".parse().unwrap();
-	let ws_server = rpc::start_ws(&ws_address, handler())?;
+	let mut core = ::tokio_core::reactor::Core::new().expect("Unable to spawn event loop.");
+
+	let _rpc_servers = {
+		let handler = || {
+			let chain = rpc::apis::chain::Chain::new(client.clone(), core.remote());
+			rpc::rpc_handler(client.clone(), chain, DummyPool)
+		};
+		let http_address = "127.0.0.1:9933".parse().unwrap();
+		let ws_address = "127.0.0.1:9944".parse().unwrap();
+
+		(
+			rpc::start_http(&http_address, handler())?,
+			rpc::start_ws(&ws_address, handler())?
+		)
+	};
 
 	if let Some(_) = matches.subcommand_matches("validator") {
 		info!("Starting validator.");
-		http_server.wait();
-		return ws_server.wait().map_err(|e| format!("{}", e).into());
+		let (exit_send, exit) = futures::sync::mpsc::channel(1);
+		ctrlc::CtrlC::set_handler(move || {
+			exit_send.clone().send(()).wait().expect("Error sending exit notification");
+		});
+		core.run(exit.into_future()).expect("Error running informant event loop");
+		return Ok(())
 	}
 
 	println!("No command given.\n");

polkadot/api/src/lib.rs

@@ -355,7 +355,6 @@ impl<S: state_machine::Backend> BlockBuilder for ClientBlockBuilder<S>
 #[cfg(test)]
 mod tests {
 	use super::*;
-	use runtime_io::with_externalities;
 	use keyring::Keyring;
 	use codec::Slicable;
 	use client::in_mem::Backend as InMemory;
@@ -388,7 +387,7 @@ mod tests {
 		::client::new_in_mem(
 			LocalDispatch::new(),
 			|| {
-				let mut storage = genesis_config.build_externalities();
+				let storage = genesis_config.build_externalities();
 				let block = ::client::genesis::construct_genesis_block(&storage);
 				(substrate_primitives::block::Header::decode(&mut block.header.encode().as_ref()).expect("to_vec() always gives a valid serialisation; qed"), storage.into_iter().collect())
 			}

polkadot/cli/src/lib.rs

@@ -141,8 +141,10 @@ pub fn run<I, T>(args: I) -> error::Result<()> where
 		let http_address = parse_address("127.0.0.1:9933", "rpc-port", &matches);
 		let ws_address = parse_address("127.0.0.1:9944", "ws-port", &matches);
 
-		let chain = rpc::apis::chain::Chain::new()
-		let handler = || rpc::rpc_handler(service.client(), service.client(), service.transaction_pool());
+		let handler = || {
+			let chain = rpc::apis::chain::Chain::new(service.client(), core.remote());
+			rpc::rpc_handler(service.client(), chain, service.transaction_pool())
+		};
 		(
 			rpc::start_http(&http_address, handler())?,
 			rpc::start_ws(&ws_address, handler())?,

substrate/rpc-servers/src/lib.rs

@@ -18,7 +18,7 @@
 
 #[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;

substrate/rpc/src/chain/mod.rs

@@ -26,6 +26,7 @@ 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;
 
@@ -63,9 +64,19 @@ build_rpc_trait! {
 /// Chain API with subscriptions support.
 pub struct Chain<B, E> {
 	/// Substrate client.
-	pub client: Arc<Client<B, E>>,
+	client: Arc<Client<B, E>>,
 	/// Current subscriptions.
-	pub subscriptions: 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