ethers-rs v0.1.3 <> v0.2.0
112 lines
use async_trait::async_trait;
use async_trait::async_trait;
use ethers_core::types::*;
use ethers_core::types::*;
use ethers_providers::{FromErr, Middleware};
use ethers_providers::{FromErr, Middleware, PendingTransaction};
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use thiserror::Error;
use thiserror::Error;
#[derive(Debug)]
#[derive(Debug)]
pub struct NonceManager<M> {
/// Middleware used for calculating nonces locally, useful for signing multiple
pub inner: M,
/// consecutive transactions without waiting for them to hit the mempool
pub initialized: AtomicBool,
pub struct NonceManagerMiddleware<M> {
pub nonce: AtomicU64,
inner: M,
pub address: Address,
initialized: AtomicBool,
nonce: AtomicU64,
address: Address,
}
}
impl<M> NonceManager<M>
impl<M> NonceManagerMiddleware<M>
where
where
M: Middleware,
M: Middleware,
{
{
/// Instantiates the nonce manager with a 0 nonce.
/// Instantiates the nonce manager with a 0 nonce. The `address` should be the
/// address which you'll be sending transactions from
pub fn new(inner: M, address: Address) -> Self {
pub fn new(inner: M, address: Address) -> Self {
NonceManager {
Self {
initialized: false.into(),
initialized: false.into(),
nonce: 0.into(),
nonce: 0.into(),
inner,
inner,
address,
address,
}
}
}
}
/// Returns the next nonce to be used
/// Returns the next nonce to be used
pub fn next(&self) -> U256 {
pub fn next(&self) -> U256 {
let nonce = self.nonce.fetch_add(1, Ordering::SeqCst);
let nonce = self.nonce.fetch_add(1, Ordering::SeqCst);
nonce.into()
nonce.into()
}
}
async fn get_transaction_count_with_manager(
async fn get_transaction_count_with_manager(
&self,
&self,
block: Option<BlockNumber>,
block: Option<BlockNumber>,
) -> Result<U256, NonceManagerError<M>> {
) -> Result<U256, NonceManagerError<M>> {
// initialize the nonce the first time the manager is called
// initialize the nonce the first time the manager is called
if !self.initialized.load(Ordering::SeqCst) {
if !self.initialized.load(Ordering::SeqCst) {
let nonce = self
let nonce = self
.inner
.inner
.get_transaction_count(self.address, block)
.get_transaction_count(self.address, block)
.await
.await
.map_err(FromErr::from)?;
.map_err(FromErr::from)?;
self.nonce.store(nonce.as_u64(), Ordering::SeqCst);
self.nonce.store(nonce.as_u64(), Ordering::SeqCst);
self.initialized.store(true, Ordering::SeqCst);
self.initialized.store(true, Ordering::SeqCst);
}
}
Ok(self.next())
Ok(self.next())
}
}
}
}
#[derive(Error, Debug)]
#[derive(Error, Debug)]
/// Thrown when an error happens at the Nonce Manager
pub enum NonceManagerError<M: Middleware> {
pub enum NonceManagerError<M: Middleware> {
/// Thrown when the internal middleware errors
#[error("{0}")]
#[error("{0}")]
MiddlewareError(M::Error),
MiddlewareError(M::Error),
}
}
impl<M: Middleware> FromErr<M::Error> for NonceManagerError<M> {
impl<M: Middleware> FromErr<M::Error> for NonceManagerError<M> {
fn from(src: M::Error) -> Self {
fn from(src: M::Error) -> Self {
NonceManagerError::MiddlewareError(src)
NonceManagerError::MiddlewareError(src)
}
}
}
}
#[async_trait(?Send)]
#[async_trait]
impl<M> Middleware for NonceManager<M>
impl<M> Middleware for NonceManagerMiddleware<M>
where
where
M: Middleware,
M: Middleware,
{
{
type Error = NonceManagerError<M>;
type Error = NonceManagerError<M>;
type Provider = M::Provider;
type Provider = M::Provider;
type Inner = M;
type Inner = M;
fn inner(&self) -> &M {
fn inner(&self) -> &M {
&self.inner
&self.inner
}
}
/// Signs and broadcasts the transaction. The optional parameter `block` can be passed so that
/// Signs and broadcasts the transaction. The optional parameter `block` can be passed so that
/// gas cost and nonce calculations take it into account. For simple transactions this can be
/// gas cost and nonce calculations take it into account. For simple transactions this can be
/// left to `None`.
/// left to `None`.
async fn send_transaction(
async fn send_transaction(
&self,
&self,
mut tx: TransactionRequest,
mut tx: TransactionRequest,
block: Option<BlockNumber>,
block: Option<BlockNumber>,
) -> Result<TxHash, Self::Error> {
) -> Result<PendingTransaction<'_, Self::Provider>, Self::Error> {
if tx.nonce.is_none() {
if tx.nonce.is_none() {
tx.nonce = Some(self.get_transaction_count_with_manager(block).await?);
tx.nonce = Some(self.get_transaction_count_with_manager(block).await?);
}
}
let mut tx_clone = tx.clone();
let mut tx_clone = tx.clone();
match self.inner.send_transaction(tx, block).await {
match self.inner.send_transaction(tx, block).await {
Ok(tx_hash) => Ok(tx_hash),
Ok(tx_hash) => Ok(tx_hash),
Err(err) => {
Err(err) => {
let nonce = self.get_transaction_count(self.address, block).await?;
let nonce = self.get_transaction_count(self.address, block).await?;
if nonce != self.nonce.load(Ordering::SeqCst).into() {
if nonce != self.nonce.load(Ordering::SeqCst).into() {
// try re-submitting the transaction with the correct nonce if there
// try re-submitting the transaction with the correct nonce if there
// was a nonce mismatch
// was a nonce mismatch
self.nonce.store(nonce.as_u64(), Ordering::SeqCst);
self.nonce.store(nonce.as_u64(), Ordering::SeqCst);
tx_clone.nonce = Some(nonce);
tx_clone.nonce = Some(nonce);
self.inner
self.inner
.send_transaction(tx_clone, block)
.send_transaction(tx_clone, block)
.await
.await
.map_err(FromErr::from)
.map_err(FromErr::from)
} else {
} else {
// propagate the error otherwise
// propagate the error otherwise
Err(FromErr::from(err))
Err(FromErr::from(err))
}
}
}
}
}
}
}
}
}
}