Merge branch 'tx_queue_integration' into tx_queue_rpc

ethcore/src/client.rs

@@ -413,7 +413,9 @@ impl<V> Client<V> where V: Verifier {
 			if !good_blocks.is_empty() && block_queue.queue_info().is_empty() {
 				io.send(NetworkIoMessage::User(SyncMessage::NewChainBlocks {
 					good: good_blocks,
-					retracted: bad_blocks,
+					bad: bad_blocks,
+					// TODO [todr] were to take those from?
+					retracted: vec![],
 				})).unwrap();
 			}
 		}

ethcore/src/service.rs

@@ -30,6 +30,8 @@ pub enum SyncMessage {
 		/// Hashes of blocks imported to blockchain
 		good: Vec<H256>,
 		/// Hashes of blocks not imported to blockchain
+		bad: Vec<H256>,
+		/// Hashes of blocks that were removed from canonical chain
 		retracted: Vec<H256>,
 	},
 	/// A block is ready

sync/src/chain.rs

@@ -936,9 +936,11 @@ impl ChainSync {
 		let item_count = r.item_count();
 		trace!(target: "sync", "{} -> Transactions ({} entries)", peer_id, item_count);
 		let fetch_latest_nonce = |a : &Address| chain.nonce(a);
+
+		let mut transaction_queue = self.transaction_queue.lock().unwrap();
 		for i in 0..item_count {
 			let tx: SignedTransaction = try!(r.val_at(i));
-			self.transaction_queue.lock().unwrap().add(tx, &fetch_latest_nonce);
+			transaction_queue.add(tx, &fetch_latest_nonce);
 		}
  		Ok(())
 	}
@@ -1268,7 +1270,7 @@ impl ChainSync {
 	}
 
 	/// called when block is imported to chain, updates transactions queue
-	pub fn chain_new_blocks(&mut self, io: &SyncIo, good: &[H256], retracted: &[H256]) {
+	pub fn chain_new_blocks(&mut self, io: &SyncIo, good: &[H256], bad: &[H256], _retracted: &[H256]) {
 		fn fetch_transactions(chain: &BlockChainClient, hash: &H256) -> Vec<SignedTransaction> {
 			let block = chain
 				.block(BlockId::Hash(hash.clone()))
@@ -1281,14 +1283,14 @@ impl ChainSync {
 
 		let chain = io.chain();
 		let good = good.par_iter().map(|h| fetch_transactions(chain, h));
-		let retracted = retracted.par_iter().map(|h| fetch_transactions(chain, h));
+		let bad = bad.par_iter().map(|h| fetch_transactions(chain, h));
 
 		good.for_each(|txs| {
 			let mut transaction_queue = self.transaction_queue.lock().unwrap();
 			let hashes = txs.iter().map(|tx| tx.hash()).collect::<Vec<H256>>();
 			transaction_queue.remove_all(&hashes, |a| chain.nonce(a));
 		});
-		retracted.for_each(|txs| {
+		bad.for_each(|txs| {
 			// populate sender
 			for tx in &txs {
 				let _sender = tx.sender();
@@ -1638,10 +1640,10 @@ mod tests {
 		let io = TestIo::new(&mut client, &mut queue, None);
 
 		// when
-		sync.chain_new_blocks(&io, &[], &good_blocks);
+		sync.chain_new_blocks(&io, &[], &good_blocks, &[]);
 		assert_eq!(sync.transaction_queue.lock().unwrap().status().future, 0);
 		assert_eq!(sync.transaction_queue.lock().unwrap().status().pending, 1);
-		sync.chain_new_blocks(&io, &good_blocks, &retracted_blocks);
+		sync.chain_new_blocks(&io, &good_blocks, &retracted_blocks, &[]);
 
 		// then
 		let status = sync.transaction_queue.lock().unwrap().status();

sync/src/lib.rs

@@ -157,9 +157,9 @@ impl NetworkProtocolHandler<SyncMessage> for EthSync {
 			SyncMessage::BlockVerified => {
 				self.sync.write().unwrap().chain_blocks_verified(&mut NetSyncIo::new(io, self.chain.deref()));
 			},
-			SyncMessage::NewChainBlocks { ref good, ref retracted } => {
+			SyncMessage::NewChainBlocks { ref good, ref bad, ref retracted } => {
 				let sync_io = NetSyncIo::new(io, self.chain.deref());
-				self.sync.write().unwrap().chain_new_blocks(&sync_io, good, retracted);
+				self.sync.write().unwrap().chain_new_blocks(&sync_io, good, bad, retracted);
 			}
 		}
 	}

sync/src/transaction_queue.rs

@@ -813,6 +813,4 @@ mod test {
 		assert_eq!(stats.future, 0);
 		assert_eq!(stats.pending, 1);
 	}
-
-
 }