aleph-client: Fetch contract events

aleph-client/Cargo.toml

@@ -1,7 +1,7 @@
 [package]
 name = "aleph_client"
 # TODO bump major version when API stablize
-version = "2.9.1"
+version = "2.10.0"
 edition = "2021"
 license = "Apache 2.0"
 

aleph-client/src/connections.rs

@@ -113,7 +113,9 @@ pub trait ConnectionApi: Sync {
 /// Data regarding submitted transaction.
 #[derive(Copy, Clone, Eq, PartialEq, Debug, Deserialize, Serialize)]
 pub struct TxInfo {
+    /// Hash of the block containing tx.
     pub block_hash: BlockHash,
+    /// Hash of the transaction itself.
     pub tx_hash: TxHash,
 }
 

aleph-client/src/contract/event.rs

@@ -1,8 +1,8 @@
 //! Utilities for listening for contract events.
 //!
-//! To use the module you will need to pass a connection, some contracts and an `UnboundedSender` to the
-//! [listen_contract_events] function. You most likely want to `tokio::spawn` the resulting future, so that it runs
-//! concurrently.
+//! To use the listening feature you will need to pass a connection, some contracts and an
+//! `UnboundedSender` to the [listen_contract_events] function. You most likely want to
+//! `tokio::spawn` the resulting future, so that it runs concurrently.
 //!
 //! ```no_run
 //! # use std::sync::Arc;
@@ -41,14 +41,39 @@
 //! #   Ok(())
 //! # }
 //! ```
+//!
+//! To use the fetching feature you will need to pass a connection, transaction coordinates and
+//! a corresponding contract to the [get_contract_events] function.
+//!
+//! ```no_run
+//! # use aleph_client::{AccountId, Connection, SignedConnection};
+//! # use aleph_client::contract::ContractInstance;
+//! # use aleph_client::contract::event::{get_contract_events, listen_contract_events};
+//! # use anyhow::Result;
+//! use futures::{channel::mpsc::unbounded, StreamExt};
+//!
+//! # async fn example(conn: Connection, signed_conn: SignedConnection, address: AccountId, path: &str) -> Result<()> {
+//! let contract = ContractInstance::new(address, path)?;
+//!
+//! let tx_info = contract.contract_exec0(&signed_conn, "some_method").await?;
+//!
+//! println!("Received events {:?}", get_contract_events(&conn, &contract, tx_info).await);
+//!
+//! #   Ok(())
+//! # }
+//! ```
 
-use std::collections::HashMap;
+use std::{collections::HashMap, error::Error};
 
-use anyhow::{bail, Result};
+use anyhow::{anyhow, bail, Result};
 use contract_transcode::Value;
 use futures::{channel::mpsc::UnboundedSender, StreamExt};
+use subxt::events::EventDetails;
 
-use crate::{contract::ContractInstance, AccountId, Connection};
+use crate::{
+    api::contracts::events::ContractEmitted, connections::TxInfo, contract::ContractInstance,
+    utility::BlocksApi, AccountId, Connection,
+};
 
 /// Represents a single event emitted by a contract.
 #[derive(Debug, Clone, Eq, PartialEq)]
@@ -61,6 +86,18 @@ pub struct ContractEvent {
     pub data: HashMap<String, Value>,
 }
 
+/// Fetch all events that corresponds to the contract call identified by `tx_info`.
+pub async fn get_contract_events(
+    conn: &Connection,
+    contract: &ContractInstance,
+    tx_info: TxInfo,
+) -> Result<Vec<ContractEvent>> {
+    let events = conn.get_tx_events(tx_info).await?;
+    translate_events(events.iter(), &[contract])
+        .into_iter()
+        .collect()
+}
+
 /// Starts an event listening loop.
 ///
 /// Will send contract event and every error encountered while fetching through the provided [UnboundedSender].
@@ -79,33 +116,54 @@ pub async fn listen_contract_events(
         if sender.is_closed() {
             break;
         }
+        let events = block?.events().await?;
+        for event in translate_events(events.iter(), contracts) {
+            sender.unbounded_send(event)?;
+        }
+    }
 
-        let block = block?;
+    Ok(())
+}
 
