diff --git a/Cargo.lock b/Cargo.lock index 323f046..edae602 100644 --- a/Cargo.lock +++ b/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", diff --git a/Cargo.toml b/Cargo.toml index ee3756f..7fc8728 100644 --- a/Cargo.toml +++ b/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" \ No newline at end of file diff --git a/examples/ssb-client.rs b/examples/ssb-client.rs index 00f9759..d1216a5 100644 --- a/examples/ssb-client.rs +++ b/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(()) diff --git a/examples/ssb-stream-example.rs b/examples/ssb-stream-example.rs new file mode 100644 index 0000000..22ef92b --- /dev/null +++ b/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 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 = 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); + } +} diff --git a/src/error.rs b/src/error.rs index 7060222..1c34248 100644 --- a/src/error.rs +++ b/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 for GolgiError { impl From for GolgiError { fn from(err: Utf8Error) -> Self { - GolgiError::Utf8Parse { source: err} + GolgiError::Utf8Parse { source: err } } } diff --git a/src/messages.rs b/src/messages.rs index 6ea9cc3..0b3c538 100644 --- a/src/messages.rs +++ b/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 { - 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 { + 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, } } diff --git a/src/sbot.rs b/src/sbot.rs index a7b2250..361f3c6 100644 --- a/src/sbot.rs +++ b/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, + rpc_reader: RpcReader, +} + /// 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, - rpc_reader: RpcReader, + // 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 { + 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 { 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, GolgiError> { - let req_id = self.client.getsubset_req_send(query).await?; + pub async fn get_subset( + &mut self, + query: SubsetQuery, + ) -> Result, 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 { - 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 { - 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, GolgiError> { - let query = SubsetQuery::Author{ + pub async fn get_about_messages( + &mut self, + ssb_id: &str, + ) -> Result, 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 = messages.into_iter().filter(|msg| { - msg.is_message_type(SsbMessageContentType::About) - }).collect(); + let mut about_messages: Vec = 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, GolgiError> { + pub async fn get_latest_about_message( + &mut self, + ssb_id: &str, + key: &str, + ) -> Result, 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, 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, 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 pub async fn create_history_stream( &mut self, id: String, - ) -> Result, GolgiError> { + ) -> Result>, 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) } } diff --git a/src/utils.rs b/src/utils.rs index 34f5ec6..97f8323 100644 --- a/src/utils.rs +++ b/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. +/// 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, + req_no: RequestNo, + f: F, +) -> impl Stream> +where + F: Fn(&[u8]) -> Result, + 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 +}