Improve dynamic events type sizes API (#61)

* Do not error if type sizes missing: log warn

* Split out EventsSubscriber, allow updating events decoder

* Format code

* Check missing type sizes after registration

* Add example which for submit and watching events

examples/submit_and_watch.rs

@@ -0,0 +1,54 @@
+// Copyright 2019 Parity Technologies (UK) Ltd.
+// This file is part of substrate-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 substrate-subxt.  If not, see <http://www.gnu.org/licenses/>.
+
+use futures::future::Future;
+use substrate_subxt::{balances, system::System, DefaultNodeRuntime as Runtime};
+use sp_keyring::AccountKeyring;
+
+type AccountId = <Runtime as System>::AccountId;
+type Balance = <Runtime as balances::Balances>::Balance;
+
+fn main() {
+    env_logger::init();
+    let signer = AccountKeyring::Alice.pair();
+
+    let dest = AccountKeyring::Bob.to_account_id();
+
+    let fut = substrate_subxt::ClientBuilder::<Runtime>::new()
+        .build()
+        .and_then(|cli| cli.xt(signer, None))
+        .and_then(move |xt| {
+            xt.watch()
+                .events_decoder(|decoder| {
+                    // for any primitive event with no type size registered
+                    decoder.register_type_size::<(u64, u64)>("IdentificationTuple")
+                })
+                .submit(balances::transfer::<Runtime>(dest.clone().into(), 10_000))
+        });
+
+    let mut rt = tokio::runtime::Runtime::new().unwrap();
+    match rt.block_on(fut) {
+        Ok(extrinsic_success) => {
+            match extrinsic_success.find_event::<(AccountId, AccountId, Balance, Balance)>("Balances", "Transfer") {
+                Some(Ok((_from, _to, value, _fees))) =>
+                    println!("Balance transfer success: value: {:?}", value),
+                Some(Err(err)) => println!("Failed to decode code hash: {}", err),
+                None => println!("Failed to find Contracts::CodeStored Event"),
+            }
+        },
+        Err(err) => println!("Error: {}", err)
+    }
+}

src/events.rs

@@ -43,9 +43,9 @@ use crate::{
         Metadata,
         MetadataError,
     },
+    Phase,
     System,
     SystemEvent,
-    Phase,
 };
 
 /// Top level Event that can be produced by a substrate runtime
@@ -72,18 +72,10 @@ pub enum EventsError {
     CodecError(#[from] CodecError),
     #[error("Metadata error: {0:?}")]
     Metadata(#[from] MetadataError),
-    #[error("Type Sizes Missing: {0:?}")]
-    TypeSizesMissing(Vec<String>),
     #[error("Type Sizes Unavailable: {0:?}")]
     TypeSizeUnavailable(String),
 }
 
-impl From<Vec<String>> for EventsError {
-    fn from(error: Vec<String>) -> Self {
-        EventsError::TypeSizesMissing(error)
-    }
-}
-
 pub struct EventsDecoder<T> {
     metadata: Metadata, // todo: [AJ] borrow?
     type_sizes: HashMap<String, usize>,
@@ -120,16 +112,6 @@ impl<T: System + Balances + 'static> TryFrom<Metadata> for EventsDecoder<T> {
         // VoteThreshold enum index
         decoder.register_type_size::<u8>("VoteThreshold")?;
 
-        // Ignore these unregistered types, which are not fixed size primitives
-        decoder.check_missing_type_sizes(vec![
-            "DispatchInfo",
-            "DispatchError",
-            "Result<(), DispatchError>",
-            "OpaqueTimeSlot",
-            // FIXME: determine type size for the following if necessary/possible
-            "IdentificationTuple",
-            "AuthorityList",
-        ])?;
         Ok(decoder)
     }
 }
@@ -148,31 +130,28 @@ impl<T: System + Balances + 'static> EventsDecoder<T> {
         }
     }
 
