fix(client): adapt internal client_request API to be easier to implement correctly

This also fixes an issue with handling DB errors when reading events

Author: Snapstromegon

src/client/client_request.rs

@@ -25,8 +25,9 @@ use futures::{
 };
 use futures_util::io;
 use reqwest::Method;
-use serde::Serialize;
 use serde::de::DeserializeOwned;
+use serde::{Deserialize, Serialize};
+use std::pin::Pin;
 use tokio::io::{AsyncBufReadExt, BufReader};
 use tokio_stream::wrappers::LinesStream;
 use tokio_util::io::StreamReader;
@@ -61,19 +62,54 @@ pub trait OneShotRequest: ClientRequest {
         Ok(())
     }
 }
+#[derive(Deserialize, Debug)]
+#[serde(tag = "type", content = "payload", rename_all = "camelCase")]
+enum StreamLineItem<T> {
+    Error {
+        error: String,
+    },
+    Heartbeat,
+    #[serde(untagged)]
+    Ok {
+        #[serde(rename = "type")]
+        ty: String,
+        payload: T,
+    },
+}
+
+impl<T> StreamLineItem<T> {
+    pub fn into_result_option(self, expected_type: &str) -> Result<Option<T>, ClientError> {
+        match self {
+            StreamLineItem::Error { error } => Err(ClientError::DBError(error)),
+            StreamLineItem::Heartbeat => Ok(None),
+            StreamLineItem::Ok { ty, payload } => {
+                if ty == expected_type {
+                    Ok(Some(payload))
+                } else {
+                    Err(ClientError::InvalidResponseType(ty))
+                }
+            }
+        }
+    }
+}
 
 /// Represents a request to the database that expects a stream of responses
 pub trait StreamingRequest: ClientRequest {
     type ItemType: DeserializeOwned;
+    const ITEM_TYPE_NAME: &'static str;
 
     fn build_stream(
         response: reqwest::Response,
-    ) -> impl Stream<Item = Result<Self::ItemType, ClientError>> {
-        Self::lines_stream(response).map(|line| {
-            let line = line?;
-            let item = serde_json::from_str(line.as_str())?;
-            Ok(item)
-        })
+    ) -> Pin<Box<impl Stream<Item = Result<Self::ItemType, ClientError>>>> {
+        Box::pin(
+            Self::lines_stream(response)
+                .map(|line| {
+                    let line = line?;
+                    let item: StreamLineItem<Self::ItemType> = serde_json::from_str(line.as_str())?;
+                    item.into_result_option(Self::ITEM_TYPE_NAME)
+                })
+                .filter_map(|o| async { o.transpose() }),
+        )
     }
 
     fn lines_stream(

src/client/client_request/list_event_types.rs

@@ -1,4 +1,3 @@
-use futures::{Stream, stream::StreamExt};
 use reqwest::Method;
 use serde::{Deserialize, Serialize};
 use serde_json::Value;
@@ -29,30 +28,5 @@ impl ClientRequest for ListEventTypesRequest {
 }
 impl StreamingRequest for ListEventTypesRequest {
     type ItemType = EventType;
-
-    fn build_stream(
-        response: reqwest::Response,
-    ) -> impl Stream<Item = Result<Self::ItemType, ClientError>> {
-        #[derive(Deserialize, Debug)]
-        #[serde(tag = "type", content = "payload", rename_all = "camelCase")]
-        enum LineItem {
-            Error { error: String },
-            EventType(EventType),
-        }
-
-        impl From<LineItem> for Result<EventType, ClientError> {
-            fn from(item: LineItem) -> Self {
-                match item {
-                    LineItem::Error { error } => Err(ClientError::DBError(error)),
-                    LineItem::EventType(event_type) => Ok(event_type),
-                }
-            }
-        }
-
-        Self::lines_stream(response).map(|line| {
-            let line = line?;
-            let item: LineItem = serde_json::from_str(line.as_str())?;
-            item.into()
-        })
-    }
+    const ITEM_TYPE_NAME: &'static str = "eventType";
 }

src/client/client_request/list_subjects.rs

@@ -1,6 +1,5 @@
-use futures::{Stream, stream::StreamExt};
 use reqwest::Method;
-use serde::{Deserialize, Serialize};
+use serde::Serialize;
 
 use crate::error::ClientError;
 
@@ -22,26 +21,5 @@ impl ClientRequest for ListSubjectsRequest<'_> {
 }
 impl StreamingRequest for ListSubjectsRequest<'_> {
     type ItemType = String;
-
-    fn build_stream(
-        response: reqwest::Response,
-    ) -> impl Stream<Item = Result<Self::ItemType, ClientError>> {
-        #[derive(Deserialize, Debug)]
-        struct LineItem {
-            payload: LineItemPayload,
-            r#type: String,
-        }
-        #[derive(Deserialize, Debug)]
-        struct LineItemPayload {
-            subject: String,
-        }
-        Self::lines_stream(response).map(|line| {
-            let line = line?;
-            let item: LineItem = serde_json::from_str(line.as_str())?;
-            if item.r#type != "subject" {
-                return Err(ClientError::InvalidEventType);
-            }
-            Ok(item.payload.subject)
-        })
-    }
+    const ITEM_TYPE_NAME: &'static str = "subject";
 }

src/client/client_request/observe_events.rs

@@ -25,4 +25,5 @@ impl ClientRequest for ObserveEventsRequest<'_> {
 
 impl StreamingRequest for ObserveEventsRequest<'_> {
     type ItemType = Event;
+    const ITEM_TYPE_NAME: &'static str = "event";
 }

src/client/client_request/read_events.rs

@@ -1,6 +1,5 @@
-use futures::{Stream, stream::StreamExt};
 use reqwest::Method;
-use serde::{Deserialize, Serialize};
+use serde::Serialize;
 
 use crate::{client::request_options::ReadEventsRequestOptions, error::ClientError, event::Event};
 
@@ -24,30 +23,5 @@ impl ClientRequest for ReadEventsRequest<'_> {
 
 impl StreamingRequest for ReadEventsRequest<'_> {
     type ItemType = Event;
-
-    fn build_stream(
-        response: reqwest::Response,
-    ) -> impl Stream<Item = Result<Self::ItemType, ClientError>> {
-        #[derive(Deserialize, Debug)]
-        #[serde(tag = "type", content = "payload", rename_all = "camelCase")]
-        enum LineItem {
-            Error { error: String },
-            Event(Box<Event>),
-        }
-
-        impl From<LineItem> for Result<Event, ClientError> {
-            fn from(item: LineItem) -> Self {
-                match item {
-                    LineItem::Error { error } => Err(ClientError::DBError(error)),
-                    LineItem::Event(event_type) => Ok(*event_type),
-                }
-            }
-        }
-
-        Self::lines_stream(response).map(|line| {
-            let line = line?;
-            let item: LineItem = serde_json::from_str(line.as_str())?;
-            item.into()
-        })
-    }
+    const ITEM_TYPE_NAME: &'static str = "event";
 }

src/error.rs

@@ -44,6 +44,9 @@ pub enum ClientError {
     #[cfg(feature = "cloudevents")]
     #[error("The CloudEvents message is invalid: {0}")]
     CloudeventsMessageError(#[from] cloudevents::message::Error),
+    /// The database returned an invalid response type
+    #[error("The DB returned an invalid response type: {0}")]
+    InvalidResponseType(String),
 }
 
 /// Error type for the [`crate::container`] feature.