fix: manually impl deser of pubsubitem

alloy-rs · prestwich · Nov 9, 2023 · Nov 6, 2023 · Nov 6, 2023 · Nov 7, 2023
commit 06e2606d0c088fa83fad25d777372b28120ae0fc
@@ -1,5 +1,8 @@
 use alloy_primitives::U256;
-use serde::{Deserialize, Serialize};
+use serde::{
+    de::{MapAccess, Visitor},
+    Deserialize, Serialize,
+};
 
 use crate::Response;
 
@@ -14,9 +17,105 @@ pub struct EthNotification<T = Box<serde_json::value::RawValue>> {
 /// An item received over an Ethereum pubsub transport. Ethereum pubsub uses a
 /// non-standard JSON-RPC notification format. An item received over a pubsub
 /// transport may be a JSON-RPC response or an Ethereum-style notification.
-#[derive(Debug, Clone, Deserialize)]
-#[serde(untagged)]
+#[derive(Debug, Clone)]
 pub enum PubSubItem {
     Response(Response),
     Notification(EthNotification),
 }
+
+impl<'de> Deserialize<'de> for PubSubItem {
+    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
+    where
+        D: serde::Deserializer<'de>,
+    {
+        struct PubSubItemVisitor;
+
+        impl<'de> Visitor<'de> for PubSubItemVisitor {
+            type Value = PubSubItem;
+
+            fn expecting(&self, formatter: &mut std::fmt::Formatter) -> std::fmt::Result {
+                formatter.write_str("a JSON-RPC response or an Ethereum-style notification")
+            }
+
+            fn visit_map<A>(self, mut map: A) -> Result<Self::Value, A::Error>
+            where
+                A: MapAccess<'de>,
+            {
+                let mut id = None;
+                let mut subscription = None;
+                let mut result = None;
+                let mut error = None;
+
+                while let Ok(Some(key)) = map.next_key() {
+                    match key {
+                        "id" => {
+                            if id.is_some() {
+                                return Err(serde::de::Error::duplicate_field("id"));
+                            }
+                            id = Some(map.next_value()?);
+                        }
+                        "subscription" => {
+                            if subscription.is_some() {
+                                return Err(serde::de::Error::duplicate_field("subscription"));
+                            }
+                            subscription = Some(map.next_value()?);
+                        }
+                        "result" => {
+                            if result.is_some() {
+                                return Err(serde::de::Error::duplicate_field("result"));
+                            }
+                            result = Some(map.next_value()?);
+                        }
+                        "error" => {
+                            if error.is_some() {
+                                return Err(serde::de::Error::duplicate_field("error"));
+                            }
+                            error = Some(map.next_value()?);
+                        }
+                        _ => {
+                            let _ = map.next_value::<serde_json::Value>()?;
+                        }
+                    }
+                }
+
+                if let Some(id) = id {
+                    if subscription.is_some() {
+                        return Err(serde::de::Error::custom(
+                            "unexpected subscription in pubsub item",
+                        ));
+                    }
+
+                    let payload = if error.is_some() {
+                        crate::ResponsePayload::Failure(error.unwrap())
+                    } else {
+                        if result.is_none() {
+                            return Err(serde::de::Error::missing_field("result"));
+                        }
+                        crate::ResponsePayload::Success(result.unwrap())
+                    };
+                    Ok(PubSubItem::Response(Response { id, payload }))
+                } else {
+                    if error.is_some() {
+                        return Err(serde::de::Error::custom(
+                            "unexpected `error` field in subscription notification",
+                        ));
+                    }
+                    if subscription.is_none() {
+                        return Err(serde::de::Error::missing_field("subscription"));
+                    }
+
+                    if result.is_none() {
+                        return Err(serde::de::Error::missing_field("result"));
+                    }
+
+                    Ok(PubSubItem::Notification(EthNotification {
+                        subscription: subscription.unwrap(),
+                        result: result.unwrap(),
+                    }))
+                }
+            }
+        }
+
+        deserializer.deserialize_any(PubSubItemVisitor)
+    }
+}
@@ -55,3 +55,8 @@ tokio-tungstenite = { version = "0.20.1", features = ["rustls-tls-webpki-roots"]
 default = ["reqwest", "hyper"]
 reqwest = ["dep:reqwest"]
 hyper = ["dep:hyper", "hyper/client"]
+
+[dev-dependencies]
+test-log = { version = "0.2.13", default-features = false, features = ["trace"] }
+tracing-subscriber = "0.3.17"
+tracing-test = "0.2.4"
@@ -4,20 +4,22 @@ use alloy_json_rpc::{Request, RequestPacket, ResponsePacket, RpcParam, RpcResult
 use core::panic;
 use serde_json::value::RawValue;
 use std::{
+    fmt::Debug,
     future::Future,
     marker::PhantomData,
     pin::Pin,
     task::{self, Poll::Ready},
 };
 use tower::Service;
+use tracing::{instrument, trace};
 
 /// The states of the [`RpcCall`] future.
 #[must_use = "futures do nothing unless you `.await` or poll them"]
 #[pin_project::pin_project(project = CallStateProj)]
 enum CallState<Params, Conn>
 where
-    Conn: Transport + Clone,
     Params: RpcParam,
+    Conn: Transport + Clone,
 {
     Prepared {
         request: Option<Request<Params>>,
@@ -30,6 +32,20 @@ where
     Complete,
 }
 
+impl<Params, Conn> Debug for CallState<Params, Conn>
+where
+    Params: RpcParam,
+    Conn: Transport + Clone,
+{
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            Self::Prepared { .. } => f.debug_struct("Prepared").finish(),
+            Self::AwaitingResponse { .. } => f.debug_struct("AwaitingResponse").finish(),
+            Self::Complete => write!(f, "Complete"),
+        }
+    }
+}
+
 impl<Params, Conn> CallState<Params, Conn>
 where
     Conn: Transport + Clone,
@@ -39,6 +55,7 @@ where
         mut self: Pin<&mut Self>,
         cx: &mut task::Context<'_>,
     ) -> task::Poll<<Self as Future>::Output> {
+        trace!("Polling prepared");
         let fut = {
             let CallStateProj::Prepared {
                 connection,
@@ -76,6 +93,7 @@ where
         mut self: Pin<&mut Self>,
         cx: &mut task::Context<'_>,
     ) -> task::Poll<<Self as Future>::Output> {
+        trace!("Polling awaiting");
         let CallStateProj::AwaitingResponse { fut } = self.as_mut().project() else {
             unreachable!("Called poll_awaiting in incorrect state")
         };
@@ -95,6 +113,7 @@ where
 {
     type Output = RpcResult<Box<RawValue>, Box<RawValue>, TransportError>;
 
+    #[instrument(skip(self, cx))]
     fn poll(mut self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> task::Poll<Self::Output> {
         if matches!(*self.as_mut(), CallState::Prepared { .. }) {
             return self.poll_prepared(cx);
@@ -191,6 +210,7 @@ where
     type Output = RpcResult<Resp, Box<RawValue>, TransportError>;
 
     fn poll(self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> task::Poll<Self::Output> {
+        tracing::trace!(?self.state, "Polling RpcCall");
         let this = self.project();
 
         let resp = task::ready!(this.state.poll(cx));

diff --git a/crates/transports/src/transports/ws/mod.rs b/crates/transports/src/transports/ws/mod.rs
@@ -10,7 +10,7 @@ pub use wasm::WsConnect;
 
 use crate::pubsub::ConnectionInterface;
 
-use tracing::{error, trace};
+use tracing::{debug, error, trace};
 
 /// An ongoing connection to a backend.
 ///
@@ -25,8 +25,9 @@ pub struct WsBackend<T> {
 }
 
 impl<T> WsBackend<T> {
+    #[tracing::instrument(skip(self))]
     pub async fn handle_text(&mut self, t: String) -> Result<(), ()> {
-        trace!(text = t, "Received message");
+        debug!(text = t, "Received message from websocket");
 
         match serde_json::from_str(&t) {
             Ok(item) => {

diff --git a/crates/transports/tests/http.rs → crates/transports/tests/it/http.rs b/crates/transports/tests/http.rs → crates/transports/tests/it/http.rs
diff --git a/crates/transports/tests/it/main.rs b/crates/transports/tests/it/main.rs
@@ -0,0 +1,2 @@
+mod http;
+mod ws;
diff --git a/crates/transports/tests/ws.rs → crates/transports/tests/it/ws.rs b/crates/transports/tests/ws.rs → crates/transports/tests/it/ws.rs
@@ -1,9 +1,9 @@
-use std::borrow::Cow;
+use alloy_transports::{ClientBuilder, RpcCall, WsConnect};
 
 use alloy_primitives::U64;
-use alloy_transports::{ClientBuilder, RpcCall, WsConnect};
+use std::borrow::Cow;
 
-#[tokio::test]
+#[test_log::test(tokio::test)]
 async fn it_makes_a_request() {
     let infura = std::env::var("INFURA_WS").unwrap();
 
@@ -12,13 +12,20 @@ async fn it_makes_a_request() {
         auth: None,
     };
 
+    dbg!("have connector");
+
     let client = ClientBuilder::default().connect(connector).await.unwrap();
 
+    dbg!("have client");
+
     let params: Cow<'static, _> = Cow::Owned(());
 
     let req: RpcCall<_, Cow<'static, ()>, U64> = client.prepare("eth_blockNumber", params);
-    let res = req.await;
+
+    let timeout = tokio::time::timeout(std::time::Duration::from_secs(2), req);
+
+    let res = timeout.await;
 
     dbg!(&res);
-    res.unwrap();
+    res.unwrap().unwrap();
 }