-        for event in block.events().await?.iter() {
-            let event = event?;
+/// Try to convert `events` to `ContractEvent` using matching contract from `contracts`.
+fn translate_events<
+    Err: Error + Into<anyhow::Error> + Send + Sync + 'static,
+    E: Iterator<Item = Result<EventDetails, Err>>,
+>(
+    events: E,
+    contracts: &[&ContractInstance],
+) -> Vec<Result<ContractEvent>> {
+    events
+        .filter_map(|maybe_event| {
+            maybe_event
+                .map(|e| e.as_event::<ContractEmitted>().ok().flatten())
+                .transpose()
+        })
+        .map(|maybe_event| match maybe_event {
+            Ok(e) => translate_event(&e, contracts),
+            Err(e) => Err(anyhow::Error::from(e)),
+        })
+        .collect()
+}
 
-            if let Some(event) =
-                event.as_event::<crate::api::contracts::events::ContractEmitted>()?
-            {
-                if let Some(contract) = contracts
-                    .iter()
-                    .find(|contract| contract.address() == &event.contract)
-                {
-                    let data = zero_prefixed(&event.data);
-                    let event = contract
-                        .transcoder
-                        .decode_contract_event(&mut data.as_slice());
+/// Try to convert `event` to `ContractEvent` using matching contract from `contracts`.
+fn translate_event(
+    event: &ContractEmitted,
+    contracts: &[&ContractInstance],
+) -> Result<ContractEvent> {
+    let matching_contract = contracts
+        .iter()
+        .find(|contract| contract.address() == &event.contract)
+        .ok_or(anyhow!(
+            "The event wasn't emitted by any of the provided contracts"
+        ))?;
 
-                    sender.unbounded_send(
-                        event.and_then(|event| build_event(contract.address().clone(), event)),
-                    )?;
-                }
-            }
-        }
-    }
+    let data = zero_prefixed(&event.data);
+    let data = matching_contract
+        .transcoder
+        .decode_contract_event(&mut data.as_slice())?;
 
-    Ok(())
+    build_event(matching_contract.address.clone(), data)
 }
 
 /// The contract transcoder assumes there is an extra byte (that it discards) indicating the size of the data. However,

aleph-client/src/contract/mod.rs

@@ -6,7 +6,7 @@
 //! ```no_run
 //! # use anyhow::{Result, Context};
 //! # use aleph_client::{AccountId, Balance};
-//! # use aleph_client::{Connection, SignedConnection};
+//! # use aleph_client::{Connection, SignedConnection, TxInfo};
 //! # use aleph_client::contract::ContractInstance;
 //! #
 //! #[derive(Debug)]
@@ -24,7 +24,7 @@
 //!         })
 //!     }
 //!
-//!     async fn transfer(&self, conn: &SignedConnection, to: AccountId, amount: Balance) -> Result<()> {
+//!     async fn transfer(&self, conn: &SignedConnection, to: AccountId, amount: Balance) -> Result<TxInfo> {
 //!         self.contract.contract_exec(
 //!             conn,
 //!             "PSP22::transfer",
