Get contract events and make Transport::Out: Unpin (#356)

* Query contract events. * Parse logs. * Add events function.
tomusdrw · Jun 16, 2020 · befcb2f · befcb2f
2 parents 2144d13 + 3975e6d
commit befcb2f
Show file tree

Hide file tree

Showing 9 changed files with 79 additions and 60 deletions.
diff --git a/examples/bench.rs b/examples/bench.rs
@@ -18,7 +18,7 @@ async fn main() -> web3::Result {
 
 fn bench<T: web3::Transport>(id: &str, transport: T, max: usize)
 where
-    T::Out: Send + Unpin + 'static,
+    T::Out: Send + 'static,
 {
     use futures::FutureExt;
 

diff --git a/src/api/accounts.rs b/src/api/accounts.rs
@@ -16,7 +16,6 @@ use rlp::RlpStream;
 use secp256k1::key::ONE_KEY;
 use secp256k1::{Message, PublicKey, Secp256k1, SecretKey};
 use std::convert::TryInto;
-use std::marker::Unpin;
 use std::mem;
 use std::ops::Deref;
 use std::pin::Pin;
@@ -49,10 +48,7 @@ impl<T: Transport> Accounts<T> {
     }
 
     /// Signs an Ethereum transaction with a given private key.
-    pub fn sign_transaction(&self, tx: TransactionParameters, key: &SecretKey) -> SignTransactionFuture<T>
-    where
-        T::Out: Unpin,
-    {
+    pub fn sign_transaction(&self, tx: TransactionParameters, key: &SecretKey) -> SignTransactionFuture<T> {
         SignTransactionFuture::new(self, tx, key)
     }
 
@@ -179,19 +175,13 @@ type TxParams<T> = Join3<MaybeReady<T, U256>, MaybeReady<T, U256>, MaybeReady<T,
 /// parameters required for signing `nonce`, `gas_price` and `chain_id`. Note
 /// that if all transaction parameters were provided, this future will resolve
 /// immediately.
-pub struct SignTransactionFuture<T: Transport>
-where
-    T::Out: Unpin,
-{
+pub struct SignTransactionFuture<T: Transport> {
     tx: TransactionParameters,
     key: ZeroizeSecretKey,
     inner: TxParams<T>,
 }
 
-impl<T: Transport> SignTransactionFuture<T>
-where
-    T::Out: Unpin,
-{
+impl<T: Transport> SignTransactionFuture<T> {
     /// Creates a new SignTransactionFuture with accounts and transaction data.
     pub fn new(accounts: &Accounts<T>, tx: TransactionParameters, key: &SecretKey) -> SignTransactionFuture<T> {
         macro_rules! maybe {
@@ -218,10 +208,7 @@ where
     }
 }
 
-impl<T: Transport> Future for SignTransactionFuture<T>
-where
-    T::Out: Unpin,
-{
+impl<T: Transport> Future for SignTransactionFuture<T> {
     type Output = error::Result<SignedTransaction>;
 
     fn poll(mut self: Pin<&mut Self>, ctx: &mut Context) -> Poll<Self::Output> {
@@ -243,10 +230,7 @@ where
     }
 }
 
-impl<T: Transport> Drop for SignTransactionFuture<T>
-where
-    T::Out: Unpin,
-{
+impl<T: Transport> Drop for SignTransactionFuture<T> {
     fn drop(&mut self) {
         self.key.zeroize();
     }

diff --git a/src/api/eth_filter.rs b/src/api/eth_filter.rs
@@ -67,10 +67,7 @@ enum FilterStreamState<I, O> {
     NextItem(vec::IntoIter<I>),
 }
 
-impl<T: Transport, I: DeserializeOwned + Unpin> Stream for FilterStream<T, I>
-where
-    T::Out: Unpin,
-{
+impl<T: Transport, I: DeserializeOwned + Unpin> Stream for FilterStream<T, I> {
     type Item = error::Result<I>;
 
     fn poll_next(mut self: Pin<&mut Self>, ctx: &mut Context) -> Poll<Option<Self::Item>> {
@@ -232,7 +229,6 @@ pub struct CreateFilter<T: Transport, I> {
 impl<T, I> Future for CreateFilter<T, I>
 where
     T: Transport,
-    T::Out: Unpin,
     I: Unpin,
 {
     type Output = error::Result<BaseFilter<T, I>>;

diff --git a/src/api/eth_subscribe.rs b/src/api/eth_subscribe.rs
@@ -81,7 +81,6 @@ impl<T: DuplexTransport, I> SubscriptionStream<T, I> {
 impl<T, I> Stream for SubscriptionStream<T, I>
 where
     T: DuplexTransport,
-    T::Out: Unpin,
     T::NotificationStream: Unpin,
     I: serde::de::DeserializeOwned + Unpin,
 {
@@ -122,7 +121,6 @@ impl<T, I> Future for SubscriptionResult<T, I>
 where
     T: DuplexTransport,
     I: serde::de::DeserializeOwned + Unpin,
-    T::Out: Unpin,
 {
     type Output = error::Result<SubscriptionStream<T, I>>;
 

diff --git a/src/confirm.rs b/src/confirm.rs
@@ -54,7 +54,6 @@ where
 impl<T, V, F> Future for WaitForConfirmations<T, V, F>
 where
     T: Transport,
-    T::Out: Unpin,
     V: ConfirmationCheck<Check = F> + Unpin,
     F: Future<Output = error::Result<Option<U64>>> + Unpin,
 {
@@ -126,7 +125,6 @@ impl<T: Transport, V, F> Confirmations<T, V, F> {
 impl<T, V, F> Future for Confirmations<T, V, F>
 where
     T: Transport,
-    T::Out: Unpin,
     V: ConfirmationCheck<Check = F> + Unpin,
     F: Future<Output = error::Result<Option<U64>>> + Unpin,
 {
@@ -176,10 +174,7 @@ struct TransactionReceiptBlockNumber<T: Transport> {
     future: CallFuture<Option<TransactionReceipt>, T::Out>,
 }
 
-impl<T: Transport> Future for TransactionReceiptBlockNumber<T>
-where
-    T::Out: Unpin,
-{
+impl<T: Transport> Future for TransactionReceiptBlockNumber<T> {
     type Output = error::Result<Option<U64>>;
 
     fn poll(mut self: Pin<&mut Self>, ctx: &mut Context) -> Poll<Self::Output> {
@@ -199,10 +194,7 @@ impl<T: Transport> TransactionReceiptBlockNumberCheck<T> {
     }
 }
 
-impl<T: Transport> ConfirmationCheck for TransactionReceiptBlockNumberCheck<T>
-where
-    T::Out: Unpin,
-{
+impl<T: Transport> ConfirmationCheck for TransactionReceiptBlockNumberCheck<T> {
     type Check = TransactionReceiptBlockNumber<T>;
 
     fn check(&self) -> Self::Check {
@@ -259,10 +251,7 @@ impl<T: Transport> SendTransactionWithConfirmation<T> {
     }
 }
 
-impl<T: Transport> Future for SendTransactionWithConfirmation<T>
-where
-    T::Out: Unpin,
-{
+impl<T: Transport> Future for SendTransactionWithConfirmation<T> {
     type Output = error::Result<TransactionReceipt>;
 
     fn poll(mut self: Pin<&mut Self>, ctx: &mut Context) -> Poll<Self::Output> {

diff --git a/src/contract/deploy.rs b/src/contract/deploy.rs
@@ -55,7 +55,6 @@ impl<T: Transport> Builder<T> {
     where
         P: Tokenize,
         V: AsRef<str>,
-        T::Out: Unpin,
     {
         let transport = self.eth.transport().clone();
         let poll_interval = self.poll_interval;
@@ -81,7 +80,6 @@ impl<T: Transport> Builder<T> {
     where
         P: Tokenize,
         V: AsRef<str>,
-        T::Out: Unpin,
     {
         let transport = self.eth.transport().clone();
         let poll_interval = self.poll_interval;
@@ -176,7 +174,6 @@ impl<T, F> Future for PendingContract<T, F>
 where
     F: Future<Output = error::Result<TransactionReceipt>> + Unpin,
     T: Transport,
-    T::Out: Unpin,
 {
     type Output = Result<Contract<T>, Error>;
 

diff --git a/src/contract/mod.rs b/src/contract/mod.rs
@@ -1,16 +1,17 @@
 //! Ethereum Contract Interface
 
-use ethabi;
-
 use crate::api::{Accounts, Eth, Namespace};
 use crate::confirm;
 use crate::contract::tokens::{Detokenize, Tokenize};
 use crate::types::{
-    Address, BlockId, Bytes, CallRequest, TransactionCondition, TransactionParameters, TransactionReceipt,
-    TransactionRequest, H256, U256,
+    Address, BlockId, Bytes, CallRequest, FilterBuilder, TransactionCondition, TransactionParameters,
+    TransactionReceipt, TransactionRequest, H256, U256,
 };
 use crate::Transport;
-use futures::future::Either;
+use futures::{
+    future::{self, Either},
+    Future, FutureExt, TryFutureExt,
+};
 use secp256k1::key::SecretKey;
 use std::{collections::HashMap, hash::Hash, time};
 
@@ -159,12 +160,7 @@ impl<T: Transport> Contract<T> {
         options: Options,
         confirmations: usize,
         key: &'a SecretKey,
-    ) -> impl futures::Future<Output = crate::Result<TransactionReceipt>> + 'a
-    where
-        T::Out: Unpin,
-    {
-        use futures::TryFutureExt;
-
+    ) -> impl Future<Output = crate::Result<TransactionReceipt>> + 'a {
         let poll_interval = time::Duration::from_secs(1);
 
         self.abi
@@ -200,7 +196,7 @@ impl<T: Transport> Contract<T> {
                 // TODO [ToDr] SendTransactionWithConfirmation should support custom error type (so that we can return
                 // `contract::Error` instead of more generic `Error`.
                 let err = crate::error::Error::Decoder(format!("{:?}", e));
-                Either::Right(futures::future::ready(Err(err)))
+                Either::Right(future::ready(Err(err)))
             })
     }
 
@@ -311,6 +307,65 @@ impl<T: Transport> Contract<T> {
             })
             .unwrap_or_else(Into::into)
     }
+
+    /// Find events matching the topics.
+    pub fn events<A, B, C, R>(
+        &self,
+        event: &str,
+        topic0: A,
+        topic1: B,
+        topic2: C,
+    ) -> impl Future<Output = Result<Vec<R>>>
+    where
+        A: Tokenize,
+        B: Tokenize,
+        C: Tokenize,
+        R: Detokenize,
+    {
+        fn to_topic<A: Tokenize>(x: A) -> ethabi::Topic<ethabi::Token> {
+            let tokens = x.into_tokens();
+            if tokens.is_empty() {
+                ethabi::Topic::Any
+            } else {
+                tokens.into()
+            }
+        }
+
+        let res = self.abi.event(event).and_then(|ev| {
+            let filter = ev.filter(ethabi::RawTopicFilter {
+                topic0: to_topic(topic0),
+                topic1: to_topic(topic1),
+                topic2: to_topic(topic2),
+            })?;
+            Ok((ev.clone(), filter))
+        });
+        let (ev, filter) = match res {
+            Ok(x) => x,
+            Err(e) => return Either::Left(future::ready(Err(e.into()))),
+        };
+
+        Either::Right(
+            self.eth
+                .logs(FilterBuilder::default().topic_filter(filter).build())
+                .map_err(Into::into)
+                .map(move |logs| {
+                    logs.and_then(|logs| {
+                        logs.into_iter()
+                            .map(move |l| {
+                                let log = ev.parse_log(ethabi::RawLog {
+                                    topics: l.topics,
+                                    data: l.data.0,
+                                })?;
+
+                                Ok(R::from_tokens(
+                                    log.params.into_iter().map(|x| x.value).collect::<Vec<_>>(),
+                                )?)
+                            })
+                            .collect::<Result<Vec<R>>>()
+                    })
+                }),
+        )
+    }
 }
 
 #[cfg(test)]

diff --git a/src/lib.rs b/src/lib.rs
@@ -41,7 +41,7 @@ pub type RequestId = usize;
 /// Transport implementation
 pub trait Transport: std::fmt::Debug + Clone + Unpin {
     /// The type of future this transport returns when a call is made.
-    type Out: futures::Future<Output = error::Result<rpc::Value>>;
+    type Out: futures::Future<Output = error::Result<rpc::Value>> + Unpin;
 
     /// Prepare serializable RPC call for given method with parameters.
     fn prepare(&self, method: &str, params: Vec<rpc::Value>) -> (RequestId, rpc::Call);

diff --git a/src/types/traces.rs b/src/types/traces.rs
@@ -1,8 +1,8 @@
 //! Types for the Parity Ad-Hoc Trace API
-use std::collections::BTreeMap;
 
 use crate::types::{Action, ActionType, Bytes, Res, H160, H256, U256};
 use serde::{Deserialize, Serialize};
+use std::collections::BTreeMap;
 
 #[derive(Debug, Clone, Serialize)]
 /// Description of the type of trace to make