feat(rpc): add eth_pubsub skeleton (#354)

crates/net/rpc-api/src/eth_pubsub.rs

@@ -6,9 +6,9 @@ use reth_rpc_types::pubsub::{Kind, Params};
 pub trait EthPubSubApi {
     /// Create an ethereum subscription.
     #[subscription(
-        name = "eth_subscribe" => "eth_subscription",
+        name = "eth_subscribe",
         unsubscribe = "eth_unsubscribe",
         item = reth_rpc_types::pubsub::SubscriptionResult
     )]
-    fn eth_subscribe(&self, kind: Kind, params: Option<Params>);
+    fn subscribe(&self, kind: Kind, params: Option<Params>);
 }

crates/net/rpc/src/eth/api/mod.rs

@@ -0,0 +1,64 @@
+//! Provides everything related to `eth_` namespace
+
+use reth_interfaces::{
+    provider::{BlockProvider, StateProviderFactory},
+    Result,
+};
+use reth_primitives::{Transaction, U256, U64};
+use reth_transaction_pool::TransactionPool;
+use std::sync::Arc;
+
+mod server;
+
+/// `Eth` API implementation.
+///
+/// This type provides the functionality for handling `eth_` related requests.
+/// These are implemented two-fold: Core functionality is implemented as functions directly on this
+/// type. Additionally, the required server implementations (e.g. [`reth_rpc_api::EthApiServer`])
+/// are implemented separately in submodules. The rpc handler implementation can then delegate to
+/// the main impls. This way [`EthApi`] is not limited to [`jsonrpsee`] and can be used standalone
+/// or in other network handlers (for example ipc).
+#[derive(Debug, Clone)]
+pub struct EthApi<Pool, Client> {
+    /// All nested fields bundled together.
+    inner: Arc<EthApiInner<Pool, Client>>,
+}
+
+impl<Pool, Client> EthApi<Pool, Client>
+where
+    Pool: TransactionPool<Transaction = Transaction> + Clone,
+    Client: BlockProvider + StateProviderFactory,
+{
+    /// Creates a new, shareable instance.
+    pub fn new(client: Arc<Client>, pool: Pool) -> Self {
+        let inner = EthApiInner { client, pool };
+        Self { inner: Arc::new(inner) }
+    }
+
+    /// Returns the inner `Client`
+    fn client(&self) -> &Arc<Client> {
+        &self.inner.client
+    }
+
+    /// Returns the current ethereum protocol version.
+    ///
+    /// Note: This returns an `U64`, since this should return as hex string.
+    pub fn protocol_version(&self) -> U64 {
+        1u64.into()
+    }
+
+    /// Returns the best block number
+    pub fn block_number(&self) -> Result<U256> {
+        Ok(self.client().chain_info()?.best_number.into())
+    }
+}
+
+/// Container type `EthApi`
+#[derive(Debug)]
+struct EthApiInner<Pool, Client> {
+    /// The transaction pool.
+    pool: Pool,
+    /// The client that can interact with the chain.
+    client: Arc<Client>,
+    // TODO needs network access to handle things like `eth_syncing`
+}

crates/net/rpc/src/eth/api/server.rs

@@ -1,7 +1,7 @@
 //! Implementation of the [`jsonrpsee`] generated [`reth_rpc_api::EthApiServer`] trait
-//! implementation for handling RPC requests for he `eth_` namespace.
+//! Handles RPC requests for he `eth_` namespace.
 
