feat(example): use changeset staging with rpc polling example

Author: evanlinjin

example-crates/example_bitcoind_rpc_polling/src/main.rs

@@ -11,12 +11,12 @@ use bdk_bitcoind_rpc::{
     bitcoincore_rpc::{Auth, Client, RpcApi},
     Emitter,
 };
-use bdk_chain::persist::PersistBackend;
+use bdk_chain::persist::{PersistBackend, StageExt};
 use bdk_chain::{
     bitcoin::{constants::genesis_block, Block, Transaction},
     indexed_tx_graph, keychain,
     local_chain::{self, LocalChain},
-    ConfirmationTimeHeightAnchor, IndexedTxGraph,
+    Append, ConfirmationTimeHeightAnchor, IndexedTxGraph,
 };
 use example_cli::{
     anyhow,
@@ -176,6 +176,7 @@ fn main() -> anyhow::Result<()> {
             let chain_tip = chain.lock().unwrap().tip();
             let rpc_client = rpc_args.new_client()?;
             let mut emitter = Emitter::new(&rpc_client, chain_tip, fallback_height);
+            let mut db_stage = ChangeSet::default();
 
             let mut last_db_commit = Instant::now();
             let mut last_print = Instant::now();
@@ -185,17 +186,18 @@ fn main() -> anyhow::Result<()> {
 
                 let mut chain = chain.lock().unwrap();
                 let mut graph = graph.lock().unwrap();
-                let mut db = db.lock().unwrap();
 
                 let chain_changeset = chain
                     .apply_update(emission.checkpoint)
                     .expect("must always apply as we receive blocks in order from emitter");
                 let graph_changeset = graph.apply_block_relevant(&emission.block, height);
-                db.write_changes(&(chain_changeset, graph_changeset))?;
+                db_stage.append((chain_changeset, graph_changeset));
 
                 // commit staged db changes in intervals
                 if last_db_commit.elapsed() >= DB_COMMIT_DELAY {
+                    let db = &mut *db.lock().unwrap();
                     last_db_commit = Instant::now();
+                    db_stage.commit_to(db)?;
                     println!(
                         "[{:>10}s] committed to db (took {}s)",
                         start.elapsed().as_secs_f32(),
@@ -230,8 +232,11 @@ fn main() -> anyhow::Result<()> {
                 mempool_txs.iter().map(|(tx, time)| (tx, *time)),
             );
             {
-                let mut db = db.lock().unwrap();
-                db.write_changes(&(local_chain::ChangeSet::default(), graph_changeset))?;
+                let db = &mut *db.lock().unwrap();
+                db_stage.append_and_commit_to(
+                    (local_chain::ChangeSet::default(), graph_changeset),
+                    db,
+                )?;
             }
         }
         RpcCommands::Live { rpc_args } => {
@@ -287,9 +292,9 @@ fn main() -> anyhow::Result<()> {
             let mut tip_height = 0_u32;
             let mut last_db_commit = Instant::now();
             let mut last_print = Option::<Instant>::None;
+            let mut db_stage = ChangeSet::default();
 
             for emission in rx {
-                let mut db = db.lock().unwrap();
                 let mut graph = graph.lock().unwrap();
                 let mut chain = chain.lock().unwrap();
 
@@ -314,11 +319,12 @@ fn main() -> anyhow::Result<()> {
                         continue;
                     }
                 };
-
-                db.write_changes(&changeset)?;
+                db_stage.append(changeset);
 
                 if last_db_commit.elapsed() >= DB_COMMIT_DELAY {
+                    let db = &mut *db.lock().unwrap();
                     last_db_commit = Instant::now();
+                    db_stage.commit_to(db)?;
                     println!(
                         "[{:>10}s] committed to db (took {}s)",
                         start.elapsed().as_secs_f32(),