-
Notifications
You must be signed in to change notification settings - Fork 2
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
fix(cardano-chain-follower): Refactor cardano-chain-follower to fix c…
…lient concurrency bug (#182) * fix(cardano-chain-follower): Refactor cardano-chain-follower to fix client concurrency bug * chore(cardano-chain-follower): fix fmt * chore(cardano-chain-follower): fix spelling * chore(cardano-chain-follower): Add set read pointer example in cardano-chain-follower * chore(cardano-chain-follower): Add missing docs for struct fields * chore(cardano-chain-follower): Remove read task in favor of spawning read block and read block range futures * chore(cardano-chain-follower): Fix set_read_pointer example * chore(cardano-chain-follower): Fix examples docs * chore(cardano-chain-follower): Add example showing how to execute concurrent reads * chore(cardano-chain-follower): Update architecture docs * Update hermes/crates/cardano-chain-follower/src/follow.rs * Update hermes/crates/cardano-chain-follower/examples/set_read_pointer.rs --------- Co-authored-by: Steven Johnson <stevenj@users.noreply.github.com>
- Loading branch information
1 parent
9e17eaa
commit a9516fc
Showing
10 changed files
with
640 additions
and
618 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
78 changes: 78 additions & 0 deletions
78
hermes/crates/cardano-chain-follower/examples/concurrent_reads.rs
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,78 @@ | ||
//! This example shows how to use the chain follower to download arbitrary blocks | ||
//! from the chain concurrently. | ||
|
||
use std::error::Error; | ||
|
||
use cardano_chain_follower::{Follower, FollowerConfigBuilder, Network, Point}; | ||
use tracing::level_filters::LevelFilter; | ||
use tracing_subscriber::EnvFilter; | ||
|
||
#[tokio::main] | ||
async fn main() -> Result<(), Box<dyn Error>> { | ||
tracing_subscriber::fmt() | ||
.with_env_filter( | ||
EnvFilter::builder() | ||
.with_default_directive(LevelFilter::INFO.into()) | ||
.from_env_lossy(), | ||
) | ||
.init(); | ||
|
||
let config = FollowerConfigBuilder::default().build(); | ||
|
||
let follower = Follower::connect( | ||
"relays-new.cardano-mainnet.iohk.io:3001", | ||
Network::Mainnet, | ||
config, | ||
) | ||
.await?; | ||
|
||
let points = vec![ | ||
Point::Specific( | ||
110_908_236, | ||
hex::decode("ad3798a1db2b6097c71f35609399e4b2ff834f0f45939803d563bf9d660df2f2")?, | ||
), | ||
Point::Specific( | ||
110_908_582, | ||
hex::decode("16e97a73e866280582ee1201a5e1815993978eede956af1869b0733bedc131f2")?, | ||
), | ||
]; | ||
let mut point_count = points.len(); | ||
|
||
let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel(); | ||
|
||
for p in points { | ||
let slot_no = p.slot_or_default(); | ||
let r = follower.read_block(p); | ||
let r_tx = tx.clone(); | ||
|
||
tokio::spawn(async move { | ||
tracing::info!(slot_no, "Reading block"); | ||
let result = r.await; | ||
drop(r_tx.send(result)); | ||
}); | ||
} | ||
|
||
while let Some(result) = rx.recv().await { | ||
let block_data = result?; | ||
let block = block_data.decode()?; | ||
|
||
let total_fee = block | ||
.txs() | ||
.iter() | ||
.map(|tx| tx.fee().unwrap_or_default()) | ||
.sum::<u64>(); | ||
|
||
println!( | ||
"Block {} (slot {}) => total fee: {total_fee}", | ||
block.number(), | ||
block.slot() | ||
); | ||
|
||
point_count -= 1; | ||
if point_count == 0 { | ||
break; | ||
} | ||
} | ||
|
||
Ok(()) | ||
} |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
74 changes: 74 additions & 0 deletions
74
hermes/crates/cardano-chain-follower/examples/set_read_pointer.rs
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,74 @@ | ||
//! This example shows how to set the follower's read pointer without stopping it. | ||
|
||
use std::error::Error; | ||
|
||
use cardano_chain_follower::{ChainUpdate, Follower, FollowerConfigBuilder, Network, Point}; | ||
use tracing::level_filters::LevelFilter; | ||
use tracing_subscriber::EnvFilter; | ||
|
||
#[tokio::main] | ||
async fn main() -> Result<(), Box<dyn Error>> { | ||
tracing_subscriber::fmt() | ||
.with_env_filter( | ||
EnvFilter::builder() | ||
.with_default_directive(LevelFilter::INFO.into()) | ||
.from_env_lossy(), | ||
) | ||
.init(); | ||
|
||
// Defaults to start following from the tip. | ||
let config = FollowerConfigBuilder::default().build(); | ||
|
||
let mut follower = Follower::connect( | ||
"relays-new.cardano-mainnet.iohk.io:3001", | ||
Network::Mainnet, | ||
config, | ||
) | ||
.await?; | ||
|
||
let (tx, mut rx) = tokio::sync::oneshot::channel::<()>(); | ||
let mut pointer_set = false; | ||
tokio::spawn(async move { | ||
let _tx = tx; | ||
tokio::time::sleep(std::time::Duration::from_secs(2)).await; | ||
}); | ||
|
||
loop { | ||
tokio::select! { | ||
_ = &mut rx, if !pointer_set => { | ||
follower.set_read_pointer(Point::Specific( | ||
110_908_236, | ||
hex::decode("ad3798a1db2b6097c71f35609399e4b2ff834f0f45939803d563bf9d660df2f2")?, | ||
)).await?; | ||
println!("set read pointer"); | ||
|
||
pointer_set = true; | ||
} | ||
|
||
chain_update = follower.next() => { | ||
match chain_update? { | ||
ChainUpdate::Block(data) => { | ||
let block = data.decode()?; | ||
|
||
println!( | ||
"New block NUMBER={} SLOT={} HASH={}", | ||
block.number(), | ||
block.slot(), | ||
hex::encode(block.hash()), | ||
); | ||
}, | ||
ChainUpdate::Rollback(data) => { | ||
let block = data.decode()?; | ||
|
||
println!( | ||
"Rollback block NUMBER={} SLOT={} HASH={}", | ||
block.number(), | ||
block.slot(), | ||
hex::encode(block.hash()), | ||
); | ||
}, | ||
} | ||
} | ||
} | ||
} | ||
} |
Oops, something went wrong.