-use crate::{eth::EthApi, result::ToRpcResult};
+use crate::{eth::api::EthApi, result::ToRpcResult};
 use jsonrpsee::core::RpcResult as Result;
 use reth_interfaces::provider::{BlockProvider, StateProviderFactory};
 use reth_primitives::{

crates/net/rpc/src/eth/mod.rs

@@ -1,64 +1,7 @@
-//! Provides everything related to `eth_` namespace
+//! `eth` namespace handler implementation.
 
-use reth_interfaces::{
-    provider::{BlockProvider, StateProviderFactory},
-    Result,
-};
-use reth_primitives::{Transaction, U256, U64};
-use reth_transaction_pool::TransactionPool;
-use std::sync::Arc;
+mod api;
+mod pubsub;
 
-mod eth_server;
-
-/// `Eth` API implementation.
-///
-/// This type provides the functionality for handling `eth_` related requests.
-/// These are implemented two-fold: Core functionality is implemented as functions directly on this
-/// type. Additionally, the required server implementations (e.g. [`reth_rpc_api::EthApiServer`])
-/// are implemented separately in submodules. The rpc handler implementation can then delegate to
-/// the main impls. This way [`EthApi`] is not limited to [`jsonrpsee`] and can be used standalone
-/// or in other network handlers (for example ipc).
-#[derive(Debug, Clone)]
-pub struct EthApi<Pool, Client> {
-    /// All nested fields bundled together.
-    inner: Arc<EthApiInner<Pool, Client>>,
-}
-
-impl<Pool, Client> EthApi<Pool, Client>
-where
-    Pool: TransactionPool<Transaction = Transaction> + Clone,
-    Client: BlockProvider + StateProviderFactory,
-{
-    /// Creates a new, shareable instance.
-    pub fn new(client: Arc<Client>, pool: Pool) -> Self {
-        let inner = EthApiInner { client, pool };
-        Self { inner: Arc::new(inner) }
-    }
-
-    /// Returns the inner `Client`
-    fn client(&self) -> &Arc<Client> {
-        &self.inner.client
-    }
-
-    /// Returns the current ethereum protocol version.
-    ///
-    /// Note: This returns an `U64`, since this should return as hex string.
-    pub fn protocol_version(&self) -> U64 {
-        1u64.into()
-    }
-
-    /// Returns the best block number
-    pub fn block_number(&self) -> Result<U256> {
-        Ok(self.client().chain_info()?.best_number.into())
-    }
-}
-
-/// Container type `EthApi`
-#[derive(Debug)]
-struct EthApiInner<Pool, Client> {
-    /// The transaction pool.
-    pool: Pool,
-    /// The client that can interact with the chain.
-    client: Arc<Client>,
-    // TODO needs network access to handle things like `eth_syncing`
-}
+pub use api::EthApi;
+pub use pubsub::EthPubSub;

crates/net/rpc/src/eth/pubsub.rs

@@ -0,0 +1,63 @@
+//! `eth_` PubSub RPC handler implementation
+
+use jsonrpsee::{types::SubscriptionResult, SubscriptionSink};
+use reth_interfaces::provider::BlockProvider;
+use reth_rpc_api::EthPubSubApiServer;
+use reth_rpc_types::pubsub::{Kind, Params};
+use reth_transaction_pool::TransactionPool;
+use std::sync::Arc;
+
+/// `Eth` pubsub RPC implementation.
+///
+/// This handles
+#[derive(Debug, Clone)]
+pub struct EthPubSub<Pool, Client> {
+    /// All nested fields bundled together.
+    inner: Arc<EthPubSubInner<Pool, Client>>,
+}
+
+// === impl EthPubSub ===
+
+impl<Pool, Client> EthPubSub<Pool, Client> {
+    /// Creates a new, shareable instance.
+    pub fn new(client: Arc<Client>, pool: Pool) -> Self {
+        let inner = EthPubSubInner { client, pool };
+        Self { inner: Arc::new(inner) }
+    }
+}
+
+impl<Pool, Client> EthPubSubApiServer for EthPubSub<Pool, Client>
+where
+    Pool: TransactionPool + Clone,
+    Client: BlockProvider + 'static,
+{
+    fn subscribe(
+        &self,
+        mut sink: SubscriptionSink,
+        _kind: Kind,
+        _params: Option<Params>,
+    ) -> SubscriptionResult {
+        sink.accept()?;
+        todo!()
+    }
+}
+
+/// The actual handler for and accepted [`EthPubSub::subscribe`] call.
+async fn handle_accepted<Pool, Client>(
+    _pool: Pool,
+    _client: Arc<Client>,
+    _accepted_sink: SubscriptionSink,
+    _kind: Kind,
+    _params: Option<Params>,
+) {
+}
+
+/// Container type `EthApi`
+#[derive(Debug)]
+struct EthPubSubInner<Pool, Client> {
+    /// The transaction pool.
+    pool: Pool,
+    /// The client that can interact with the chain.
+    client: Arc<Client>,
+    // TODO needs spawn access
+}

crates/net/rpc/src/lib.rs

@@ -13,6 +13,6 @@
 
 mod eth;
 
-pub use eth::EthApi;
+pub use eth::{EthApi, EthPubSub};
 
 pub(crate) mod result;