support for polling

This commit is contained in:
debris 2016-02-23 18:51:29 +01:00
parent 52ed69b619
commit 35cabecad8
9 changed files with 267 additions and 9 deletions

9
Cargo.lock generated
View File

@ -211,6 +211,7 @@ dependencies = [
"serde_codegen 0.6.14 (registry+https://github.com/rust-lang/crates.io-index)", "serde_codegen 0.6.14 (registry+https://github.com/rust-lang/crates.io-index)",
"serde_json 0.6.0 (registry+https://github.com/rust-lang/crates.io-index)", "serde_json 0.6.0 (registry+https://github.com/rust-lang/crates.io-index)",
"syntex 0.29.0 (registry+https://github.com/rust-lang/crates.io-index)", "syntex 0.29.0 (registry+https://github.com/rust-lang/crates.io-index)",
"transient-hashmap 0.1.0 (git+https://github.com/debris/transient-hashmap)",
] ]
[[package]] [[package]]
@ -728,6 +729,14 @@ name = "traitobject"
version = "0.0.1" version = "0.0.1"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
[[package]]
name = "transient-hashmap"
version = "0.1.0"
source = "git+https://github.com/debris/transient-hashmap#e21bf844277785504ddc30ee22d2a709103d4992"
dependencies = [
"time 0.1.34 (registry+https://github.com/rust-lang/crates.io-index)",
]
[[package]] [[package]]
name = "typeable" name = "typeable"
version = "0.1.2" version = "0.1.2"

View File

