Fix pubsub new_blocks notifications to include all blocks

ethcore/private-tx/src/lib.rs

@@ -69,7 +69,6 @@ pub use error::{Error, ErrorKind};
 
 use std::sync::{Arc, Weak};
 use std::collections::{HashMap, HashSet, BTreeMap};
-use std::time::Duration;
 use ethereum_types::{H128, H256, U256, Address};
 use hash::keccak;
 use rlp::*;
@@ -82,7 +81,7 @@ use ethcore::executed::{Executed};
 use transaction::{SignedTransaction, Transaction, Action, UnverifiedTransaction};
 use ethcore::{contract_address as ethcore_contract_address};
 use ethcore::client::{
-	Client, ChainNotify, ChainRoute, ChainMessageType, ClientIoMessage, BlockId,
+	Client, ChainNotify, NewBlocks, ChainMessageType, ClientIoMessage, BlockId,
 	CallContract, Call, BlockInfo
 };
 use ethcore::account_provider::AccountProvider;
@@ -733,12 +732,11 @@ fn find_account_password(passwords: &Vec<Password>, account_provider: &AccountPr
 }
 
 impl ChainNotify for Provider {
-	fn new_blocks(&self, imported: Vec<H256>, _invalid: Vec<H256>, _route: ChainRoute, _sealed: Vec<H256>, _proposed: Vec<Bytes>, _duration: Duration) {
-		if !imported.is_empty() {
-			trace!(target: "privatetx", "New blocks imported, try to prune the queue");
-			if let Err(err) = self.process_verification_queue() {
-				warn!(target: "privatetx", "Cannot prune private transactions queue. error: {:?}", err);
-			}
+	fn new_blocks(&self, new_blocks: NewBlocks) {
+		if new_blocks.imported.is_empty() || new_blocks.has_more_blocks_to_import { return }
+		trace!(target: "privatetx", "New blocks imported, try to prune the queue");
+		if let Err(err) = self.process_verification_queue() {
+			warn!(target: "privatetx", "Cannot prune private transactions queue. error: {:?}", err);
 		}
 	}
 }

ethcore/src/client/chain_notify.rs

@@ -114,19 +114,51 @@ impl ChainRoute {
 	}
 }
 
+/// Used by `ChainNotify` `new_blocks()`
+pub struct NewBlocks {
+	/// Imported blocks
+	pub imported: Vec<H256>,
+	/// Invalid blocks
+	pub invalid: Vec<H256>,
+	/// Route
+	pub route: ChainRoute,
+	/// Sealed
+	pub sealed: Vec<H256>,
+	/// Block bytes.
+	pub proposed: Vec<Bytes>,
+	/// Duration
+	pub duration: Duration,
+	/// Has more blocks to import
+	pub has_more_blocks_to_import: bool,
+}
+
+impl NewBlocks {
+	/// Constructor
+	pub fn new (
+		imported: Vec<H256>,
+		invalid: Vec<H256>,
+		route: ChainRoute,
+		sealed: Vec<H256>,
+		proposed: Vec<Bytes>,
+		duration: Duration,
+		has_more_blocks_to_import: bool,
+	) -> NewBlocks {
+		NewBlocks {
+			imported,
+			invalid,
+			route,
+			sealed,
+			proposed,
+			duration,
+			has_more_blocks_to_import,
+		}
+	}
+}
+
 /// Represents what has to be handled by actor listening to chain events
 pub trait ChainNotify : Send + Sync {
 	/// fires when chain has new blocks.
-	fn new_blocks(
-		&self,
-		_imported: Vec<H256>,
-		_invalid: Vec<H256>,
-		_route: ChainRoute,
-		_sealed: Vec<H256>,
-		// Block bytes.
-		_proposed: Vec<Bytes>,
-		_duration: Duration,
-	) {
+	fn new_blocks( &self, _new_blocks: NewBlocks) {
 		// does nothing by default
 	}
 

ethcore/src/client/client.rs

@@ -44,7 +44,7 @@ use client::{
 use client::{
 	BlockId, TransactionId, UncleId, TraceId, ClientConfig, BlockChainClient,
 	TraceFilter, CallAnalytics, Mode,
-	ChainNotify, ChainRoute, PruningInfo, ProvingBlockChainClient, EngineInfo, ChainMessageType,
+	ChainNotify, NewBlocks, ChainRoute, PruningInfo, ProvingBlockChainClient, EngineInfo, ChainMessageType,
 	IoClient, BadBlocks,
 };
 use client::bad_blocks;
@@ -268,7 +268,7 @@ impl Importer {
 		}
 
 		let max_blocks_to_import = client.config.max_round_blocks_to_import;
-		let (imported_blocks, import_results, invalid_blocks, imported, proposed_blocks, duration, is_empty) = {
+		let (imported_blocks, import_results, invalid_blocks, imported, proposed_blocks, duration, has_more_blocks_to_import) = {
 			let mut imported_blocks = Vec::with_capacity(max_blocks_to_import);
 			let mut invalid_blocks = HashSet::new();
 			let mut proposed_blocks = Vec::with_capacity(max_blocks_to_import);
@@ -322,26 +322,29 @@ impl Importer {
 			if !invalid_blocks.is_empty() {
 				self.block_queue.mark_as_bad(&invalid_blocks);
 			}
-			let is_empty = self.block_queue.mark_as_good(&imported_blocks);
-			(imported_blocks, import_results, invalid_blocks, imported, proposed_blocks, start.elapsed(), is_empty)
+			let has_more_blocks_to_import = !self.block_queue.mark_as_good(&imported_blocks);
+			(imported_blocks, import_results, invalid_blocks, imported, proposed_blocks, start.elapsed(), has_more_blocks_to_import)
 		};
 
 		{
-			if !imported_blocks.is_empty() && is_empty {
+			if !imported_blocks.is_empty() {
 				let route = ChainRoute::from(import_results.as_ref());
 
-				if is_empty {
+				if !has_more_blocks_to_import {
 					self.miner.chain_new_blocks(client, &imported_blocks, &invalid_blocks, route.enacted(), route.retracted(), false);
 				}
 
 				client.notify(|notify| {
 					notify.new_blocks(
-						imported_blocks.clone(),
-						invalid_blocks.clone(),
-						route.clone(),
-						Vec::new(),
-						proposed_blocks.clone(),
-						duration,
+						NewBlocks::new(
+							imported_blocks.clone(),
+							invalid_blocks.clone(),
+							route.clone(),
+							Vec::new(),
+							proposed_blocks.clone(),
+							duration,
+							has_more_blocks_to_import,
+						)
 					);
 				});
 			}
@@ -2342,12 +2345,15 @@ impl ImportSealedBlock for Client {
 		);
 		self.notify(|notify| {
 			notify.new_blocks(
-				vec![hash],
-				vec![],
-				route.clone(),
-				vec![hash],
-				vec![],
-				start.elapsed(),
+				NewBlocks::new(
+					vec![hash],
+					vec![],
+					route.clone(),
+					vec![hash],
+					vec![],
+					start.elapsed(),
+					false
+				)
 			);
 		});
 		self.db.read().key_value().flush().expect("DB flush failed.");
@@ -2360,12 +2366,15 @@ impl BroadcastProposalBlock for Client {
 		const DURATION_ZERO: Duration = Duration::from_millis(0);
 		self.notify(|notify| {
 			notify.new_blocks(
-				vec![],
-				vec![],
-				ChainRoute::default(),
-				vec![],
-				vec![block.rlp_bytes()],
-				DURATION_ZERO,
+				NewBlocks::new(
+					vec![],
+					vec![],
+					ChainRoute::default(),
+					vec![],
+					vec![block.rlp_bytes()],
+					DURATION_ZERO,
+					false
+				)
 			);
 		});
 	}

ethcore/src/client/mod.rs

@@ -34,7 +34,7 @@ pub use self::evm_test_client::{EvmTestClient, EvmTestError, TransactResult};
 pub use self::io_message::ClientIoMessage;
 #[cfg(any(test, feature = "test-helpers"))]
 pub use self::test_client::{TestBlockChainClient, EachBlockWith};
-pub use self::chain_notify::{ChainNotify, ChainRoute, ChainRouteType, ChainMessageType};
+pub use self::chain_notify::{ChainNotify, NewBlocks, ChainRoute, ChainRouteType, ChainMessageType};
 pub use self::traits::{
     Nonce, Balance, ChainInfo, BlockInfo, ReopenBlock, PrepareOpenBlock, CallContract, TransactionInfo, RegistryInfo, ScheduleInfo, ImportSealedBlock, BroadcastProposalBlock, ImportBlock,
     StateOrBlock, StateClient, Call, EngineInfo, AccountData, BlockChain, BlockProducer, SealedBlockImporter, BadBlocks,

ethcore/src/snapshot/watcher.rs

@@ -17,14 +17,13 @@
 //! Watcher for snapshot-related chain events.
 
 use parking_lot::Mutex;
-use client::{BlockInfo, Client, ChainNotify, ChainRoute, ClientIoMessage};
+use client::{BlockInfo, Client, ChainNotify, NewBlocks, ClientIoMessage};
 use ids::BlockId;
 
 use io::IoChannel;
 use ethereum_types::H256;
-use bytes::Bytes;
 
-use std::{sync::Arc, time::Duration};
+use std::sync::Arc;
 
 // helper trait for transforming hashes to numbers and checking if syncing.
 trait Oracle: Send + Sync {
@@ -99,20 +98,12 @@ impl Watcher {
 }
 
 impl ChainNotify for Watcher {
-	fn new_blocks(
-		&self,
-		imported: Vec<H256>,
-		_: Vec<H256>,
-		_: ChainRoute,
-		_: Vec<H256>,
-		_: Vec<Bytes>,
-		_duration: Duration)
-	{
-		if self.oracle.is_major_importing() { return }
+	fn new_blocks(&self, new_blocks: NewBlocks) {
+		if self.oracle.is_major_importing() || new_blocks.has_more_blocks_to_import { return }
 
-		trace!(target: "snapshot_watcher", "{} imported", imported.len());
+		trace!(target: "snapshot_watcher", "{} imported", new_blocks.imported.len());
 
-		let highest = imported.into_iter()
+		let highest = new_blocks.imported.into_iter()
 			.filter_map(|h| self.oracle.to_number(h))
 			.filter(|&num| num >= self.period + self.history)
 			.map(|num| num - self.history)
@@ -130,7 +121,7 @@ impl ChainNotify for Watcher {
 mod tests {
 	use super::{Broadcast, Oracle, Watcher};
 
-	use client::{ChainNotify, ChainRoute};
+	use client::{ChainNotify, NewBlocks, ChainRoute};
 
 	use ethereum_types::{H256, U256};
 
@@ -170,14 +161,15 @@ mod tests {
 			history: history,
 		};
 
-		watcher.new_blocks(
+		watcher.new_blocks(NewBlocks::new(
 			hashes,
 			vec![],
 			ChainRoute::default(),
 			vec![],
 			vec![],
 			DURATION_ZERO,
-		);
+			false
+		));
 	}
 
 	// helper

ethcore/sync/src/api.rs

@@ -29,7 +29,7 @@ use types::pruning_info::PruningInfo;
 use ethereum_types::{H256, H512, U256};
 use io::{TimerToken};
 use ethcore::ethstore::ethkey::Secret;
-use ethcore::client::{BlockChainClient, ChainNotify, ChainRoute, ChainMessageType};
+use ethcore::client::{BlockChainClient, ChainNotify, NewBlocks, ChainMessageType};
 use ethcore::snapshot::SnapshotService;
 use ethcore::header::BlockNumber;
 use sync_io::NetSyncIo;
@@ -498,27 +498,22 @@ impl ChainNotify for EthSync {
 		}
 	}
 
-	fn new_blocks(&self,
-		imported: Vec<H256>,
-		invalid: Vec<H256>,
-		route: ChainRoute,
-		sealed: Vec<H256>,
-		proposed: Vec<Bytes>,
-		_duration: Duration)
+	fn new_blocks(&self, new_blocks: NewBlocks)
 	{
+		if new_blocks.has_more_blocks_to_import { return }
 		use light::net::Announcement;
 
 		self.network.with_context(self.subprotocol_name, |context| {
 			let mut sync_io = NetSyncIo::new(context, &*self.eth_handler.chain, &*self.eth_handler.snapshot_service,
 				&self.eth_handler.overlay);
 			self.eth_handler.sync.write().chain_new_blocks(
 				&mut sync_io,
-				&imported,
-				&invalid,
-				route.enacted(),
-				route.retracted(),
-				&sealed,
-				&proposed);
+				&new_blocks.imported,
+				&new_blocks.invalid,
+				new_blocks.route.enacted(),
+				new_blocks.route.retracted(),
+				&new_blocks.sealed,
+				&new_blocks.proposed);
 		});
 
 		self.network.with_context(self.light_subprotocol_name, |context| {

ethcore/sync/src/tests/helpers.rs

@@ -16,14 +16,13 @@
 
 use std::collections::{VecDeque, HashSet, HashMap};
 use std::sync::Arc;
-use std::time::Duration;
 use ethereum_types::H256;
 use parking_lot::{RwLock, Mutex};
 use bytes::Bytes;
 use network::{self, PeerId, ProtocolId, PacketId, SessionInfo};
 use tests::snapshot::*;
 use ethcore::client::{TestBlockChainClient, BlockChainClient, Client as EthcoreClient,
-	ClientConfig, ChainNotify, ChainRoute, ChainMessageType, ClientIoMessage};
+	ClientConfig, ChainNotify, NewBlocks, ChainMessageType, ClientIoMessage};
 use ethcore::header::BlockNumber;
 use ethcore::snapshot::SnapshotService;
 use ethcore::spec::Spec;
@@ -535,23 +534,18 @@ impl IoHandler<ClientIoMessage> for TestIoHandler {
 }
 
 impl ChainNotify for EthPeer<EthcoreClient> {
-	fn new_blocks(&self,
-		imported: Vec<H256>,
-		invalid: Vec<H256>,
-		route: ChainRoute,
-		sealed: Vec<H256>,
-		proposed: Vec<Bytes>,
-		_duration: Duration)
+	fn new_blocks(&self, new_blocks: NewBlocks)
 	{
-		let (enacted, retracted) = route.into_enacted_retracted();
+		if new_blocks.has_more_blocks_to_import { return }
+		let (enacted, retracted) = new_blocks.route.into_enacted_retracted();
 
 		self.new_blocks_queue.write().push_back(NewBlockMessage {
-			imported,
-			invalid,
+			imported: new_blocks.imported,
+			invalid: new_blocks.invalid,
 			enacted,
 			retracted,
-			sealed,
-			proposed,
+			sealed: new_blocks.sealed,
+			proposed: new_blocks.proposed,
 		});
 	}