Cargo.lock

@@ -151,7 +151,17 @@ version = "0.2.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "22068c0c19514942eefcfd4daf8976ef1aad84e61539f95cd200c35202f80af5"
 dependencies = [
- "async-stream-impl",
+ "async-stream-impl 0.2.1",
+ "futures-core",
+]
+
+[[package]]
+name = "async-stream"
+version = "0.3.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "171374e7e3b2504e0e5236e3b59260560f9fe94bfe9ac39ba5e4e929c5590625"
+dependencies = [
+ "async-stream-impl 0.3.2",
  "futures-core",
 ]
 
@@ -166,6 +176,17 @@ dependencies = [
  "syn",
 ]
 
+[[package]]
+name = "async-stream-impl"
+version = "0.3.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "648ed8c8d2ce5409ccd57453d9d1b214b342a0d69376a6feda1fd6cae3299308"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn",
+]
+
 [[package]]
 name = "async-task"
 version = "4.0.3"
@@ -478,6 +499,7 @@ name = "golgi"
 version = "0.1.0"
 dependencies = [
  "async-std",
+ "async-stream 0.3.2",
  "base64 0.13.0",
  "futures",
  "hex",
@@ -572,7 +594,7 @@ name = "kuska-ssb"
 version = "0.2.0"
 dependencies = [
  "async-std",
- "async-stream",
+ "async-stream 0.2.1",
  "base64 0.11.0",
  "dirs",
  "futures",

Cargo.toml

@@ -15,3 +15,4 @@ kuska-sodiumoxide = "0.2.5-0"
 kuska-ssb = { path = "../ssb" }
 serde = { version = "1", features = ["derive"] }
 serde_json = "1"
+async-stream = "0.3.2"

examples/ssb-client.rs

@@ -1,7 +1,7 @@
 use std::process;
 
-use golgi::messages::SsbMessageContent;
 use golgi::error::GolgiError;
+use golgi::messages::SsbMessageContent;
 use golgi::sbot::Sbot;
 
 async fn run() -> Result<(), GolgiError> {
@@ -32,7 +32,9 @@ async fn run() -> Result<(), GolgiError> {
     let post_msg_ref = sbot_client.publish(post).await?;
     println!("{}", post_msg_ref);
 
-    let post_msg_ref = sbot_client.publish_description("this is a description").await?;
+    let post_msg_ref = sbot_client
+        .publish_description("this is a description")
+        .await?;
     println!("description: {}", post_msg_ref);
 
     Ok(())

examples/ssb-stream-example.rs

@@ -0,0 +1,85 @@
+use std::process;
+
+use async_std::stream::StreamExt;
+use futures::{pin_mut, TryStreamExt};
+
+use golgi::error::GolgiError;
+use golgi::messages::{SsbMessageContentType, SsbMessageValue};
+use golgi::sbot::Sbot;
+
+async fn run() -> Result<(), GolgiError> {
+    let mut sbot_client = Sbot::init(None, None).await?;
+
+    let id = sbot_client.whoami().await?;
+    println!("{}", id);
+
+    let author = "@L/z54cbc8V1kL1/MiBhpEKuN3QJkSoZYNaukny3ghIs=.ed25519";
+
+    // create a history stream
+    let history_stream = sbot_client
+        .create_history_stream(author.to_string())
+        .await?;
+
+    // loop through the results until the end of the stream
+    pin_mut!(history_stream); // needed for iteration
+    println!("looping through stream");
+    while let Some(res) = history_stream.next().await {
+        match res {
+            Ok(value) => {
+                println!("value: {:?}", value);
+            }
+            Err(err) => {
+                println!("err: {:?}", err);
+            }
+        }
+    }
+    println!("reached end of stream");
+
+    // create a history stream and convert it into a Vec<SsbMessageValue> using try_collect
+    //  (if there is any error in the results, it will be raised)
+    let history_stream = sbot_client
+        .create_history_stream(author.to_string())
+        .await?;
+    let results: Vec<SsbMessageValue> = history_stream.try_collect().await?;
+    for x in results {
+        println!("x: {:?}", x);
+    }
+
+    // example to create a history stream and use a map to convert stream of SsbMessageValue
+    // into a stream of tuples of (String, SsbMessageContentType)
+    let history_stream = sbot_client
+        .create_history_stream(author.to_string())
+        .await?;
+    let type_stream = history_stream.map(|msg| match msg {
+        Ok(val) => {
+            let message_type = val.get_message_type()?;
+            let tuple: (String, SsbMessageContentType) = (val.signature, message_type);
+            Ok(tuple)
+        }
+        Err(err) => Err(err),
+    });
+    pin_mut!(type_stream); // needed for iteration
+    println!("looping through type stream");
+    while let Some(res) = type_stream.next().await {
+        match res {
+            Ok(value) => {
+                println!("value: {:?}", value);
+            }
+            Err(err) => {
+                println!("err: {:?}", err);
+            }
+        }
+    }
+    println!("reached end of type stream");
+
+    // return Ok
+    Ok(())
+}
+
+#[async_std::main]
+async fn main() {
+    if let Err(e) = run().await {
+        eprintln!("Application error: {}", e);
+        process::exit(1);
+    }
+}

src/error.rs

@@ -57,7 +57,7 @@ impl std::error::Error for GolgiError {
             GolgiError::Sbot(_) => None,
             GolgiError::SerdeJson(ref err) => Some(err),
             GolgiError::ContentType(_) => None,
-            GolgiError::Utf8Parse{ ref source} => Some(source),
+            GolgiError::Utf8Parse { ref source } => Some(source),
         }
     }
 }
@@ -82,7 +82,9 @@ impl std::fmt::Display for GolgiError {
                 "Failed to decode typed message from ssb message content: {}",
                 err
             ),
-            GolgiError::Utf8Parse{ source } => write!(f, "Failed to deserialize UTF8 from bytes: {}", source),
+            GolgiError::Utf8Parse { source } => {
+                write!(f, "Failed to deserialize UTF8 from bytes: {}", source)
+            }
         }
     }
 }
