2017-05-23 12:26:39 +02:00
|
|
|
// Copyright 2015-2017 Parity Technologies (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/>.
|
|
|
|
|
|
|
|
//! Eth PUB-SUB rpc implementation.
|
|
|
|
|
2018-02-16 16:51:34 +01:00
|
|
|
use std::sync::{Arc, Weak};
|
2017-05-23 12:26:39 +02:00
|
|
|
use std::collections::BTreeMap;
|
|
|
|
|
2017-11-14 11:38:17 +01:00
|
|
|
use jsonrpc_core::{BoxFuture, Result, Error};
|
2017-10-05 12:35:01 +02:00
|
|
|
use jsonrpc_core::futures::{self, Future, IntoFuture};
|
2017-05-23 12:26:39 +02:00
|
|
|
use jsonrpc_macros::Trailing;
|
|
|
|
use jsonrpc_macros::pubsub::{Sink, Subscriber};
|
|
|
|
use jsonrpc_pubsub::SubscriptionId;
|
|
|
|
|
2017-06-28 12:21:13 +02:00
|
|
|
use v1::helpers::{errors, limit_logs, Subscribers};
|
|
|
|
use v1::helpers::light_fetch::LightFetch;
|
2017-05-23 12:26:39 +02:00
|
|
|
use v1::metadata::Metadata;
|
|
|
|
use v1::traits::EthPubSub;
|
2017-06-28 12:21:13 +02:00
|
|
|
use v1::types::{pubsub, RichHeader, Log};
|
2017-05-23 12:26:39 +02:00
|
|
|
|
|
|
|
use ethcore::encoded;
|
2017-06-28 12:21:13 +02:00
|
|
|
use ethcore::filter::Filter as EthFilter;
|
2017-05-23 12:26:39 +02:00
|
|
|
use ethcore::client::{BlockChainClient, ChainNotify, BlockId};
|
2017-06-28 12:21:13 +02:00
|
|
|
use ethsync::LightSync;
|
|
|
|
use light::cache::Cache;
|
|
|
|
use light::on_demand::OnDemand;
|
2017-05-23 12:26:39 +02:00
|
|
|
use light::client::{LightChainClient, LightChainNotify};
|
|
|
|
use parity_reactor::Remote;
|
2018-01-10 13:35:18 +01:00
|
|
|
use ethereum_types::H256;
|
2017-09-06 20:47:45 +02:00
|
|
|
use bytes::Bytes;
|
2017-09-02 20:09:13 +02:00
|
|
|
use parking_lot::{RwLock, Mutex};
|
2017-06-28 12:21:13 +02:00
|
|
|
|
|
|
|
type Client = Sink<pubsub::Result>;
|
2017-05-23 12:26:39 +02:00
|
|
|
|
|
|
|
/// Eth PubSub implementation.
|
|
|
|
pub struct EthPubSubClient<C> {
|
|
|
|
handler: Arc<ChainNotificationHandler<C>>,
|
2017-06-28 12:21:13 +02:00
|
|
|
heads_subscribers: Arc<RwLock<Subscribers<Client>>>,
|
|
|
|
logs_subscribers: Arc<RwLock<Subscribers<(Client, EthFilter)>>>,
|
2018-02-16 16:51:34 +01:00
|
|
|
transactions_subscribers: Arc<RwLock<Subscribers<Client>>>,
|
2017-05-23 12:26:39 +02:00
|
|
|
}
|
|
|
|
|
|
|
|
impl<C> EthPubSubClient<C> {
|
|
|
|
/// Creates new `EthPubSubClient`.
|
|
|
|
pub fn new(client: Arc<C>, remote: Remote) -> Self {
|
2017-06-28 12:21:13 +02:00
|
|
|
let heads_subscribers = Arc::new(RwLock::new(Subscribers::default()));
|
|
|
|
let logs_subscribers = Arc::new(RwLock::new(Subscribers::default()));
|
2018-02-16 16:51:34 +01:00
|
|
|
let transactions_subscribers = Arc::new(RwLock::new(Subscribers::default()));
|
|
|
|
|
2017-05-23 12:26:39 +02:00
|
|
|
EthPubSubClient {
|
|
|
|
handler: Arc::new(ChainNotificationHandler {
|
2017-06-13 17:36:39 +02:00
|
|
|
client,
|
|
|
|
remote,
|
2017-05-23 12:26:39 +02:00
|
|
|
heads_subscribers: heads_subscribers.clone(),
|
2017-06-28 12:21:13 +02:00
|
|
|
logs_subscribers: logs_subscribers.clone(),
|
2018-02-16 16:51:34 +01:00
|
|
|
transactions_subscribers: transactions_subscribers.clone(),
|
2017-05-23 12:26:39 +02:00
|
|
|
}),
|
2017-06-13 17:36:39 +02:00
|
|
|
heads_subscribers,
|
2017-06-28 12:21:13 +02:00
|
|
|
logs_subscribers,
|
2018-02-16 16:51:34 +01:00
|
|
|
transactions_subscribers,
|
2017-05-23 12:26:39 +02:00
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2017-06-13 17:36:39 +02:00
|
|
|
/// Creates new `EthPubSubCient` with deterministic subscription ids.
|
|
|
|
#[cfg(test)]
|
|
|
|
pub fn new_test(client: Arc<C>, remote: Remote) -> Self {
|
|
|
|
let client = Self::new(client, remote);
|
2017-06-28 12:21:13 +02:00
|
|
|
*client.heads_subscribers.write() = Subscribers::new_test();
|
|
|
|
*client.logs_subscribers.write() = Subscribers::new_test();
|
2018-02-16 16:51:34 +01:00
|
|
|
*client.transactions_subscribers.write() = Subscribers::new_test();
|
2017-06-13 17:36:39 +02:00
|
|
|
client
|
|
|
|
}
|
|
|
|
|
2017-05-23 12:26:39 +02:00
|
|
|
/// Returns a chain notification handler.
|
2018-02-16 16:51:34 +01:00
|
|
|
pub fn handler(&self) -> Weak<ChainNotificationHandler<C>> {
|
|
|
|
Arc::downgrade(&self.handler)
|
2017-05-23 12:26:39 +02:00
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2017-06-28 12:21:13 +02:00
|
|
|
impl EthPubSubClient<LightFetch> {
|
|
|
|
/// Creates a new `EthPubSubClient` for `LightClient`.
|
|
|
|
pub fn light(
|
|
|
|
client: Arc<LightChainClient>,
|
|
|
|
on_demand: Arc<OnDemand>,
|
|
|
|
sync: Arc<LightSync>,
|
|
|
|
cache: Arc<Mutex<Cache>>,
|
|
|
|
remote: Remote,
|
2018-01-09 12:43:36 +01:00
|
|
|
gas_price_percentile: usize,
|
2017-06-28 12:21:13 +02:00
|
|
|
) -> Self {
|
|
|
|
let fetch = LightFetch {
|
|
|
|
client,
|
|
|
|
on_demand,
|
|
|
|
sync,
|
2018-01-09 12:43:36 +01:00
|
|
|
cache,
|
|
|
|
gas_price_percentile,
|
2017-06-28 12:21:13 +02:00
|
|
|
};
|
|
|
|
EthPubSubClient::new(Arc::new(fetch), remote)
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2017-05-23 12:26:39 +02:00
|
|
|
/// PubSub Notification handler.
|
|
|
|
pub struct ChainNotificationHandler<C> {
|
|
|
|
client: Arc<C>,
|
|
|
|
remote: Remote,
|
2017-06-28 12:21:13 +02:00
|
|
|
heads_subscribers: Arc<RwLock<Subscribers<Client>>>,
|
|
|
|
logs_subscribers: Arc<RwLock<Subscribers<(Client, EthFilter)>>>,
|
2018-02-16 16:51:34 +01:00
|
|
|
transactions_subscribers: Arc<RwLock<Subscribers<Client>>>,
|
2017-05-23 12:26:39 +02:00
|
|
|
}
|
|
|
|
|
|
|
|
impl<C> ChainNotificationHandler<C> {
|
2017-06-28 12:21:13 +02:00
|
|
|
fn notify(remote: &Remote, subscriber: &Client, result: pubsub::Result) {
|
|
|
|
remote.spawn(subscriber
|
|
|
|
.notify(Ok(result))
|
|
|
|
.map(|_| ())
|
|
|
|
.map_err(|e| warn!(target: "rpc", "Unable to send notification: {}", e))
|
|
|
|
);
|
|
|
|
}
|
|
|
|
|
|
|
|
fn notify_heads(&self, headers: &[(encoded::Header, BTreeMap<String, String>)]) {
|
|
|
|
for subscriber in self.heads_subscribers.read().values() {
|
|
|
|
for &(ref header, ref extra_info) in headers {
|
|
|
|
Self::notify(&self.remote, subscriber, pubsub::Result::Header(RichHeader {
|
|
|
|
inner: header.into(),
|
|
|
|
extra_info: extra_info.clone(),
|
|
|
|
}));
|
2017-05-23 12:26:39 +02:00
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
2017-06-28 12:21:13 +02:00
|
|
|
|
2017-10-05 12:35:01 +02:00
|
|
|
fn notify_logs<F, T>(&self, enacted: &[H256], logs: F) where
|
|
|
|
F: Fn(EthFilter) -> T,
|
|
|
|
T: IntoFuture<Item = Vec<Log>, Error = Error>,
|
|
|
|
T::Future: Send + 'static,
|
2017-06-28 12:21:13 +02:00
|
|
|
{
|
|
|
|
for &(ref subscriber, ref filter) in self.logs_subscribers.read().values() {
|
|
|
|
let logs = futures::future::join_all(enacted
|
|
|
|
.iter()
|
|
|
|
.map(|hash| {
|
|
|
|
let mut filter = filter.clone();
|
|
|
|
filter.from_block = BlockId::Hash(*hash);
|
|
|
|
filter.to_block = filter.from_block.clone();
|
2017-10-05 12:35:01 +02:00
|
|
|
logs(filter).into_future()
|
2017-06-28 12:21:13 +02:00
|
|
|
})
|
|
|
|
.collect::<Vec<_>>()
|
|
|
|
);
|
|
|
|
let limit = filter.limit;
|
|
|
|
let remote = self.remote.clone();
|
|
|
|
let subscriber = subscriber.clone();
|
|
|
|
self.remote.spawn(logs
|
|
|
|
.map(move |logs| {
|
|
|
|
let logs = logs.into_iter().flat_map(|log| log).collect();
|
2017-12-05 18:14:50 +01:00
|
|
|
|
|
|
|
for log in limit_logs(logs, limit) {
|
|
|
|
Self::notify(&remote, &subscriber, pubsub::Result::Log(log))
|
2017-06-28 12:21:13 +02:00
|
|
|
}
|
|
|
|
})
|
|
|
|
.map_err(|e| warn!("Unable to fetch latest logs: {:?}", e))
|
|
|
|
);
|
|
|
|
}
|
|
|
|
}
|
2018-02-16 16:51:34 +01:00
|
|
|
|
|
|
|
/// Notify all subscribers about new transaction hashes.
|
|
|
|
pub fn new_transactions(&self, hashes: &[H256]) {
|
|
|
|
for subscriber in self.transactions_subscribers.read().values() {
|
|
|
|
for hash in hashes {
|
|
|
|
Self::notify(&self.remote, subscriber, pubsub::Result::TransactionHash((*hash).into()));
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
2017-06-28 12:21:13 +02:00
|
|
|
}
|
|
|
|
|
|
|
|
/// A light client wrapper struct.
|
|
|
|
pub trait LightClient: Send + Sync {
|
|
|
|
/// Get a recent block header.
|
|
|
|
fn block_header(&self, id: BlockId) -> Option<encoded::Header>;
|
|
|
|
|
|
|
|
/// Fetch logs.
|
2017-11-14 11:38:17 +01:00
|
|
|
fn logs(&self, filter: EthFilter) -> BoxFuture<Vec<Log>>;
|
2017-06-28 12:21:13 +02:00
|
|
|
}
|
|
|
|
|
|
|
|
impl LightClient for LightFetch {
|
|
|
|
fn block_header(&self, id: BlockId) -> Option<encoded::Header> {
|
|
|
|
self.client.block_header(id)
|
|
|
|
}
|
|
|
|
|
2017-11-14 11:38:17 +01:00
|
|
|
fn logs(&self, filter: EthFilter) -> BoxFuture<Vec<Log>> {
|
2017-06-28 12:21:13 +02:00
|
|
|
LightFetch::logs(self, filter)
|
|
|
|
}
|
2017-05-23 12:26:39 +02:00
|
|
|
}
|
|
|
|
|
2017-06-28 12:21:13 +02:00
|
|
|
impl<C: LightClient> LightChainNotify for ChainNotificationHandler<C> {
|
2017-05-23 12:26:39 +02:00
|
|
|
fn new_headers(
|
|
|
|
&self,
|
2017-06-28 12:21:13 +02:00
|
|
|
enacted: &[H256],
|
2017-05-23 12:26:39 +02:00
|
|
|
) {
|
2017-06-28 12:21:13 +02:00
|
|
|
let headers = enacted
|
2017-05-23 12:26:39 +02:00
|
|
|
.iter()
|
|
|
|
.filter_map(|hash| self.client.block_header(BlockId::Hash(*hash)))
|
|
|
|
.map(|header| (header, Default::default()))
|
2017-06-28 12:21:13 +02:00
|
|
|
.collect::<Vec<_>>();
|
2017-05-23 12:26:39 +02:00
|
|
|
|
2017-06-28 12:21:13 +02:00
|
|
|
self.notify_heads(&headers);
|
|
|
|
self.notify_logs(&enacted, |filter| self.client.logs(filter))
|
2017-05-23 12:26:39 +02:00
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
impl<C: BlockChainClient> ChainNotify for ChainNotificationHandler<C> {
|
|
|
|
fn new_blocks(
|
|
|
|
&self,
|
|
|
|
_imported: Vec<H256>,
|
|
|
|
_invalid: Vec<H256>,
|
|
|
|
enacted: Vec<H256>,
|
2017-06-28 12:21:13 +02:00
|
|
|
retracted: Vec<H256>,
|
2017-05-23 12:26:39 +02:00
|
|
|
_sealed: Vec<H256>,
|
|
|
|
// Block bytes.
|
|
|
|
_proposed: Vec<Bytes>,
|
|
|
|
_duration: u64,
|
|
|
|
) {
|
|
|
|
const EXTRA_INFO_PROOF: &'static str = "Object exists in in blockchain (fetched earlier), extra_info is always available if object exists; qed";
|
2017-06-28 12:21:13 +02:00
|
|
|
let headers = enacted
|
|
|
|
.iter()
|
|
|
|
.filter_map(|hash| self.client.block_header(BlockId::Hash(*hash)))
|
2017-05-23 12:26:39 +02:00
|
|
|
.map(|header| {
|
|
|
|
let hash = header.hash();
|
|
|
|
(header, self.client.block_extra_info(BlockId::Hash(hash)).expect(EXTRA_INFO_PROOF))
|
|
|
|
})
|
2017-06-28 12:21:13 +02:00
|
|
|
.collect::<Vec<_>>();
|
|
|
|
|
|
|
|
// Headers
|
|
|
|
self.notify_heads(&headers);
|
|
|
|
|
|
|
|
// Enacted logs
|
|
|
|
self.notify_logs(&enacted, |filter| {
|
2017-10-05 12:35:01 +02:00
|
|
|
Ok(self.client.logs(filter).into_iter().map(Into::into).collect())
|
2017-06-28 12:21:13 +02:00
|
|
|
});
|
|
|
|
|
|
|
|
// Retracted logs
|
|
|
|
self.notify_logs(&retracted, |filter| {
|
2017-10-05 12:35:01 +02:00
|
|
|
Ok(self.client.logs(filter).into_iter().map(Into::into).map(|mut log: Log| {
|
2017-06-28 12:21:13 +02:00
|
|
|
log.log_type = "removed".into();
|
|
|
|
log
|
2017-10-05 12:35:01 +02:00
|
|
|
}).collect())
|
2017-06-28 12:21:13 +02:00
|
|
|
});
|
2017-05-23 12:26:39 +02:00
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
impl<C: Send + Sync + 'static> EthPubSub for EthPubSubClient<C> {
|
|
|
|
type Metadata = Metadata;
|
|
|
|
|
|
|
|
fn subscribe(
|
|
|
|
&self,
|
|
|
|
_meta: Metadata,
|
|
|
|
subscriber: Subscriber<pubsub::Result>,
|
|
|
|
kind: pubsub::Kind,
|
|
|
|
params: Trailing<pubsub::Params>,
|
|
|
|
) {
|
2017-08-20 06:13:00 +02:00
|
|
|
let error = match (kind, params.into()) {
|
2017-06-13 16:29:35 +02:00
|
|
|
(pubsub::Kind::NewHeads, None) => {
|
2017-08-20 06:13:00 +02:00
|
|
|
self.heads_subscribers.write().push(subscriber);
|
|
|
|
return;
|
2017-06-28 12:21:13 +02:00
|
|
|
},
|
2018-02-16 16:51:34 +01:00
|
|
|
(pubsub::Kind::NewHeads, _) => {
|
|
|
|
errors::invalid_params("newHeads", "Expected no parameters.")
|
|
|
|
},
|
2017-06-28 12:21:13 +02:00
|
|
|
(pubsub::Kind::Logs, Some(pubsub::Params::Logs(filter))) => {
|
|
|
|
self.logs_subscribers.write().push(subscriber, filter.into());
|
2017-08-20 06:13:00 +02:00
|
|
|
return;
|
|
|
|
},
|
|
|
|
(pubsub::Kind::Logs, _) => {
|
|
|
|
errors::invalid_params("logs", "Expected a filter object.")
|
2017-05-23 12:26:39 +02:00
|
|
|
},
|
2018-02-16 16:51:34 +01:00
|
|
|
(pubsub::Kind::NewPendingTransactions, None) => {
|
|
|
|
self.transactions_subscribers.write().push(subscriber);
|
|
|
|
return;
|
|
|
|
},
|
|
|
|
(pubsub::Kind::NewPendingTransactions, _) => {
|
|
|
|
errors::invalid_params("newPendingTransactions", "Expected no parameters.")
|
|
|
|
},
|
2017-05-23 12:26:39 +02:00
|
|
|
_ => {
|
2017-08-20 06:13:00 +02:00
|
|
|
errors::unimplemented(None)
|
2017-05-23 12:26:39 +02:00
|
|
|
},
|
2017-08-20 06:13:00 +02:00
|
|
|
};
|
|
|
|
|
|
|
|
let _ = subscriber.reject(error);
|
2017-05-23 12:26:39 +02:00
|
|
|
}
|
|
|
|
|
2017-11-14 11:38:17 +01:00
|
|
|
fn unsubscribe(&self, id: SubscriptionId) -> Result<bool> {
|
2017-06-28 12:21:13 +02:00
|
|
|
let res = self.heads_subscribers.write().remove(&id).is_some();
|
|
|
|
let res2 = self.logs_subscribers.write().remove(&id).is_some();
|
2018-02-16 16:51:34 +01:00
|
|
|
let res3 = self.transactions_subscribers.write().remove(&id).is_some();
|
2017-06-28 12:21:13 +02:00
|
|
|
|
2018-02-16 16:51:34 +01:00
|
|
|
Ok(res || res2 || res3)
|
2017-05-23 12:26:39 +02:00
|
|
|
}
|
|
|
|
}
|