ClientRefs, Poller, and Streams #179
Conversation
prestwich
force-pushed
the
prestwich/block-heart
branch
from
a2a7f5d to
0188a78
Compare
crates/providers/src/chain.rs
Outdated
| // Then try to fill as many blocks as possible | ||
| while !known_blocks.contains(&block_number) { | ||
| let block = provider.get_block_by_number(block_number, false).await; |
There was a problem hiding this comment.
would consider doing a tokio::future::join_all if the gap between block_number and next_yield is >N blocks
crates/providers/src/heart.rs
Outdated
| impl WatchConfig { | ||
| /// Create a new watch for a transaction. | ||
| pub fn new(tx_hash: B256, tx: oneshot::Sender<()>) -> Self { | ||
| Self { tx_hash, confirmations: 0, timeout: Default::default(), tx } |
There was a problem hiding this comment.
I think?
| Self { tx_hash, confirmations: 0, timeout: Default::default(), tx } | |
| Self { tx_hash, confirmations: 1, timeout: Default::default(), tx } |
crates/rpc-client/src/poller.rs
Outdated
| let fut = async move { | ||
| let limit = self.limit.unwrap_or(usize::MAX); | ||
| for _ in 0..limit { | ||
| let client = match self.client.upgrade() { |
There was a problem hiding this comment.
do we want to upgrade each time in the loop?
There was a problem hiding this comment.
I believe we want to break if this loop is the only provider left
| /// An [`RpcClient`] in a [`Weak`] reference. | ||
| pub type WeakClient<T> = Weak<RpcClientInner<T>>; | ||
|
|
||
| /// A borrowed [`RpcClient`]. | ||
| pub type ClientRef<'a, T> = &'a RpcClientInner<T>; | ||
|
|
||
| /// A JSON-RPC client. | ||
| #[derive(Debug)] | ||
| pub struct RpcClient<T>(Arc<RpcClientInner<T>>); |
There was a problem hiding this comment.
Is there any amazing readout for how to think about Weak vs full Arc? https://doc.rust-lang.org/std/sync/struct.Weak.html
There was a problem hiding this comment.
Allows breaking out of task loops when the provider is dropped in addition to the channel dropping, makes sense with broadcast channels ig
| Some(Ok(value)) => match serde_json::from_str(value.get()) { | ||
| Ok(item) => return task::Poll::Ready(Some(Ok(item))), | ||
| Err(e) => { | ||
| trace!(value = value.get(), error = ?e, "Received unexpected value in subscription."); | ||
| continue; | ||
| } | ||
| }, |
| /// stream will attempt to deserialize the notifications and yield the | ||
| /// `serde_json::Result` of the deserialization. | ||
| #[derive(Debug)] | ||
| pub struct SubResultStream<T> { |
There was a problem hiding this comment.
Should these be called SubscriptionStream & SubscriptionStreamResult?
| } | ||
|
|
||
| /// A stream of notifications from the server, identified by a local ID. This | ||
| /// stream may yield unexpected types. |
There was a problem hiding this comment.
what's an example of an unexpected type?
crates/providers/src/heart.rs
Outdated
|
|
||
| /// Reap any timeout | ||
| fn reap_timeouts(&mut self) { | ||
| let now = std::time::Instant::now(); | ||
| let to_keep = self.reap_at.split_off(&now); | ||
| let to_reap = std::mem::replace(&mut self.reap_at, to_keep); | ||
|
|
||
| for tx_hash in to_reap.values() { | ||
| self.unconfirmed.remove(tx_hash); | ||
| } | ||
| } |
There was a problem hiding this comment.
Reap is basically "unsubscribe from transactions that have been pending for longer than T"?
crates/providers/src/heart.rs
Outdated
| /// Handle a new block by checking if any of the transactions we're | ||
| /// watching are in it, and if so, notifying the watcher. Also updates | ||
| /// the latest block. | ||
| fn handle_new_block(&mut self, block: Block, latest: &watch::Sender<Block>) { |
| let http = Http::<Client>::new(url); | ||
| let provider = RootProvider::<TmpNetwork, _>::new(RpcClient::new(http, true)); |
There was a problem hiding this comment.
we should sugar this to Provider::new(url) and have it work for all HTTP/WS/IPC etc.
crates/providers/src/new.rs
Outdated
| let pending_tx = provider.send_transaction(&TxLegacy(tx)).await.expect("failed to send tx"); | ||
| eprintln!("{pending_tx:?}"); | ||
| let () = pending_tx.await.expect("failed to await pending tx"); |
| @@ -552,7 +552,7 @@ impl<'a> TryFrom<&'a String> for Provider<Http<Client>> { | |||
| #[cfg(test)] | |||
| mod tests { | |||
| use crate::{ | |||
| provider::{Provider, TempProvider}, | |||
| tmp::{Provider, TempProvider}, | |||
There was a problem hiding this comment.
note we're gonna get out of the tmp stuff once @onbjerg work is done
| #[cfg_attr(target_arch = "wasm32", async_trait::async_trait(?Send))] | ||
| #[cfg_attr(not(target_arch = "wasm32"), async_trait::async_trait)] |
There was a problem hiding this comment.
nit should we start replacing these with native async
|
@DaniPopes let's review & get this in if feature complete and proceed with events? |
Separate
RpcClientfromRpcClientInnerto allow weak and lifetime-borrowed use of RpcClient without holding a reference beyondMotivation
Solution
PR Checklist