New Event Subscription API (#442)

* Add reworked event types

* first pass implementing event subscriptions

* make clear that some methods are private

* comment tidy

* use Events in transaction stuff

* align transaction and event APIs

* remove __private_ prefixes; they are ugly

* fix examples and remove old events and subscription code

* better comments on hidden event functions

* re-add find_first_event; it's used a bunch in tests and examples

* cargo check --all-targets now passes

* Fix up existing event tests

* cargo fmt

* change todo to note

* clippy and doc niggles

* revert to find_first_event

* Add specific subscription related tests

* cargo fmt

* Update tests and add/fix examples

* cargo fmt

* add a little to subscribe_all_events example

* cargo fmt

* move an example comment

* easy access to root mod for more clarity

* add a couple of tests to ensure that events properly decoded until naff bytes

* Simplify EventSubscription Stream impl a little

* Address some PR feedback

codegen/src/api/calls.rs

@@ -90,7 +90,7 @@ pub fn generate_calls(
                 pub fn #fn_name(
                     &self,
                     #( #call_fn_args, )*
-                ) -> ::subxt::SubmittableExtrinsic<'a, T, X, A, #call_struct_name, DispatchError> {
+                ) -> ::subxt::SubmittableExtrinsic<'a, T, X, A, #call_struct_name, DispatchError, root_mod::Event> {
                     let call = #call_struct_name { #( #call_args, )* };
                     ::subxt::SubmittableExtrinsic::new(self.client, call)
                 }
