feat: define subscription type (#2203)

* feat: define subscription type

* feat: `FallbackLayer` transport (#2135)

* feat: initial impl of fallback layer and service

* chore: released trait bounds

* chore: use random ping id

* chore: update fn name

* chore: update fn name

* chore: rm health check task and sample count

* fix: lock scope

* chore: adjust logs

* chore: small cleanup

* chore: cargo fmt

* chore: cargo fmt again

* chore: addressed review, dont take transports from the list on each request

* chore: rm private doc

* chore: update mod doc comment ordering

* chore: addressed review

---------

Co-authored-by: Jennifer <[email protected]>

* feat: add BlobAndProofV2 (#2202)

* feat: add BlobAndProofV2

* Update crates/eips/src/eip4844/engine.rs

---------

Co-authored-by: Matthias Seitz <[email protected]>

* fix: buffer size go brrr

* refactor: rename EthSuscribe to GetSubscription, naming is hard sigh

* feat:
- fix description
- add `method` member
- make params optional
- handle optional channel size and also handle params case and no params case separately

* chore: fmt, rename buffer to channel_size

---------

Co-authored-by: nicolas <[email protected]>
Co-authored-by: Jennifer <[email protected]>
Co-authored-by: Varun Doshi <[email protected]>
Co-authored-by: Matthias Seitz <[email protected]>

Author: 5 people

crates/provider/src/provider/mod.rs

@@ -27,3 +27,6 @@ pub use multicall::*;
 
 mod erased;
 pub use erased::DynProvider;
+
+mod subscription;
+pub use subscription::GetSubscription;

crates/provider/src/provider/subscription.rs

@@ -0,0 +1,99 @@
+use crate::{Provider, RootProvider};
+use alloy_json_rpc::{RpcRecv, RpcSend};
+use alloy_network::{Ethereum, Network};
+use alloy_transport::TransportResult;
+use std::{borrow::Cow, future::Future, pin::Pin};
+
+/// A general-purpose subscription request builder
+///
+/// This struct allows configuring subscription parameters and channel size
+/// before initiating a request to subscribe to Ethereum events.
+pub struct GetSubscription<P, R, N = Ethereum>
+where
+    P: RpcSend,
+    R: RpcRecv,
+    N: Network,
+{
+    root: RootProvider<N>,
+    method: Cow<'static, str>,
+    params: Option<P>,
+    channel_size: Option<usize>,
+    _marker: std::marker::PhantomData<fn() -> R>,
+}
+
+impl<P, R, N> GetSubscription<P, R, N>
+where
+    N: Network,
+    P: RpcSend,
+    R: RpcRecv,
+{
+    /// Creates a new [`GetSubscription`] instance
+    pub fn new(
+        root: RootProvider<N>,
+        method: impl Into<Cow<'static, str>>,
+        params: Option<P>,
+    ) -> Self {
+        Self {
+            root,
+            method: method.into(),
+            channel_size: None,
+            params,
+            _marker: std::marker::PhantomData,
+        }
+    }
+
+    /// Set the channel_size for the subscription stream.
+    pub fn channel_size(mut self, size: usize) -> Self {
+        self.channel_size = Some(size);
+        self
+    }
+}
+
+impl<P, R, N> core::fmt::Debug for GetSubscription<P, R, N>
+where
+    N: Network,
+    P: RpcSend,
+    R: RpcRecv,
+{
+    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
+        f.debug_struct("GetSubscription")
+            .field("channel_size", &self.channel_size)
+            .field("method", &self.method)
+            .finish()
+    }
+}
+
+#[cfg(feature = "pubsub")]
+impl<P, R, N> std::future::IntoFuture for GetSubscription<P, R, N>
+where
+    N: Network,
+    P: RpcSend + 'static,
+    R: RpcRecv,
+{
+    type Output = TransportResult<alloy_pubsub::Subscription<R>>;
+    type IntoFuture = Pin<Box<dyn Future<Output = Self::Output> + Send + 'static>>;
+
+    fn into_future(self) -> Self::IntoFuture {
+        Box::pin(async move {
+            let pubsub = self.root.pubsub_frontend()?;
+
+            // Set config channel size if any
+            if let Some(size) = self.channel_size {
+                pubsub.set_channel_size(size);
+            }
+
+            // Handle params and no-params case separately
+            let id = if let Some(params) = self.params {
+                let mut call = self.root.client().request(self.method, params);
+                call.set_is_subscription();
+                call.await?
+            } else {
+                let mut call = self.root.client().request_noparams(self.method);
+                call.set_is_subscription();
+                call.await?
+            };
+
+            self.root.get_subscription(id).await
+        })
+    }
+}