@@ -125,6 +127,6 @@ impl From<JsonError> for GolgiError {
 
 impl From<Utf8Error> for GolgiError {
     fn from(err: Utf8Error) -> Self {
-        GolgiError::Utf8Parse { source: err}
+        GolgiError::Utf8Parse { source: err }
     }
 }

src/messages.rs

@@ -1,10 +1,9 @@
 //! Message types and conversion methods for `golgi`.
 
-use std::fmt::Debug;
-
 use kuska_ssb::api::dto::content::TypedMessage;
 use serde::{Deserialize, Serialize};
 use serde_json::Value;
+use std::fmt::Debug;
 
 use crate::error::GolgiError;
 
@@ -31,34 +30,36 @@ pub struct SsbMessageValue {
 
 // Enum representing the different possible message content types
 #[derive(Debug)]
+#[allow(missing_docs)]
 pub enum SsbMessageContentType {
     About,
     Vote,
     Post,
     Contact,
-    Unrecognized
+    Unrecognized,
 }
 
 impl SsbMessageValue {
-
-     /// Gets the type field of the message content as an enum, if found.
-     /// if no type field is found or the type field is not a string, it returns an Err(GolgiError::ContentType)
-     /// if a type field is found but with an unknown string it returns an Ok(SsbMessageContentType::Unrecognized)
-     pub fn get_message_type(&self) -> Result<SsbMessageContentType, GolgiError> {
-         let msg_type = self
-             .content
-             .get("type")
-             .ok_or(GolgiError::ContentType("type field not found".to_string()))?;
-         let mtype_str: &str = msg_type.as_str().ok_or(GolgiError::ContentType("type field value is not a string as expected".to_string()))?;
-         let enum_type = match mtype_str {
-             "about" => SsbMessageContentType::About,
-             "post" => SsbMessageContentType::Post,
-             "vote" => SsbMessageContentType::Vote,
-             "contact" => SsbMessageContentType::Contact,
-             _ => SsbMessageContentType::Unrecognized,
-         };
-         Ok(enum_type)
-     }
+    /// Gets the type field of the message content as an enum, if found.
+    /// if no type field is found or the type field is not a string, it returns an Err(GolgiError::ContentType)
+    /// if a type field is found but with an unknown string it returns an Ok(SsbMessageContentType::Unrecognized)
+    pub fn get_message_type(&self) -> Result<SsbMessageContentType, GolgiError> {
+        let msg_type = self
+            .content
+            .get("type")
+            .ok_or(GolgiError::ContentType("type field not found".to_string()))?;
+        let mtype_str: &str = msg_type.as_str().ok_or(GolgiError::ContentType(
+            "type field value is not a string as expected".to_string(),
+        ))?;
+        let enum_type = match mtype_str {
+            "about" => SsbMessageContentType::About,
+            "post" => SsbMessageContentType::Post,
+            "vote" => SsbMessageContentType::Vote,
+            "contact" => SsbMessageContentType::Contact,
+            _ => SsbMessageContentType::Unrecognized,
+        };
+        Ok(enum_type)
+    }
 
     /// Helper function which returns true if this message is of the given type,
     /// and false if the type does not match or is not found
@@ -68,9 +69,7 @@ impl SsbMessageValue {
             Ok(mtype) => {
                 matches!(mtype, message_type)
             }
-            Err(_err) => {
-                false
-            }
+            Err(_err) => false,
         }
     }
 

src/sbot.rs

@@ -1,25 +1,32 @@
 //! Sbot type and associated methods.
-use async_std::net::TcpStream;
+use async_std::{net::TcpStream, stream::Stream};
 
 use kuska_handshake::async_std::BoxStream;
 use kuska_sodiumoxide::crypto::{auth, sign::ed25519};
 use kuska_ssb::{
-    api::{
-        dto::{CreateHistoryStreamIn},
-        ApiCaller,
-    },
+    api::{dto::CreateHistoryStreamIn, ApiCaller},
     discovery, keystore,
     keystore::OwnedIdentity,
     rpc::{RpcReader, RpcWriter},
 };
 
 use crate::error::GolgiError;
-use crate::messages::{SsbMessageKVT, SsbMessageContent, SsbMessageValue, SsbMessageContentType};
+use crate::messages::{SsbMessageContent, SsbMessageContentType, SsbMessageKVT, SsbMessageValue};
 use crate::utils;
+use crate::utils::get_source_stream;
 
 // re-export types from kuska
 pub use kuska_ssb::api::dto::content::SubsetQuery;
 
+/// A struct representing a connection with a running sbot.
+/// A client and an rpc_reader can together be used to make requests to the sbot
+/// and read the responses.
+/// Note there can be multiple SbotConnection at the same time.
+pub struct SbotConnection {
+    client: ApiCaller<TcpStream>,
+    rpc_reader: RpcReader<TcpStream>,
+}
+
 /// The Scuttlebutt identity, keys and configuration parameters for connecting to a local sbot
 /// instance, as well as handles for calling RPC methods and receiving responses.
 pub struct Sbot {
@@ -29,8 +36,9 @@ pub struct Sbot {
     address: String,
     // aka caps key (scuttleverse identifier)
     network_id: auth::Key,
-    client: ApiCaller<TcpStream>,
-    rpc_reader: RpcReader<TcpStream>,
+    // the primary connection with sbot which can be re-used for non-stream calls
+    // note that stream calls will each need their own SbotConnection
+    sbot_connection: SbotConnection,
 }
 
 impl Sbot {
@@ -54,6 +62,40 @@ impl Sbot {
             .await
             .expect("couldn't read local secret");
 
+        let sbot_connection =
+            Sbot::_get_sbot_connection_helper(address.clone(), network_id.clone(), pk, sk.clone())
+                .await?;
+
+        Ok(Self {
+            id,
+            public_key: pk,
+            private_key: sk,
+            address,
+            network_id,
+            sbot_connection,
+        })
+    }
+
+    /// Creates a new connection with the sbot,
+    /// using the address, network_id, public_key and private_key supplied when Sbot was initialized.
+    ///
+    /// Note that a single Sbot can have multiple SbotConnection at the same time.
+    pub async fn get_sbot_connection(&self) -> Result<SbotConnection, GolgiError> {
+        let address = self.address.clone();
+        let network_id = self.network_id.clone();
+        let public_key = self.public_key;
+        let private_key = self.private_key.clone();
+        Sbot::_get_sbot_connection_helper(address, network_id, public_key, private_key).await
+    }
+
+    /// Private helper function which creates a new connection with sbot,
+    /// but with all variables passed as arguments.
+    async fn _get_sbot_connection_helper(
+        address: String,
+        network_id: auth::Key,
+        public_key: ed25519::PublicKey,
+        private_key: ed25519::SecretKey,
+    ) -> Result<SbotConnection, GolgiError> {
         let socket = TcpStream::connect(&address)
             .await
             .map_err(|source| GolgiError::Io {
@@ -64,9 +106,9 @@ impl Sbot {
         let handshake = kuska_handshake::async_std::handshake_client(
             &mut &socket,
             network_id.clone(),
-            pk,
-            sk.clone(),
-            pk,
+            public_key,
+            private_key.clone(),
+            public_key,
         )
         .await
         .map_err(GolgiError::Handshake)?;
@@ -76,32 +118,41 @@ impl Sbot {
 
         let rpc_reader = RpcReader::new(box_stream_read);
         let client = ApiCaller::new(RpcWriter::new(box_stream_write));
-
-        Ok(Self {
-            id,
-            public_key: pk,
-            private_key: sk,
-            address,
-            network_id,
-            client,
-            rpc_reader,
-        })
+        let sbot_connection = SbotConnection { rpc_reader, client };
+        Ok(sbot_connection)
     }
 
     /// Call the `partialReplication getSubset` RPC method and return a vector
     /// of messages as KVTs (key, value, timestamp).
     // TODO: add args for `descending` and `page` (max number of msgs in response)
-    pub async fn get_subset(&mut self, query: SubsetQuery) -> Result<Vec<SsbMessageKVT>, GolgiError> {
-        let req_id = self.client.getsubset_req_send(query).await?;
+    pub async fn get_subset(
+        &mut self,
+        query: SubsetQuery,
+    ) -> Result<Vec<SsbMessageKVT>, GolgiError> {
+        let req_id = self
+            .sbot_connection
+            .client
+            .getsubset_req_send(query)
+            .await?;
 
-        utils::get_source_until_eof(&mut self.rpc_reader, req_id, utils::kvt_res_parse).await
+        utils::get_source_until_eof(
+            &mut self.sbot_connection.rpc_reader,
+            req_id,
+            utils::kvt_res_parse,
+        )
+        .await
     }
 
     /// Call the `whoami` RPC method and return an `id`.
     pub async fn whoami(&mut self) -> Result<String, GolgiError> {
-        let req_id = self.client.whoami_req_send().await?;
+        let req_id = self.sbot_connection.client.whoami_req_send().await?;
 
-        utils::get_async(&mut self.rpc_reader, req_id, utils::string_res_parse).await
+        utils::get_async(
+            &mut self.sbot_connection.rpc_reader,
+            req_id,
+            utils::string_res_parse,
+        )
+        .await
     }
 
     /// Call the `publish` RPC method and return a message reference.
@@ -112,9 +163,14 @@ impl Sbot {
     /// `Channel` and `Vote`. See the `kuska_ssb` documentation for further details such as field
     /// names and accepted values for each variant.
     pub async fn publish(&mut self, msg: SsbMessageContent) -> Result<String, GolgiError> {
-        let req_id = self.client.publish_req_send(msg).await?;
+        let req_id = self.sbot_connection.client.publish_req_send(msg).await?;
 
-        utils::get_async(&mut self.rpc_reader, req_id, utils::string_res_parse).await
+        utils::get_async(
+            &mut self.sbot_connection.rpc_reader,
+            req_id,
+            utils::string_res_parse,
+        )
+        .await
     }
 
     /// Wrapper for publish which constructs and publishes a post message appropriately from a string.
@@ -169,8 +225,11 @@ impl Sbot {
     }
 
     /// Get the about messages for a particular user in order of recency.
-    pub async fn get_about_messages(&mut self, ssb_id: &str) -> Result<Vec<SsbMessageValue>, GolgiError> {
-        let query = SubsetQuery::Author{
+    pub async fn get_about_messages(
+        &mut self,
+        ssb_id: &str,
+    ) -> Result<Vec<SsbMessageValue>, GolgiError> {
+        let query = SubsetQuery::Author {
             op: "author".to_string(),
             feed: ssb_id.to_string(),
         };
@@ -180,19 +239,22 @@ impl Sbot {
         // change this subset query to filter by type about in addition to author
         // and remove this filter section
         // filter down to about  messages
-        let mut about_messages: Vec<SsbMessageValue> = messages.into_iter().filter(|msg| {
-            msg.is_message_type(SsbMessageContentType::About)
-        }).collect();
+        let mut about_messages: Vec<SsbMessageValue> = messages
+            .into_iter()
+            .filter(|msg| msg.is_message_type(SsbMessageContentType::About))
+            .collect();
         // TODO: use subset query to order messages instead of doing it this way
-        about_messages.sort_by(|a, b| {
-            b.timestamp.partial_cmp(&a.timestamp).unwrap()
-        });
+        about_messages.sort_by(|a, b| b.timestamp.partial_cmp(&a.timestamp).unwrap());
         // return about messages
         Ok(about_messages)
     }
 
     /// Get value of latest about message with given key from given user
-    pub async fn get_latest_about_message(&mut self, ssb_id: &str, key: &str) -> Result<Option<String>, GolgiError> {
+    pub async fn get_latest_about_message(
+        &mut self,
+        ssb_id: &str,
+        key: &str,
+    ) -> Result<Option<String>, GolgiError> {
         // vector of about messages with most recent at the front of the vector
         let about_messages = self.get_about_messages(ssb_id).await?;
         // iterate through the vector looking for most recent about message with the given key
@@ -215,22 +277,44 @@ impl Sbot {
         Ok(latest_about)
     }
 
+    /// Get latest about name from given user
+    ///
+    /// # Arguments
+    ///
+    /// * `ssb_id` - A reference to a string slice which represents the ssb user
+    /// to lookup the about name for.
     pub async fn get_name(&mut self, ssb_id: &str) -> Result<Option<String>, GolgiError> {
         self.get_latest_about_message(ssb_id, "name").await
     }
 
+    /// Get lateset about description from given user
+    ///
+    /// # Arguments
+    ///
+    /// * `ssb_id` - A reference to a string slice which represents the ssb user
+    /// to lookup the about description for.
     pub async fn get_description(&mut self, ssb_id: &str) -> Result<Option<String>, GolgiError> {
         self.get_latest_about_message(ssb_id, "description").await
     }
 
-    /// Call the `createHistoryStream` RPC method and return a vector
-    /// of SsbMessageValue.
+    /// Call the `createHistoryStream` RPC method
+    /// and return a Stream of Result<SsbMessageValue, GolgiError>
     pub async fn create_history_stream(
         &mut self,
         id: String,
-    ) -> Result<Vec<SsbMessageValue>, GolgiError> {
+    ) -> Result<impl Stream<Item = Result<SsbMessageValue, GolgiError>>, GolgiError> {
+        let mut sbot_connection = self.get_sbot_connection().await.unwrap();
         let args = CreateHistoryStreamIn::new(id);
-        let req_id = self.client.create_history_stream_req_send(&args).await?;
-        utils::get_source_until_eof(&mut self.rpc_reader, req_id, utils::ssb_message_res_parse).await
+        let req_id = sbot_connection
+            .client
+            .create_history_stream_req_send(&args)
+            .await?;
+        let history_stream = get_source_stream(
+            sbot_connection.rpc_reader,
+            req_id,
+            utils::ssb_message_res_parse,
+        )
+        .await;
+        Ok(history_stream)
     }
 }

src/utils.rs

@@ -1,7 +1,8 @@
 //! Utility methods for `golgi`.
-use async_std::io::Read;
 use std::fmt::Debug;
 
+use async_std::{io::Read, net::TcpStream, stream::Stream};
+use async_stream::stream;
 use kuska_ssb::rpc::{RecvMsg, RequestNo, RpcReader};
 use serde_json::Value;
 
@@ -140,8 +141,6 @@ where
     Ok(messages)
 }
 
-
-
 /// Takes in an rpc request number, and a handling function,
 /// and calls the handling function on all responses which match the request number,
 /// and prints out the result of the handling function.
@@ -183,3 +182,60 @@ where
     }
     Ok(())
 }
+
+/// Takes in an rpc request number, and a handling function (parsing results of type T),
+/// and produces an async_std::stream::Stream
+/// of results of type T where the handling function is called
+/// on all rpc_reader responses which match the request number.
+///
+/// # Arguments
+///
+/// * `req_no` - A `RequestNo` of the response to listen for
+/// * `f` - A function which takes in an array of u8 and returns a Result<T, GolgiError>.
+/// This is a function which parses the response from the RpcReader. T is a generic type,
+/// so this parse function can return multiple possible types (String, json, custom struct etc.)
+pub async fn get_source_stream<'a, F, T>(
+    mut rpc_reader: RpcReader<TcpStream>,
+    req_no: RequestNo,
+    f: F,
+) -> impl Stream<Item = Result<T, GolgiError>>
+where
+    F: Fn(&[u8]) -> Result<T, GolgiError>,
+    T: Debug + serde::Deserialize<'a>,
+{
+    // we use the async_stream::stream macro to allow for creating a stream which calls async functions
+    // see https://users.rust-lang.org/t/how-to-create-async-std-stream-which-calls-async-function-in-poll-next/69760
+    let source_stream = stream! {
+        loop {
+            // get the next message from the rpc_reader
+            let (id, msg) = rpc_reader.recv().await?;
+            let x : i32 = id.clone();
+            // check if the next message from rpc_reader matches the req_no we are looking for
+            // if it matches, then this rpc response is for the given request
+            // and if it doesn't match, then we ignore it
+            if x == req_no {
+                match msg {
+                    RecvMsg::RpcResponse(_type, body) => {
+                        // parse an item of type T from the message body using the provided
+                        // function for parsing
+                        let item = f(&body)?;
+                        // return Ok(item) as the next value in the stream
+                        yield Ok(item)
+                    }
+                    RecvMsg::ErrorResponse(message) => {
+                        // if an error is received
+                        // return an Err(err) as the next value in the stream
+                        yield Err(GolgiError::Sbot(message.to_string()));
+                    }
+                    // if we find a CancelStreamResponse
+                    // this is the end of the stream
+                    RecvMsg::CancelStreamRespose() => break,
+                    // if we find an unknown response, we just continue the loop
+                    _ => {}
+                }
+            }
+        }
+    };
+    // finally return the stream object
+    source_stream
+}