From 0addd8dc472c96874aad611011bc9b640345df54 Mon Sep 17 00:00:00 2001
From: notplants <max@mfowler.info>
Date: Wed, 12 Jan 2022 15:01:45 -0500
Subject: [PATCH 1/2] Convert result to option

---
 src/sbot.rs | 33 +++++++++++++++------------------
 1 file changed, 15 insertions(+), 18 deletions(-)

diff --git a/src/sbot.rs b/src/sbot.rs
index 2c1abb3..f935e80 100644
--- a/src/sbot.rs
+++ b/src/sbot.rs
@@ -281,29 +281,26 @@ impl Sbot {
         // now we have a stream of about messages with most recent at the front of the vector
         pin_mut!(about_message_stream);
         // iterate through the vector looking for most recent about message with the given key
-        let latest_about_message: Result<SsbMessageValue, GolgiError> = about_message_stream
+        let latest_about_message_res: Option<Result<SsbMessageValue, GolgiError>> = about_message_stream
             // find the first msg that contains the field `key`
             .find(|res| match res {
                 Ok(msg) => msg.content.get(key).is_some(),
                 Err(_) => false,
             })
-            .await
-            .ok_or_else(|| {
-                GolgiError::Sbot("error while looking for about message with given key".to_string())
-            })?;
-        let latest_about_value = match latest_about_message {
-            Ok(msg) => {
-                msg
-                    // SsbMessageValue -> Option<&Value>
-                    .content
-                    .get(key)
-                    // Option<&Value> -> <Option<&str>
-                    .and_then(|value| value.as_str())
-                    // Option<&str> -> Option<String>
-                    .map(|value| value.to_string())
-            }
-            Err(_) => None,
-        };
+            .await;
+        // Option<Result<SsbMessageValue, GolgiError>> -> Option<SsbMessageValue>
+        let latest_about_message = latest_about_message_res.and_then(|msg| msg.ok());
+        // Option<SsbMessageValue> -> Option<String>
+        let latest_about_value = latest_about_message.and_then(|msg| {
+            msg
+                // SsbMessageValue -> Option<&Value>
+                .content
+                .get(key)
+                // Option<&Value> -> <Option<&str>
+                .and_then(|value| value.as_str())
+                // Option<&str> -> Option<String>
+                .map(|value| value.to_string())
+        });
         // return value is either `Ok(Some(String))` or `Ok(None)`
         Ok(latest_about_value)
     }

From 1879341499d1b447d131e07ee8bd4f9a27b7dea7 Mon Sep 17 00:00:00 2001
From: notplants <max@mfowler.info>
Date: Wed, 12 Jan 2022 15:22:26 -0500
Subject: [PATCH 2/2] Chang get_subset_stream to return SsbMessageValue instead
 of SsbMessageKVT

---
 src/sbot.rs | 19 ++++++++-----------
 1 file changed, 8 insertions(+), 11 deletions(-)

diff --git a/src/sbot.rs b/src/sbot.rs
index f935e80..621fb8e 100644
--- a/src/sbot.rs
+++ b/src/sbot.rs
@@ -127,7 +127,7 @@ impl Sbot {
     }
 
     /// Call the `partialReplication getSubset` RPC method
-    /// and return a Stream of Result<SsbMessageKVT, GolgiError>
+    /// and return a Stream of Result<SsbMessageValue, GolgiError>
     ///
     /// # Arguments
     ///
@@ -138,14 +138,14 @@ impl Sbot {
         &mut self,
         query: SubsetQuery,
         options: Option<SubsetQueryOptions>,
-    ) -> Result<impl Stream<Item = Result<SsbMessageKVT, GolgiError>>, GolgiError> {
+    ) -> Result<impl Stream<Item = Result<SsbMessageValue, GolgiError>>, GolgiError> {
         let mut sbot_connection = self.get_sbot_connection().await?;
         let req_id = sbot_connection
             .client
             .getsubset_req_send(query, options)
             .await?;
         let get_subset_stream =
-            get_source_stream(sbot_connection.rpc_reader, req_id, utils::kvt_res_parse).await;
+            get_source_stream(sbot_connection.rpc_reader, req_id, utils::ssb_message_res_parse).await;
         Ok(get_subset_stream)
     }
 
@@ -252,18 +252,15 @@ impl Sbot {
             keys: None,
             page_limit: None,
         };
-        let get_subset_kvt_stream = self.get_subset_stream(query, Some(query_options)).await?;
-        // map into Stream<Item=Result<SsbMessageValue, GolgiError>>
-        let ssb_message_stream = get_subset_kvt_stream.map(|msg| match msg {
-            Ok(val) => Ok(val.value),
-            Err(err) => Err(err),
-        });
+        let get_subset_stream = self.get_subset_stream(query, Some(query_options)).await?;
         // TODO: after fixing sbot regression,
         // change this subset query to filter by type about in addition to author
         // and remove this filter section
         // filter down to about  messages
-        let about_message_stream = ssb_message_stream.filter(|msg| match msg {
-            Ok(val) => val.is_message_type(SsbMessageContentType::About),
+        let about_message_stream = get_subset_stream.filter(|msg| match msg {
+            Ok(val) => {
+                val.is_message_type(SsbMessageContentType::About)
+            }
             Err(_err) => false,
         });
         // return about message stream