Stop using #[serde(untagged)] for Message enum

chromiumoxide_types/src/lib.rs

@@ -2,6 +2,7 @@ use std::borrow::Cow;
 use std::fmt;
 use std::fmt::Debug;
 use std::ops::Deref;
+use std::str::FromStr;
 
 use serde::de::DeserializeOwned;
 use serde::{Deserialize, Serialize};
@@ -199,8 +200,7 @@ pub struct Response {
 /// An incoming message read from the web socket can either be a response to a
 /// previously submitted `Request`, identified by an identifier `id`, or an
 /// `Event` emitted by the server.
-#[derive(Deserialize, Debug, Clone)]
-#[serde(untagged)]
+#[derive(Debug, Clone)]
 #[allow(clippy::large_enum_variant)]
 pub enum Message<T = CdpJsonEventMessage> {
     /// A response for a request
@@ -209,6 +209,33 @@ pub enum Message<T = CdpJsonEventMessage> {
     Event(T),
 }
 
+// `#[serde(untagged)]` does not work correctly with serde_json and some numeric types in the schema,
+// without activating `arbitrary_precision` feature of serde_json and using `serde_json::Number` instead of all usage of `f64` or else.
+// For now, we use a `FromStr` trait to deserialize the message, instead of `Deserialize` with
+// using `serde_json` to deserialize the message to `serde_json::Value` at first and then trying to
+// deserialize both `Response` or `Event` from the `serde_json::Value`.
+// - https://github.com/serde-rs/serde/issues/2661
+// Also, this has a win in error reporting compared to `untagged` because we can return the
+// detailed error instead of just "did not match any variant of untagged enum".
+impl<T> FromStr for Message<T>
+where
+    T: DeserializeOwned + Debug,
+{
+    type Err = serde_json::Error;
+
+    fn from_str(s: &str) -> Result<Self, Self::Err> {
+        if let Ok(response) = serde_json::from_str::<Response>(s) {
+            Ok(Message::Response(response))
+        } else {
+            // For now, returns an error only about the event deserialization.
+            // Ideally, we can return an custom error type that contains errors for both response and event.
+            // It seems not necessary for now because the response seems to be deserialized correctly in most cases.
+            let event = serde_json::from_str::<T>(s)?;
+            Ok(Message::Event(event))
+        }
+    }
+}
+
 /// A response can either contain the `Command::Response` type in the `result`
 /// field of the payload or an `Error` in the `error` field if the request
 /// resulted in an error.

src/conn.rs

@@ -3,6 +3,7 @@ use std::marker::PhantomData;
 use std::pin::Pin;
 use std::task::ready;
 
+use async_tungstenite::tungstenite::Message as WsMessage;
 use async_tungstenite::{tungstenite::protocol::WebSocketConfig, WebSocketStream};
 use futures::stream::Stream;
 use futures::task::{Context, Poll};
@@ -132,21 +133,33 @@ impl<T: EventMessage + Unpin> Stream for Connection<T> {
                     pin.pending_flush = Some(call);
                 }
             }
+
             break;
         }
 
         // read from the ws
         match ready!(pin.ws.poll_next_unpin(cx)) {
-            Some(Ok(msg)) => match serde_json::from_slice::<Message<T>>(&msg.into_data()) {
-                Ok(msg) => {
-                    tracing::trace!("Received {:?}", msg);
-                    Poll::Ready(Some(Ok(msg)))
-                }
-                Err(err) => {
-                    tracing::error!("Failed to deserialize WS response {}", err);
-                    Poll::Ready(Some(Err(err.into())))
-                }
-            },
+            Some(Ok(WsMessage::Text(text))) => {
+                let ready = match text.parse::<Message<T>>() {
+                    Ok(msg) => {
+                        tracing::trace!("Received {:?}", msg);
+                        Ok(msg)
+                    }
+                    Err(err) => {
+                        tracing::debug!(target: "chromiumoxide::conn::raw_ws::parse_errors", msg = text, "Failed to parse raw WS message");
+                        tracing::error!("Failed to deserialize WS response {}", err);
+                        Err(err.into())
+                    }
+                };
+                Poll::Ready(Some(ready))
+            }
+            Some(Ok(WsMessage::Close(_))) => Poll::Ready(None),
+            // ignore ping and pong
+            Some(Ok(WsMessage::Ping(_))) | Some(Ok(WsMessage::Pong(_))) => {
+                cx.waker().wake_by_ref();
+                Poll::Pending
+            }
+            Some(Ok(msg)) => Poll::Ready(Some(Err(CdpError::UnexpectedWsMessage(msg)))),
             Some(Err(err)) => Poll::Ready(Some(Err(CdpError::Ws(err)))),
             None => {
                 // ws connection closed

src/error.rs

@@ -4,6 +4,7 @@ use std::process::ExitStatus;
 use std::time::Instant;
 
 use async_tungstenite::tungstenite;
+use async_tungstenite::tungstenite::Message;
 use base64::DecodeError;
 use futures::channel::mpsc::SendError;
 use futures::channel::oneshot::Canceled;
@@ -28,6 +29,8 @@ pub enum CdpError {
     Chrome(#[from] chromiumoxide_types::Error),
     #[error("Received no response from the chromium instance.")]
     NoResponse,
+    #[error("Received unexpected ws message: {0:?}")]
+    UnexpectedWsMessage(Message),
     #[error("{0}")]
     ChannelSendError(#[from] ChannelError),
     #[error("Browser process exited with status {0:?} before websocket URL could be resolved, stderr: {1:?}")]