@@ -52,6 +52,7 @@ use contract_transcode::ContractMessageTranscoder;
 pub use convertible_value::ConvertibleValue;
 
 use crate::{
+    connections::TxInfo,
     contract_transcode::Value,
     pallets::contract::{ContractCallArgs, ContractRpc, ContractsUserApi},
     sp_weights::weight_v2::Weight,
@@ -128,7 +129,7 @@ impl ContractInstance {
         &self,
         conn: &C,
         message: &str,
-    ) -> Result<()> {
+    ) -> Result<TxInfo> {
         self.contract_exec::<C, String>(conn, message, &[]).await
     }
 
@@ -138,7 +139,7 @@ impl ContractInstance {
         conn: &C,
         message: &str,
         args: &[S],
-    ) -> Result<()> {
+    ) -> Result<TxInfo> {
         self.contract_exec_value::<C, S>(conn, message, args, 0)
             .await
     }
@@ -149,7 +150,7 @@ impl ContractInstance {
         conn: &C,
         message: &str,
         value: Balance,
-    ) -> Result<()> {
+    ) -> Result<TxInfo> {
         self.contract_exec_value::<C, String>(conn, message, &[], value)
             .await
     }
@@ -161,7 +162,7 @@ impl ContractInstance {
         message: &str,
         args: &[S],
         value: Balance,
-    ) -> Result<()> {
+    ) -> Result<TxInfo> {
         let data = self.encode(message, args)?;
         conn.call(
             self.address.clone(),
@@ -175,7 +176,6 @@ impl ContractInstance {
             TxStatus::InBlock,
         )
         .await
-        .map(|_| ())
     }
 
     fn encode<S: AsRef<str> + Debug>(&self, message: &str, args: &[S]) -> Result<Vec<u8>> {

aleph-client/src/lib.rs

@@ -61,7 +61,7 @@ pub type SubxtClient = OnlineClient<AlephConfig>;
 
 pub use connections::{
     AsConnection, AsSigned, Connection, ConnectionApi, RootConnection, SignedConnection,
-    SignedConnectionApi, SudoCall,
+    SignedConnectionApi, SudoCall, TxInfo,
 };
 
 /// When submitting a transaction, wait for given status before proceeding.

e2e-tests/src/test/adder.rs

@@ -1,13 +1,12 @@
-use std::{fmt::Debug, str::FromStr, sync::Arc};
+use std::{fmt::Debug, str::FromStr};
 
 use aleph_client::{
-    contract::{event::listen_contract_events, ContractInstance},
+    contract::{event::get_contract_events, ContractInstance},
     contract_transcode::Value,
-    AccountId, ConnectionApi, SignedConnectionApi,
+    AccountId, ConnectionApi, SignedConnectionApi, TxInfo,
 };
-use anyhow::{Context, Result};
+use anyhow::{anyhow, Context, Result};
 use assert2::assert;
-use futures::{channel::mpsc::unbounded, StreamExt};
 
 use crate::{config::setup_test, test::helpers::basic_test_context};
 
@@ -19,26 +18,22 @@ pub async fn adder() -> Result<()> {
 
     let (conn, _authority, account) = basic_test_context(config).await?;
 
-    let contract = Arc::new(AdderInstance::new(
+    let contract = AdderInstance::new(
         &config.test_case_params.adder,
         &config.test_case_params.adder_metadata,
-    )?);
-
-    let listen_conn = conn.clone();
-    let listen_contract = contract.clone();
-    let (tx, mut rx) = unbounded();
-    let listen = || async move {
-        listen_contract_events(&listen_conn, &[listen_contract.as_ref().into()], tx).await?;
-        <Result<(), anyhow::Error>>::Ok(())
-    };
-    let join = tokio::spawn(listen());
+    )?;
 
     let increment = 10;
     let before = contract.get(&conn).await?;
 
-    contract.add(&account.sign(&conn), increment).await?;
+    let tx_info = contract.add(&account.sign(&conn), increment).await?;
+    let events = get_contract_events(&conn, &contract.contract, tx_info).await?;
+    let event = match events.len() {
+        0 => return Err(anyhow!("No event received")),
+        1 => events[0].clone(),
+        _ => return Err(anyhow!("Too many events received")),
+    };
 
-    let event = rx.next().await.context("No event received")??;
     assert!(event.name == Some("ValueChanged".to_string()));
     assert!(event.contract == *contract.contract.address());
     assert!(event.data["new_value"] == Value::UInt(before as u128 + 10));
@@ -54,9 +49,6 @@ pub async fn adder() -> Result<()> {
         .await?;
     assert!(contract.get_name(&conn).await? == Some(new_name.to_string()));
 
-    rx.close();
-    join.await??;
-
     Ok(())
 }
 
@@ -95,7 +87,7 @@ impl AdderInstance {
         self.contract.contract_read0(conn, "get").await
     }
 
-    pub async fn add<S: SignedConnectionApi>(&self, conn: &S, value: u32) -> Result<()> {
+    pub async fn add<S: SignedConnectionApi>(&self, conn: &S, value: u32) -> Result<TxInfo> {
         self.contract
             .contract_exec(conn, "add", &[value.to_string()])
             .await
@@ -105,7 +97,7 @@ impl AdderInstance {
         &self,
         conn: &S,
         name: Option<&str>,
-    ) -> Result<()> {
+    ) -> Result<TxInfo> {
         let name = name.map_or_else(
             || "None".to_string(),
             |name| {