373 lines
13 KiB
Rust
373 lines
13 KiB
Rust
// Copyright 2015-2020 Parity Technologies (UK) Ltd.
|
|
// This file is part of OpenEthereum.
|
|
|
|
// OpenEthereum 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.
|
|
|
|
// OpenEthereum 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 OpenEthereum. If not, see <http://www.gnu.org/licenses/>.
|
|
|
|
//! Eth Filter RPC implementation
|
|
|
|
use std::{
|
|
collections::{BTreeSet, VecDeque},
|
|
sync::Arc,
|
|
};
|
|
|
|
use ethcore::{
|
|
client::{BlockChainClient, BlockId},
|
|
miner::{self, MinerService},
|
|
};
|
|
use ethereum_types::{H256, U256};
|
|
use parking_lot::Mutex;
|
|
use types::filter::Filter as EthcoreFilter;
|
|
|
|
use jsonrpc_core::{
|
|
futures::{future, future::Either, Future},
|
|
BoxFuture, Result,
|
|
};
|
|
use v1::{
|
|
helpers::{errors, limit_logs, PollFilter, PollManager, SyncPollFilter},
|
|
impls::eth::pending_logs,
|
|
traits::EthFilter,
|
|
types::{BlockNumber, Filter, FilterChanges, Index, Log},
|
|
};
|
|
|
|
/// Something which provides data that can be filtered over.
|
|
pub trait Filterable {
|
|
/// Current best block number.
|
|
fn best_block_number(&self) -> u64;
|
|
|
|
/// Get a block hash by block id.
|
|
fn block_hash(&self, id: BlockId) -> Option<H256>;
|
|
|
|
/// pending transaction hashes at the given block (unordered).
|
|
fn pending_transaction_hashes(&self) -> BTreeSet<H256>;
|
|
|
|
/// Get logs that match the given filter.
|
|
fn logs(&self, filter: EthcoreFilter) -> BoxFuture<Vec<Log>>;
|
|
|
|
/// Get logs from the pending block.
|
|
fn pending_logs(&self, block_number: u64, filter: &EthcoreFilter) -> Vec<Log>;
|
|
|
|
/// Get a reference to the poll manager.
|
|
fn polls(&self) -> &Mutex<PollManager<SyncPollFilter>>;
|
|
|
|
/// Get removed logs within route from the given block to the nearest canon block, not including the canon block. Also returns how many logs have been traversed.
|
|
fn removed_logs(&self, block_hash: H256, filter: &EthcoreFilter) -> (Vec<Log>, u64);
|
|
}
|
|
|
|
/// Eth filter rpc implementation for a full node.
|
|
pub struct EthFilterClient<C, M> {
|
|
client: Arc<C>,
|
|
miner: Arc<M>,
|
|
polls: Mutex<PollManager<SyncPollFilter>>,
|
|
}
|
|
|
|
impl<C, M> EthFilterClient<C, M> {
|
|
/// Creates new Eth filter client.
|
|
pub fn new(client: Arc<C>, miner: Arc<M>, poll_lifetime: u32) -> Self {
|
|
EthFilterClient {
|
|
client,
|
|
miner,
|
|
polls: Mutex::new(PollManager::new(poll_lifetime)),
|
|
}
|
|
}
|
|
}
|
|
|
|
impl<C, M> Filterable for EthFilterClient<C, M>
|
|
where
|
|
C: miner::BlockChainClient + BlockChainClient,
|
|
M: MinerService,
|
|
{
|
|
fn best_block_number(&self) -> u64 {
|
|
self.client.chain_info().best_block_number
|
|
}
|
|
|
|
fn block_hash(&self, id: BlockId) -> Option<H256> {
|
|
self.client.block_hash(id)
|
|
}
|
|
|
|
fn pending_transaction_hashes(&self) -> BTreeSet<H256> {
|
|
self.miner.pending_transaction_hashes(&*self.client)
|
|
}
|
|
|
|
fn logs(&self, filter: EthcoreFilter) -> BoxFuture<Vec<Log>> {
|
|
Box::new(future::ok(
|
|
self.client
|
|
.logs(filter)
|
|
.unwrap_or_default()
|
|
.into_iter()
|
|
.map(Into::into)
|
|
.collect(),
|
|
))
|
|
}
|
|
|
|
fn pending_logs(&self, block_number: u64, filter: &EthcoreFilter) -> Vec<Log> {
|
|
pending_logs(&*self.miner, block_number, filter)
|
|
}
|
|
|
|
fn polls(&self) -> &Mutex<PollManager<SyncPollFilter>> {
|
|
&self.polls
|
|
}
|
|
|
|
fn removed_logs(&self, block_hash: H256, filter: &EthcoreFilter) -> (Vec<Log>, u64) {
|
|
let inner = || -> Option<Vec<H256>> {
|
|
let mut route = Vec::new();
|
|
|
|
let mut current_block_hash = block_hash;
|
|
let mut current_block_header = self
|
|
.client
|
|
.block_header(BlockId::Hash(current_block_hash))?;
|
|
|
|
while current_block_hash
|
|
!= self
|
|
.client
|
|
.block_hash(BlockId::Number(current_block_header.number()))?
|
|
{
|
|
route.push(current_block_hash);
|
|
|
|
current_block_hash = current_block_header.parent_hash();
|
|
current_block_header = self
|
|
.client
|
|
.block_header(BlockId::Hash(current_block_hash))?;
|
|
}
|
|
|
|
Some(route)
|
|
};
|
|
|
|
let route = inner().unwrap_or_default();
|
|
let route_len = route.len() as u64;
|
|
(
|
|
route
|
|
.into_iter()
|
|
.flat_map(|block_hash| {
|
|
let mut filter = filter.clone();
|
|
filter.from_block = BlockId::Hash(block_hash);
|
|
filter.to_block = filter.from_block;
|
|
|
|
self.client
|
|
.logs(filter)
|
|
.unwrap_or_default()
|
|
.into_iter()
|
|
.map(|log| {
|
|
let mut log: Log = log.into();
|
|
log.log_type = "removed".into();
|
|
log.removed = true;
|
|
|
|
log
|
|
})
|
|
})
|
|
.collect(),
|
|
route_len,
|
|
)
|
|
}
|
|
}
|
|
|
|
impl<T: Filterable + Send + Sync + 'static> EthFilter for T {
|
|
fn new_filter(&self, filter: Filter) -> Result<U256> {
|
|
let mut polls = self.polls().lock();
|
|
let block_number = self.best_block_number();
|
|
let include_pending = filter.to_block == Some(BlockNumber::Pending);
|
|
let filter = filter.try_into()?;
|
|
let id = polls.create_poll(SyncPollFilter::new(PollFilter::Logs {
|
|
block_number,
|
|
filter,
|
|
include_pending,
|
|
last_block_hash: None,
|
|
previous_logs: Default::default(),
|
|
}));
|
|
Ok(id.into())
|
|
}
|
|
|
|
fn new_block_filter(&self) -> Result<U256> {
|
|
let mut polls = self.polls().lock();
|
|
// +1, since we don't want to include the current block
|
|
let id = polls.create_poll(SyncPollFilter::new(PollFilter::Block {
|
|
last_block_number: self.best_block_number(),
|
|
recent_reported_hashes: VecDeque::with_capacity(PollFilter::MAX_BLOCK_HISTORY_SIZE),
|
|
}));
|
|
Ok(id.into())
|
|
}
|
|
|
|
fn new_pending_transaction_filter(&self) -> Result<U256> {
|
|
let mut polls = self.polls().lock();
|
|
let pending_transactions = self.pending_transaction_hashes();
|
|
let id = polls.create_poll(SyncPollFilter::new(PollFilter::PendingTransaction(
|
|
pending_transactions,
|
|
)));
|
|
Ok(id.into())
|
|
}
|
|
|
|
fn filter_changes(&self, index: Index) -> BoxFuture<FilterChanges> {
|
|
let filter = match self.polls().lock().poll_mut(&index.value()) {
|
|
Some(filter) => filter.clone(),
|
|
None => return Box::new(future::err(errors::filter_not_found())),
|
|
};
|
|
|
|
Box::new(filter.modify(|filter| match *filter {
|
|
PollFilter::Block {
|
|
ref mut last_block_number,
|
|
ref mut recent_reported_hashes,
|
|
} => {
|
|
// Check validity of recently reported blocks -- in case of re-org, rewind block to last valid
|
|
while let Some((num, hash)) = recent_reported_hashes.front().cloned() {
|
|
if self.block_hash(BlockId::Number(num)) == Some(hash) {
|
|
break;
|
|
}
|
|
*last_block_number = num - 1;
|
|
recent_reported_hashes.pop_front();
|
|
}
|
|
let current_number = self.best_block_number();
|
|
let mut hashes = Vec::new();
|
|
for n in (*last_block_number + 1)..=current_number {
|
|
let block_number = BlockId::Number(n);
|
|
if let Some(hash) = self.block_hash(block_number) {
|
|
*last_block_number = n;
|
|
hashes.push(hash);
|
|
// Only keep the most recent history
|
|
if recent_reported_hashes.len() >= PollFilter::MAX_BLOCK_HISTORY_SIZE {
|
|
recent_reported_hashes.pop_back();
|
|
}
|
|
recent_reported_hashes.push_front((n, hash));
|
|
}
|
|
}
|
|
|
|
Either::A(future::ok(FilterChanges::Hashes(hashes)))
|
|
}
|
|
PollFilter::PendingTransaction(ref mut previous_hashes) => {
|
|
// get hashes of pending transactions
|
|
let current_hashes = self.pending_transaction_hashes();
|
|
|
|
let new_hashes = {
|
|
// find all new hashes
|
|
current_hashes
|
|
.difference(previous_hashes)
|
|
.cloned()
|
|
.map(Into::into)
|
|
.collect()
|
|
};
|
|
|
|
// save all hashes of pending transactions
|
|
*previous_hashes = current_hashes;
|
|
|
|
// return new hashes
|
|
Either::A(future::ok(FilterChanges::Hashes(new_hashes)))
|
|
}
|
|
PollFilter::Logs {
|
|
ref mut block_number,
|
|
ref mut last_block_hash,
|
|
ref mut previous_logs,
|
|
ref filter,
|
|
include_pending,
|
|
} => {
|
|
// retrive the current block number
|
|
let current_number = self.best_block_number();
|
|
|
|
let mut filter = filter.clone();
|
|
|
|
// retrieve reorg logs
|
|
let (mut reorg, reorg_len) = last_block_hash
|
|
.map_or_else(|| (Vec::new(), 0), |h| self.removed_logs(h, &filter));
|
|
*block_number -= reorg_len as u64;
|
|
|
|
filter.from_block = BlockId::Number(*block_number);
|
|
filter.to_block = BlockId::Latest;
|
|
|
|
// retrieve pending logs
|
|
let pending = if include_pending {
|
|
let pending_logs = self.pending_logs(current_number, &filter);
|
|
|
|
// remove logs about which client was already notified about
|
|
let new_pending_logs: Vec<_> = pending_logs
|
|
.iter()
|
|
.filter(|p| !previous_logs.contains(p))
|
|
.cloned()
|
|
.collect();
|
|
|
|
// save all logs retrieved by client
|
|
*previous_logs = pending_logs.into_iter().collect();
|
|
|
|
new_pending_logs
|
|
} else {
|
|
Vec::new()
|
|
};
|
|
|
|
// save the number of the next block as a first block from which
|
|
// we want to get logs
|
|
*block_number = current_number + 1;
|
|
|
|
// save the current block hash, which we used to get back to the
|
|
// canon chain in case of reorg.
|
|
*last_block_hash = self.block_hash(BlockId::Number(current_number));
|
|
|
|
// retrieve logs in range from_block..min(BlockId::Latest..to_block)
|
|
let limit = filter.limit;
|
|
Either::B(
|
|
self.logs(filter)
|
|
.map(move |logs| {
|
|
reorg.extend(logs);
|
|
reorg
|
|
}) // append reorg logs in the front
|
|
.map(move |mut logs| {
|
|
logs.extend(pending);
|
|
logs
|
|
}) // append fetched pending logs
|
|
.map(move |logs| limit_logs(logs, limit)) // limit the logs
|
|
.map(FilterChanges::Logs),
|
|
)
|
|
}
|
|
}))
|
|
}
|
|
|
|
fn filter_logs(&self, index: Index) -> BoxFuture<Vec<Log>> {
|
|
let (filter, include_pending) = {
|
|
let mut polls = self.polls().lock();
|
|
|
|
match polls.poll(&index.value()).and_then(|f| {
|
|
f.modify(|filter| match *filter {
|
|
PollFilter::Logs {
|
|
ref filter,
|
|
include_pending,
|
|
..
|
|
} => Some((filter.clone(), include_pending)),
|
|
_ => None,
|
|
})
|
|
}) {
|
|
Some((filter, include_pending)) => (filter, include_pending),
|
|
None => return Box::new(future::err(errors::filter_not_found())),
|
|
}
|
|
};
|
|
|
|
// fetch pending logs.
|
|
let pending = if include_pending {
|
|
let best_block = self.best_block_number();
|
|
self.pending_logs(best_block, &filter)
|
|
} else {
|
|
Vec::new()
|
|
};
|
|
|
|
// retrieve logs asynchronously, appending pending logs.
|
|
let limit = filter.limit;
|
|
let logs = self.logs(filter);
|
|
Box::new(
|
|
logs.map(move |mut logs| {
|
|
logs.extend(pending);
|
|
logs
|
|
})
|
|
.map(move |logs| limit_logs(logs, limit)),
|
|
)
|
|
}
|
|
|
|
fn uninstall_filter(&self, index: Index) -> Result<bool> {
|
|
Ok(self.polls().lock().remove_poll(&index.value()))
|
|
}
|
|
}
|