chore(client): add comments to stream parsing to make complex parts easier to understand especially for new rustaceans

Signed-off-by: Raphael Höser <[email protected]>

Author: Snapstromegon

src/client/client_request.rs

@@ -61,13 +61,20 @@ pub trait OneShotRequest: ClientRequest {
         Ok(())
     }
 }
+
+/// A line in any json-nd stream coming from the database
 #[derive(Deserialize, Debug)]
 #[serde(tag = "type", content = "payload", rename_all = "camelCase")]
 enum StreamLineItem<T> {
-    Error {
-        error: String,
-    },
+    /// An error occured during the request
+    Error { error: String },
+    /// A heardbeat message was sent to keep the connection alive.
+    /// This is only used when observing events, but it does not hurt to have it everywhere.
     Heartbeat,
+    /// A successful response from the database
+    /// Since the exact type of the payload is not known at this point, we use this as a fallback case.
+    /// Every request item gets put in here and the type can be checked later on.
+    /// The type name checking is only for semantic reasons, as the payload is already parsed as the correct type at this point.
     #[serde(untagged)]
     Ok {
         #[serde(rename = "type")]
@@ -76,22 +83,6 @@ enum StreamLineItem<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;
@@ -102,12 +93,38 @@ pub trait StreamingRequest: ClientRequest {
     ) -> 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)
+                .map(|maybe_line| {
+                    let line = maybe_line?;
+                    // This line does the heavy lifting of parsing the json-nd line into the correct type.
+                    // There's some Rust typesystem glory involved here, so let's break it down:
+                    // First of all `serde_json::from_str` is used to parse any json `&str` into the type we want to have (in this case a `StreamLineItem`).
+                    // `StreamLineItem` in turn is generic over `Self::ItemType`, which is the type that is expected by the exact response implementation and can change.
+                    // This means, that this will throw an error if the line is invalid json or the string cannot be parsed into an error, heartbeat or the expected type.
+                    // Because of this, we can guarantee after this line, that the payload of the `StreamLineItem` is of the correct type and no further checks are needed.
+                    Ok(serde_json::from_str::<StreamLineItem<Self::ItemType>>(
+                        line.as_str(),
+                    )?)
                 })
-                .filter_map(|o| async { o.transpose() }),
+                .filter_map(|o| async {
+                    match o {
+                        // An error was passed by the database, so we forward it as an error.
+                        Ok(StreamLineItem::Error { error }) => {
+                            Some(Err(ClientError::DBError(error)))
+                        }
+                        // A heartbeat message was sent, which we ignore.
+                        Ok(StreamLineItem::Heartbeat) => None,
+                        // A successful response was sent with the correct type.
+                        Ok(StreamLineItem::Ok { payload, ty }) if ty == Self::ITEM_TYPE_NAME => {
+                            Some(Ok(payload))
+                        }
+                        // A successful response was sent, but the type does not match the expected type.
+                        Ok(StreamLineItem::Ok { ty, .. }) => {
+                            Some(Err(ClientError::InvalidResponseType(ty)))
+                        }
+                        // An error occured while parsing the line, which we forward as an error.
+                        Err(e) => Some(Err(e)),
+                    }
+                }),
         )
     }