Limit for logs filter. (#2180)

* Limit for logs filter.

* Moving limit inside the filter object

* Fixing tests

ethcore/src/client/client.rs

@@ -957,7 +957,7 @@ impl BlockChainClient for Client {
 		}
 	}
 
-	fn logs(&self, filter: Filter, limit: Option<usize>) -> Vec<LocalizedLogEntry> {
+	fn logs(&self, filter: Filter) -> Vec<LocalizedLogEntry> {
 		let blocks = filter.bloom_possibilities().iter()
 			.filter_map(|bloom| self.blocks_with_bloom(bloom, filter.from_block.clone(), filter.to_block.clone()))
 			.flat_map(|m| m)
@@ -966,7 +966,7 @@ impl BlockChainClient for Client {
 			.into_iter()
 			.collect::<Vec<u64>>();
 
-		self.chain.read().logs(blocks, |entry| filter.matches(entry), limit)
+		self.chain.read().logs(blocks, |entry| filter.matches(entry), filter.limit)
 	}
 
 	fn filter_traces(&self, filter: TraceFilter) -> Option<Vec<LocalizedTrace>> {

ethcore/src/client/test_client.rs

@@ -67,6 +67,8 @@ pub struct TestBlockChainClient {
 	pub execution_result: RwLock<Option<Result<Executed, CallError>>>,
 	/// Transaction receipts.
 	pub receipts: RwLock<HashMap<TransactionID, LocalizedReceipt>>,
+	/// Logs
+	pub logs: RwLock<Vec<LocalizedLogEntry>>,
 	/// Block queue size.
 	pub queue_size: AtomicUsize,
 	/// Miner
@@ -114,6 +116,7 @@ impl TestBlockChainClient {
 			code: RwLock::new(HashMap::new()),
 			execution_result: RwLock::new(None),
 			receipts: RwLock::new(HashMap::new()),
+			logs: RwLock::new(Vec::new()),
 			queue_size: AtomicUsize::new(0),
 			miner: Arc::new(Miner::with_spec(&spec)),
 			spec: spec,
@@ -165,6 +168,11 @@ impl TestBlockChainClient {
 		*self.latest_block_timestamp.write() = ts;
 	}
 
+	/// Set logs to return for each logs call.
+	pub fn set_logs(&self, logs: Vec<LocalizedLogEntry>) {
+		*self.logs.write() = logs;
+	}
+
 	/// Add blocks to test client.
 	pub fn add_blocks(&self, count: usize, with: EachBlockWith) {
 		let len = self.numbers.read().len();
@@ -390,8 +398,13 @@ impl BlockChainClient for TestBlockChainClient {
 		unimplemented!();
 	}
 
-	fn logs(&self, _filter: Filter, _limit: Option<usize>) -> Vec<LocalizedLogEntry> {
+	fn logs(&self, filter: Filter) -> Vec<LocalizedLogEntry> {
-		Vec::new()
+		let mut logs = self.logs.read().clone();
+		let len = logs.len();
+		match filter.limit {
+			Some(limit) if limit <= len => logs.split_off(len - limit),
+			_ => logs,
+		}
 	}
 
 	fn last_hashes(&self) -> LastHashes {

ethcore/src/client/traits.rs

@@ -156,7 +156,7 @@ pub trait BlockChainClient : Sync + Send {
 	fn blocks_with_bloom(&self, bloom: &H2048, from_block: BlockID, to_block: BlockID) -> Option<Vec<BlockNumber>>;
 
 	/// Returns logs matching given filter.
-	fn logs(&self, filter: Filter, limit: Option<usize>) -> Vec<LocalizedLogEntry>;
+	fn logs(&self, filter: Filter) -> Vec<LocalizedLogEntry>;
 
 	/// Makes a non-persistent transaction call.
 	fn call(&self, t: &SignedTransaction, block: BlockID, analytics: CallAnalytics) -> Result<Executed, CallError>;

ethcore/src/tests/client.rs

@@ -142,7 +142,8 @@ fn returns_logs() {
 		to_block: BlockID::Latest,
 		address: None,
 		topics: vec![],
-	}, None);
+		limit: None,
+	});
 	assert_eq!(logs.len(), 0);
 }
 
@@ -156,7 +157,8 @@ fn returns_logs_with_limit() {
 		to_block: BlockID::Latest,
 		address: None,
 		topics: vec![],
-	}, Some(2));
+		limit: Some(2),
+	});
 	assert_eq!(logs.len(), 0);
 }
 

ethcore/src/types/filter.rs

@@ -41,6 +41,12 @@ pub struct Filter {
 	/// If None, match all.
 	/// If specified, log must contain one of these topics.
 	pub topics: Vec<Option<Vec<H256>>>,
+
+	/// Logs limit
+	///
+	/// If None, return all logs
+	/// If specified, should only return *last* `n` logs.
+	pub limit: Option<usize>,
 }
 
 impl Clone for Filter {
@@ -59,7 +65,8 @@ impl Clone for Filter {
 			from_block: self.from_block.clone(),
 			to_block: self.to_block.clone(),
 			address: self.address.clone(),
-			topics: topics[..].to_vec()
+			topics: topics[..].to_vec(),
+			limit: self.limit,
 		}
 	}
 }
