XXX/lightclient: Use unstable backend

Signed-off-by: Alexandru Vasile <alexandru.vasile@parity.io>

subxt/Cargo.toml

@@ -61,7 +61,7 @@ unstable-metadata = []
 
 # Activate this to expose the Light Client functionality.
 # Note that this feature is experimental and things may break or not work as expected.
-unstable-light-client = ["subxt-lightclient", "tokio-stream"]
+unstable-light-client = ["subxt-lightclient", "tokio-stream", "tokio"]
 
 [dependencies]
 async-trait = { workspace = true }
@@ -118,6 +118,8 @@ getrandom = { workspace = true, optional = true }
 # Included if "native" feature is enabled
 tokio-util = { workspace = true, features = ["compat"], optional = true }
 
+tokio = { workspace = true, features = ["macros", "time", "rt-multi-thread"], optional = true }
+
 [dev-dependencies]
 bitvec = { workspace = true }
 codec = { workspace = true, features = ["derive", "bit-vec"] }

subxt/examples/light_client_parachains.rs

@@ -3,14 +3,14 @@
 //! Run: `cargo run --example light_client_parachains --features="unstable-light-client native"`.
 
 #![allow(missing_docs)]
+use sp_core::crypto::{AccountId32, Ss58Codec};
+use sp_core::ByteArray;
+use std::collections::BTreeSet;
 use std::{iter, num::NonZeroU32};
 use subxt::{
     client::{LightClient, RawLightClient},
     PolkadotConfig,
 };
-use sp_core::crypto::{AccountId32, Ss58Codec};
-use std::collections::BTreeSet;
-use sp_core::ByteArray;
 
 // Generate an interface that we can use from the node's metadata.
 #[subxt::subxt(runtime_metadata_path = "../artifacts/polkadot_metadata_small.scale")]
@@ -89,18 +89,38 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
         raw_light_client.for_chain(parachain_chain_id).await?;
 
     // Step 6. Subscribe to the finalized blocks of the chains.
-    let key = collectives::storage().fellowship_collective().members_iter();
-    let mut query = parachain_api.storage().at_latest().await.unwrap().iter(key).await?;
-    let mut members = BTreeSet::new();
+    {
+        let key = collectives::storage()
+            .fellowship_collective()
+            .members_iter();
+        let mut query = parachain_api
+            .storage()
+            .at_latest()
+            .await
+            .unwrap()
+            .iter(key)
+            .await?;
+        let mut members = BTreeSet::new();
+        let mut members_by_key = BTreeSet::new();
 
-    while let Some(Ok((id, fellow))) = query.next().await {
-        let account = AccountId32::from_slice(&id[id.len() - 32..]).unwrap();
+        while let Some(Ok((id, fellow))) = query.next().await {
+            let account = AccountId32::from_slice(&id[id.len() - 32..]).unwrap();
 
-        println!("Fetched member: {} rank {}", account.to_ss58check(), fellow.rank);
-        if members.contains(&account) {
-            println!("ERROR: Fetched account twice");
+            println!(
+                "Fetched member: {} rank {}",
+                account.to_ss58check(),
+                fellow.rank
+            );
+            if members.contains(&account) {
+                let cont = members_by_key.contains(&id);
+                println!(
+                    "ERROR: Fetched account twice. However is the key already inserted? {}",
+                    cont
+                );
+            }
+            members.insert(account);
+            members_by_key.insert(id);
         }
-        members.insert(account);
     }
 
     Ok(())

subxt/src/client/light_client/mod.rs

@@ -93,7 +93,33 @@ impl RawLightClient {
     ) -> Result<LightClient<TChainConfig>, crate::Error> {
         let raw_rpc = self.raw_rpc.for_chain(chain_id);
         let rpc_client = RpcClient::new(raw_rpc.clone());
-        let client = OnlineClient::<TChainConfig>::from_rpc_client(rpc_client).await?;
+
+        use crate::backend::unstable::UnstableBackend;
+        use std::sync::Arc;
+
+        let (backend, mut driver) = UnstableBackend::builder().build(rpc_client);
+
+        let future = async move {
+            use futures::StreamExt;
+            while let Some(val) = driver.next().await {
+                if let Err(e) = val {
+                    // This is a test; bail if something does wrong and try to
+                    // ensure that the message makes it to some logs.
+                    eprintln!("Error driving unstable backend in tests (will panic): {e}");
+                    panic!("Error driving unstable backend in tests: {e}");
+                }
+            }
+        };
+
+        // The unstable backend needs driving:
+        #[cfg(feature = "native")]
+        tokio::spawn(future);
+        #[cfg(feature = "web")]
+        wasm_bindgen_futures::spawn_local(future);
+
+        let client = OnlineClient::from_backend(Arc::new(backend))
+            .await
+            .map_err(|e| format!("Cannot construct OnlineClient from backend: {e}"))?;
 
         Ok(LightClient { client, chain_id })
     }