From 054877794894e4a483f95cb1afe7ece0caa7c2ea Mon Sep 17 00:00:00 2001 From: notplants Date: Tue, 4 Jan 2022 11:07:51 -0500 Subject: [PATCH 01/16] Working on lifetimes --- Cargo.lock | 73 ++++++++++++++++++++++++++++++++++++- Cargo.toml | 2 + src/sbot.rs | 23 +++++++++++- src/utils.rs | 101 +++++++++++++++++++++++++++++++++++++++++++++------ 4 files changed, 183 insertions(+), 16 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 323f046..c873136 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,12 +499,14 @@ name = "golgi" version = "0.1.0" dependencies = [ "async-std", + "async-stream 0.3.2", "base64 0.13.0", "futures", "hex", "kuska-handshake", "kuska-sodiumoxide", "kuska-ssb", + "rand", "serde", "serde_json", ] @@ -572,7 +595,7 @@ name = "kuska-ssb" version = "0.2.0" dependencies = [ "async-std", - "async-stream", + "async-stream 0.2.1", "base64 0.11.0", "dirs", "futures", @@ -690,6 +713,12 @@ dependencies = [ "winapi 0.3.9", ] +[[package]] +name = "ppv-lite86" +version = "0.2.16" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "eb9f9e6e233e5c4a35559a617bf40a4ec447db2e84c20b55a6f83167b7e57872" + [[package]] name = "proc-macro2" version = "1.0.32" @@ -708,6 +737,46 @@ dependencies = [ "proc-macro2", ] +[[package]] +name = "rand" +version = "0.8.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2e7573632e6454cf6b99d7aac4ccca54be06da05aca2ef7423d22d27d4d4bcd8" +dependencies = [ + "libc", + "rand_chacha", + "rand_core", + "rand_hc", +] + +[[package]] +name = "rand_chacha" +version = "0.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e6c10a63a0fa32252be49d21e7709d4d4baf8d231c2dbce1eaa8141b9b127d88" +dependencies = [ + "ppv-lite86", + "rand_core", +] + +[[package]] +name = "rand_core" +version = "0.6.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d34f1408f55294453790c48b2f1ebbb1c5b4b7563eb1f418bcfcfdbb06ebb4e7" +dependencies = [ + "getrandom", +] + +[[package]] +name = "rand_hc" +version = "0.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d51e9f596de227fda2ea6c84607f5558e196eeaf43c986b724ba4fb8fdf497e7" +dependencies = [ + "rand_core", +] + [[package]] name = "redox_syscall" version = "0.2.10" diff --git a/Cargo.toml b/Cargo.toml index ee3756f..d5eca42 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -15,3 +15,5 @@ kuska-sodiumoxide = "0.2.5-0" kuska-ssb = { path = "../ssb" } serde = { version = "1", features = ["derive"] } serde_json = "1" +async-stream = "0.3.2" +rand = "0.8.4" \ No newline at end of file diff --git a/src/sbot.rs b/src/sbot.rs index a7b2250..dc6def4 100644 --- a/src/sbot.rs +++ b/src/sbot.rs @@ -1,5 +1,7 @@ //! Sbot type and associated methods. use async_std::net::TcpStream; +use futures::pin_mut; +use async_std::stream::StreamExt; use kuska_handshake::async_std::BoxStream; use kuska_sodiumoxide::crypto::{auth, sign::ed25519}; @@ -16,6 +18,7 @@ use kuska_ssb::{ use crate::error::GolgiError; use crate::messages::{SsbMessageKVT, SsbMessageContent, SsbMessageValue, SsbMessageContentType}; use crate::utils; +use crate::utils::get_source_stream; // re-export types from kuska pub use kuska_ssb::api::dto::content::SubsetQuery; @@ -228,9 +231,25 @@ impl Sbot { pub async fn create_history_stream( &mut self, id: String, - ) -> Result, GolgiError> { + ) -> Result { 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 source_stream = get_source_stream(&self.rpc_reader, req_id); + pin_mut!(source_stream); // needed for iteration + + while let Some(res) = source_stream.next().await { + match res { + Ok(value) => { + println!("value: {:?}", value); + }, + Err(err) => { + println!("err: {:?}", err); + } + } + + } + println!("exit loop"); + Ok(2) + // utils::get_source_until_eof(&mut self.rpc_reader, req_id, utils::ssb_message_res_parse).await } } diff --git a/src/utils.rs b/src/utils.rs index 34f5ec6..9a46288 100644 --- a/src/utils.rs +++ b/src/utils.rs @@ -1,6 +1,12 @@ //! Utility methods for `golgi`. use async_std::io::Read; use std::fmt::Debug; +use async_std::task; +use std::time::Duration; +use async_std::net::TcpStream; +use async_std::stream::Stream; +use async_stream::stream; +use rand::distributions::{Distribution, Uniform}; use kuska_ssb::rpc::{RecvMsg, RequestNo, RpcReader}; use serde_json::Value; @@ -64,10 +70,10 @@ pub async fn get_async<'a, R, T, F>( req_no: RequestNo, f: F, ) -> Result -where - R: Read + Unpin, - F: Fn(&[u8]) -> Result, - T: Debug, + where + R: Read + Unpin, + F: Fn(&[u8]) -> Result, + T: Debug, { loop { let (id, msg) = rpc_reader.recv().await?; @@ -108,10 +114,10 @@ pub async fn get_source_until_eof<'a, R, T, F>( req_no: RequestNo, f: F, ) -> Result, GolgiError> -where - R: Read + Unpin, - F: Fn(&[u8]) -> Result, - T: Debug, + where + R: Read + Unpin, + F: Fn(&[u8]) -> Result, + T: Debug, { let mut messages: Vec = Vec::new(); loop { @@ -160,10 +166,10 @@ pub async fn print_source_until_eof<'a, R, T, F>( req_no: RequestNo, f: F, ) -> Result<(), GolgiError> -where - R: Read + Unpin, - F: Fn(&[u8]) -> Result, - T: Debug + serde::Deserialize<'a>, + where + R: Read + Unpin, + F: Fn(&[u8]) -> Result, + T: Debug + serde::Deserialize<'a>, { loop { let (id, msg) = rpc_reader.recv().await?; @@ -183,3 +189,74 @@ where } Ok(()) } + + +pub async fn get_async_num() -> i32 { + let millis = Uniform::from(0..10).sample(&mut rand::thread_rng()); + println!("get_async_num will complete in {} ms", millis); + + task::sleep(Duration::from_millis(millis)).await; + let i: i32 = rand::random(); + i +} + + +// pub fn get_source_stream() -> impl Stream { +// let s = stream! { +// for i in 0..3 { +// let x = get_async_num().await; +// yield x; +// } +// }; +// s +// } + +// pub fn get_source_stream(rpc_reader: &RpcReader, req_no: RequestNo) -> impl Stream { +// let s = stream! { +// loop { +// let (id, msg) = rpc_reader.recv().await?; +// yield Ok(id); +// // if id == req_no { +// // match msg { +// // RecvMsg::RpcResponse(_type, body) => { +// // let display = f(&body)?; +// // println!("{:?}", display); +// // } +// // RecvMsg::ErrorResponse(message) => { +// // return Err(GolgiError::Sbot(message)); +// // } +// // RecvMsg::CancelStreamRespose() => break, +// // _ => {} +// // } +// // } +// // } +// // yield x; +// } +// }; +// s +// } + +pub fn<'a> get_source_stream(mut rpc_reader: &<'a> RpcReader, req_no: RequestNo) -> impl Stream> { + let s = stream! { + loop { + let (id, msg) = rpc_reader.recv().await?; + yield Ok(id); + // if id == req_no { + // match msg { + // RecvMsg::RpcResponse(_type, body) => { + // let display = f(&body)?; + // println!("{:?}", display); + // } + // RecvMsg::ErrorResponse(message) => { + // return Err(GolgiError::Sbot(message)); + // } + // RecvMsg::CancelStreamRespose() => break, + // _ => {} + // } + // } + // } + // yield x; + } + }; + s +} \ No newline at end of file -- 2.54.0 From 58d133b5c8452e40cbf08041590a811f75310078 Mon Sep 17 00:00:00 2001 From: notplants Date: Tue, 4 Jan 2022 12:01:22 -0500 Subject: [PATCH 02/16] Working recv stream --- src/sbot.rs | 38 ++++++++++++++++++++++++++++++++++---- src/utils.rs | 49 ------------------------------------------------- 2 files changed, 34 insertions(+), 53 deletions(-) diff --git a/src/sbot.rs b/src/sbot.rs index dc6def4..a3d876d 100644 --- a/src/sbot.rs +++ b/src/sbot.rs @@ -1,7 +1,9 @@ //! Sbot type and associated methods. +use std::fmt::Debug; use async_std::net::TcpStream; use futures::pin_mut; -use async_std::stream::StreamExt; +use async_std::stream::{Stream, StreamExt}; +use async_stream::stream; use kuska_handshake::async_std::BoxStream; use kuska_sodiumoxide::crypto::{auth, sign::ed25519}; @@ -12,16 +14,16 @@ use kuska_ssb::{ }, discovery, keystore, keystore::OwnedIdentity, - rpc::{RpcReader, RpcWriter}, + rpc::{RpcReader, RpcWriter, RecvMsg}, }; use crate::error::GolgiError; use crate::messages::{SsbMessageKVT, SsbMessageContent, SsbMessageValue, SsbMessageContentType}; use crate::utils; -use crate::utils::get_source_stream; // re-export types from kuska pub use kuska_ssb::api::dto::content::SubsetQuery; +use kuska_ssb::rpc::RequestNo; /// 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. @@ -234,7 +236,7 @@ impl Sbot { ) -> Result { let args = CreateHistoryStreamIn::new(id); let req_id = self.client.create_history_stream_req_send(&args).await?; - let source_stream = get_source_stream(&self.rpc_reader, req_id); + let source_stream = self.get_source_stream(req_id, utils::ssb_message_res_parse); pin_mut!(source_stream); // needed for iteration while let Some(res) = source_stream.next().await { @@ -252,4 +254,32 @@ impl Sbot { Ok(2) // utils::get_source_until_eof(&mut self.rpc_reader, req_id, utils::ssb_message_res_parse).await } + + pub fn get_source_stream<'a, F, T>(&'a mut self, req_no: RequestNo, f: F) -> impl Stream> + 'a + where + F: Fn(&[u8]) -> Result + 'a, + T: Debug + serde::Deserialize<'a>, + { + let s = stream! { + loop { + let (id, msg) = &self.rpc_reader.recv().await?; + let x : i32 = id.clone(); + if x == req_no { + match msg { + RecvMsg::RpcResponse(_type, body) => { + let display = f(&body)?; + println!("display: {:?}", display); + yield Ok(x) + } + RecvMsg::ErrorResponse(message) => { + yield Err(GolgiError::Sbot(message.to_string())); + } + RecvMsg::CancelStreamRespose() => break, + _ => break + } + } + } + }; + s + } } diff --git a/src/utils.rs b/src/utils.rs index 9a46288..3478030 100644 --- a/src/utils.rs +++ b/src/utils.rs @@ -211,52 +211,3 @@ pub async fn get_async_num() -> i32 { // s // } -// pub fn get_source_stream(rpc_reader: &RpcReader, req_no: RequestNo) -> impl Stream { -// let s = stream! { -// loop { -// let (id, msg) = rpc_reader.recv().await?; -// yield Ok(id); -// // if id == req_no { -// // match msg { -// // RecvMsg::RpcResponse(_type, body) => { -// // let display = f(&body)?; -// // println!("{:?}", display); -// // } -// // RecvMsg::ErrorResponse(message) => { -// // return Err(GolgiError::Sbot(message)); -// // } -// // RecvMsg::CancelStreamRespose() => break, -// // _ => {} -// // } -// // } -// // } -// // yield x; -// } -// }; -// s -// } - -pub fn<'a> get_source_stream(mut rpc_reader: &<'a> RpcReader, req_no: RequestNo) -> impl Stream> { - let s = stream! { - loop { - let (id, msg) = rpc_reader.recv().await?; - yield Ok(id); - // if id == req_no { - // match msg { - // RecvMsg::RpcResponse(_type, body) => { - // let display = f(&body)?; - // println!("{:?}", display); - // } - // RecvMsg::ErrorResponse(message) => { - // return Err(GolgiError::Sbot(message)); - // } - // RecvMsg::CancelStreamRespose() => break, - // _ => {} - // } - // } - // } - // yield x; - } - }; - s -} \ No newline at end of file -- 2.54.0 From 8ad65f785d511b1cacdf1107afc295a2b3bb83be Mon Sep 17 00:00:00 2001 From: notplants Date: Tue, 4 Jan 2022 14:09:49 -0500 Subject: [PATCH 03/16] Working create_history_stream --- examples/ssb-stream-example.rs | 97 ++++++++++++++++++++++++++++++++++ src/sbot.rs | 70 +++++++++++++----------- 2 files changed, 137 insertions(+), 30 deletions(-) create mode 100644 examples/ssb-stream-example.rs diff --git a/examples/ssb-stream-example.rs b/examples/ssb-stream-example.rs new file mode 100644 index 0000000..2e82ea8 --- /dev/null +++ b/examples/ssb-stream-example.rs @@ -0,0 +1,97 @@ +use std::process; + +use golgi::error::GolgiError; + +use golgi::sbot::Sbot; +use async_std::stream::StreamExt; +use futures::pin_mut; + + +async fn run() -> Result<(), GolgiError> { + let mut sbot_client = Sbot::init(None, None).await?; + + let id = sbot_client.whoami().await?; + println!("{}", id); + // + // let name = SsbMessageContent::About { + // about: id, + // name: Some("golgi".to_string()), + // title: None, + // branch: None, + // image: None, + // description: None, + // location: None, + // start_datetime: None, + // }; + // + // let name_msg_ref = sbot_client.publish(name).await?; + // println!("{}", name_msg_ref); + // + // let post = SsbMessageContent::Post { + // text: "golgi go womp womp".to_string(), + // mentions: None, + // }; + // + // 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?; + // println!("description: {}", post_msg_ref); + + let author = "@L/z54cbc8V1kL1/MiBhpEKuN3QJkSoZYNaukny3ghIs=.ed25519"; + + // let query = SubsetQuery::Author{ + // op: "author".to_string(), + // feed: author + // }; + // + // let kvts = sbot_client.get_subset(query).await?; + // + // for kvt in kvts { + // println!("kvt: {:?}", kvt); + // } + + // let about_messages = sbot_client.get_about_messages(author).await?; + // for msg in about_messages { + // println!("msg: {:?}", msg); + // } + // + // sbot_client.publish_name("this is my new name").await?; + // let name = sbot_client.get_name(author).await?; + // println!("name: {:?}", name); + // + // // sbot_client.publish_description("this is test description 99").await?; + // + // let description = sbot_client.get_description(author).await?; + // println!("desc: {:?}", description); + + // + // let description2 = sbot_client.get_latest_about_message(author, "description").await?; + // println!("desc2: {:?}", description2); + + let mut history_stream = sbot_client.create_history_stream(author.to_string()).await?; + pin_mut!(history_stream); // needed for iteration + + while let Some(res) = history_stream.next().await { + match res { + Ok(value) => { + println!("value: {:?}", value); + }, + Err(err) => { + println!("err: {:?}", err); + } + } + + } + println!("exit loop"); + + Ok(()) +} + +#[async_std::main] +async fn main() { + if let Err(e) = run().await { + eprintln!("Application error: {}", e); + process::exit(1); + } +} diff --git a/src/sbot.rs b/src/sbot.rs index a3d876d..5022977 100644 --- a/src/sbot.rs +++ b/src/sbot.rs @@ -228,58 +228,68 @@ impl Sbot { self.get_latest_about_message(ssb_id, "description").await } - /// Call the `createHistoryStream` RPC method and return a vector - /// of SsbMessageValue. - pub async fn create_history_stream( - &mut self, + /// Call the `createHistoryStream` RPC method + /// and return a Stream of Result + pub async fn create_history_stream<'a>( + &'a mut self, id: String, - ) -> Result { + ) -> Result> + 'a, GolgiError> { let args = CreateHistoryStreamIn::new(id); let req_id = self.client.create_history_stream_req_send(&args).await?; - let source_stream = self.get_source_stream(req_id, utils::ssb_message_res_parse); - pin_mut!(source_stream); // needed for iteration - - while let Some(res) = source_stream.next().await { - match res { - Ok(value) => { - println!("value: {:?}", value); - }, - Err(err) => { - println!("err: {:?}", err); - } - } - - } - println!("exit loop"); - Ok(2) - // utils::get_source_until_eof(&mut self.rpc_reader, req_id, utils::ssb_message_res_parse).await + let history_stream = self.get_source_stream(req_id, utils::ssb_message_res_parse); + Ok(history_stream) } - pub fn get_source_stream<'a, F, T>(&'a mut self, req_no: RequestNo, f: F) -> impl Stream> + 'a + /// 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 functions 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 fn get_source_stream<'a, F, T>(&'a mut self, req_no: RequestNo, f: F) -> impl Stream> + 'a where F: Fn(&[u8]) -> Result + 'a, - T: Debug + serde::Deserialize<'a>, + T: Debug + serde::Deserialize<'a> + 'a, { - let s = stream! { + // 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) = &self.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) => { - let display = f(&body)?; - println!("display: {:?}", display); - yield Ok(x) + // 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, - _ => break + // if we find an unknown response, we just continue the loop + _ => {} } } } }; - s + // finally return the stream object + source_stream } } -- 2.54.0 From 2fd4c12a95f03b3078b45f9662abff8ac8c4e2e7 Mon Sep 17 00:00:00 2001 From: notplants Date: Tue, 4 Jan 2022 14:58:08 -0500 Subject: [PATCH 04/16] Working example --- examples/ssb-stream-example.rs | 113 ++++++++++++++++----------------- 1 file changed, 54 insertions(+), 59 deletions(-) diff --git a/examples/ssb-stream-example.rs b/examples/ssb-stream-example.rs index 2e82ea8..3ed0fc4 100644 --- a/examples/ssb-stream-example.rs +++ b/examples/ssb-stream-example.rs @@ -3,8 +3,9 @@ use std::process; use golgi::error::GolgiError; use golgi::sbot::Sbot; -use async_std::stream::StreamExt; -use futures::pin_mut; +use async_std::stream::{Stream, StreamExt}; +use futures::{pin_mut, TryStreamExt}; +use golgi::messages::{SsbMessageContentType, SsbMessageValue}; async fn run() -> Result<(), GolgiError> { @@ -12,66 +13,15 @@ async fn run() -> Result<(), GolgiError> { let id = sbot_client.whoami().await?; println!("{}", id); - // - // let name = SsbMessageContent::About { - // about: id, - // name: Some("golgi".to_string()), - // title: None, - // branch: None, - // image: None, - // description: None, - // location: None, - // start_datetime: None, - // }; - // - // let name_msg_ref = sbot_client.publish(name).await?; - // println!("{}", name_msg_ref); - // - // let post = SsbMessageContent::Post { - // text: "golgi go womp womp".to_string(), - // mentions: None, - // }; - // - // 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?; - // println!("description: {}", post_msg_ref); let author = "@L/z54cbc8V1kL1/MiBhpEKuN3QJkSoZYNaukny3ghIs=.ed25519"; - // let query = SubsetQuery::Author{ - // op: "author".to_string(), - // feed: author - // }; - // - // let kvts = sbot_client.get_subset(query).await?; - // - // for kvt in kvts { - // println!("kvt: {:?}", kvt); - // } - - // let about_messages = sbot_client.get_about_messages(author).await?; - // for msg in about_messages { - // println!("msg: {:?}", msg); - // } - // - // sbot_client.publish_name("this is my new name").await?; - // let name = sbot_client.get_name(author).await?; - // println!("name: {:?}", name); - // - // // sbot_client.publish_description("this is test description 99").await?; - // - // let description = sbot_client.get_description(author).await?; - // println!("desc: {:?}", description); - - // - // let description2 = sbot_client.get_latest_about_message(author, "description").await?; - // println!("desc2: {:?}", description2); - + // create a history stream let mut history_stream = sbot_client.create_history_stream(author.to_string()).await?; - pin_mut!(history_stream); // needed for iteration + // 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) => { @@ -81,10 +31,55 @@ async fn run() -> Result<(), GolgiError> { println!("err: {:?}", err); } } - } - println!("exit loop"); + 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 mut 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 KeyTypeTuple (local struct for storing message_key and message_type) + #[derive(Debug)] + struct KeyTypeTuple { + message_key: String, + message_type: SsbMessageContentType, + }; + let mut 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 = KeyTypeTuple { + message_key: val.signature, + message_type: 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(()) } -- 2.54.0 From c4b57ae813b30ae7bac21be2f901072602ddfc63 Mon Sep 17 00:00:00 2001 From: notplants Date: Tue, 4 Jan 2022 14:59:28 -0500 Subject: [PATCH 05/16] Clean up --- src/utils.rs | 24 +----------------------- 1 file changed, 1 insertion(+), 23 deletions(-) diff --git a/src/utils.rs b/src/utils.rs index 3478030..4dd0f1e 100644 --- a/src/utils.rs +++ b/src/utils.rs @@ -188,26 +188,4 @@ pub async fn print_source_until_eof<'a, R, T, F>( } } Ok(()) -} - - -pub async fn get_async_num() -> i32 { - let millis = Uniform::from(0..10).sample(&mut rand::thread_rng()); - println!("get_async_num will complete in {} ms", millis); - - task::sleep(Duration::from_millis(millis)).await; - let i: i32 = rand::random(); - i -} - - -// pub fn get_source_stream() -> impl Stream { -// let s = stream! { -// for i in 0..3 { -// let x = get_async_num().await; -// yield x; -// } -// }; -// s -// } - +} \ No newline at end of file -- 2.54.0 From 5927b7dfe60047d6aac66867b8d938e73ebf6584 Mon Sep 17 00:00:00 2001 From: notplants Date: Tue, 4 Jan 2022 15:00:05 -0500 Subject: [PATCH 06/16] Cleanup --- examples/ssb-stream-example.rs | 94 +++++++++++++++++----------------- src/messages.rs | 4 +- src/sbot.rs | 2 +- src/utils.rs | 12 ++--- 4 files changed, 56 insertions(+), 56 deletions(-) diff --git a/examples/ssb-stream-example.rs b/examples/ssb-stream-example.rs index 3ed0fc4..ee91685 100644 --- a/examples/ssb-stream-example.rs +++ b/examples/ssb-stream-example.rs @@ -34,53 +34,53 @@ async fn run() -> Result<(), GolgiError> { } 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 mut 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 KeyTypeTuple (local struct for storing message_key and message_type) - #[derive(Debug)] - struct KeyTypeTuple { - message_key: String, - message_type: SsbMessageContentType, - }; - let mut 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 = KeyTypeTuple { - message_key: val.signature, - message_type: 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(()) + // // 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 mut 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 KeyTypeTuple (local struct for storing message_key and message_type) + // #[derive(Debug)] + // struct KeyTypeTuple { + // message_key: String, + // message_type: SsbMessageContentType, + // }; + // let mut 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 = KeyTypeTuple { + // message_key: val.signature, + // message_type: 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] diff --git a/src/messages.rs b/src/messages.rs index 6ea9cc3..2f7f7b1 100644 --- a/src/messages.rs +++ b/src/messages.rs @@ -62,11 +62,11 @@ impl SsbMessageValue { /// 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 - pub fn is_message_type(&self, message_type: SsbMessageContentType) -> bool { + pub fn is_message_type(&self, _message_type: SsbMessageContentType) -> bool { let self_message_type = self.get_message_type(); match self_message_type { Ok(mtype) => { - matches!(mtype, message_type) + matches!(mtype, _message_type) } Err(_err) => { false diff --git a/src/sbot.rs b/src/sbot.rs index 5022977..9fff930 100644 --- a/src/sbot.rs +++ b/src/sbot.rs @@ -1,7 +1,7 @@ //! Sbot type and associated methods. use std::fmt::Debug; use async_std::net::TcpStream; -use futures::pin_mut; + use async_std::stream::{Stream, StreamExt}; use async_stream::stream; diff --git a/src/utils.rs b/src/utils.rs index 4dd0f1e..54e20a2 100644 --- a/src/utils.rs +++ b/src/utils.rs @@ -1,12 +1,12 @@ //! Utility methods for `golgi`. use async_std::io::Read; use std::fmt::Debug; -use async_std::task; -use std::time::Duration; -use async_std::net::TcpStream; -use async_std::stream::Stream; -use async_stream::stream; -use rand::distributions::{Distribution, Uniform}; + + + + + + use kuska_ssb::rpc::{RecvMsg, RequestNo, RpcReader}; use serde_json::Value; -- 2.54.0 From 5711efbece02743f0902c87ba658e8c880af73e3 Mon Sep 17 00:00:00 2001 From: notplants Date: Tue, 4 Jan 2022 15:00:25 -0500 Subject: [PATCH 07/16] Cleanup --- examples/ssb-stream-example.rs | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/examples/ssb-stream-example.rs b/examples/ssb-stream-example.rs index ee91685..3f056e7 100644 --- a/examples/ssb-stream-example.rs +++ b/examples/ssb-stream-example.rs @@ -3,9 +3,9 @@ use std::process; use golgi::error::GolgiError; use golgi::sbot::Sbot; -use async_std::stream::{Stream, StreamExt}; -use futures::{pin_mut, TryStreamExt}; -use golgi::messages::{SsbMessageContentType, SsbMessageValue}; +use async_std::stream::{StreamExt}; +use futures::{pin_mut}; + async fn run() -> Result<(), GolgiError> { @@ -17,7 +17,7 @@ async fn run() -> Result<(), GolgiError> { let author = "@L/z54cbc8V1kL1/MiBhpEKuN3QJkSoZYNaukny3ghIs=.ed25519"; // create a history stream - let mut history_stream = sbot_client.create_history_stream(author.to_string()).await?; + 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 @@ -80,7 +80,7 @@ async fn run() -> Result<(), GolgiError> { // println!("reached end of type stream"); // // // return Ok - // Ok(()) + Ok(()) } #[async_std::main] -- 2.54.0 From 1eb377d065a1890bb115db54e70967e206450a66 Mon Sep 17 00:00:00 2001 From: notplants Date: Tue, 4 Jan 2022 15:02:20 -0500 Subject: [PATCH 08/16] Fix clippy error --- src/messages.rs | 1 + src/sbot.rs | 2 +- 2 files changed, 2 insertions(+), 1 deletion(-) diff --git a/src/messages.rs b/src/messages.rs index 2f7f7b1..ea08802 100644 --- a/src/messages.rs +++ b/src/messages.rs @@ -31,6 +31,7 @@ pub struct SsbMessageValue { // Enum representing the different possible message content types #[derive(Debug)] +#[allow(missing_docs)] pub enum SsbMessageContentType { About, Vote, diff --git a/src/sbot.rs b/src/sbot.rs index 9fff930..d63c741 100644 --- a/src/sbot.rs +++ b/src/sbot.rs @@ -271,7 +271,7 @@ impl Sbot { RecvMsg::RpcResponse(_type, body) => { // parse an item of type T from the message body using the provided // function for parsing - let item = f(&body)?; + let item = f(body)?; // return Ok(item) as the next value in the stream yield Ok(item) } -- 2.54.0 From db146446d07247a853e715a790dd74ea32464683 Mon Sep 17 00:00:00 2001 From: notplants Date: Tue, 4 Jan 2022 15:05:13 -0500 Subject: [PATCH 09/16] Fix imports --- src/messages.rs | 1 - src/sbot.rs | 1 - src/utils.rs | 6 ------ 3 files changed, 8 deletions(-) diff --git a/src/messages.rs b/src/messages.rs index ea08802..3af22b5 100644 --- a/src/messages.rs +++ b/src/messages.rs @@ -1,7 +1,6 @@ //! 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; diff --git a/src/sbot.rs b/src/sbot.rs index d63c741..1583262 100644 --- a/src/sbot.rs +++ b/src/sbot.rs @@ -1,7 +1,6 @@ //! Sbot type and associated methods. use std::fmt::Debug; use async_std::net::TcpStream; - use async_std::stream::{Stream, StreamExt}; use async_stream::stream; diff --git a/src/utils.rs b/src/utils.rs index 54e20a2..138e605 100644 --- a/src/utils.rs +++ b/src/utils.rs @@ -2,12 +2,6 @@ use async_std::io::Read; use std::fmt::Debug; - - - - - - use kuska_ssb::rpc::{RecvMsg, RequestNo, RpcReader}; use serde_json::Value; -- 2.54.0 From a076e4741f79641edd73f2fc4359fa1aed704a6b Mon Sep 17 00:00:00 2001 From: notplants Date: Wed, 5 Jan 2022 08:50:57 -0500 Subject: [PATCH 10/16] Multiple RPC reader --- src/sbot.rs | 57 ++++++++++++++++++++++++++++++++++++++++++++--------- 1 file changed, 48 insertions(+), 9 deletions(-) diff --git a/src/sbot.rs b/src/sbot.rs index 1583262..f64f614 100644 --- a/src/sbot.rs +++ b/src/sbot.rs @@ -92,6 +92,43 @@ impl Sbot { }) } + pub async fn get_rpc_reader(&self, ip_port: Option, net_id: Option) -> Result, GolgiError> { + let address = if ip_port.is_none() { + "127.0.0.1:8008".to_string() + } else { + ip_port.unwrap() + }; + + let network_id = if net_id.is_none() { + discovery::ssb_net_id() + } else { + auth::Key::from_slice(&hex::decode(net_id.unwrap()).unwrap()).unwrap() + }; + + let socket = TcpStream::connect(&address) + .await + .map_err(|source| GolgiError::Io { + source, + context: "socket error; failed to initiate tcp stream connection".to_string(), + })?; + + let handshake = kuska_handshake::async_std::handshake_client( + &mut &socket, + network_id.clone(), + self.public_key, + self.private_key.clone(), + self.public_key, + ) + .await + .map_err(GolgiError::Handshake)?; + + let (box_stream_read, box_stream_write) = + BoxStream::from_handshake(socket.clone(), socket, handshake, 0x8000).split_read_write(); + + let rpc_reader = RpcReader::new(box_stream_read); + Ok(rpc_reader) + } + /// 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) @@ -229,13 +266,14 @@ impl Sbot { /// Call the `createHistoryStream` RPC method /// and return a Stream of Result - pub async fn create_history_stream<'a>( - &'a mut self, + pub async fn create_history_stream( + &mut self, id: String, - ) -> Result> + 'a, GolgiError> { + ) -> Result>, GolgiError> { let args = CreateHistoryStreamIn::new(id); let req_id = self.client.create_history_stream_req_send(&args).await?; - let history_stream = self.get_source_stream(req_id, utils::ssb_message_res_parse); + let mut rpc_reader = self.get_rpc_reader(None, None).await.unwrap(); + let history_stream = Sbot::get_source_stream(rpc_reader, req_id, utils::ssb_message_res_parse).await; Ok(history_stream) } @@ -250,17 +288,18 @@ impl Sbot { /// * `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 fn get_source_stream<'a, F, T>(&'a mut self, req_no: RequestNo, f: F) -> impl Stream> + 'a + 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 + 'a, - T: Debug + serde::Deserialize<'a> + 'a, + 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) = &self.rpc_reader.recv().await?; + 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 @@ -270,7 +309,7 @@ impl Sbot { RecvMsg::RpcResponse(_type, body) => { // parse an item of type T from the message body using the provided // function for parsing - let item = f(body)?; + let item = f(&body)?; // return Ok(item) as the next value in the stream yield Ok(item) } -- 2.54.0 From 03da29e3f958bfc0cf611ac105bea642b630a0fd Mon Sep 17 00:00:00 2001 From: notplants Date: Wed, 5 Jan 2022 10:58:07 -0500 Subject: [PATCH 11/16] Multiple RPC --- examples/ssb-stream-example.rs | 96 +++++++++++++++++----------------- src/sbot.rs | 28 +++++++--- 2 files changed, 70 insertions(+), 54 deletions(-) diff --git a/examples/ssb-stream-example.rs b/examples/ssb-stream-example.rs index 3f056e7..077bb39 100644 --- a/examples/ssb-stream-example.rs +++ b/examples/ssb-stream-example.rs @@ -4,8 +4,8 @@ use golgi::error::GolgiError; use golgi::sbot::Sbot; use async_std::stream::{StreamExt}; -use futures::{pin_mut}; - +use futures::{pin_mut, TryStreamExt}; +use golgi::messages::{SsbMessageContentType, SsbMessageValue}; async fn run() -> Result<(), GolgiError> { @@ -34,52 +34,52 @@ async fn run() -> Result<(), GolgiError> { } 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 mut 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 KeyTypeTuple (local struct for storing message_key and message_type) - // #[derive(Debug)] - // struct KeyTypeTuple { - // message_key: String, - // message_type: SsbMessageContentType, - // }; - // let mut 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 = KeyTypeTuple { - // message_key: val.signature, - // message_type: 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 + // 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 mut 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 KeyTypeTuple (local struct for storing message_key and message_type) + #[derive(Debug)] + struct KeyTypeTuple { + message_key: String, + message_type: SsbMessageContentType, + }; + let mut 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 = KeyTypeTuple { + message_key: val.signature, + message_type: 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(()) } diff --git a/src/sbot.rs b/src/sbot.rs index f64f614..998b498 100644 --- a/src/sbot.rs +++ b/src/sbot.rs @@ -24,6 +24,11 @@ use crate::utils; pub use kuska_ssb::api::dto::content::SubsetQuery; use kuska_ssb::rpc::RequestNo; +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 { @@ -35,6 +40,7 @@ pub struct Sbot { network_id: auth::Key, client: ApiCaller, rpc_reader: RpcReader, + sbot_connections: Vec, } impl Sbot { @@ -80,6 +86,7 @@ impl Sbot { let rpc_reader = RpcReader::new(box_stream_read); let client = ApiCaller::new(RpcWriter::new(box_stream_write)); + let mut sbot_connections = Vec::new(); Ok(Self { id, @@ -89,10 +96,11 @@ impl Sbot { network_id, client, rpc_reader, + sbot_connections }) } - pub async fn get_rpc_reader(&self, ip_port: Option, net_id: Option) -> Result, GolgiError> { + pub async fn get_sbot_connection(&self, ip_port: Option, net_id: Option) -> Result { let address = if ip_port.is_none() { "127.0.0.1:8008".to_string() } else { @@ -126,7 +134,12 @@ impl Sbot { BoxStream::from_handshake(socket.clone(), socket, handshake, 0x8000).split_read_write(); let rpc_reader = RpcReader::new(box_stream_read); - Ok(rpc_reader) + let client = ApiCaller::new(RpcWriter::new(box_stream_write)); + let sbot_connection = SbotConnection { + rpc_reader, + client, + }; + Ok(sbot_connection) } /// Call the `partialReplication getSubset` RPC method and return a vector @@ -270,10 +283,10 @@ impl Sbot { &mut self, id: String, ) -> Result>, GolgiError> { + let mut sbot_connection = self.get_sbot_connection(None, None).await.unwrap(); let args = CreateHistoryStreamIn::new(id); - let req_id = self.client.create_history_stream_req_send(&args).await?; - let mut rpc_reader = self.get_rpc_reader(None, None).await.unwrap(); - let history_stream = Sbot::get_source_stream(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 = Sbot::get_source_stream(sbot_connection.rpc_reader, req_id, utils::ssb_message_res_parse).await; Ok(history_stream) } @@ -320,7 +333,10 @@ impl Sbot { } // if we find a CancelStreamResponse // this is the end of the stream - RecvMsg::CancelStreamRespose() => break, + RecvMsg::CancelStreamRespose() => { + println!("found cancel stream"); + break + }, // if we find an unknown response, we just continue the loop _ => {} } -- 2.54.0 From 22417e4d8232bb3dc601cbc5a00c125073d8fe9d Mon Sep 17 00:00:00 2001 From: notplants Date: Wed, 5 Jan 2022 10:58:34 -0500 Subject: [PATCH 12/16] Cleanup --- src/sbot.rs | 5 +---- 1 file changed, 1 insertion(+), 4 deletions(-) diff --git a/src/sbot.rs b/src/sbot.rs index 998b498..d553128 100644 --- a/src/sbot.rs +++ b/src/sbot.rs @@ -333,10 +333,7 @@ impl Sbot { } // if we find a CancelStreamResponse // this is the end of the stream - RecvMsg::CancelStreamRespose() => { - println!("found cancel stream"); - break - }, + RecvMsg::CancelStreamRespose() => break, // if we find an unknown response, we just continue the loop _ => {} } -- 2.54.0 From 26e2809c9ae0e42085856a8b520f539220a192e4 Mon Sep 17 00:00:00 2001 From: notplants Date: Wed, 5 Jan 2022 13:58:48 -0500 Subject: [PATCH 13/16] Get one SbotConnection per stream call --- examples/ssb-client.rs | 6 +- examples/ssb-stream-example.rs | 53 +++---- src/error.rs | 8 +- src/messages.rs | 47 +++--- src/sbot.rs | 260 +++++++++++++++------------------ src/utils.rs | 88 +++++++++-- 6 files changed, 245 insertions(+), 217 deletions(-) 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 index 077bb39..22ef92b 100644 --- a/examples/ssb-stream-example.rs +++ b/examples/ssb-stream-example.rs @@ -1,12 +1,11 @@ use std::process; -use golgi::error::GolgiError; - -use golgi::sbot::Sbot; -use async_std::stream::{StreamExt}; +use async_std::stream::StreamExt; use futures::{pin_mut, TryStreamExt}; -use golgi::messages::{SsbMessageContentType, SsbMessageValue}; +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?; @@ -17,7 +16,9 @@ async fn run() -> Result<(), GolgiError> { let author = "@L/z54cbc8V1kL1/MiBhpEKuN3QJkSoZYNaukny3ghIs=.ed25519"; // create a history stream - let history_stream = sbot_client.create_history_stream(author.to_string()).await?; + 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 @@ -26,7 +27,7 @@ async fn run() -> Result<(), GolgiError> { match res { Ok(value) => { println!("value: {:?}", value); - }, + } Err(err) => { println!("err: {:?}", err); } @@ -36,34 +37,26 @@ async fn run() -> Result<(), GolgiError> { // 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 mut history_stream = sbot_client.create_history_stream(author.to_string()).await?; - let results : Vec = history_stream.try_collect().await?; + 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 KeyTypeTuple (local struct for storing message_key and message_type) - #[derive(Debug)] - struct KeyTypeTuple { - message_key: String, - message_type: SsbMessageContentType, - }; - let mut 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 = KeyTypeTuple { - message_key: val.signature, - message_type: message_type, - }; - Ok(tuple) - } - Err(err) => { - Err(err) - } + // 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"); @@ -71,7 +64,7 @@ async fn run() -> Result<(), GolgiError> { match res { Ok(value) => { println!("value: {:?}", value); - }, + } Err(err) => { println!("err: {:?}", err); } 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 3af22b5..43a8f28 100644 --- a/src/messages.rs +++ b/src/messages.rs @@ -1,9 +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; @@ -36,29 +36,30 @@ pub enum SsbMessageContentType { 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 d553128..361f3c6 100644 --- a/src/sbot.rs +++ b/src/sbot.rs @@ -1,29 +1,27 @@ //! Sbot type and associated methods. -use std::fmt::Debug; -use async_std::net::TcpStream; -use async_std::stream::{Stream, StreamExt}; -use async_stream::stream; +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, RecvMsg}, + 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; -use kuska_ssb::rpc::RequestNo; +/// 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, @@ -38,9 +36,9 @@ pub struct Sbot { address: String, // aka caps key (scuttleverse identifier) network_id: auth::Key, - client: ApiCaller, - rpc_reader: RpcReader, - sbot_connections: Vec, + // 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 { @@ -64,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 { @@ -74,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)?; @@ -86,76 +118,41 @@ impl Sbot { let rpc_reader = RpcReader::new(box_stream_read); let client = ApiCaller::new(RpcWriter::new(box_stream_write)); - let mut sbot_connections = Vec::new(); - - Ok(Self { - id, - public_key: pk, - private_key: sk, - address, - network_id, - client, - rpc_reader, - sbot_connections - }) - } - - pub async fn get_sbot_connection(&self, ip_port: Option, net_id: Option) -> Result { - let address = if ip_port.is_none() { - "127.0.0.1:8008".to_string() - } else { - ip_port.unwrap() - }; - - let network_id = if net_id.is_none() { - discovery::ssb_net_id() - } else { - auth::Key::from_slice(&hex::decode(net_id.unwrap()).unwrap()).unwrap() - }; - - let socket = TcpStream::connect(&address) - .await - .map_err(|source| GolgiError::Io { - source, - context: "socket error; failed to initiate tcp stream connection".to_string(), - })?; - - let handshake = kuska_handshake::async_std::handshake_client( - &mut &socket, - network_id.clone(), - self.public_key, - self.private_key.clone(), - self.public_key, - ) - .await - .map_err(GolgiError::Handshake)?; - - let (box_stream_read, box_stream_write) = - BoxStream::from_handshake(socket.clone(), socket, handshake, 0x8000).split_read_write(); - - let rpc_reader = RpcReader::new(box_stream_read); - let client = ApiCaller::new(RpcWriter::new(box_stream_write)); - let sbot_connection = SbotConnection { - rpc_reader, - client, - }; + 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. @@ -166,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. @@ -223,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(), }; @@ -234,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 @@ -269,10 +277,22 @@ 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 } @@ -283,64 +303,18 @@ impl Sbot { &mut self, id: String, ) -> Result>, GolgiError> { - let mut sbot_connection = self.get_sbot_connection(None, None).await.unwrap(); + let mut sbot_connection = self.get_sbot_connection().await.unwrap(); let args = CreateHistoryStreamIn::new(id); - let req_id = sbot_connection.client.create_history_stream_req_send(&args).await?; - let history_stream = Sbot::get_source_stream(sbot_connection.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) } - - /// 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 functions 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 - } } diff --git a/src/utils.rs b/src/utils.rs index 138e605..d4c871c 100644 --- a/src/utils.rs +++ b/src/utils.rs @@ -1,5 +1,8 @@ //! Utility methods for `golgi`. use async_std::io::Read; +use async_std::net::TcpStream; +use async_std::stream::Stream; +use async_stream::stream; use std::fmt::Debug; use kuska_ssb::rpc::{RecvMsg, RequestNo, RpcReader}; @@ -64,10 +67,10 @@ pub async fn get_async<'a, R, T, F>( req_no: RequestNo, f: F, ) -> Result - where - R: Read + Unpin, - F: Fn(&[u8]) -> Result, - T: Debug, +where + R: Read + Unpin, + F: Fn(&[u8]) -> Result, + T: Debug, { loop { let (id, msg) = rpc_reader.recv().await?; @@ -108,10 +111,10 @@ pub async fn get_source_until_eof<'a, R, T, F>( req_no: RequestNo, f: F, ) -> Result, GolgiError> - where - R: Read + Unpin, - F: Fn(&[u8]) -> Result, - T: Debug, +where + R: Read + Unpin, + F: Fn(&[u8]) -> Result, + T: Debug, { let mut messages: Vec = Vec::new(); loop { @@ -140,8 +143,6 @@ pub async fn get_source_until_eof<'a, R, T, F>( 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. @@ -160,10 +161,10 @@ pub async fn print_source_until_eof<'a, R, T, F>( req_no: RequestNo, f: F, ) -> Result<(), GolgiError> - where - R: Read + Unpin, - F: Fn(&[u8]) -> Result, - T: Debug + serde::Deserialize<'a>, +where + R: Read + Unpin, + F: Fn(&[u8]) -> Result, + T: Debug + serde::Deserialize<'a>, { loop { let (id, msg) = rpc_reader.recv().await?; @@ -182,4 +183,61 @@ pub async fn print_source_until_eof<'a, R, T, F>( } } Ok(()) -} \ No newline at end of file +} + +/// 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 +} -- 2.54.0 From 84add5baacd29d4b7a1c8ae4a3a9a055c3b6cab0 Mon Sep 17 00:00:00 2001 From: notplants Date: Wed, 5 Jan 2022 14:00:23 -0500 Subject: [PATCH 14/16] Remove rand dependency --- Cargo.lock | 47 ----------------------------------------------- Cargo.toml | 3 +-- 2 files changed, 1 insertion(+), 49 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index c873136..edae602 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -506,7 +506,6 @@ dependencies = [ "kuska-handshake", "kuska-sodiumoxide", "kuska-ssb", - "rand", "serde", "serde_json", ] @@ -713,12 +712,6 @@ dependencies = [ "winapi 0.3.9", ] -[[package]] -name = "ppv-lite86" -version = "0.2.16" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "eb9f9e6e233e5c4a35559a617bf40a4ec447db2e84c20b55a6f83167b7e57872" - [[package]] name = "proc-macro2" version = "1.0.32" @@ -737,46 +730,6 @@ dependencies = [ "proc-macro2", ] -[[package]] -name = "rand" -version = "0.8.4" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "2e7573632e6454cf6b99d7aac4ccca54be06da05aca2ef7423d22d27d4d4bcd8" -dependencies = [ - "libc", - "rand_chacha", - "rand_core", - "rand_hc", -] - -[[package]] -name = "rand_chacha" -version = "0.3.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e6c10a63a0fa32252be49d21e7709d4d4baf8d231c2dbce1eaa8141b9b127d88" -dependencies = [ - "ppv-lite86", - "rand_core", -] - -[[package]] -name = "rand_core" -version = "0.6.3" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d34f1408f55294453790c48b2f1ebbb1c5b4b7563eb1f418bcfcfdbb06ebb4e7" -dependencies = [ - "getrandom", -] - -[[package]] -name = "rand_hc" -version = "0.3.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d51e9f596de227fda2ea6c84607f5558e196eeaf43c986b724ba4fb8fdf497e7" -dependencies = [ - "rand_core", -] - [[package]] name = "redox_syscall" version = "0.2.10" diff --git a/Cargo.toml b/Cargo.toml index d5eca42..7fc8728 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -15,5 +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" -rand = "0.8.4" \ No newline at end of file +async-stream = "0.3.2" \ No newline at end of file -- 2.54.0 From 001649552567b127684fbe6bed6e1c965d396170 Mon Sep 17 00:00:00 2001 From: notplants Date: Wed, 5 Jan 2022 14:02:20 -0500 Subject: [PATCH 15/16] Change name of _message_type to message_type --- src/messages.rs | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/src/messages.rs b/src/messages.rs index 43a8f28..0b3c538 100644 --- a/src/messages.rs +++ b/src/messages.rs @@ -63,11 +63,11 @@ impl SsbMessageValue { /// 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 - pub fn is_message_type(&self, _message_type: SsbMessageContentType) -> bool { + pub fn is_message_type(&self, message_type: SsbMessageContentType) -> bool { let self_message_type = self.get_message_type(); match self_message_type { Ok(mtype) => { - matches!(mtype, _message_type) + matches!(mtype, message_type) } Err(_err) => false, } -- 2.54.0 From 386642d6f1944a364362dd342f3c73aedfad72a4 Mon Sep 17 00:00:00 2001 From: notplants Date: Wed, 5 Jan 2022 14:05:59 -0500 Subject: [PATCH 16/16] Combine imports --- src/utils.rs | 6 ++---- 1 file changed, 2 insertions(+), 4 deletions(-) diff --git a/src/utils.rs b/src/utils.rs index d4c871c..97f8323 100644 --- a/src/utils.rs +++ b/src/utils.rs @@ -1,10 +1,8 @@ //! Utility methods for `golgi`. -use async_std::io::Read; -use async_std::net::TcpStream; -use async_std::stream::Stream; -use async_stream::stream; 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; -- 2.54.0