use crate::{
ens,
http::Provider as HttpProvider,
stream::{FilterStream, FilterWatcher},
JsonRpcClient, PendingTransaction,
};
use ethers_core::{
abi::{self, Detokenize, ParamType},
types::{
Address, Block, BlockId, BlockNumber, Bytes, Filter, Log, NameOrAddress, Selector,
Signature, Transaction, TransactionReceipt, TransactionRequest, TxHash, H256, U256, U64,
},
utils,
};
use serde::Deserialize;
use thiserror::Error;
use url::{ParseError, Url};
use std::{convert::TryFrom, fmt::Debug};
#[derive(Clone, Debug)]
pub struct Provider<P>(P, Option<Address>);
#[derive(Debug, Error)]
pub enum ProviderError {
#[error(transparent)]
JsonRpcClientError(#[from] Box<dyn std::error::Error + Send + Sync>),
#[error("ens name not found: {0}")]
EnsError(String),
}
#[derive(Clone, Debug)]
pub enum FilterKind<'a> {
Logs(&'a Filter),
NewBlocks,
PendingTransactions,
}
impl<P: JsonRpcClient> Provider<P> {
pub fn new(provider: P) -> Self {
Self(provider, None)
}
pub async fn get_block_number(&self) -> Result<U64, ProviderError> {
Ok(self
.0
.request("eth_blockNumber", ())
.await
.map_err(Into::into)?)
}
pub async fn get_block(
&self,
block_hash_or_number: impl Into<BlockId>,
) -> Result<Block<TxHash>, ProviderError> {
Ok(self
.get_block_gen(block_hash_or_number.into(), false)
.await?)
}
pub async fn get_block_with_txs(
&self,
block_hash_or_number: impl Into<BlockId>,
) -> Result<Block<Transaction>, ProviderError> {
Ok(self
.get_block_gen(block_hash_or_number.into(), true)
.await?)
}
async fn get_block_gen<Tx: for<'a> Deserialize<'a>>(
&self,
id: BlockId,
include_txs: bool,
) -> Result<Block<Tx>, ProviderError> {
let include_txs = utils::serialize(&include_txs);
Ok(match id {
BlockId::Hash(hash) => {
let hash = utils::serialize(&hash);
self.0
.request("eth_getBlockByHash", [hash, include_txs])
.await
.map_err(Into::into)?
}
BlockId::Number(num) => {
let num = utils::serialize(&num);
self.0
.request("eth_getBlockByNumber", [num, include_txs])
.await
.map_err(Into::into)?
}
})
}
pub async fn get_transaction<T: Send + Sync + Into<TxHash>>(
&self,
transaction_hash: T,
) -> Result<Transaction, ProviderError> {
let hash = transaction_hash.into();
Ok(self
.0
.request("eth_getTransactionByHash", [hash])
.await
.map_err(Into::into)?)
}
pub async fn get_transaction_receipt<T: Send + Sync + Into<TxHash>>(
&self,
transaction_hash: T,
) -> Result<TransactionReceipt, ProviderError> {
let hash = transaction_hash.into();
Ok(self
.0
.request("eth_getTransactionReceipt", [hash])
.await
.map_err(Into::into)?)
}
pub async fn get_gas_price(&self) -> Result<U256, ProviderError> {
Ok(self
.0
.request("eth_gasPrice", ())
.await
.map_err(Into::into)?)
}
pub async fn get_accounts(&self) -> Result<Vec<Address>, ProviderError> {
Ok(self
.0
.request("eth_accounts", ())
.await
.map_err(Into::into)?)
}
pub async fn get_transaction_count(
&self,
from: impl Into<NameOrAddress>,
block: Option<BlockNumber>,
) -> Result<U256, ProviderError> {
let from = match from.into() {
NameOrAddress::Name(ens_name) => self.resolve_name(&ens_name).await?,
NameOrAddress::Address(addr) => addr,
};
let from = utils::serialize(&from);
let block = utils::serialize(&block.unwrap_or(BlockNumber::Latest));
Ok(self
.0
.request("eth_getTransactionCount", [from, block])
.await
.map_err(Into::into)?)
}
pub async fn get_balance(
&self,
from: impl Into<NameOrAddress>,
block: Option<BlockNumber>,
) -> Result<U256, ProviderError> {
let from = match from.into() {
NameOrAddress::Name(ens_name) => self.resolve_name(&ens_name).await?,
NameOrAddress::Address(addr) => addr,
};
let from = utils::serialize(&from);
let block = utils::serialize(&block.unwrap_or(BlockNumber::Latest));
Ok(self
.0
.request("eth_getBalance", [from, block])
.await
.map_err(Into::into)?)
}
pub async fn get_chainid(&self) -> Result<U256, ProviderError> {
Ok(self
.0
.request("eth_chainId", ())
.await
.map_err(Into::into)?)
}
pub async fn call(
&self,
tx: &TransactionRequest,
block: Option<BlockNumber>,
) -> Result<Bytes, ProviderError> {
let tx = utils::serialize(tx);
let block = utils::serialize(&block.unwrap_or(BlockNumber::Latest));
Ok(self
.0
.request("eth_call", [tx, block])
.await
.map_err(Into::into)?)
}
pub async fn estimate_gas(&self, tx: &TransactionRequest) -> Result<U256, ProviderError> {
let tx = utils::serialize(tx);
Ok(self
.0
.request("eth_estimateGas", [tx])
.await
.map_err(Into::into)?)
}
pub async fn send_transaction(
&self,
mut tx: TransactionRequest,
) -> Result<PendingTransaction<'_, P>, ProviderError> {
if let Some(ref to) = tx.to {
if let NameOrAddress::Name(ens_name) = to {
let addr = self.resolve_name(&ens_name).await?;
tx.to = Some(addr.into())
}
}
let tx_hash = self
.0
.request("eth_sendTransaction", [tx])
.await
.map_err(Into::into)?;
Ok(PendingTransaction::new(tx_hash, self))
}
pub async fn send_raw_transaction(
&self,
tx: &Transaction,
) -> Result<PendingTransaction<'_, P>, ProviderError> {
let rlp = utils::serialize(&tx.rlp());
let tx_hash = self
.0
.request("eth_sendRawTransaction", [rlp])
.await
.map_err(Into::into)?;
Ok(PendingTransaction::new(tx_hash, self))
}
pub async fn sign<T: Into<Bytes>>(
&self,
data: T,
from: &Address,
) -> Result<Signature, ProviderError> {
let data = utils::serialize(&data.into());
let from = utils::serialize(from);
Ok(self
.0
.request("eth_sign", [from, data])
.await
.map_err(Into::into)?)
}
pub async fn get_logs(&self, filter: &Filter) -> Result<Vec<Log>, ProviderError> {
Ok(self
.0
.request("eth_getLogs", [filter])
.await
.map_err(Into::into)?)
}
pub async fn watch(
&self,
filter: &Filter,
) -> Result<impl FilterStream<Log> + '_, ProviderError> {
let id = self.new_filter(FilterKind::Logs(filter)).await?;
let fut = move || Box::pin(self.get_filter_changes(id));
Ok(FilterWatcher::new(id, fut))
}
pub async fn watch_blocks(&self) -> Result<impl FilterStream<H256> + '_, ProviderError> {
let id = self.new_filter(FilterKind::NewBlocks).await?;
let fut = move || Box::pin(self.get_filter_changes(id));
Ok(FilterWatcher::new(id, fut))
}
pub async fn watch_pending_transactions(
&self,
) -> Result<impl FilterStream<H256> + '_, ProviderError> {
let id = self.new_filter(FilterKind::PendingTransactions).await?;
let fut = move || Box::pin(self.get_filter_changes(id));
Ok(FilterWatcher::new(id, fut))
}
pub async fn new_filter(&self, filter: FilterKind<'_>) -> Result<U256, ProviderError> {
let (method, args) = match filter {
FilterKind::NewBlocks => ("eth_newBlockFilter", vec![]),
FilterKind::PendingTransactions => ("eth_newPendingTransactionFilter", vec![]),
FilterKind::Logs(filter) => ("eth_newFilter", vec![utils::serialize(&filter)]),
};
Ok(self.0.request(method, args).await.map_err(Into::into)?)
}
pub async fn uninstall_filter<T: Into<U256>>(&self, id: T) -> Result<bool, ProviderError> {
let id = utils::serialize(&id.into());
Ok(self
.0
.request("eth_uninstallFilter", [id])
.await
.map_err(Into::into)?)
}
pub async fn get_filter_changes<T, R>(&self, id: T) -> Result<Vec<R>, ProviderError>
where
T: Into<U256>,
R: for<'a> Deserialize<'a>,
{
let id = utils::serialize(&id.into());
Ok(self
.0
.request("eth_getFilterChanges", [id])
.await
.map_err(Into::into)?)
}
pub async fn get_code(
&self,
at: impl Into<NameOrAddress>,
block: Option<BlockNumber>,
) -> Result<Bytes, ProviderError> {
let at = match at.into() {
NameOrAddress::Name(ens_name) => self.resolve_name(&ens_name).await?,
NameOrAddress::Address(addr) => addr,
};
let at = utils::serialize(&at);
let block = utils::serialize(&block.unwrap_or(BlockNumber::Latest));
Ok(self
.0
.request("eth_getCode", [at, block])
.await
.map_err(Into::into)?)
}
pub async fn resolve_name(&self, ens_name: &str) -> Result<Address, ProviderError> {
self.query_resolver(ParamType::Address, ens_name, ens::ADDR_SELECTOR)
.await
}
pub async fn lookup_address(&self, address: Address) -> Result<String, ProviderError> {
let ens_name = ens::reverse_address(address);
self.query_resolver(ParamType::String, &ens_name, ens::NAME_SELECTOR)
.await
}
async fn query_resolver<T: Detokenize>(
&self,
param: ParamType,
ens_name: &str,
selector: Selector,
) -> Result<T, ProviderError> {
let ens_addr = self.1.unwrap_or(ens::ENS_ADDRESS);
let data = self
.call(&ens::get_resolver(ens_addr, ens_name), None)
.await?;
let resolver_address: Address = decode_bytes(ParamType::Address, data);
if resolver_address == Address::zero() {
return Err(ProviderError::EnsError(ens_name.to_owned()));
}
let data = self
.call(&ens::resolve(resolver_address, selector, ens_name), None)
.await?;
Ok(decode_bytes(param, data))
}
#[cfg(test)]
pub async fn mine(&self, num_blocks: usize) -> Result<(), ProviderError> {
for _ in 0..num_blocks {
self.0
.request::<_, U256>("evm_mine", None::<()>)
.await
.map_err(Into::into)?;
}
Ok(())
}
pub fn ens<T: Into<Address>>(mut self, ens: T) -> Self {
self.1 = Some(ens.into());
self
}
}
fn decode_bytes<T: Detokenize>(param: ParamType, bytes: Bytes) -> T {
let tokens =
abi::decode(&[param], &bytes.0).expect("could not abi-decode bytes to address tokens");
T::from_tokens(tokens).expect("could not parse tokens as address")
}
impl TryFrom<&str> for Provider<HttpProvider> {
type Error = ParseError;
fn try_from(src: &str) -> Result<Self, Self::Error> {
Ok(Provider(HttpProvider::new(Url::parse(src)?), None))
}
}
#[cfg(test)]
mod ens_tests {
use super::*;
const INFURA: &str = "https://mainnet.infura.io/v3/c60b0bb42f8a4c6481ecd229eddaca27";
#[tokio::test]
async fn mainnet_resolve_name() {
let provider = Provider::<HttpProvider>::try_from(INFURA).unwrap();
let addr = provider
.resolve_name("registrar.firefly.eth")
.await
.unwrap();
assert_eq!(
addr,
"6fC21092DA55B392b045eD78F4732bff3C580e2c".parse().unwrap()
);
provider.resolve_name("asdfasdffads").await.unwrap_err();
provider
.resolve_name("asdfasdf.registrar.firefly.eth")
.await
.unwrap_err();
}
#[tokio::test]
async fn mainnet_lookup_address() {
let provider = Provider::<HttpProvider>::try_from(INFURA).unwrap();
let name = provider
.lookup_address("6fC21092DA55B392b045eD78F4732bff3C580e2c".parse().unwrap())
.await
.unwrap();
assert_eq!(name, "registrar.firefly.eth");
provider
.lookup_address("AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA".parse().unwrap())
.await
.unwrap_err();
}
}
#[cfg(test)]
mod tests {
use super::*;
use ethers_core::types::H256;
use futures_util::StreamExt;
#[tokio::test]
#[ignore]
async fn test_new_block_filter() {
let num_blocks = 3;
let provider = Provider::<HttpProvider>::try_from("http://localhost:8545").unwrap();
let start_block = provider.get_block_number().await.unwrap();
let stream = provider
.watch_blocks()
.await
.unwrap()
.interval(1000u64)
.stream();
let hashes: Vec<H256> = stream.take(num_blocks).collect::<Vec<H256>>().await;
for (i, hash) in hashes.iter().enumerate() {
let block = provider
.get_block(start_block + i as u64 + 1)
.await
.unwrap();
assert_eq!(*hash, block.hash.unwrap());
}
}
#[tokio::test]
#[ignore]
async fn test_new_pending_txs_filter() {
let num_txs = 5;
let provider = Provider::<HttpProvider>::try_from("http://localhost:8545").unwrap();
let accounts = provider.get_accounts().await.unwrap();
let stream = provider
.watch_pending_transactions()
.await
.unwrap()
.interval(1000u64)
.stream();
let mut tx_hashes = Vec::new();
let tx = TransactionRequest::new()
.from(accounts[0])
.to(accounts[0])
.value(1e18 as u64);
for _ in 0..num_txs {
tx_hashes.push(provider.send_transaction(tx.clone()).await.unwrap());
}
let hashes: Vec<H256> = stream.take(num_txs).collect::<Vec<H256>>().await;
assert_eq!(tx_hashes, hashes);
}
}