@@ -101,6 +101,7 @@ pub fn generate_calls(
 
     quote! {
         pub mod calls {
+            use super::root_mod;
             use super::#types_mod_ident;
 
             type DispatchError = #types_mod_ident::sp_runtime::DispatchError;

codegen/src/api/mod.rs

@@ -209,6 +209,7 @@ impl RuntimeGenerator {
 
             quote! {
                 pub mod #mod_name {
+                    use super::root_mod;
                     use super::#types_mod_ident;
                     #calls
                     #event
@@ -275,6 +276,9 @@ impl RuntimeGenerator {
         quote! {
             #[allow(dead_code, unused_imports, non_camel_case_types)]
             pub mod #mod_ident {
+                // Make it easy to access the root via `root_mod` at different levels:
+                use super::#mod_ident as root_mod;
+
                 #outer_event
                 #( #modules )*
                 #types_mod
@@ -323,12 +327,33 @@ impl RuntimeGenerator {
                     pub fn tx(&'a self) -> TransactionApi<'a, T, X, A> {
                         TransactionApi { client: &self.client, marker: ::core::marker::PhantomData }
                     }
+
+                    pub fn events(&'a self) -> EventsApi<'a, T> {
+                        EventsApi { client: &self.client }
+                    }
+                }
+
+                pub struct EventsApi<'a, T: ::subxt::Config> {
+                    client: &'a ::subxt::Client<T>,
+                }
+
+                impl <'a, T: ::subxt::Config> EventsApi<'a, T> {
+                    pub async fn at(&self, block_hash: T::Hash) -> Result<::subxt::events::Events<'a, T, Event>, ::subxt::BasicError> {
+                        ::subxt::events::at::<T, Event>(self.client, block_hash).await
+                    }
+
+                    pub async fn subscribe(&self) -> Result<::subxt::events::EventSubscription<'a, T, Event>, ::subxt::BasicError> {
+                        ::subxt::events::subscribe::<T, Event>(self.client).await
+                    }
+
+                    pub async fn subscribe_finalized(&self) -> Result<::subxt::events::EventSubscription<'a, T, Event>, ::subxt::BasicError> {
+                        ::subxt::events::subscribe_finalized::<T, Event>(self.client).await
+                    }
                 }
 
                 pub struct ConstantsApi;
 
-                impl ConstantsApi
+                impl ConstantsApi {
-                {
                     #(
                         pub fn #pallets_with_constants(&self) -> #pallets_with_constants::constants::ConstantsApi {
                             #pallets_with_constants::constants::ConstantsApi

examples/examples/balance_transfer.rs

@@ -14,11 +14,11 @@
 // You should have received a copy of the GNU General Public License
 // along with subxt.  If not, see <http://www.gnu.org/licenses/>.
 
-//! To run this example, a local polkadot node should be running.
+//! To run this example, a local polkadot node should be running. Example verified against polkadot 0.9.13-82616422d0-aarch64-macos.
 //!
 //! E.g.
 //! ```bash
-//! curl "https://github.com/paritytech/polkadot/releases/download/v0.9.11/polkadot" --output /usr/local/bin/polkadot --location
+//! curl "https://github.com/paritytech/polkadot/releases/download/v0.9.13/polkadot" --output /usr/local/bin/polkadot --location
 //! polkadot --dev --tmp
 //! ```
 

examples/examples/custom_config.rs

@@ -14,6 +14,14 @@
 // You should have received a copy of the GNU General Public License
 // along with subxt.  If not, see <http://www.gnu.org/licenses/>.
 
+//! To run this example, a local polkadot node should be running. Example verified against polkadot 0.9.13-82616422d0-aarch64-macos.
+//!
+//! E.g.
+//! ```bash
+//! curl "https://github.com/paritytech/polkadot/releases/download/v0.9.13/polkadot" --output /usr/local/bin/polkadot --location
+//! polkadot --dev --tmp
+//! ```
+
 use sp_keyring::AccountKeyring;
 use subxt::{
     ClientBuilder,

examples/examples/custom_type_derives.rs

@@ -14,13 +14,17 @@
 // You should have received a copy of the GNU General Public License
 // along with subxt.  If not, see <http://www.gnu.org/licenses/>.
 
+//! Example verified against polkadot 0.9.13-82616422d0-aarch64-macos.
+
 #![allow(clippy::redundant_clone)]
 
 #[subxt::subxt(
     runtime_metadata_path = "examples/polkadot_metadata.scale",
     // We can add (certain) custom derives to the generated types by providing
-    // a comma separated list to the below attribute. Most useful for adding `Clone`:
+    // a comma separated list to the below attribute. Most useful for adding `Clone`.
-    generated_type_derives = "Clone, Hash"
+    // The derives that we can add ultiamtely is limited to the traits that the base
+    // types relied upon by the codegen implement.
+    generated_type_derives = "Clone, PartialEq"
 )]
 pub mod polkadot {}
 

examples/examples/fetch_all_accounts.rs

@@ -14,11 +14,11 @@
 // You should have received a copy of the GNU General Public License
 // along with subxt.  If not, see <http://www.gnu.org/licenses/>.
 
-//! To run this example, a local polkadot node should be running.
+//! To run this example, a local polkadot node should be running. Example verified against polkadot 0.9.13-82616422d0-aarch64-macos.
 //!
 //! E.g.
 //! ```bash
-//! curl "https://github.com/paritytech/polkadot/releases/download/v0.9.11/polkadot" --output /usr/local/bin/polkadot --location
+//! curl "https://github.com/paritytech/polkadot/releases/download/v0.9.13/polkadot" --output /usr/local/bin/polkadot --location
 //! polkadot --dev --tmp
 //! ```
 

examples/examples/rpc_call.rs

@@ -14,6 +14,14 @@
 // You should have received a copy of the GNU General Public License
 // along with subxt.  If not, see <http://www.gnu.org/licenses/>.
 
+//! To run this example, a local polkadot node should be running. Example verified against polkadot 0.9.13-82616422d0-aarch64-macos.
+//!
+//! E.g.
+//! ```bash
+//! curl "https://github.com/paritytech/polkadot/releases/download/v0.9.13/polkadot" --output /usr/local/bin/polkadot --location
+//! polkadot --dev --tmp
+//! ```
+
 use subxt::{
     ClientBuilder,
     DefaultConfig,
@@ -42,9 +50,9 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
         .await?;
 
     if let Some(hash) = block_hash {
-        println!("Block hash for block number {}: {}", block_number, hash);
+        println!("Block hash for block number {block_number}: {hash}");
     } else {
-        println!("Block number {} not found.", block_number);
+        println!("Block number {block_number} not found.");
     }
 
     Ok(())

examples/examples/submit_and_watch.rs

@@ -14,11 +14,11 @@
 // You should have received a copy of the GNU General Public License
 // along with subxt.  If not, see <http://www.gnu.org/licenses/>.
 
-//! To run this example, a local polkadot node should be running.
+//! To run this example, a local polkadot node should be running. Example verified against polkadot 0.9.13-82616422d0-aarch64-macos.
 //!
 //! E.g.
 //! ```bash
-//! curl "https://github.com/paritytech/polkadot/releases/download/v0.9.11/polkadot" --output /usr/local/bin/polkadot --location
+//! curl "https://github.com/paritytech/polkadot/releases/download/v0.9.13/polkadot" --output /usr/local/bin/polkadot --location
 //! polkadot --dev --tmp
 //! ```
 
@@ -70,7 +70,7 @@ async fn simple_transfer() -> Result<(), Box<dyn std::error::Error>> {
         balance_transfer.find_first_event::<polkadot::balances::events::Transfer>()?;
 
     if let Some(event) = transfer_event {
-        println!("Balance transfer success: value: {:?}", event.2);
+        println!("Balance transfer success: {event:?}");
     } else {
         println!("Failed to find Balances::Transfer Event");
     }
@@ -119,7 +119,7 @@ async fn simple_transfer_separate_events() -> Result<(), Box<dyn std::error::Err
         let transfer_event =
             events.find_first_event::<polkadot::balances::events::Transfer>()?;
         if let Some(event) = transfer_event {
-            println!("Balance transfer success: value: {:?}", event.2);
+            println!("Balance transfer success: {event:?}");
         } else {
             println!("Failed to find Balances::Transfer Event");
         }
@@ -165,8 +165,7 @@ async fn handle_transfer_events() -> Result<(), Box<dyn std::error::Error>> {
 
             if let Some(event) = transfer_event {
                 println!(
-                    "Balance transfer is now in block (but not finalized): value: {:?}",
+                    "Balance transfer is now in block (but not finalized): {event:?}"
-                    event.2
                 );
             } else {
                 println!("Failed to find Balances::Transfer Event");
@@ -185,7 +184,7 @@ async fn handle_transfer_events() -> Result<(), Box<dyn std::error::Error>> {
                 events.find_first_event::<polkadot::balances::events::Transfer>()?;
 
             if let Some(event) = transfer_event {
-                println!("Balance transfer success: value: {:?}", event.2);
+                println!("Balance transfer success: {event:?}");
             } else {
                 println!("Failed to find Balances::Transfer Event");
             }

examples/examples/subscribe_all_events.rs

@@ -0,0 +1,116 @@
+// Copyright 2019-2022 Parity Technologies (UK) Ltd.
+// This file is part of subxt.
+//
+// subxt 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.
+//
+// subxt 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 subxt.  If not, see <http://www.gnu.org/licenses/>.
+
+//! To run this example, a local polkadot node should be running. Example verified against polkadot 0.9.13-82616422d0-aarch64-macos.
+//!
+//! E.g.
+//! ```bash
+//! curl "https://github.com/paritytech/polkadot/releases/download/v0.9.13/polkadot" --output /usr/local/bin/polkadot --location
+//! polkadot --dev --tmp
+//! ```
+
+use futures::StreamExt;
+use sp_keyring::AccountKeyring;
+use std::time::Duration;
+use subxt::{
+    ClientBuilder,
+    DefaultConfig,
+    DefaultExtra,
+    PairSigner,
+};
+
+#[subxt::subxt(runtime_metadata_path = "examples/polkadot_metadata.scale")]
+pub mod polkadot {}
+
+/// Subscribe to all events, and then manually look through them and
+/// pluck out the events that we care about.
+#[async_std::main]
+async fn main() -> Result<(), Box<dyn std::error::Error>> {
+    env_logger::init();
+
+    let api = ClientBuilder::new()
+        .build()
+        .await?
+        .to_runtime_api::<polkadot::RuntimeApi<DefaultConfig, DefaultExtra<DefaultConfig>>>();
+
+    // Subscribe to any events that occur:
+    let mut event_sub = api.events().subscribe().await?;
+
+    // While this subscription is active, balance transfers are made somewhere:
+    async_std::task::spawn(async {
+        let signer = PairSigner::new(AccountKeyring::Alice.pair());
+        let api = ClientBuilder::new()
+            .build()
+            .await
+            .unwrap()
+            .to_runtime_api::<polkadot::RuntimeApi<DefaultConfig, DefaultExtra<DefaultConfig>>>();
+
+        let mut transfer_amount = 1_000_000_000;
+
+        // Make small balance transfers from Alice to Bob in a loop:
+        loop {
+            api.tx()
+                .balances()
+                .transfer(AccountKeyring::Bob.to_account_id().into(), transfer_amount)
+                .sign_and_submit(&signer)
+                .await
+                .unwrap();
+
+            async_std::task::sleep(Duration::from_secs(10)).await;
+            transfer_amount += 100_000_000;
+        }
+    });
+
+    // Our subscription will see the events emitted as a result of this:
+    while let Some(events) = event_sub.next().await {
+        let events = events?;
+        let block_hash = events.block_hash();
+
+        // We can iterate, statically decoding all events if we want:
+        println!("All events in block {block_hash:?}:");
+        println!("  Static event details:");
+        for event in events.iter() {
+            let event = event?;
+            println!("    {event:?}");
+        }
+
+        // Or we can dynamically decode events:
+        println!("  Dynamic event details: {block_hash:?}:");
+        for event in events.iter_raw() {
+            let event = event?;
+            let is_balance_transfer = event
+                .as_event::<polkadot::balances::events::Transfer>()?
+                .is_some();
+            let pallet = event.pallet;
+            let variant = event.variant;
+            println!(
+                "    {pallet}::{variant} (is balance transfer? {is_balance_transfer})"
+            );
+        }
+
+        // Or we can dynamically find the first transfer event, ignoring any others:
+        let transfer_event =
+            events.find_first_event::<polkadot::balances::events::Transfer>()?;
+
+        if let Some(ev) = transfer_event {
+            println!("  - Balance transfer success: value: {:?}", ev.amount);
+        } else {
+            println!("  - No balance transfer event found in this block");
+        }
+    }
+
+    Ok(())
+}

examples/examples/subscribe_one_event.rs

@@ -0,0 +1,97 @@
+// Copyright 2019-2022 Parity Technologies (UK) Ltd.
+// This file is part of subxt.
+//
+// subxt 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.
+//
+// subxt 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 subxt.  If not, see <http://www.gnu.org/licenses/>.
+
+//! To run this example, a local polkadot node should be running. Example verified against polkadot 0.9.13-82616422d0-aarch64-macos.
+//!
+//! E.g.
+//! ```bash
+//! curl "https://github.com/paritytech/polkadot/releases/download/v0.9.13/polkadot" --output /usr/local/bin/polkadot --location
+//! polkadot --dev --tmp
+//! ```
+
+use futures::{
+    future,
+    stream,
+    StreamExt,
+};
+use sp_keyring::AccountKeyring;
+use std::time::Duration;
+use subxt::{
+    ClientBuilder,
+    DefaultConfig,
+    DefaultExtra,
+    PairSigner,
+};
+
+#[subxt::subxt(runtime_metadata_path = "examples/polkadot_metadata.scale")]
+pub mod polkadot {}
+
+/// Subscribe to all events, and then manually look through them and
+/// pluck out the events that we care about.
+#[async_std::main]
+async fn main() -> Result<(), Box<dyn std::error::Error>> {
+    env_logger::init();
+
+    // Subscribe to any events that occur:
+    let api = ClientBuilder::new()
+        .build()
+        .await?
+        .to_runtime_api::<polkadot::RuntimeApi<DefaultConfig, DefaultExtra<DefaultConfig>>>();
+
+    // Subscribe to just balance transfer events, making use of `flat_map` and
+    // `filter_map` from the StreamExt trait to filter everything else out.
+    let mut transfer_events = api
+        .events()
+        .subscribe()
+        .await?
+        // Ignore errors returning events:
+        .filter_map(|events| future::ready(events.ok()))
+        // Map events to just the one we care about:
+        .flat_map(|events| {
+            let transfer_events = events
+                .find::<polkadot::balances::events::Transfer>()
+                .collect::<Vec<_>>();
+            stream::iter(transfer_events)
+        });
+
+    // While this subscription is active, we imagine some balance transfers are made somewhere else:
+    async_std::task::spawn(async {
+        let signer = PairSigner::new(AccountKeyring::Alice.pair());
+        let api = ClientBuilder::new()
+            .build()
+            .await
+            .unwrap()
+            .to_runtime_api::<polkadot::RuntimeApi<DefaultConfig, DefaultExtra<DefaultConfig>>>();
+
+        // Make small balance transfers from Alice to Bob in a loop:
+        loop {
+            api.tx()
+                .balances()
+                .transfer(AccountKeyring::Bob.to_account_id().into(), 1_000_000_000)
+                .sign_and_submit(&signer)
+                .await
+                .unwrap();
+            async_std::task::sleep(Duration::from_secs(10)).await;
+        }
+    });
+
+    // Our subscription will see all of the transfer events emitted as a result of this:
+    while let Some(transfer_event) = transfer_events.next().await {
+        println!("Balance transfer event: {transfer_event:?}");
+    }
+
+    Ok(())
+}

examples/examples/transfer_subscribe.rs

@@ -1,70 +0,0 @@
-// Copyright 2019-2022 Parity Technologies (UK) Ltd.
-// This file is part of subxt.
-//
-// subxt 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.
-//
-// subxt 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 subxt.  If not, see <http://www.gnu.org/licenses/>.
-
-//! To run this example, a local polkadot node should be running.
-//!
-//! E.g.
-//! ```bash
-//! curl "https://github.com/paritytech/polkadot/releases/download/v0.9.11/polkadot" --output /usr/local/bin/polkadot --location
-//! polkadot --dev --tmp
-//! ```
-
-use sp_keyring::AccountKeyring;
-use subxt::{
-    ClientBuilder,
-    DefaultConfig,
-    DefaultExtra,
-    EventSubscription,
-    PairSigner,
-};
-
-#[subxt::subxt(runtime_metadata_path = "examples/polkadot_metadata.scale")]
-pub mod polkadot {}
-
-#[async_std::main]
-async fn main() -> Result<(), Box<dyn std::error::Error>> {
-    env_logger::init();
-
-    let signer = PairSigner::new(AccountKeyring::Alice.pair());
-    let dest = AccountKeyring::Bob.to_account_id().into();
-
-    let api = ClientBuilder::new()
-        .build()
-        .await?
-        .to_runtime_api::<polkadot::RuntimeApi<DefaultConfig, DefaultExtra<DefaultConfig>>>();
-
-    let sub = api.client.rpc().subscribe_events().await?;
-    let decoder = api.client.events_decoder();
-    let mut sub = EventSubscription::<DefaultConfig>::new(sub, decoder);
-    sub.filter_event::<polkadot::balances::events::Transfer>();
-
-    api.tx()
-        .balances()
-        .transfer(dest, 10_000)
-        .sign_and_submit(&signer)
-        .await?;
-
-    let raw = sub.next().await.unwrap().unwrap();
-    let event = <polkadot::balances::events::Transfer as subxt::codec::Decode>::decode(
-        &mut &raw.data[..],
-    );
-    if let Ok(e) = event {
-        println!("Balance transfer success: value: {:?}", e.2);
-    } else {
-        println!("Failed to subscribe to Balances::Transfer Event");
-    }
-    Ok(())
-}

subxt/src/client.rs

@@ -20,7 +20,6 @@ pub use sp_runtime::traits::SignedExtension;
 
 use crate::{
     error::BasicError,
-    events::EventsDecoder,
     extrinsic::{
         self,
         SignedExtra,
@@ -98,13 +97,10 @@ impl ClientBuilder {
         .await;
         let metadata = metadata?;
 
-        let events_decoder = EventsDecoder::new(metadata.clone());
-
         Ok(Client {
             rpc,
             genesis_hash: genesis_hash?,
             metadata: Arc::new(metadata),
-            events_decoder,
             properties: properties.unwrap_or_else(|_| Default::default()),
             runtime_version: runtime_version?,
             iter_page_size: self.page_size.unwrap_or(10),
@@ -119,7 +115,6 @@ pub struct Client<T: Config> {
     rpc: Rpc<T>,
     genesis_hash: T::Hash,
     metadata: Arc<Metadata>,
-    events_decoder: EventsDecoder<T>,
     properties: SystemProperties,
     runtime_version: RuntimeVersion,
     iter_page_size: u32,
@@ -179,27 +174,23 @@ impl<T: Config> Client<T> {
     pub fn to_runtime_api<R: From<Self>>(self) -> R {
         self.into()
     }
-
-    /// Returns the events decoder.
-    pub fn events_decoder(&self) -> &EventsDecoder<T> {
-        &self.events_decoder
-    }
 }
 
 /// A constructed call ready to be signed and submitted.
-pub struct SubmittableExtrinsic<'client, T: Config, X, A, C, E: Decode> {
+pub struct SubmittableExtrinsic<'client, T: Config, X, A, C, E: Decode, Evs: Decode> {
     client: &'client Client<T>,
     call: C,
-    marker: std::marker::PhantomData<(X, A, E)>,
+    marker: std::marker::PhantomData<(X, A, E, Evs)>,
 }
 
-impl<'client, T, X, A, C, E> SubmittableExtrinsic<'client, T, X, A, C, E>
+impl<'client, T, X, A, C, E, Evs> SubmittableExtrinsic<'client, T, X, A, C, E, Evs>
 where
     T: Config,
     X: SignedExtra<T>,
     A: AccountData,
     C: Call + Send + Sync,
     E: Decode,
+    Evs: Decode,
 {
     /// Create a new [`SubmittableExtrinsic`].
     pub fn new(client: &'client Client<T>, call: C) -> Self {
@@ -217,7 +208,7 @@ where
     pub async fn sign_and_submit_then_watch(
         self,
         signer: &(dyn Signer<T, X> + Send + Sync),
-    ) -> Result<TransactionProgress<'client, T, E>, BasicError>
+    ) -> Result<TransactionProgress<'client, T, E, Evs>, BasicError>
     where
         <<X as SignedExtra<T>>::Extra as SignedExtension>::AdditionalSigned:
             Send + Sync + 'static,

subxt/src/lib.rs

@@ -59,12 +59,11 @@ use derivative::Derivative;
 mod client;
 mod config;
 mod error;
-mod events;
+pub mod events;
 pub mod extrinsic;
 mod metadata;
 pub mod rpc;
 pub mod storage;
-mod subscription;
 mod transaction;
 
 pub use crate::{
@@ -86,8 +85,9 @@ pub use crate::{
         TransactionError,
     },
     events::{
-        EventsDecoder,
+        EventDetails,
-        RawEvent,
+        Events,
+        RawEventDetails,
     },
     extrinsic::{
         DefaultExtra,
@@ -114,11 +114,6 @@ pub use crate::{
         StorageEntryKey,
         StorageMapKey,
     },
-    subscription::{
-        EventStorageSubscription,
-        EventSubscription,
-        FinalizedEventStorageSubscription,
-    },
     transaction::{
         TransactionEvents,
         TransactionInBlock,

subxt/src/rpc.rs

@@ -29,11 +29,6 @@ use std::{
 use crate::{
     error::BasicError,
     storage::StorageKeyPrefix,
-    subscription::{
-        EventStorageSubscription,
-        FinalizedEventStorageSubscription,
-        SystemEvents,
-    },
     Config,
     Metadata,
 };
@@ -406,35 +401,6 @@ impl<T: Config> Rpc<T> {
         Ok(version)
     }
 
-    /// Subscribe to System Events that are imported into blocks.
-    ///
-    /// *WARNING* these may not be included in the finalized chain, use
-    /// `subscribe_finalized_events` to ensure events are finalized.
-    pub async fn subscribe_events(
-        &self,
-    ) -> Result<EventStorageSubscription<T>, BasicError> {
-        let keys = Some(vec![StorageKey::from(SystemEvents::new())]);
-        let params = rpc_params![keys];
-
-        let subscription = self
-            .client
-            .subscribe("state_subscribeStorage", params, "state_unsubscribeStorage")
-            .await?;
-        Ok(EventStorageSubscription::Imported(subscription))
-    }
-
-    /// Subscribe to finalized events.
-    pub async fn subscribe_finalized_events(
-        &self,
-    ) -> Result<EventStorageSubscription<T>, BasicError> {
-        Ok(EventStorageSubscription::Finalized(
-            FinalizedEventStorageSubscription::new(
-                self.clone(),
-                self.subscribe_finalized_blocks().await?,
-            ),
-        ))
-    }
-
     /// Subscribe to blocks.
     pub async fn subscribe_blocks(&self) -> Result<Subscription<T::Header>, BasicError> {
         let subscription = self

subxt/src/subscription.rs

@@ -1,481 +0,0 @@
-// Copyright 2019-2022 Parity Technologies (UK) Ltd.
-// This file is part of subxt.
-//
-// subxt 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.
-//
-// subxt 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 subxt.  If not, see <http://www.gnu.org/licenses/>.
-
-use crate::{
-    error::BasicError,
-    events::{
-        EventsDecoder,
-        RawEvent,
-    },
-    rpc::Rpc,
-    Config,
-    Event,
-    Phase,
-};
-use jsonrpsee::core::{
-    client::Subscription,
-    DeserializeOwned,
-};
-use sp_core::{
-    storage::{
-        StorageChangeSet,
-        StorageKey,
-    },
-    twox_128,
-};
-use sp_runtime::traits::Header;
-use std::collections::VecDeque;
-
-/// Raw bytes for an Event, including the block hash where it occurred and its
-/// corresponding event index.
-#[derive(Debug)]
-#[cfg_attr(test, derive(PartialEq, Clone))]
-pub struct EventContext<Hash> {
-    pub block_hash: Hash,
-    pub event_idx: usize,
-    pub event: RawEvent,
-}
-
-/// Event subscription simplifies filtering a storage change set stream for
-/// events of interest.
-pub struct EventSubscription<'a, T: Config> {
-    block_reader: BlockReader<'a, T>,
-    block: Option<T::Hash>,
-    extrinsic: Option<usize>,
-    event: Option<(&'static str, &'static str)>,
-    events: VecDeque<EventContext<T::Hash>>,
-    finished: bool,
-}
-
-enum BlockReader<'a, T: Config> {
-    Decoder {
-        subscription: EventStorageSubscription<T>,
-        decoder: &'a EventsDecoder<T>,
-    },
-    /// Mock event listener for unit tests
-    #[cfg(test)]
-    Mock(
-        Box<
-            dyn Iterator<
-                Item = (T::Hash, Result<Vec<(Phase, usize, RawEvent)>, BasicError>),
-            >,
-        >,
-    ),
-}
-
-impl<'a, T: Config> BlockReader<'a, T> {
-    async fn next(
-        &mut self,
-    ) -> Option<(T::Hash, Result<Vec<(Phase, usize, RawEvent)>, BasicError>)> {
-        match self {
-            BlockReader::Decoder {
-                subscription,
-                decoder,
-            } => {
-                let change_set = subscription.next().await?;
-                let events: Result<Vec<_>, _> = change_set
-                    .changes
-                    .into_iter()
-                    .filter_map(|(_key, change)| {
-                        Some(decoder.decode_events(&mut change?.0.as_slice()))
-                    })
-                    .collect();
-
-                let flattened_events = events.map(|x| {
-                    x.into_iter()
-                        .flatten()
-                        .enumerate()
-                        .map(|(event_idx, (phase, raw))| (phase, event_idx, raw))
-                        .collect()
-                });
-                Some((change_set.block, flattened_events))
-            }
-            #[cfg(test)]
-            BlockReader::Mock(it) => it.next(),
-        }
-    }
-}
-
-impl<'a, T: Config> EventSubscription<'a, T> {
-    /// Creates a new event subscription.
-    pub fn new(
-        subscription: EventStorageSubscription<T>,
-        decoder: &'a EventsDecoder<T>,
-    ) -> Self {
-        Self {
-            block_reader: BlockReader::Decoder {
-                subscription,
-                decoder,
-            },
-            block: None,
-            extrinsic: None,
-            event: None,
-            events: Default::default(),
-            finished: false,
-        }
-    }
-
-    /// Only returns events contained in the block with the given hash.
-    pub fn filter_block(&mut self, block: T::Hash) {
-        self.block = Some(block);
-    }
-
-    /// Only returns events from block emitted by extrinsic with index.
-    pub fn filter_extrinsic(&mut self, block: T::Hash, ext_index: usize) {
-        self.block = Some(block);
-        self.extrinsic = Some(ext_index);
-    }
-
-    /// Filters events by type.
-    pub fn filter_event<Ev: Event>(&mut self) {
-        self.event = Some((Ev::PALLET, Ev::EVENT));
-    }
-
-    /// Gets the next event.
-    pub async fn next(&mut self) -> Option<Result<RawEvent, BasicError>> {
-        self.next_context()
-            .await
-            .map(|res| res.map(|ctx| ctx.event))
-    }
-    /// Gets the next event with the associated block hash and its corresponding
-    /// event index.
-    pub async fn next_context(
-        &mut self,
-    ) -> Option<Result<EventContext<T::Hash>, BasicError>> {
-        loop {
-            if let Some(raw_event) = self.events.pop_front() {
-                return Some(Ok(raw_event))
-            }
-            if self.finished {
-                return None
-            }
-            // always return None if subscription has closed
-            let (received_hash, events) = self.block_reader.next().await?;
-            if let Some(hash) = self.block.as_ref() {
-                if &received_hash == hash {
-                    self.finished = true;
-                } else {
-                    continue
-                }
-            }
-
-            match events {
-                Err(err) => return Some(Err(err)),
-                Ok(raw_events) => {
-                    for (phase, event_idx, raw) in raw_events {
-                        if let Some(ext_index) = self.extrinsic {
-                            if !matches!(phase, Phase::ApplyExtrinsic(i) if i as usize == ext_index)
-                            {
-                                continue
-                            }
-                        }
-                        if let Some((module, variant)) = self.event {
-                            if raw.pallet != module || raw.variant != variant {
-                                continue
-                            }
-                        }
-                        self.events.push_back(EventContext {
-                            block_hash: received_hash,
-                            event_idx,
-                            event: raw,
-                        });
-                    }
-                }
-            }
-        }
-    }
-}
-
-pub(crate) struct SystemEvents(StorageKey);
-
-impl SystemEvents {
-    pub(crate) fn new() -> Self {
-        let mut storage_key = twox_128(b"System").to_vec();
-        storage_key.extend(twox_128(b"Events").to_vec());
-        log::debug!("Events storage key {:?}", hex::encode(&storage_key));
-        Self(StorageKey(storage_key))
-    }
-}
-
-impl From<SystemEvents> for StorageKey {
-    fn from(key: SystemEvents) -> Self {
-        key.0
-    }
-}
-
-/// Event subscription to only fetch finalized storage changes.
-pub struct FinalizedEventStorageSubscription<T: Config> {
-    rpc: Rpc<T>,
-    subscription: Subscription<T::Header>,
-    storage_changes: VecDeque<StorageChangeSet<T::Hash>>,
-    storage_key: StorageKey,
-}
-
-impl<T: Config> FinalizedEventStorageSubscription<T> {
-    /// Creates a new finalized event storage subscription.
-    pub fn new(rpc: Rpc<T>, subscription: Subscription<T::Header>) -> Self {
-        Self {
-            rpc,
-            subscription,
-            storage_changes: Default::default(),
-            storage_key: SystemEvents::new().into(),
-        }
-    }
-
-    /// Gets the next change_set.
-    pub async fn next(&mut self) -> Option<StorageChangeSet<T::Hash>> {
-        loop {
-            if let Some(storage_change) = self.storage_changes.pop_front() {
-                return Some(storage_change)
-            }
-            let header: T::Header =
-                read_subscription_response("HeaderSubscription", &mut self.subscription)
-                    .await?;
-            self.storage_changes.extend(
-                self.rpc
-                    .query_storage_at(&[self.storage_key.clone()], Some(header.hash()))
-                    .await
-                    .ok()?,
-            );
-        }
-    }
-}
-
-/// Wrapper over imported and finalized event subscriptions.
-pub enum EventStorageSubscription<T: Config> {
-    /// Events that are InBlock
-    Imported(Subscription<StorageChangeSet<T::Hash>>),
-    /// Events that are Finalized
-    Finalized(FinalizedEventStorageSubscription<T>),
-}
-
-impl<T: Config> EventStorageSubscription<T> {
-    /// Gets the next change_set from the subscription.
-    pub async fn next(&mut self) -> Option<StorageChangeSet<T::Hash>> {
-        match self {
-            Self::Imported(event_sub) => {
-                read_subscription_response("StorageChangeSetSubscription", event_sub)
-                    .await
-            }
-            Self::Finalized(event_sub) => event_sub.next().await,
-        }
-    }
-}
-
-async fn read_subscription_response<T>(
-    sub_name: &str,
-    sub: &mut Subscription<T>,
-) -> Option<T>
-where
-    T: DeserializeOwned,
-{
-    match sub.next().await {
-        Some(Ok(next)) => Some(next),
-        Some(Err(e)) => {
-            log::error!("Subscription {} failed: {:?} dropping", sub_name, e);
-            None
-        }
-        None => None,
-    }
-}
-
-#[cfg(test)]
-mod tests {
-    use super::*;
-    use crate::DefaultConfig;
-    use sp_core::H256;
-
-    fn named_event(event_name: &str) -> RawEvent {
-        RawEvent {
-            data: sp_core::Bytes::from(Vec::new()),
-            pallet: event_name.to_string(),
-            variant: event_name.to_string(),
-            pallet_index: 0,
-            variant_index: 0,
-        }
-    }
-
-    #[async_std::test]
-    /// test that filters work correctly, and are independent of each other
-    async fn test_filters() {
-        let mut events: Vec<(H256, Phase, usize, RawEvent)> = vec![];
-        // create all events
-        for block_hash in [H256::from([0; 32]), H256::from([1; 32])] {
-            for phase in [
-                Phase::Initialization,
-                Phase::ApplyExtrinsic(0),
-                Phase::ApplyExtrinsic(1),
-                Phase::Finalization,
-            ] {
-                [named_event("a"), named_event("b")]
-                    .iter()
-                    .enumerate()
-                    .for_each(|(idx, event)| {
-                        events.push((
-                            block_hash,
-                            phase.clone(),
-                            // The event index
-                            idx,
-                            event.clone(),
-                        ))
-                    });
-            }
-        }
-
-        // set variant index so we can uniquely identify the event
-        events.iter_mut().enumerate().for_each(|(idx, event)| {
-            event.3.variant_index = idx as u8;
-        });
-
-        let half_len = events.len() / 2;
-
-        for block_filter in [None, Some(H256::from([1; 32]))] {
-            for extrinsic_filter in [None, Some(1)] {
-                for event_filter in [None, Some(("b", "b"))] {
-                    let mut subscription: EventSubscription<DefaultConfig> =
-                        EventSubscription {
-                            block_reader: BlockReader::Mock(Box::new(
-                                vec![
-                                    (
-                                        events[0].0,
-                                        Ok(events
-                                            .iter()
-                                            .take(half_len)
-                                            .map(|(_, phase, idx, event)| {
-                                                (phase.clone(), *idx, event.clone())
-                                            })
-                                            .collect()),
-                                    ),
-                                    (
-                                        events[half_len].0,
-                                        Ok(events
-                                            .iter()
-                                            .skip(half_len)
-                                            .map(|(_, phase, idx, event)| {
-                                                (phase.clone(), *idx, event.clone())
-                                            })
-                                            .collect()),
-                                    ),
-                                ]
-                                .into_iter(),
-                            )),
-                            block: block_filter,
-                            extrinsic: extrinsic_filter,
-                            event: event_filter,
-                            events: Default::default(),
-                            finished: false,
-                        };
-
-                    let mut expected_events: Vec<(H256, Phase, usize, RawEvent)> =
-                        events.clone();
-
-                    if let Some(hash) = block_filter {
-                        expected_events.retain(|(h, _, _, _)| h == &hash);
-                    }
-                    if let Some(idx) = extrinsic_filter {
-                        expected_events.retain(|(_, phase, _,  _)| matches!(phase, Phase::ApplyExtrinsic(i) if *i as usize == idx));
-                    }
-                    if let Some(name) = event_filter {
-                        expected_events.retain(|(_, _, _, event)| event.pallet == name.0);
-                    }
-
-                    for expected_event in expected_events {
-                        assert_eq!(
-                            subscription.next().await.unwrap().unwrap(),
-                            expected_event.3
-                        );
-                    }
-                    assert!(subscription.next().await.is_none());
-                }
-            }
-        }
-    }
-
-    #[async_std::test]
-    async fn test_context() {
-        let mut events = vec![];
-        // create all events
-        for block_hash in [H256::from([0; 32]), H256::from([1; 32])] {
-            for phase in [
-                Phase::Initialization,
-                Phase::ApplyExtrinsic(0),
-                Phase::ApplyExtrinsic(1),
-                Phase::Finalization,
-            ] {
-                [named_event("a"), named_event("b")]
-                    .iter()
-                    .enumerate()
-                    .for_each(|(idx, event)| {
-                        events.push((
-                            phase.clone(),
-                            EventContext {
-                                block_hash,
-                                event_idx: idx,
-                                event: event.clone(),
-                            },
-                        ));
-                    });
-            }
-        }
-
-        // set variant index so we can uniquely identify the event
-        events.iter_mut().enumerate().for_each(|(idx, (_, ctx))| {
-            ctx.event.variant_index = idx as u8;
-        });
-
-        let half_len = events.len() / 2;
-
-        let mut subscription: EventSubscription<DefaultConfig> = EventSubscription {
-            block_reader: BlockReader::Mock(Box::new(
-                vec![
-                    (
-                        events[0].1.block_hash,
-                        Ok(events
-                            .iter()
-                            .take(half_len)
-                            .map(|(phase, ctx)| {
-                                (phase.clone(), ctx.event_idx, ctx.event.clone())
-                            })
-                            .collect()),
-                    ),
-                    (
-                        events[half_len].1.block_hash,
-                        Ok(events
-                            .iter()
-                            .skip(half_len)
-                            .map(|(phase, ctx)| {
-                                (phase.clone(), ctx.event_idx, ctx.event.clone())
-                            })
-                            .collect()),
-                    ),
-                ]
-                .into_iter(),
-            )),
-            block: None,
-            extrinsic: None,
-            event: None,
-            events: Default::default(),
-            finished: false,
-        };
-
-        let expected_events = events.clone();
-
-        for exp in expected_events {
-            assert_eq!(subscription.next_context().await.unwrap().unwrap(), exp.1);
-        }
-        assert!(subscription.next().await.is_none());
-    }
-}

subxt/src/transaction.rs

@@ -18,7 +18,6 @@ use std::task::Poll;
 
 use crate::PhantomDataSendSync;
 use codec::Decode;
-use sp_core::storage::StorageKey;
 use sp_runtime::traits::Hash;
 pub use sp_runtime::traits::SignedExtension;
 pub use sp_version::RuntimeVersion;
@@ -31,8 +30,13 @@ use crate::{
         RuntimeError,
         TransactionError,
     },
+    events::{
+        self,
+        EventDetails,
+        Events,
+        RawEventDetails,
+    },
     rpc::SubstrateTransactionStatus,
-    subscription::SystemEvents,
     Config,
     Phase,
 };
@@ -50,19 +54,22 @@ use jsonrpsee::core::{
 /// returned from [`crate::SubmittableExtrinsic::sign_and_submit_then_watch()`].
 #[derive(Derivative)]
 #[derivative(Debug(bound = ""))]
-pub struct TransactionProgress<'client, T: Config, E: Decode> {
+pub struct TransactionProgress<'client, T: Config, E: Decode, Evs: Decode> {
     sub: Option<RpcSubscription<SubstrateTransactionStatus<T::Hash, T::Hash>>>,
     ext_hash: T::Hash,
     client: &'client Client<T>,
-    _error: PhantomDataSendSync<E>,
+    _error: PhantomDataSendSync<(E, Evs)>,
 }
 
 // The above type is not `Unpin` by default unless the generic param `T` is,
 // so we manually make it clear that Unpin is actually fine regardless of `T`
 // (we don't care if this moves around in memory while it's "pinned").
-impl<'client, T: Config, E: Decode> Unpin for TransactionProgress<'client, T, E> {}
+impl<'client, T: Config, E: Decode, Evs: Decode> Unpin
+    for TransactionProgress<'client, T, E, Evs>
+{
+}
 
-impl<'client, T: Config, E: Decode> TransactionProgress<'client, T, E> {
+impl<'client, T: Config, E: Decode, Evs: Decode> TransactionProgress<'client, T, E, Evs> {
     /// Instantiate a new [`TransactionProgress`] from a custom subscription.
     pub fn new(
         sub: RpcSubscription<SubstrateTransactionStatus<T::Hash, T::Hash>>,
@@ -82,7 +89,7 @@ impl<'client, T: Config, E: Decode> TransactionProgress<'client, T, E> {
     /// avoid importing that trait if you don't otherwise need it.
     pub async fn next_item(
         &mut self,
-    ) -> Option<Result<TransactionStatus<'client, T, E>, BasicError>> {
+    ) -> Option<Result<TransactionStatus<'client, T, E, Evs>, BasicError>> {
         self.next().await
     }
 
@@ -99,7 +106,7 @@ impl<'client, T: Config, E: Decode> TransactionProgress<'client, T, E> {
     /// level [`TransactionProgress::next_item()`] API if you'd like to handle these statuses yourself.
     pub async fn wait_for_in_block(
         mut self,
-    ) -> Result<TransactionInBlock<'client, T, E>, BasicError> {
+    ) -> Result<TransactionInBlock<'client, T, E, Evs>, BasicError> {
         while let Some(status) = self.next_item().await {
             match status? {
                 // Finalized or otherwise in a block! Return.
@@ -129,7 +136,7 @@ impl<'client, T: Config, E: Decode> TransactionProgress<'client, T, E> {
     /// level [`TransactionProgress::next_item()`] API if you'd like to handle these statuses yourself.
     pub async fn wait_for_finalized(
         mut self,
-    ) -> Result<TransactionInBlock<'client, T, E>, BasicError> {
+    ) -> Result<TransactionInBlock<'client, T, E, Evs>, BasicError> {
         while let Some(status) = self.next_item().await {
             match status? {
                 // Finalized! Return.
@@ -158,14 +165,16 @@ impl<'client, T: Config, E: Decode> TransactionProgress<'client, T, E> {
     /// level [`TransactionProgress::next_item()`] API if you'd like to handle these statuses yourself.
     pub async fn wait_for_finalized_success(
         self,
-    ) -> Result<TransactionEvents<T>, Error<E>> {
+    ) -> Result<TransactionEvents<'client, T, Evs>, Error<E>> {
         let evs = self.wait_for_finalized().await?.wait_for_success().await?;
         Ok(evs)
     }
 }
 
-impl<'client, T: Config, E: Decode> Stream for TransactionProgress<'client, T, E> {
+impl<'client, T: Config, E: Decode, Evs: Decode> Stream
-    type Item = Result<TransactionStatus<'client, T, E>, BasicError>;
+    for TransactionProgress<'client, T, E, Evs>
+{
+    type Item = Result<TransactionStatus<'client, T, E, Evs>, BasicError>;
 
     fn poll_next(
         mut self: std::pin::Pin<&mut Self>,
@@ -274,7 +283,7 @@ impl<'client, T: Config, E: Decode> Stream for TransactionProgress<'client, T, E
 /// or that finality gadget is lagging behind.
 #[derive(Derivative)]
 #[derivative(Debug(bound = ""))]
-pub enum TransactionStatus<'client, T: Config, E: Decode> {
+pub enum TransactionStatus<'client, T: Config, E: Decode, Evs: Decode> {
     /// The transaction is part of the "future" queue.
     Future,
     /// The transaction is part of the "ready" queue.
@@ -282,7 +291,7 @@ pub enum TransactionStatus<'client, T: Config, E: Decode> {
     /// The transaction has been broadcast to the given peers.
     Broadcast(Vec<String>),
     /// The transaction has been included in a block with given hash.
-    InBlock(TransactionInBlock<'client, T, E>),
+    InBlock(TransactionInBlock<'client, T, E, Evs>),
     /// The block this transaction was included in has been retracted,
     /// probably because it did not make it onto the blocks which were
     /// finalized.
@@ -291,7 +300,7 @@ pub enum TransactionStatus<'client, T: Config, E: Decode> {
     /// blocks, and so the subscription has ended.
     FinalityTimeout(T::Hash),
     /// The transaction has been finalized by a finality-gadget, e.g GRANDPA.
-    Finalized(TransactionInBlock<'client, T, E>),
+    Finalized(TransactionInBlock<'client, T, E, Evs>),
     /// The transaction has been replaced in the pool by another transaction
     /// that provides the same tags. (e.g. same (sender, nonce)).
     Usurped(T::Hash),
@@ -301,10 +310,10 @@ pub enum TransactionStatus<'client, T: Config, E: Decode> {
     Invalid,
 }
 
-impl<'client, T: Config, E: Decode> TransactionStatus<'client, T, E> {
+impl<'client, T: Config, E: Decode, Evs: Decode> TransactionStatus<'client, T, E, Evs> {
     /// A convenience method to return the `Finalized` details. Returns
     /// [`None`] if the enum variant is not [`TransactionStatus::Finalized`].
-    pub fn as_finalized(&self) -> Option<&TransactionInBlock<'client, T, E>> {
+    pub fn as_finalized(&self) -> Option<&TransactionInBlock<'client, T, E, Evs>> {
         match self {
             Self::Finalized(val) => Some(val),
             _ => None,
@@ -313,7 +322,7 @@ impl<'client, T: Config, E: Decode> TransactionStatus<'client, T, E> {
 
     /// A convenience method to return the `InBlock` details. Returns
     /// [`None`] if the enum variant is not [`TransactionStatus::InBlock`].
-    pub fn as_in_block(&self) -> Option<&TransactionInBlock<'client, T, E>> {
+    pub fn as_in_block(&self) -> Option<&TransactionInBlock<'client, T, E, Evs>> {
         match self {
             Self::InBlock(val) => Some(val),
             _ => None,
@@ -324,14 +333,14 @@ impl<'client, T: Config, E: Decode> TransactionStatus<'client, T, E> {
 /// This struct represents a transaction that has made it into a block.
 #[derive(Derivative)]
 #[derivative(Debug(bound = ""))]
-pub struct TransactionInBlock<'client, T: Config, E: Decode> {
+pub struct TransactionInBlock<'client, T: Config, E: Decode, Evs: Decode> {
     block_hash: T::Hash,
     ext_hash: T::Hash,
     client: &'client Client<T>,
-    _error: PhantomDataSendSync<E>,
+    _error: PhantomDataSendSync<(E, Evs)>,
 }
 
-impl<'client, T: Config, E: Decode> TransactionInBlock<'client, T, E> {
+impl<'client, T: Config, E: Decode, Evs: Decode> TransactionInBlock<'client, T, E, Evs> {
     pub(crate) fn new(
         block_hash: T::Hash,
         ext_hash: T::Hash,
@@ -368,11 +377,14 @@ impl<'client, T: Config, E: Decode> TransactionInBlock<'client, T, E> {
     ///
     /// **Note:** This has to download block details from the node and decode events
     /// from them.
-    pub async fn wait_for_success(&self) -> Result<TransactionEvents<T>, Error<E>> {
+    pub async fn wait_for_success(
+        &self,
+    ) -> Result<TransactionEvents<'client, T, Evs>, Error<E>> {
         let events = self.fetch_events().await?;
 
         // Try to find any errors; return the first one we encounter.
-        for ev in events.as_slice() {
+        for ev in events.iter_raw() {
+            let ev = ev?;
             if &ev.pallet == "System" && &ev.variant == "ExtrinsicFailed" {
                 let dispatch_error = E::decode(&mut &*ev.data)?;
                 return Err(Error::Runtime(RuntimeError(dispatch_error)))
@@ -388,7 +400,9 @@ impl<'client, T: Config, E: Decode> TransactionInBlock<'client, T, E> {
     ///
     /// **Note:** This has to download block details from the node and decode events
     /// from them.
-    pub async fn fetch_events(&self) -> Result<TransactionEvents<T>, BasicError> {
+    pub async fn fetch_events(
+        &self,
+    ) -> Result<TransactionEvents<'client, T, Evs>, BasicError> {
         let block = self
             .client
             .rpc()
@@ -406,31 +420,11 @@ impl<'client, T: Config, E: Decode> TransactionInBlock<'client, T, E> {
             // extrinsic, the extrinsic should be in there somewhere..
             .ok_or(BasicError::Transaction(TransactionError::BlockHashNotFound))?;
 
-        let raw_events = self
+        let events = events::at::<T, Evs>(self.client, self.block_hash).await?;
-            .client
-            .rpc()
-            .storage(
-                &StorageKey::from(SystemEvents::new()),
-                Some(self.block_hash),
-            )
-            .await?
-            .map(|s| s.0)
-            .unwrap_or_else(Vec::new);
-
-        let events = self
-            .client
-            .events_decoder()
-            .decode_events(&mut &*raw_events)?
-            .into_iter()
-            .filter(move |(phase, _raw)| {
-                phase == &Phase::ApplyExtrinsic(extrinsic_idx as u32)
-            })
-            .map(|(_phase, event)| event)
-            .collect();
 
         Ok(TransactionEvents {
-            block_hash: self.block_hash,
             ext_hash: self.ext_hash,
+            ext_idx: extrinsic_idx as u32,
             events,
         })
     }
@@ -440,16 +434,16 @@ impl<'client, T: Config, E: Decode> TransactionInBlock<'client, T, E> {
 /// We can iterate over the events, or look for a specific one.
 #[derive(Derivative)]
 #[derivative(Debug(bound = ""))]
-pub struct TransactionEvents<T: Config> {
+pub struct TransactionEvents<'client, T: Config, Evs: Decode> {
-    block_hash: T::Hash,
     ext_hash: T::Hash,
-    events: Vec<crate::RawEvent>,
+    ext_idx: u32,
+    events: Events<'client, T, Evs>,
 }
 
-impl<T: Config> TransactionEvents<T> {
+impl<'client, T: Config, Evs: Decode> TransactionEvents<'client, T, Evs> {
     /// Return the hash of the block that the transaction has made it into.
     pub fn block_hash(&self) -> T::Hash {
-        self.block_hash
+        self.events.block_hash()
     }
 
     /// Return the hash of the extrinsic.
@@ -457,43 +451,66 @@ impl<T: Config> TransactionEvents<T> {
         self.ext_hash
     }
 
-    /// Return a slice of the returned events.
+    /// Return all of the events in the block that the transaction made it into.
-    pub fn as_slice(&self) -> &[crate::RawEvent] {
+    pub fn all_events_in_block(&self) -> &events::Events<'client, T, Evs> {
         &self.events
     }
 
-    /// Find all of the events matching the event type provided as a generic parameter. This
+    /// Iterate over the statically decoded events associated with this transaction.
-    /// will return an error if a matching event is found but cannot be properly decoded.
-    pub fn find_events<Ev: crate::Event>(&self) -> Result<Vec<Ev>, BasicError> {
-        self.events
-            .iter()
-            .filter_map(|e| e.as_event::<Ev>().map_err(Into::into).transpose())
-            .collect()
-    }
-
-    /// Find the first event that matches the event type provided as a generic parameter. This
-    /// will return an error if a matching event is found but cannot be properly decoded.
     ///
-    /// Use [`TransactionEvents::find_events`], or iterate over [`TransactionEvents`] yourself
+    /// This works in the same way that [`events::Events::iter()`] does, with the
-    /// if you'd like to handle multiple events of the same type.
+    /// exception that it filters out events not related to the submitted extrinsic.
+    pub fn iter(
+        &self,
+    ) -> impl Iterator<Item = Result<EventDetails<Evs>, BasicError>> + '_ {
+        self.events.iter().filter(|ev| {
+            ev.as_ref()
+                .map(|ev| ev.phase == Phase::ApplyExtrinsic(self.ext_idx))
+                .unwrap_or(true) // Keep any errors
+        })
+    }
+
+    /// Iterate over all of the raw events associated with this transaction.
+    ///
+    /// This works in the same way that [`events::Events::iter_raw()`] does, with the
+    /// exception that it filters out events not related to the submitted extrinsic.
+    pub fn iter_raw(
+        &self,
+    ) -> impl Iterator<Item = Result<RawEventDetails, BasicError>> + '_ {
+        self.events.iter_raw().filter(|ev| {
+            ev.as_ref()
+                .map(|ev| ev.phase == Phase::ApplyExtrinsic(self.ext_idx))
+                .unwrap_or(true) // Keep any errors.
+        })
+    }
+
+    /// Find all of the transaction events matching the event type provided as a generic parameter.
+    ///
+    /// This works in the same way that [`events::Events::find()`] does, with the
+    /// exception that it filters out events not related to the submitted extrinsic.
+    pub fn find<Ev: crate::Event>(
+        &self,
+    ) -> impl Iterator<Item = Result<Ev, BasicError>> + '_ {
+        self.iter_raw().filter_map(|ev| {
+            ev.and_then(|ev| ev.as_event::<Ev>().map_err(Into::into))
+                .transpose()
+        })
+    }
+
+    /// Iterate through the transaction events using metadata to dynamically decode and skip
+    /// them, and return the first event found which decodes to the provided `Ev` type.
+    ///
+    /// This works in the same way that [`events::Events::find_first_event()`] does, with the
+    /// exception that it ignores events not related to the submitted extrinsic.
     pub fn find_first_event<Ev: crate::Event>(&self) -> Result<Option<Ev>, BasicError> {
-        self.events
+        self.find::<Ev>().next().transpose()
-            .iter()
-            .filter_map(|e| e.as_event::<Ev>().transpose())
-            .next()
-            .transpose()
-            .map_err(Into::into)
     }
 
-    /// Find an event. Returns true if it was found.
+    /// Find an event in those associated with this transaction. Returns true if it was found.
-    pub fn has_event<Ev: crate::Event>(&self) -> Result<bool, BasicError> {
+    ///
-        Ok(self.find_first_event::<Ev>()?.is_some())
+    /// This works in the same way that [`events::Events::has()`] does, with the
-    }
+    /// exception that it ignores events not related to the submitted extrinsic.
-}
+    pub fn has<Ev: crate::Event>(&self) -> Result<bool, BasicError> {
-
+        Ok(self.find::<Ev>().next().transpose()?.is_some())
-impl<T: Config> std::ops::Deref for TransactionEvents<T> {
-    type Target = [crate::RawEvent];
-    fn deref(&self) -> &Self::Target {
-        &self.events
     }
 }

subxt/tests/integration/events.rs

@@ -0,0 +1,139 @@
+// Copyright 2019-2022 Parity Technologies (UK) Ltd.
+// This file is part of subxt.
+//
+// subxt 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.
+//
+// subxt 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 subxt.  If not, see <http://www.gnu.org/licenses/>.
+
+use crate::{
+    node_runtime::{
+        balances,
+        system,
+    },
+    pair_signer,
+    test_context,
+};
+use futures::StreamExt;
+use sp_keyring::AccountKeyring;
+use subxt::Signer;
+
+// Check that we can subscribe to non-finalized block events.
+#[async_std::test]
+async fn non_finalized_block_subscription() -> Result<(), subxt::BasicError> {
+    env_logger::try_init().ok();
+    let ctx = test_context().await;
+
+    let mut event_sub = ctx.api.events().subscribe().await?;
+
+    // Wait for the next set of events, and check that the
+    // associated block hash is not finalized yet.
+    let events = event_sub.next().await.unwrap()?;
+    let event_block_hash = events.block_hash();
+    let finalized_hash = ctx.api.client.rpc().finalized_head().await?;
+
+    assert_ne!(event_block_hash, finalized_hash);
+    Ok(())
+}
+
+// Check that we can subscribe to finalized block events.
+#[async_std::test]
+async fn finalized_block_subscription() -> Result<(), subxt::BasicError> {
+    env_logger::try_init().ok();
+    let ctx = test_context().await;
+
+    let mut event_sub = ctx.api.events().subscribe_finalized().await?;
+
+    // Wait for the next set of events, and check that the
+    // associated block hash is the one we just finalized.
+    // (this can be a bit slow as we have to wait for finalization)
+    let events = event_sub.next().await.unwrap()?;
+    let event_block_hash = events.block_hash();
+    let finalized_hash = ctx.api.client.rpc().finalized_head().await?;
+
+    assert_eq!(event_block_hash, finalized_hash);
+    Ok(())
+}
+
+// Check that our subscription actually keeps producing events for
+// a few blocks.
+#[async_std::test]
+async fn subscription_produces_events_each_block() -> Result<(), subxt::BasicError> {
+    env_logger::try_init().ok();
+    let ctx = test_context().await;
+
+    let mut event_sub = ctx.api.events().subscribe().await?;
+
+    for i in 0..3 {
+        let events = event_sub
+            .next()
+            .await
+            .expect("events expected each block")?;
+        let success_event = events
+            .find_first_event::<system::events::ExtrinsicSuccess>()
+            .expect("decode error");
+        // Every now and then I get no bytes back for the first block events;
+        // I assume that this might be the case for the genesis block, so don't
+        // worry if no event found (but we should have no decode errors etc either way).
+        if i > 0 && success_event.is_none() {
+            let n = events.len();
+            panic!("Expected an extrinsic success event on iteration {i} (saw {n} other events)")
+        }
+    }
+
+    Ok(())
+}
+
+// Check that our subscription receives events, and we can filter them based on
+// it's Stream impl, and ultimately see the event we expect.
+#[async_std::test]
+async fn balance_transfer_subscription() -> Result<(), subxt::BasicError> {
+    env_logger::try_init().ok();
+    let ctx = test_context().await;
+
+    // Subscribe to balance transfer events, ignoring all else.
+    let event_sub = ctx.api.events().subscribe().await?.filter_map(|events| {
+        async move {
+            let events = events.ok()?;
+            events
+                .find_first_event::<balances::events::Transfer>()
+                .ok()?
+        }
+    });
+
+    // Calling `.next()` on the above borrows it, and the `filter_map`
+    // means it's no longer `Unpin`, so we pin it on the stack:
+    futures::pin_mut!(event_sub);
+
+    // Make a transfer:
+    let alice = pair_signer(AccountKeyring::Alice.pair());
+    let bob = AccountKeyring::Bob.to_account_id();
+    ctx.api
+        .tx()
+        .balances()
+        .transfer(bob.clone().into(), 10_000)
+        .sign_and_submit_then_watch(&alice)
+        .await?;
+
+    // Wait for the next balance transfer event in our subscription stream
+    // and check that it lines up:
+    let event = event_sub.next().await.unwrap();
+    assert_eq!(
+        event,
+        balances::events::Transfer {
+            from: alice.account_id().clone(),
+            to: bob.clone(),
+            amount: 10_000
+        }
+    );
+
+    Ok(())
+}

subxt/tests/integration/frame/balances.rs

@@ -31,9 +31,7 @@ use sp_core::{
 };
 use sp_keyring::AccountKeyring;
 use subxt::{
-    DefaultConfig,
     Error,
-    EventSubscription,
     Signer,
 };
 
@@ -189,38 +187,6 @@ async fn transfer_error() {
     }
 }
 
-#[async_std::test]
-async fn transfer_subscription() {
-    env_logger::try_init().ok();
-    let alice = pair_signer(AccountKeyring::Alice.pair());
-    let bob = AccountKeyring::Bob.to_account_id();
-    let bob_addr = bob.clone().into();
-    let cxt = test_context().await;
-    let sub = cxt.client().rpc().subscribe_events().await.unwrap();
-    let decoder = cxt.client().events_decoder();
-    let mut sub = EventSubscription::<DefaultConfig>::new(sub, decoder);
-    sub.filter_event::<balances::events::Transfer>();
-
-    cxt.api
-        .tx()
-        .balances()
-        .transfer(bob_addr, 10_000)
-        .sign_and_submit_then_watch(&alice)
-        .await
-        .unwrap();
-
-    let raw = sub.next().await.unwrap().unwrap();
-    let event = balances::events::Transfer::decode(&mut &raw.data[..]).unwrap();
-    assert_eq!(
-        event,
-        balances::events::Transfer {
-            from: alice.account_id().clone(),
-            to: bob.clone(),
-            amount: 10_000
-        }
-    );
-}
-
 #[async_std::test]
 async fn transfer_implicit_subscription() {
     env_logger::try_init().ok();

subxt/tests/integration/frame/contracts.rs

@@ -18,6 +18,7 @@ use sp_keyring::AccountKeyring;
 
 use crate::{
     node_runtime::{
+        self,
         contracts::{
             calls::TransactionApi,
             events,
@@ -150,8 +151,10 @@ impl ContractsTestContext {
         &self,
         contract: AccountId,
         input_data: Vec<u8>,
-    ) -> Result<TransactionProgress<'_, DefaultConfig, DispatchError>, Error<DispatchError>>
+    ) -> Result<
-    {
+        TransactionProgress<'_, DefaultConfig, DispatchError, node_runtime::Event>,
+        Error<DispatchError>,
+    > {
         log::info!("call: {:?}", contract);
         let result = self
             .contracts_tx()

subxt/tests/integration/frame/staking.rs

@@ -180,7 +180,7 @@ async fn chill_works_for_controller_only() -> Result<(), Error<DispatchError>> {
         .await?
         .wait_for_finalized_success()
         .await?
-        .has_event::<staking::events::Chilled>()?;
+        .has::<staking::events::Chilled>()?;
     assert!(is_chilled);
 
     Ok(())

subxt/tests/integration/frame/sudo.rs

@@ -48,7 +48,7 @@ async fn test_sudo() -> Result<(), subxt::Error<DispatchError>> {
         .await?
         .wait_for_finalized_success()
         .await?
-        .has_event::<sudo::events::Sudid>()?;
+        .has::<sudo::events::Sudid>()?;
 
     assert!(found_event);
     Ok(())
@@ -74,7 +74,7 @@ async fn test_sudo_unchecked_weight() -> Result<(), subxt::Error<DispatchError>>
         .await?
         .wait_for_finalized_success()
         .await?
-        .has_event::<sudo::events::Sudid>()?;
+        .has::<sudo::events::Sudid>()?;
 
     assert!(found_event);
     Ok(())

subxt/tests/integration/frame/system.rs

@@ -56,7 +56,7 @@ async fn tx_remark_with_event() -> Result<(), subxt::Error<DispatchError>> {
         .await?
         .wait_for_finalized_success()
         .await?
-        .has_event::<system::events::Remarked>()?;
+        .has::<system::events::Remarked>()?;
 
     assert!(found_event);
     Ok(())

subxt/tests/integration/main.rs

@@ -20,7 +20,9 @@ mod utils;
 #[cfg(test)]
 mod client;
 #[cfg(test)]
+mod events;
+#[cfg(test)]
 mod frame;
 
-pub use test_runtime::node_runtime;
+use test_runtime::node_runtime;
-pub use utils::*;
+use utils::*;

subxt/tests/integration/utils/context.rs

@@ -14,7 +14,7 @@
 // You should have received a copy of the GNU General Public License
 // along with subxt.  If not, see <http://www.gnu.org/licenses/>.
 
-pub use crate::{
+pub(crate) use crate::{
     node_runtime,
     TestNodeProcess,
 };