@ -42,6 +42,22 @@ pub struct Filter {
pub topics: [Option<Vec<H256>>; 4], pub topics: [Option<Vec<H256>>; 4],
} }
impl Clone for Filter {
fn clone(&self) -> Self {
let mut topics = [None, None, None, None];
for i in 0..4 {
topics[i] = self.topics[i].clone();
}
Filter {
from_block: self.from_block.clone(),
to_block: self.to_block.clone(),
address: self.address.clone(),
topics: topics
}
}
}
impl Filter { impl Filter {
/// Returns combinations of each address and topic. /// Returns combinations of each address and topic.
pub fn bloom_possibilities(&self) -> Vec<H2048> { pub fn bloom_possibilities(&self) -> Vec<H2048> {

View File

@ -19,6 +19,7 @@ ethsync = { path = "../sync" }
clippy = { version = "0.0.44", optional = true } clippy = { version = "0.0.44", optional = true }
rustc-serialize = "0.3" rustc-serialize = "0.3"
serde_macros = { version = "0.6.13", optional = true } serde_macros = { version = "0.6.13", optional = true }
transient-hashmap = { git = "https://github.com/debris/transient-hashmap" }
[build-dependencies] [build-dependencies]
serde_codegen = { version = "0.6.13", optional = true } serde_codegen = { version = "0.6.13", optional = true }

View File

@ -27,6 +27,7 @@ extern crate jsonrpc_http_server;
extern crate ethcore_util as util; extern crate ethcore_util as util;
extern crate ethcore; extern crate ethcore;
extern crate ethsync; extern crate ethsync;
extern crate transient_hashmap;
#[cfg(feature = "serde_macros")] #[cfg(feature = "serde_macros")]
include!("lib.rs.in"); include!("lib.rs.in");

21
rpc/src/v1/helpers/mod.rs Normal file
View File

@ -0,0 +1,21 @@
// Copyright 2015, 2016 Ethcore (UK) Ltd.
// This file is part of Parity.
// Parity 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.
// Parity 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 Parity. If not, see <http://www.gnu.org/licenses/>.
mod poll_indexer;
mod poll_filter;
pub use self::poll_indexer::PollIndexer;
pub use self::poll_filter::PollFilter;

View File

@ -0,0 +1,10 @@
//! Helper type with all filter possibilities.
use ethcore::filter::Filter;
#[derive(Clone)]
pub enum PollFilter {
Block,
PendingTransaction,
Logs(Filter)
}

View File

@ -0,0 +1,144 @@
// Copyright 2015, 2016 Ethcore (UK) Ltd.
// This file is part of Parity.
// Parity 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.
// Parity 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 Parity. If not, see <http://www.gnu.org/licenses/>.
//! Indexes all rpc poll requests.
use transient_hashmap::{TransientHashMap, Timer, StandardTimer};
/// Lifetime of poll (in seconds).
const POLL_LIFETIME: u64 = 60;
pub type PollId = usize;
pub type BlockNumber = u64;
pub struct PollInfo<F> {
pub filter: F,
pub block_number: BlockNumber
}
impl<F> Clone for PollInfo<F> where F: Clone {
fn clone(&self) -> Self {
PollInfo {
filter: self.filter.clone(),
block_number: self.block_number.clone()
}
}
}
/// Indexes all poll requests.
///
/// Lazily garbage collects unused polls info.
pub struct PollIndexer<F, T = StandardTimer> where T: Timer {
polls: TransientHashMap<PollId, PollInfo<F>, T>,
next_available_id: PollId
}
impl<F> PollIndexer<F, StandardTimer> {
/// Creates new instance of indexer.
pub fn new() -> Self {
PollIndexer::new_with_timer(Default::default())
}
}
impl<F, T> PollIndexer<F, T> where T: Timer {
pub fn new_with_timer(timer: T) -> Self {
PollIndexer {
polls: TransientHashMap::new_with_timer(POLL_LIFETIME, timer),
next_available_id: 0
}
}
/// Returns id which can be used for new poll.
///
/// Stores information when last poll happend.
pub fn create_poll(&mut self, filter: F, block: BlockNumber) -> PollId {
self.polls.prune();
let id = self.next_available_id;
self.next_available_id += 1;
self.polls.insert(id, PollInfo {
filter: filter,
block_number: block
});
id
}
/// Updates information when last poll happend.
pub fn update_poll(&mut self, id: &PollId, block: BlockNumber) {
self.polls.prune();
if let Some(info) = self.polls.get_mut(id) {
info.block_number = block;
}
}
/// Returns number of block when last poll happend.
pub fn get_poll(&mut self, id: &PollId) -> Option<&PollInfo<F>> {
self.polls.prune();
self.polls.get(id)
}
/// Removes poll info.
pub fn remove_poll(&mut self, id: &PollId) {
self.polls.remove(id);
}
}
#[cfg(test)]
mod tests {
use std::cell::RefCell;
use transient_hashmap::Timer;
use v1::helpers::PollIndexer;
struct TestTimer<'a> {
time: &'a RefCell<i64>
}
impl<'a> Timer for TestTimer<'a> {
fn get_time(&self) -> i64 {
*self.time.borrow()
}
}
#[test]
fn test_poll_indexer() {
let time = RefCell::new(0);
let timer = TestTimer {
time: &time
};
let mut indexer = PollIndexer::new_with_timer(timer);
assert_eq!(indexer.create_poll(false, 20), 0);
assert_eq!(indexer.create_poll(true, 20), 1);
*time.borrow_mut() = 10;
indexer.update_poll(&0, 21);
assert_eq!(indexer.get_poll(&0).unwrap().filter, false);
assert_eq!(indexer.get_poll(&0).unwrap().block_number, 21);
*time.borrow_mut() = 30;
indexer.update_poll(&1, 23);
assert_eq!(indexer.get_poll(&1).unwrap().filter, true);
assert_eq!(indexer.get_poll(&1).unwrap().block_number, 23);
*time.borrow_mut() = 75;
indexer.update_poll(&0, 30);
assert!(indexer.get_poll(&0).is_none());
assert_eq!(indexer.get_poll(&1).unwrap().filter, true);
assert_eq!(indexer.get_poll(&1).unwrap().block_number, 23);
indexer.remove_poll(&1);
assert!(indexer.get_poll(&1).is_none());
}
}

View File

@ -15,7 +15,7 @@
// along with Parity. If not, see <http://www.gnu.org/licenses/>. // along with Parity. If not, see <http://www.gnu.org/licenses/>.
//! Eth rpc implementation. //! Eth rpc implementation.
use std::sync::Arc; use std::sync::{Mutex, Arc};
use ethsync::{EthSync, SyncState}; use ethsync::{EthSync, SyncState};
use jsonrpc_core::*; use jsonrpc_core::*;
use util::hash::*; use util::hash::*;
@ -26,6 +26,7 @@ use ethcore::views::*;
use ethcore::ethereum::denominations::shannon; use ethcore::ethereum::denominations::shannon;
use v1::traits::{Eth, EthFilter}; use v1::traits::{Eth, EthFilter};
use v1::types::{Block, BlockTransactions, BlockNumber, Bytes, SyncStatus, SyncInfo, Transaction, OptionalValue, Index, Filter, Log}; use v1::types::{Block, BlockTransactions, BlockNumber, Bytes, SyncStatus, SyncInfo, Transaction, OptionalValue, Index, Filter, Log};
use v1::helpers::{PollFilter, PollIndexer};
/// Eth rpc implementation. /// Eth rpc implementation.
pub struct EthClient { pub struct EthClient {
@ -212,28 +213,82 @@ impl Eth for EthClient {
/// Eth filter rpc implementation. /// Eth filter rpc implementation.
pub struct EthFilterClient { pub struct EthFilterClient {
client: Arc<Client> client: Arc<Client>,
polls: Mutex<PollIndexer<PollFilter>>
} }
impl EthFilterClient { impl EthFilterClient {
/// Creates new Eth filter client. /// Creates new Eth filter client.
pub fn new(client: Arc<Client>) -> Self { pub fn new(client: Arc<Client>) -> Self {
EthFilterClient { EthFilterClient {
client: client client: client,
polls: Mutex::new(PollIndexer::new())
} }
} }
} }
impl EthFilter for EthFilterClient { impl EthFilter for EthFilterClient {
fn new_block_filter(&self, _params: Params) -> Result<Value, Error> { fn new_filter(&self, params: Params) -> Result<Value, Error> {
Ok(Value::U64(0)) from_params::<(Filter,)>(params)
.and_then(|(filter,)| {
let mut polls = self.polls.lock().unwrap();
let id = polls.create_poll(PollFilter::Logs(filter.into()), self.client.chain_info().best_block_number);
to_value(&U256::from(id))
})
} }
fn new_pending_transaction_filter(&self, _params: Params) -> Result<Value, Error> { fn new_block_filter(&self, params: Params) -> Result<Value, Error> {
Ok(Value::U64(1)) match params {
Params::None => {
let mut polls = self.polls.lock().unwrap();
let id = polls.create_poll(PollFilter::Block, self.client.chain_info().best_block_number);
to_value(&U256::from(id))
},
_ => Err(Error::invalid_params())
}
} }
fn filter_changes(&self, _: Params) -> Result<Value, Error> { fn new_pending_transaction_filter(&self, params: Params) -> Result<Value, Error> {
match params {
Params::None => {
let mut polls = self.polls.lock().unwrap();
let id = polls.create_poll(PollFilter::PendingTransaction, self.client.chain_info().best_block_number);
to_value(&U256::from(id))
},
_ => Err(Error::invalid_params())
}
}
fn filter_changes(&self, params: Params) -> Result<Value, Error> {
from_params::<(Index,)>(params)
.and_then(|(index,)| {
let info = self.polls.lock().unwrap().get_poll(&index.value()).cloned();
match info {
None => Ok(Value::Array(vec![] as Vec<Value>)),
Some(info) => match info.filter {
PollFilter::Block => {
//unimplemented!()
to_value(&self.client.chain_info().best_block_hash).map(|v| Value::Array(vec![v])) to_value(&self.client.chain_info().best_block_hash).map(|v| Value::Array(vec![v]))
},
PollFilter::PendingTransaction => {
//unimplemented!()
to_value(&self.client.chain_info().best_block_hash).map(|v| Value::Array(vec![v]))
},
PollFilter::Logs(mut filter) => {
filter.from_block = BlockId::Number(info.block_number);
filter.to_block = BlockId::Latest;
let logs = self.client.logs(filter)
.into_iter()
.map(From::from)
.collect::<Vec<Log>>();
let current_number = self.client.chain_info().best_block_number;
self.polls.lock().unwrap().update_poll(&index.value(), current_number);
to_value(&logs)
}
}
}
})
} }
} }

View File

@ -23,6 +23,7 @@ mod impls;
mod types; mod types;
#[cfg(test)] #[cfg(test)]
mod tests; mod tests;
mod helpers;
pub use self::traits::{Web3, Eth, EthFilter, Net}; pub use self::traits::{Web3, Eth, EthFilter, Net};
pub use self::impls::*; pub use self::impls::*;