[WIP] Notifications and pubsub transports

crates/json-rpc/Cargo.toml

@@ -12,5 +12,6 @@ exclude.workspace = true
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
 [dependencies]
+alloy-primitives.workspace = true
 serde = { workspace = true, features = ["derive"] }
-serde_json = { version = "1.0.103", features = ["raw_value"] }
+serde_json = { version = "1.0.103", features = ["raw_value"] }

crates/json-rpc/src/common.rs

@@ -9,6 +9,31 @@ pub enum Id {
     None,
 }
 
+impl PartialOrd for Id {
+    fn partial_cmp(&self, other: &Self) -> Option<std::cmp::Ordering> {
+        Some(self.cmp(other))
+    }
+}
+
+impl Ord for Id {
+    fn cmp(&self, other: &Self) -> std::cmp::Ordering {
+        // numbers < strings
+        // strings < null
+        // null == null
+        match (self, other) {
+            (Id::Number(a), Id::Number(b)) => a.cmp(b),
+            (Id::Number(_), _) => std::cmp::Ordering::Less,
+
+            (Id::String(_), Id::Number(_)) => std::cmp::Ordering::Greater,
+            (Id::String(a), Id::String(b)) => a.cmp(b),
+            (Id::String(_), Id::None) => std::cmp::Ordering::Less,
+
+            (Id::None, Id::None) => std::cmp::Ordering::Equal,
+            (Id::None, _) => std::cmp::Ordering::Greater,
+        }
+    }
+}
+
 impl Id {
     /// Returns `true` if the ID is a number.
     pub fn is_number(&self) -> bool {

crates/json-rpc/src/lib.rs

@@ -11,6 +11,9 @@
 
 use serde::{de::DeserializeOwned, Serialize};
 
+mod notification;
+pub use notification::{EthNotification, PubSubItem};
+
 mod request;
 pub use request::Request;
 

crates/json-rpc/src/notification.rs

@@ -0,0 +1,35 @@
+use alloy_primitives::U256;
+use serde::{ser::SerializeMap, Deserialize, Serialize, Serializer};
+
+use crate::Response;
+
+/// An ethereum-style notification, not to be confused with a JSON-RPC
+/// notification.
+#[derive(Debug, Clone, Deserialize)]
+pub struct EthNotification<T = Box<serde_json::value::RawValue>> {
+    pub subscription: U256,
+    pub result: T,
+}
+
+impl<T> Serialize for EthNotification<T>
+where
+    T: Serialize,
+{
+    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
+    where
+        S: Serializer,
+    {
+        let mut state = serializer.serialize_map(Some(2))?;
+        state.serialize_entry("subscription", &self.subscription)?;
+        state.serialize_entry("result", &self.result)?;
+        state.end()
+    }
+}
+
+/// An item received from a pubsub transport.
+#[derive(Debug, Clone, Deserialize)]
+#[serde(untagged)]
+pub enum PubSubItem {
+    Response(Response),
+    Notification(EthNotification),
+}

crates/json-rpc/src/request.rs

@@ -1,6 +1,7 @@
 use crate::{common::Id, RpcParam};
 
 use serde::{ser::SerializeMap, Deserialize, Serialize};
+use serde_json::value::RawValue;
 
 /// A JSON-RPC 2.0 request object.
 ///
@@ -12,12 +13,35 @@ use serde::{ser::SerializeMap, Deserialize, Serialize};
 ///
 /// The value of `method` should be known at compile time.
 #[derive(Debug, Deserialize, Clone)]
-pub struct Request<Params> {
+pub struct Request<Params = Box<RawValue>> {
     pub method: &'static str,
     pub params: Params,
     pub id: Id,
 }
 
+impl<Params> Request<Params>
+where
+    Params: RpcParam,
+{
+    /// Serialize the request parameters as a boxed [`RawValue`].
+    ///
+    /// # Panics
+    ///
+    /// If serialization of the params fails.
+    pub fn box_params(self) -> Request {
+        Request {
+            method: self.method,
+            params: RawValue::from_string(serde_json::to_string(&self.params).unwrap()).unwrap(),
+            id: self.id,
+        }
+    }
+
+    /// Convert the entire request to a boxed [`RawValue`].
+    pub fn to_boxed_raw_value(&self) -> serde_json::Result<Box<RawValue>> {
+        serde_json::to_string(&self).and_then(RawValue::from_string)
+    }
+}
+
 // manually implemented to avoid adding a type for the protocol-required
 // `jsonrpc` field
 impl<Params> Serialize for Request<Params>
@@ -32,7 +56,7 @@ where
         map.serialize_entry("method", self.method)?;
 
         // Params may be omitted if it is 0-sized
-        if !is_zst::<Params>() {
+        if std::mem::size_of::<Params>() != 0 {
             map.serialize_entry("params", &self.params)?;
         }
 
@@ -41,7 +65,3 @@ where
         map.end()
     }
 }
-
-fn is_zst<T>() -> bool {
-    std::mem::size_of::<T>() == 0
-}

crates/json-rpc/src/response.rs

@@ -1,12 +1,12 @@
-use std::fmt;
+use std::{fmt, marker::PhantomData};
 
 use serde::{
     de::{MapAccess, Visitor},
     Deserialize, Deserializer, Serialize,
 };
 use serde_json::value::RawValue;
 
-use crate::common::Id;
+use crate::{common::Id, RpcReturn};
 
 /// A JSONRPC-2.0 error object.
 ///
@@ -32,8 +32,8 @@ pub struct ErrorPayload {
 /// This type does not implement `Serialize` or `Deserialize` directly. It is
 /// deserialized as part of the [`Response`] type.
 #[derive(Debug, Clone)]
-pub enum ResponsePayload {
-    Success(Box<RawValue>),
+pub enum ResponsePayload<T = Box<RawValue>> {
+    Success(T),
     Error(ErrorPayload),
 }
 
@@ -43,12 +43,16 @@ pub enum ResponsePayload {
 /// either a successful result or an error. The `id` field is used to match
 /// the response to the request that it is responding to, and should be
 /// mirrored from the response.
-pub struct Response {
+#[derive(Debug, Clone)]
+pub struct Response<T = Box<RawValue>> {
     pub id: Id,
-    pub payload: ResponsePayload,
+    pub payload: ResponsePayload<T>,
 }
 
-impl<'de> Deserialize<'de> for Response {
+impl<'de, T> Deserialize<'de> for Response<T>
+where
+    T: RpcReturn,
+{
     fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
     where
         D: serde::Deserializer<'de>,
@@ -90,10 +94,13 @@ impl<'de> Deserialize<'de> for Response {
             }
         }
 
-        struct JsonRpcResponseVisitor;
+        struct JsonRpcResponseVisitor<T>(PhantomData<T>);
 
-        impl<'de> Visitor<'de> for JsonRpcResponseVisitor {
-            type Value = Response;
+        impl<'de, T> Visitor<'de> for JsonRpcResponseVisitor<T>
+        where
+            T: RpcReturn,
+        {
+            type Value = Response<T>;
 
             fn expecting(&self, formatter: &mut std::fmt::Formatter) -> std::fmt::Result {
                 formatter.write_str(
@@ -153,7 +160,7 @@ impl<'de> Deserialize<'de> for Response {
             }
         }
 
-        deserializer.deserialize_map(JsonRpcResponseVisitor)
+        deserializer.deserialize_map(JsonRpcResponseVisitor(PhantomData))
     }
 }
 

crates/transports/Cargo.toml

@@ -27,6 +27,13 @@ pin-project.workspace = true
 
 # feature deps
 reqwest = { version = "0.11.18", features = ["serde_json", "json"], optional = true }
+tokio = { version = "1.32.0", features = ["macros"] }
+bimap = "0.6.3"
+alloy-primitives.workspace = true
+tokio-tungstenite = "0.20.1"
+http = "0.2.9"
+tracing = "0.1.37"
+futures-util.workspace = true
 
 [target.'cfg(not(target_arch = "wasm32"))'.dependencies.hyper]
 version = "0.14.27"
@@ -36,4 +43,4 @@ features = ["full"]
 [features]
 default = ["reqwest", "hyper"]
 reqwest = ["dep:reqwest"]
-hyper = ["dep:hyper", "hyper/client"]
+hyper = ["dep:hyper", "hyper/client"]

crates/transports/src/error.rs

@@ -18,18 +18,27 @@ pub enum TransportError {
     #[error("Missing response in batch request")]
     MissingBatchResponse,
 
+    /// Custom error
     #[error(transparent)]
     Custom(Box<dyn StdError + Send + Sync + 'static>),
 
-    /// Hyper http transport
+    /// Hyper http transport.
     #[error(transparent)]
     #[cfg(feature = "reqwest")]
     Reqwest(#[from] reqwest::Error),
 
-    /// Hyper http transport
+    /// Hyper http transport.
     #[error(transparent)]
     #[cfg(all(not(target_arch = "wasm32"), feature = "hyper"))]
     Hyper(#[from] hyper::Error),
+
+    /// Tungstenite websocket transport.
+    #[error(transparent)]
+    Tungstenite(#[from] tokio_tungstenite::tungstenite::Error),
+
+    /// PubSub backend connection task has stopped.
+    #[error("PubSub backend connection task has stopped.")]
+    BackendGone,
 }
 
 impl TransportError {

crates/transports/src/lib.rs

@@ -17,10 +17,12 @@ pub use batch::BatchRequest;
 mod transports;
 pub use transports::{BoxTransport, Http, Transport};
 
-pub use alloy_json_rpc::RpcResult;
+mod pubsub;
+pub use pubsub::{BoxPubSub, ConnectionHandle, ConnectionInterface, PubSub, PubSubConnect};
 
 pub(crate) mod utils;
 
+pub use alloy_json_rpc::RpcResult;
 pub use type_aliases::*;
 
 #[cfg(not(target_arch = "wasm32"))]

crates/transports/src/pubsub/connect.rs

@@ -0,0 +1,79 @@
+use std::{future::Future, pin::Pin};
+
+use alloy_json_rpc::{Request, ResponsePayload, RpcParam};
+use tokio::sync::mpsc;
+
+use crate::{
+    pubsub::{
+        handle::ConnectionHandle, ix::PubSubInstruction, managers::InFlight, service::PubSubService,
+    },
+    TransportError,
+};
+
+#[derive(Debug, Clone)]
+pub struct ServiceFrontend {
+    pub tx: mpsc::UnboundedSender<PubSubInstruction>,
+}
+
+impl ServiceFrontend {
+    /// Send a request.
+    pub fn send<T>(
+        &self,
+        req: Request<T>,
+    ) -> Pin<Box<dyn Future<Output = Result<ResponsePayload, TransportError>> + Send>>
+    where
+        T: RpcParam,
+    {
+        let (in_flight, rx) = InFlight::new(req.box_params());
+        let ix = PubSubInstruction::Request(in_flight);
+        let tx = self.tx.clone();
+
+        Box::pin(async move {
+            tx.send(ix).map_err(|_| TransportError::BackendGone)?;
+            rx.await.map_err(|_| TransportError::BackendGone)?
+        })
+    }
+}
+
+/// Configuration objects that contain connection details for a backend.
+///
+/// Implementers should contain configuration options for the underlying
+/// transport.
+pub trait PubSubConnect: Sized + Send + Sync + 'static {
+    /// Returned by the `connect` and `into_service` methods if connection
+    /// fails.
+    type Error: std::error::Error + Send + Sync + 'static;
+
+    /// Spawn the backend, returning a handle to it.
+    ///
+    /// This function MUST create a long-lived task containing a
+    /// [`ConnectionInterface`], and return the corresponding handle.
+    ///
+    /// [`ConnectionInterface`]: crate::pubsub::ConnectionInterface
+    fn connect<'a: 'b, 'b>(
+        &'a self,
+    ) -> Pin<Box<dyn Future<Output = Result<ConnectionHandle, Self::Error>> + Send + 'b>>;
+
+    /// Convert the configuration object into a service with a running backend.
+    fn into_service(
+        self,
+    ) -> Pin<Box<dyn Future<Output = Result<ServiceFrontend, Self::Error>> + Send>> {
+        Box::pin(async move {
+            let handle = self.connect().await?;
+            let (tx, reqs) = mpsc::unbounded_channel();
+
+            let service_handle = ServiceFrontend { tx };
+            let service = PubSubService {
+                handle,
+                connector: self,
+                reqs,
+                subs: Default::default(),
+                in_flights: Default::default(),
+            };
+
+            service.spawn();
+
+            Ok(service_handle)
+        })
+    }
+}