alloy_provider/provider/
root.rs

1use crate::{
2    blocks::NewBlocks,
3    heart::{Heartbeat, HeartbeatHandle},
4    Identity, ProviderBuilder,
5};
6use alloy_network::{Ethereum, Network};
7use alloy_rpc_client::{BuiltInConnectionString, ClientBuilder, ClientRef, RpcClient, WeakClient};
8use alloy_transport::{TransportConnect, TransportError};
9use std::{
10    fmt,
11    marker::PhantomData,
12    sync::{Arc, OnceLock},
13};
14
15#[cfg(feature = "pubsub")]
16use alloy_pubsub::{PubSubFrontend, Subscription};
17
18/// The root provider manages the RPC client and the heartbeat. It is at the
19/// base of every provider stack.
20///
21/// Cloning a root provider is cheap: clones share the client and heartbeat. Some deferred APIs,
22/// including default call builders, filter pollers, block or log watch builders, and subscription
23/// request builders, keep only a weak client handle. Keep at least one provider clone alive until
24/// those builders are awaited or their streams are consumed, or they may report that the backend
25/// was dropped or end early. A [`PendingTransactionBuilder`](crate::PendingTransactionBuilder)
26/// instead owns a root-provider clone. Layers can also retain additional state; for example,
27/// batched calls keep their batching backend alive.
28pub struct RootProvider<N: Network = Ethereum> {
29    /// The inner state of the root provider.
30    pub(crate) inner: Arc<RootProviderInner<N>>,
31}
32
33impl<N: Network> Clone for RootProvider<N> {
34    fn clone(&self) -> Self {
35        Self { inner: self.inner.clone() }
36    }
37}
38
39impl<N: Network> fmt::Debug for RootProvider<N> {
40    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
41        f.debug_struct("RootProvider").field("client", &self.inner.client).finish_non_exhaustive()
42    }
43}
44
45/// Helper function to directly access [`ProviderBuilder`] with minimal
46/// generics.
47pub fn builder<N: Network>() -> ProviderBuilder<Identity, Identity, N> {
48    ProviderBuilder::default()
49}
50
51impl<N: Network> RootProvider<N> {
52    /// Creates a new HTTP root provider from the given URL.
53    #[cfg(all(feature = "reqwest", not(all(target_os = "wasi", target_env = "p1"))))]
54    pub fn new_http(url: url::Url) -> Self {
55        Self::new(RpcClient::new_http(url))
56    }
57
58    /// Creates a new root provider from the given RPC client.
59    pub fn new(client: RpcClient) -> Self {
60        Self { inner: Arc::new(RootProviderInner::new(client)) }
61    }
62
63    /// Creates a new root provider from the provided string.
64    ///
65    /// See [`BuiltInConnectionString`] for more information.
66    pub async fn connect(s: &str) -> Result<Self, TransportError> {
67        Self::connect_with(s.parse::<BuiltInConnectionString>()?).await
68    }
69
70    /// Connects to a transport with the given connector.
71    pub async fn connect_with<C: TransportConnect>(conn: C) -> Result<Self, TransportError> {
72        ClientBuilder::default().connect_with(conn).await.map(Self::new)
73    }
74}
75
76impl<N: Network> RootProvider<N> {
77    /// Gets the subscription corresponding to the given RPC subscription ID.
78    #[cfg(feature = "pubsub")]
79    pub async fn get_subscription<R: alloy_json_rpc::RpcRecv>(
80        &self,
81        id: alloy_primitives::B256,
82    ) -> alloy_transport::TransportResult<Subscription<R>> {
83        self.pubsub_frontend()?.get_subscription(id).await.map(Subscription::from)
84    }
85
86    /// Unsubscribes from the subscription corresponding to the given RPC subscription ID.
87    #[cfg(feature = "pubsub")]
88    pub fn unsubscribe(&self, id: alloy_primitives::B256) -> alloy_transport::TransportResult<()> {
89        self.pubsub_frontend()?.unsubscribe(id)
90    }
91
92    #[cfg(feature = "pubsub")]
93    pub(crate) fn pubsub_frontend(&self) -> alloy_transport::TransportResult<&PubSubFrontend> {
94        self.inner
95            .client_ref()
96            .pubsub_frontend()
97            .ok_or_else(alloy_transport::TransportErrorKind::pubsub_unavailable)
98    }
99
100    #[inline]
101    pub(crate) fn get_heart(&self) -> &HeartbeatHandle {
102        self.inner.heart.get_or_init(|| {
103            let new_blocks = NewBlocks::<N>::new(self.inner.weak_client());
104            let paused = new_blocks.paused.clone();
105            let stream = new_blocks.into_stream();
106            Heartbeat::<N, _>::new(Box::pin(stream), paused).spawn()
107        })
108    }
109}
110
111/// The root provider manages the RPC client and the heartbeat. It is at the
112/// base of every provider stack.
113pub(crate) struct RootProviderInner<N: Network = Ethereum> {
114    client: RpcClient,
115    heart: OnceLock<HeartbeatHandle>,
116    _network: PhantomData<N>,
117}
118
119impl<N: Network> Clone for RootProviderInner<N> {
120    fn clone(&self) -> Self {
121        Self { client: self.client.clone(), heart: self.heart.clone(), _network: PhantomData }
122    }
123}
124
125impl<N: Network> RootProviderInner<N> {
126    pub(crate) fn new(client: RpcClient) -> Self {
127        Self { client, heart: Default::default(), _network: PhantomData }
128    }
129
130    pub(crate) fn weak_client(&self) -> WeakClient {
131        self.client.get_weak()
132    }
133
134    pub(crate) fn client_ref(&self) -> ClientRef<'_> {
135        self.client.get_ref()
136    }
137}
alloy_provider/provider/root.rs

alloy_provider/provider/
root.rs