@@ -117,6 +124,7 @@ mod tests {
 			to_block: BlockID::Latest,
 			address: None,
 			topics: vec![None, None, None, None],
+			limit: None,
 		};
 
 		let possibilities = none_filter.bloom_possibilities();
@@ -136,7 +144,8 @@ mod tests {
 				None,
 				None,
 				None,
-			]
+			],
+			limit: None,
 		};
 
 		let possibilities = filter.bloom_possibilities();
@@ -154,7 +163,8 @@ mod tests {
 				Some(vec!["ff74e91598aed6ae5d2fdcf8b24cd2c7be49a0808112a305069355b7160f23f9".into()]),
 				None,
 				None,
-			]
+			],
+			limit: None,
 		};
 
 		let possibilities = filter.bloom_possibilities();
@@ -181,7 +191,8 @@ mod tests {
 				]),
 				Some(vec!["ff74e91598aed6ae5d2fdcf8b24cd2c7be49a0808112a305069355b7160f23f9".into()]),
 				None
-			]
+			],
+			limit: None,
 		};
 
 		// number of possibilites should be equal 2 * 2 * 2 * 1 = 8
@@ -201,7 +212,8 @@ mod tests {
 				Some(vec!["ff74e91598aed6ae5d2fdcf8b24cd2c7be49a0808112a305069355b7160f23fa".into()]),
 				None,
 				None,
-			]
+			],
+			limit: None,
 		};
 
 		let entry0 = LogEntry {

rpc/src/v1/helpers/mod.rs

@@ -26,7 +26,7 @@ mod signing_queue;
 mod network_settings;
 
 pub use self::poll_manager::PollManager;
-pub use self::poll_filter::PollFilter;
+pub use self::poll_filter::{PollFilter, limit_logs};
 pub use self::requests::{TransactionRequest, FilledTransactionRequest, ConfirmationRequest, ConfirmationPayload, CallRequest};
 pub use self::signing_queue::{ConfirmationsQueue, ConfirmationPromise, ConfirmationResult, SigningQueue, QueueEvent};
 pub use self::signer::SignerService;

rpc/src/v1/helpers/poll_filter.rs

@@ -13,6 +13,15 @@ pub enum PollFilter {
 	Block(BlockNumber),
 	/// Hashes of all transactions which client was notified about.
 	PendingTransaction(Vec<H256>),
-	/// Number of From block number, pending logs and log filter iself.
+	/// Number of From block number, pending logs and log filter itself.
 	Logs(BlockNumber, HashSet<Log>, Filter)
 }
+
+/// Returns only last `n` logs
+pub fn limit_logs(mut logs: Vec<Log>, limit: Option<usize>) -> Vec<Log> {
+	let len = logs.len();
+	match limit {
+		Some(limit) if len >= limit => logs.split_off(len - limit),
+		_ => logs,
+	}
+}

rpc/src/v1/impls/eth.rs

@@ -43,7 +43,7 @@ use ethcore::filter::Filter as EthcoreFilter;
 use self::ethash::SeedHashCompute;
 use v1::traits::Eth;
 use v1::types::{Block, BlockTransactions, BlockNumber, Bytes, SyncStatus, SyncInfo, Transaction, CallRequest, Index, Filter, Log, Receipt, H64 as RpcH64, H256 as RpcH256, H160 as RpcH160, U256 as RpcU256};
-use v1::helpers::{CallRequest as CRequest, errors};
+use v1::helpers::{CallRequest as CRequest, errors, limit_logs};
 use v1::helpers::dispatch::{default_gas_price, dispatch_transaction};
 use v1::helpers::params::{expect_no_params, params_len, from_params_default_second, from_params_default_third};
 
@@ -498,14 +498,10 @@ impl<C, S: ?Sized, M, EM> Eth for EthClient<C, S, M, EM> where
 
 	fn logs(&self, params: Params) -> Result<Value, Error> {
 		try!(self.active());
-		let params = match params_len(&params) {
+		from_params::<(Filter, )>(params).and_then(|(filter,)| {
-			1 => from_params::<(Filter, )>(params).map(|(filter, )| (filter, None)),
-			_ => from_params::<(Filter, usize)>(params).map(|(filter, val)| (filter, Some(val))),
-		};
-		params.and_then(|(filter, limit)| {
 			let include_pending = filter.to_block == Some(BlockNumber::Pending);
 			let filter: EthcoreFilter = filter.into();
-			let mut logs = take_weak!(self.client).logs(filter.clone(), limit)
+			let mut logs = take_weak!(self.client).logs(filter.clone())
 				.into_iter()
 				.map(From::from)
 				.collect::<Vec<Log>>();
@@ -515,14 +511,7 @@ impl<C, S: ?Sized, M, EM> Eth for EthClient<C, S, M, EM> where
 				logs.extend(pending);
 			}
 
-			let len = logs.len();
+			let logs = limit_logs(logs, filter.limit);
-			match limit {
-				Some(limit) if len >= limit => {
-					logs = logs.split_off(len - limit);
-				},
-				_ => {},
-			}
-
 			Ok(to_value(&logs))
 		})
 	}

rpc/src/v1/impls/eth_filter.rs

@@ -25,8 +25,8 @@ use ethcore::client::{BlockChainClient, BlockID};
 use util::Mutex;
 use v1::traits::EthFilter;
 use v1::types::{BlockNumber, Index, Filter, Log, H256 as RpcH256, U256 as RpcU256};
-use v1::helpers::{PollFilter, PollManager};
+use v1::helpers::{PollFilter, PollManager, limit_logs};
-use v1::helpers::params::expect_no_params;
+use v1::helpers::params::{expect_no_params, params_len};
 use v1::impls::eth::pending_logs;
 
 /// Eth filter rpc implementation.
@@ -65,8 +65,8 @@ impl<C, M> EthFilter for EthFilterClient<C, M> where
 
 	fn new_filter(&self, params: Params) -> Result<Value, Error> {
 		try!(self.active());
-		from_params::<(Filter,)>(params)
+		from_params::<(Filter, )>(params)
-			.and_then(|(filter,)| {
+			.and_then(|(filter, )| {
 				let mut polls = self.polls.lock();
 				let block_number = take_weak!(self.client).chain_info().best_block_number;
 				let id = polls.create_poll(PollFilter::Logs(block_number, Default::default(), filter));
@@ -152,7 +152,7 @@ impl<C, M> EthFilter for EthFilterClient<C, M> where
 							filter.to_block = BlockID::Latest;
 
 							// retrieve logs in range from_block..min(BlockID::Latest..to_block)
-							let mut logs = client.logs(filter.clone(), None)
+							let mut logs = client.logs(filter.clone())
 								.into_iter()
 								.map(From::from)
 								.collect::<Vec<Log>>();
@@ -174,6 +174,8 @@ impl<C, M> EthFilter for EthFilterClient<C, M> where
 								logs.extend(new_pending_logs);
 							}
 
+							let logs = limit_logs(logs, filter.limit);
+
 							// save the number of the next block as a first block from which
 							// we want to get logs
 							*block_number = current_number + 1;
@@ -194,7 +196,7 @@ impl<C, M> EthFilter for EthFilterClient<C, M> where
 					Some(&PollFilter::Logs(ref _block_number, ref _previous_log, ref filter)) => {
 						let include_pending = filter.to_block == Some(BlockNumber::Pending);
 						let filter: EthcoreFilter = filter.clone().into();
-						let mut logs = take_weak!(self.client).logs(filter.clone(), None)
+						let mut logs = take_weak!(self.client).logs(filter.clone())
 							.into_iter()
 							.map(From::from)
 							.collect::<Vec<Log>>();
@@ -203,6 +205,8 @@ impl<C, M> EthFilter for EthFilterClient<C, M> where
 							logs.extend(pending_logs(&*take_weak!(self.miner), &filter));
 						}
 
+						let logs = limit_logs(logs, filter.limit);
+
 						Ok(to_value(&logs))
 					},
 					// just empty array

rpc/src/v1/tests/mocked/eth.rs

@@ -27,7 +27,7 @@ use ethcore::receipt::LocalizedReceipt;
 use ethcore::transaction::{Transaction, Action};
 use ethcore::miner::{ExternalMiner, MinerService};
 use ethsync::SyncState;
-use v1::{Eth, EthClient, EthClientOptions, EthSigning, EthSigningUnsafeClient};
+use v1::{Eth, EthClient, EthClientOptions, EthFilter, EthFilterClient, EthSigning, EthSigningUnsafeClient};
 use v1::tests::helpers::{TestSyncProvider, Config, TestMinerService};
 use rustc_serialize::hex::ToHex;
 use time::get_time;
@@ -76,10 +76,12 @@ impl EthTester {
 		let hashrates = Arc::new(Mutex::new(HashMap::new()));
 		let external_miner = Arc::new(ExternalMiner::new(hashrates.clone()));
 		let eth = EthClient::new(&client, &sync, &ap, &miner, &external_miner, options).to_delegate();
+		let filter = EthFilterClient::new(&client, &miner).to_delegate();
 		let sign = EthSigningUnsafeClient::new(&client, &ap, &miner).to_delegate();
 		let io = IoHandler::new();
 		io.add_delegate(eth);
 		io.add_delegate(sign);
+		io.add_delegate(filter);
 
 		EthTester {
 			client: client,
@@ -152,23 +154,88 @@ fn rpc_eth_hashrate() {
 #[test]
 fn rpc_eth_logs() {
 	let tester = EthTester::default();
+	tester.client.set_logs(vec![LocalizedLogEntry {
+		block_number: 1,
+		block_hash: H256::default(),
+		entry: LogEntry {
+			address: Address::default(),
+			topics: vec![],
+			data: vec![1,2,3],
+		},
+		transaction_index: 0,
+		transaction_hash: H256::default(),
+		log_index: 0,
+	}, LocalizedLogEntry {
+		block_number: 1,
+		block_hash: H256::default(),
+		entry: LogEntry {
+			address: Address::default(),
+			topics: vec![],
+			data: vec![1,2,3],
+		},
+		transaction_index: 0,
+		transaction_hash: H256::default(),
+		log_index: 0,
+	}]);
 
-	let request = r#"{"jsonrpc": "2.0", "method": "eth_getLogs", "params": [{}], "id": 1}"#;
-	let response = r#"{"jsonrpc":"2.0","result":[],"id":1}"#;
 
-	assert_eq!(tester.io.handle_request_sync(request), Some(response.to_owned()));
+	let request1 = r#"{"jsonrpc": "2.0", "method": "eth_getLogs", "params": [{}], "id": 1}"#;
+	let request2 = r#"{"jsonrpc": "2.0", "method": "eth_getLogs", "params": [{"limit":1}], "id": 1}"#;
+	let request3 = r#"{"jsonrpc": "2.0", "method": "eth_getLogs", "params": [{"limit":0}], "id": 1}"#;
+
+	let response1 = r#"{"jsonrpc":"2.0","result":[{"address":"0x0000000000000000000000000000000000000000","blockHash":"0x0000000000000000000000000000000000000000000000000000000000000000","blockNumber":"0x1","data":"0x010203","logIndex":"0x0","topics":[],"transactionHash":"0x0000000000000000000000000000000000000000000000000000000000000000","transactionIndex":"0x0","type":"mined"},{"address":"0x0000000000000000000000000000000000000000","blockHash":"0x0000000000000000000000000000000000000000000000000000000000000000","blockNumber":"0x1","data":"0x010203","logIndex":"0x0","topics":[],"transactionHash":"0x0000000000000000000000000000000000000000000000000000000000000000","transactionIndex":"0x0","type":"mined"}],"id":1}"#;
+	let response2 = r#"{"jsonrpc":"2.0","result":[{"address":"0x0000000000000000000000000000000000000000","blockHash":"0x0000000000000000000000000000000000000000000000000000000000000000","blockNumber":"0x1","data":"0x010203","logIndex":"0x0","topics":[],"transactionHash":"0x0000000000000000000000000000000000000000000000000000000000000000","transactionIndex":"0x0","type":"mined"}],"id":1}"#;
+	let response3 = r#"{"jsonrpc":"2.0","result":[],"id":1}"#;
+
+	assert_eq!(tester.io.handle_request_sync(request1), Some(response1.to_owned()));
+	assert_eq!(tester.io.handle_request_sync(request2), Some(response2.to_owned()));
+	assert_eq!(tester.io.handle_request_sync(request3), Some(response3.to_owned()));
 }
 
 #[test]
-fn rpc_eth_logs_with_limit() {
+fn rpc_logs_filter() {
 	let tester = EthTester::default();
+	// Set some logs
+	tester.client.set_logs(vec![LocalizedLogEntry {
+		block_number: 1,
+		block_hash: H256::default(),
+		entry: LogEntry {
+			address: Address::default(),
+			topics: vec![],
+			data: vec![1,2,3],
+		},
+		transaction_index: 0,
+		transaction_hash: H256::default(),
+		log_index: 0,
+	}, LocalizedLogEntry {
+		block_number: 1,
+		block_hash: H256::default(),
+		entry: LogEntry {
+			address: Address::default(),
+			topics: vec![],
+			data: vec![1,2,3],
+		},
+		transaction_index: 0,
+		transaction_hash: H256::default(),
+		log_index: 0,
+	}]);
 
-	let request1 = r#"{"jsonrpc": "2.0", "method": "eth_getLogs", "params": [{}, 1], "id": 1}"#;
+	// Register filters first
-	let request2 = r#"{"jsonrpc": "2.0", "method": "eth_getLogs", "params": [{}, 0], "id": 1}"#;
+	let request_default = r#"{"jsonrpc": "2.0", "method": "eth_newFilter", "params": [{}], "id": 1}"#;
-	let response = r#"{"jsonrpc":"2.0","result":[],"id":1}"#;
+	let request_limit = r#"{"jsonrpc": "2.0", "method": "eth_newFilter", "params": [{"limit":1}], "id": 1}"#;
+	let response1 = r#"{"jsonrpc":"2.0","result":"0x0","id":1}"#;
+	let response2 = r#"{"jsonrpc":"2.0","result":"0x1","id":1}"#;
 
-	assert_eq!(tester.io.handle_request_sync(request1), Some(response.to_owned()));
+	assert_eq!(tester.io.handle_request_sync(request_default), Some(response1.to_owned()));
-	assert_eq!(tester.io.handle_request_sync(request2), Some(response.to_owned()));
+	assert_eq!(tester.io.handle_request_sync(request_limit), Some(response2.to_owned()));
+
+	let request_changes1 = r#"{"jsonrpc": "2.0", "method": "eth_getFilterChanges", "params": ["0x0"], "id": 1}"#;
+	let request_changes2 = r#"{"jsonrpc": "2.0", "method": "eth_getFilterChanges", "params": ["0x1"], "id": 1}"#;
+	let response1 = r#"{"jsonrpc":"2.0","result":[{"address":"0x0000000000000000000000000000000000000000","blockHash":"0x0000000000000000000000000000000000000000000000000000000000000000","blockNumber":"0x1","data":"0x010203","logIndex":"0x0","topics":[],"transactionHash":"0x0000000000000000000000000000000000000000000000000000000000000000","transactionIndex":"0x0","type":"mined"},{"address":"0x0000000000000000000000000000000000000000","blockHash":"0x0000000000000000000000000000000000000000000000000000000000000000","blockNumber":"0x1","data":"0x010203","logIndex":"0x0","topics":[],"transactionHash":"0x0000000000000000000000000000000000000000000000000000000000000000","transactionIndex":"0x0","type":"mined"}],"id":1}"#;
+	let response2 = r#"{"jsonrpc":"2.0","result":[{"address":"0x0000000000000000000000000000000000000000","blockHash":"0x0000000000000000000000000000000000000000000000000000000000000000","blockNumber":"0x1","data":"0x010203","logIndex":"0x0","topics":[],"transactionHash":"0x0000000000000000000000000000000000000000000000000000000000000000","transactionIndex":"0x0","type":"mined"}],"id":1}"#;
+
+	assert_eq!(tester.io.handle_request_sync(request_changes1), Some(response1.to_owned()));
+	assert_eq!(tester.io.handle_request_sync(request_changes2), Some(response2.to_owned()));
 }
 
 #[test]

rpc/src/v1/types/filter.rs

@@ -66,6 +66,8 @@ pub struct Filter {
 	pub address: Option<FilterAddress>,
 	/// Topics
 	pub topics: Option<Vec<Topic>>,
+	/// Limit
+	pub limit: Option<usize>,
 }
 
 impl Into<EthFilter> for Filter {
@@ -85,7 +87,8 @@ impl Into<EthFilter> for Filter {
 					VariadicValue::Multiple(t) => Some(t.into_iter().map(Into::into).collect())
 				}).filter_map(|m| m).collect()).into_iter();
 				vec![iter.next(), iter.next(), iter.next(), iter.next()]
-			}
+			},
+			limit: self.limit,
 		}
 	}
 }
@@ -120,7 +123,8 @@ mod tests {
 			from_block: Some(BlockNumber::Earliest),
 			to_block: Some(BlockNumber::Latest),
 			address: None,
-			topics: None
+			topics: None,
+			limit: None,
 		});
 	}
 }