-    fn check_missing_type_sizes<I: IntoIterator<Item = &'static str>>(
-        &self,
-        ignore: I,
-    ) -> Result<(), Vec<String>> {
+    pub fn check_missing_type_sizes(&self) {
         let mut missing = HashSet::new();
-        let mut ignore_set = HashSet::new();
-        ignore_set.extend(ignore);
         for module in self.metadata.modules_with_events() {
             for event in module.events() {
                 for arg in event.arguments() {
                     for primitive in arg.primitives() {
-                        if !self.type_sizes.contains_key(&primitive)
-                            && !ignore_set.contains(primitive.as_str())
+                        if module.name() != "System"
+                            && !self.type_sizes.contains_key(&primitive)
                             && !primitive.contains("PhantomData")
                         {
-                            missing.insert(primitive);
+                            missing.insert(format!("{}::{}::{}", module.name(), event.name, primitive));
                         }
                     }
                 }
             }
         }
-        if missing.is_empty() {
-            Ok(())
-        } else {
-            Err(missing.into_iter().collect())
+        if missing.len() > 0 {
+            log::warn!(
+                "The following primitive types do not have registered sizes: {:?} \
+                If any of these events are received, an error will occur since we cannot decode them",
+                missing
+            );
         }
     }
 

src/frame/contracts.rs

@@ -186,7 +186,8 @@ mod tests {
         let wasm = wabt::wat2wasm(CONTRACT).expect("invalid wabt");
 
         client.xt(signer, None).and_then(|xt| {
-            xt.submit_and_watch(super::put_code(500_000, wasm))
+            xt.watch()
+                .submit(super::put_code(500_000, wasm))
                 .map(|result| result.find_event::<T::Hash>(MODULE, events::CODE_STORED))
         })
     }
@@ -224,15 +225,19 @@ mod tests {
         println!("{:?}", code_hash);
 
         let instantiate = client.xt(signer, None).and_then(move |xt| {
-            xt.submit_and_watch(super::instantiate::<Runtime>(
-                100_000_000_000_000,
-                500_000,
-                code_hash,
-                Vec::new(),
-            ))
-            .map(|result| {
-                result.find_event::<(AccountId, AccountId)>(MODULE, events::INSTANTIATED)
-            })
+            xt.watch()
+                .submit(super::instantiate::<Runtime>(
+                    100_000_000_000_000,
+                    500_000,
+                    code_hash,
+                    Vec::new(),
+                ))
+                .map(|result| {
+                    result.find_event::<(AccountId, AccountId)>(
+                        MODULE,
+                        events::INSTANTIATED,
+                    )
+                })
         });
 
         let result = rt.block_on(instantiate).unwrap();

src/lib.rs

@@ -72,7 +72,7 @@ pub use self::{
     runtimes::*,
 };
 use self::{
-    events::EventsDecoder,
+    events::{EventsDecoder, EventsError},
     extrinsic::{
         DefaultExtra,
         SignedExtra,
@@ -80,9 +80,9 @@ use self::{
     frame::{
         balances::Balances,
         system::{
+            Phase,
             System,
             SystemEvent,
-            Phase,
             SystemStore,
         },
     },
@@ -209,8 +209,7 @@ impl<T: System + Balances + 'static, S: 'static> Client<T, S> {
 
     /// Get a block hash of the latest finalized block
     pub fn finalized_head(&self) -> impl Future<Item = T::Hash, Error = Error> {
-        self.connect()
-            .and_then(|rpc| rpc.finalized_head())
+        self.connect().and_then(|rpc| rpc.finalized_head())
     }
 
     /// Get a block
@@ -280,6 +279,7 @@ impl<T: System + Balances + 'static, S: 'static> Client<T, S> {
 }
 
 /// Transaction builder.
+#[derive(Clone)]
 pub struct XtBuilder<T: System, P, S> {
     client: Client<T, S>,
     nonce: T::Index,
@@ -373,21 +373,58 @@ where
     }
 
     /// Submits transaction to the chain and watch for events.
-    pub fn submit_and_watch<C: Encode>(
-        &self,
+    pub fn watch(self) -> EventsSubscriber<T, P, S> {
+        let metadata = self.client.metadata().clone();
+        let decoder = EventsDecoder::try_from(metadata).map_err(Into::into);
+        EventsSubscriber {
+            client: self.client.clone(),
+            builder: self,
+            decoder,
+        }
+    }
+}
+
+/// Submits an extrinsic and subscribes to the triggered events
+pub struct EventsSubscriber<T: System, P, S> {
+    client: Client<T, S>,
+    builder: XtBuilder<T, P, S>,
+    decoder: Result<EventsDecoder<T>, EventsError>,
+}
+
+impl<T: System + Balances + Send + Sync + 'static, P, S: 'static>
+    EventsSubscriber<T, P, S>
+where
+    P: Pair,
+    S: Verify + Codec + From<P::Signature>,
+    S::Signer: From<P::Public> + IdentifyAccount<AccountId = T::AccountId>,
+    T::Address: From<T::AccountId>,
+{
+    /// Access the events decoder for registering custom type sizes
+    pub fn events_decoder<F: FnOnce(&mut EventsDecoder<T>) -> Result<usize, EventsError>>(self, f: F) -> Self {
+        let mut this = self;
+        if let Ok(ref mut decoder) = this.decoder {
+            if let Err(err) = f(decoder) {
+                this.decoder = Err(err)
+            }
+        }
+        this
+    }
+
+    /// Submits transaction to the chain and watch for events.
+    pub fn submit<C: Encode>(
+        self,
         call: Call<C>,
     ) -> impl Future<Item = ExtrinsicSuccess<T>, Error = Error> {
         let cli = self.client.connect();
-        let metadata = self.client.metadata().clone();
-        let decoder = EventsDecoder::try_from(metadata)
-            .into_future()
-            .map_err(Into::into);
+        let decoder = self.decoder.into_future().map_err(Into::into);
 
-        self.create_and_sign(call)
+        self.builder
+            .create_and_sign(call)
             .into_future()
             .map_err(Into::into)
             .join(decoder)
             .and_then(move |(extrinsic, decoder)| {
+                decoder.check_missing_type_sizes();
                 cli.and_then(move |rpc| {
                     rpc.submit_and_watch_extrinsic(extrinsic, decoder)
                 })

src/metadata.rs

@@ -244,7 +244,7 @@ impl<K: Encode, V: Decode + Clone> StorageMap<K, V> {
                     .chain(&encoded_key)
                     .cloned()
                     .collect::<Vec<_>>()
-            },
+            }
             StorageHasher::Blake2_256 => sp_core::blake2_256(&encoded_key).to_vec(),
             StorageHasher::Twox128 => sp_core::twox_128(&encoded_key).to_vec(),
             StorageHasher::Twox256 => sp_core::twox_256(&encoded_key).to_vec(),

src/rpc.rs

@@ -170,9 +170,7 @@ impl<T: System> Rpc<T> {
 
     /// Get a block hash of the latest finalized block
     pub fn finalized_head(&self) -> impl Future<Item = T::Hash, Error = Error> {
-        self.chain
-            .finalized_head()
-            .map_err(Into::into)
+        self.chain.finalized_head().map_err(Into::into)
     }
 
     /// Get a Block