Basic informant (#118)

* Informant

* Pass Handle instead of Core

cli/Cargo.toml

@@ -13,9 +13,13 @@ hex-literal = "0.1"
 triehash = "0.1"
 ed25519 = { path = "../../substrate/ed25519" }
 app_dirs = "1.2"
+tokio-core = "0.1.12"
+futures = "0.1.17"
+ctrlc = { git = "https://github.com/paritytech/rust-ctrlc.git" }
 substrate-client = { path = "../../substrate/client" }
+substrate-network = { path = "../../substrate/network" }
 substrate-codec = { path = "../../substrate/codec" }
-substrate-runtime-io = { path = "../../substrate/runtime-io" }
+substrate-runtime-support = { path = "../../substrate/runtime-support" }
 substrate-state-machine = { path = "../../substrate/state-machine" }
 substrate-executor = { path = "../../substrate/executor" }
 substrate-primitives = { path = "../../substrate/primitives" }

cli/src/informant.rs

+// Copyright 2017 Parity Technologies (UK) Ltd.
+// This file is part of Polkadot.
+
+// Polkadot 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.
+
+// Polkadot 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 Polkadot.  If not, see <http://www.gnu.org/licenses/>.
+
+//! Console informant. Prints sync progress and block events. Runs on the calling thread.
+
+use std::time::{Duration, Instant};
+use futures::stream::Stream;
+use service::Service;
+use tokio_core::reactor;
+use network::{SyncState, SyncProvider};
+use runtime_support::Hashable;
+use primitives::block::HeaderHash;
+use client::BlockchainEvents;
+
+const TIMER_INTERVAL_MS: u64 = 5000;
+
+/// Spawn informant on the event loop
+pub fn start(service: &Service, handle: reactor::Handle) {
+	let interval = reactor::Interval::new_at(Instant::now(), Duration::from_millis(TIMER_INTERVAL_MS), &handle)
+		.expect("Error creating informant timer");
+
+	let network = service.network();
+	let client = service.client();
+
+	let display_notifications = interval.map_err(|e| debug!("Timer error: {:?}", e)).for_each(move |_| {
+		let sync_status = network.status();
+
+		if let Ok(best_block) = client.best_block_header() {
+			let hash: HeaderHash = best_block.blake2_256().into();
+			let status = match (sync_status.sync.state, sync_status.sync.best_seen_block) {
+				(SyncState::Idle, _) => "Idle".into(),
+				(SyncState::Downloading, None) => "Syncing".into(),
+				(SyncState::Downloading, Some(n)) => format!("Syncing, target=#{}", n),
+			};
+			println!("{} ({} peers), best: #{} ({})", status, sync_status.num_peers, best_block.number, hash)
+		} else {
+			warn!("Error getting best block information");
+		}
+		Ok(())
+	});
+
+	let client = service.client();
+	let display_block_import = client.import_notification_stream().for_each(|n| {
+		info!(target: "polkadot", "Imported #{} ({})", n.header.number, n.hash);
+		Ok(())
+	});
+
+	handle.spawn(display_notifications);
+	handle.spawn(display_block_import);
+}
+
+
+

cli/src/lib.rs

@@ -20,13 +20,18 @@
 
 extern crate app_dirs;
 extern crate env_logger;
+extern crate futures;
+extern crate tokio_core;
+extern crate ctrlc;
 extern crate ed25519;
 extern crate triehash;
 extern crate substrate_codec as codec;
 extern crate substrate_state_machine as state_machine;
 extern crate substrate_client as client;
 extern crate substrate_primitives as primitives;
+extern crate substrate_network as network;
 extern crate substrate_rpc_servers as rpc;
+extern crate substrate_runtime_support as runtime_support;
 extern crate polkadot_primitives;
 extern crate polkadot_executor;
 extern crate polkadot_runtime;
@@ -40,10 +45,13 @@ extern crate error_chain;
 extern crate log;
 
 pub mod error;
+mod informant;
 
 use std::path::{Path, PathBuf};
 use std::net::SocketAddr;
-use std::sync::mpsc;
+use futures::sync::mpsc;
+use futures::{Sink, Future, Stream};
+use tokio_core::reactor;
 
 /// Parse command line arguments and start the node.
 ///
@@ -53,10 +61,12 @@ use std::sync::mpsc;
 /// 9556-9591		Unassigned
 /// 9803-9874		Unassigned
 /// 9926-9949		Unassigned
-pub fn run<I, T>(args: I, exit: mpsc::Receiver<()>) -> error::Result<()> where
+pub fn run<I, T>(args: I) -> error::Result<()> where
 	I: IntoIterator<Item = T>,
 	T: Into<std::ffi::OsString> + Clone,
 {
+	let mut core = reactor::Core::new().expect("tokio::Core could not be created");
+
 	let yaml = load_yaml!("./cli.yml");
 	let matches = match clap::App::from_yaml(yaml).version(crate_version!()).get_matches_from_safe(args) {
 		Ok(m) => m,
@@ -118,10 +128,17 @@ pub fn run<I, T>(args: I, exit: mpsc::Receiver<()>) -> error::Result<()> where
 		let rpc_port: u16 = port.parse().expect("Invalid RPC port value specified.");
 		address.set_port(rpc_port);
 	}
+
 	let handler = rpc::rpc_handler(service.client(), service.transaction_pool());
 	let _server = rpc::start_http(&address, handler)?;
 
-	exit.recv().ok();
+	informant::start(&service, core.handle());
+
+	let (exit_send, exit) = 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");
 	Ok(())
 }
 

src/main.rs

@@ -22,16 +22,9 @@ extern crate polkadot_cli as cli;
 
 #[macro_use]
 extern crate error_chain;
-extern crate ctrlc;
-
-use std::sync::mpsc;
 
 quick_main!(run);
 
 fn run() -> cli::error::Result<()> {
-	let (exit_send, exit_receive) = mpsc::channel();
-	ctrlc::CtrlC::set_handler(move || {
-		exit_send.send(()).expect("Error sending exit notification");
-	});
-	cli::run(::std::env::args(), exit_receive)
+	cli::run(::std::env::args())
 }