This PR shows a working example of changing create_history_stream to produce a Stream<Item = Result<SsbMessageValue, GolgiError>> instead of a Vec<SsbMessageValue>
Examples of how this can be used are included in examples/ssb-stream-example.rs.
Currently, two of the examples in that file are commented out, because I am running into an issue with the borrow checker, where after calling sbot_client.create_history_stream once, something is borrowed, and I cannot call it a second time. Aka the commented out examples also work, if uncommented, but you can only uncomment one at a time, currently.
I am hopeful that this is fixable without a major change to the way this PR handles streams. Similarly, I currently included get_source_stream as a method in sbot.rs instead of as a function in utils.rs next to the other similar functions, because it was easier for me to reason about borrowing working in sbot.rs, but I imagine we can also change this - just need to figure out correct way to pass rpc_reader to the function. I wanted to focus on getting the stream code working first.
One of the important commented out examples is like this:
It shows how in one line we can convert this stream into a Vec<SsbMessageValue> using try_collect. This shows that if we change our base functions to return streams, its purely 'additive' on the previous functionality. We can easily convert into Vec and work like that if we want to, and we can keep things as Stream and work like that where it would be advantageous.
figure out how to make create_history_stream callable multiple times
figure out if lifetimes are really needed for create_history_stream and get_source_stream or if there is a better way to do this (might also be related to the above two points). I haven't used lifetimes much before, and am not sure if the way I am using them is necessary or correct, but I couldn't get the code working yet without them
If you want to try experiment with any of the above three points, please feel free and let me know. Or if you have other comments etc., but hopefully having this working example is a good starting point for any further improvements
This PR shows a working example of changing create_history_stream to produce a `Stream<Item = Result<SsbMessageValue, GolgiError>>` instead of a `Vec<SsbMessageValue>`
Examples of how this can be used are included in `examples/ssb-stream-example.rs`.
Currently, two of the examples in that file are commented out, because I am running into an issue with the borrow checker, where after calling sbot_client.create_history_stream once, something is borrowed, and I cannot call it a second time. Aka the commented out examples also work, if uncommented, but you can only uncomment one at a time, currently.
I am hopeful that this is fixable without a major change to the way this PR handles streams. Similarly, I currently included get_source_stream as a method in sbot.rs instead of as a function in utils.rs next to the other similar functions, because it was easier for me to reason about borrowing working in sbot.rs, but I imagine we can also change this - just need to figure out correct way to pass rpc_reader to the function. I wanted to focus on getting the stream code working first.
One of the important commented out examples is like this:
``` rust
let mut 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);
}
```
It shows how in one line we can convert this stream into a `Vec<SsbMessageValue>` using try_collect. This shows that if we change our base functions to return streams, its purely 'additive' on the previous functionality. We can easily convert into Vec and work like that if we want to, and we can keep things as Stream and work like that where it would be advantageous.
I ended up using the stream! macro from the async-stream crate, as suggested on the rust forum: https://users.rust-lang.org/t/how-to-create-async-std-stream-which-calls-async-function-in-poll-next/69760. As described in my question, I was having issues figuring out how to produce a stream while calling an async function. Maybe there are other ways, but I didn't figure it out. This was a pretty good related article https://gendignoux.com/blog/2021/04/01/rust-async-streams-futures-part1.html, but still couldn't get something matching exactly what we need from this. But its working with the macro so maybe thats fine.
So outstanding TODOS:
- move get_source_stream to utils.rs
- figure out how to make create_history_stream callable multiple times
- figure out if lifetimes are really needed for create_history_stream and get_source_stream or if there is a better way to do this (might also be related to the above two points). I haven't used lifetimes much before, and am not sure if the way I am using them is necessary or correct, but I couldn't get the code working yet without them
If you want to try experiment with any of the above three points, please feel free and let me know. Or if you have other comments etc., but hopefully having this working example is a good starting point for any further improvements
This looks great overall! It's a major plus to have the ability of returning a Vec or a Stream; super nice to have that level of flexibility at this early stage of the library development.
I've left a number of comments, including a way of moving get_source_stream into utils.
As mentioned on cabal, I believe the best way to allow multiple stream-returning sbot function calls is to simply instantiate additional connection to the sbot (reminder that sbot::init() still needs to change to sbot::connect()). The ssb-stream-example.rs can be updated to use this pattern. The examples are really nice to get others up-and-running quickly.
Another thing I saw (when running cargo nightly clippy) but didn't mention in the comments:
warning: avoid using `collect()` when not needed
--> src/sbot.rs:187:84
|187|let messages: Vec<SsbMessageValue> = kvts.into_iter().map(|kvt| kvt.value).collect();| ^^^^^^^
...
192|let mut about_messages: Vec<SsbMessageValue> = messages
| ________________________________________________________-
193|| .into_iter()||________________________- the iterator could be used here instead
|= note: `#[warn(clippy::needless_collect)]` on by default= help: for further information visit https://rust-lang.github.io/rust-clippy/master/index.html#needless_collect
help: use the original Iterator instead of collecting it and then producing a new one
|187 ~
188| // TODO: after fixing sbot regression,
189| // change this subset query to filter by type about in addition to author
190| // and remove this filter section
191| // filter down to about messages
192 ~ let mut about_messages: Vec<SsbMessageValue> = kvts.into_iter().map(|kvt| kvt.value)
This looks great overall! It's a major plus to have the ability of returning a `Vec` or a `Stream`; super nice to have that level of flexibility at this early stage of the library development.
I've left a number of comments, including a way of moving `get_source_stream` into utils.
As mentioned on `cabal`, I believe the best way to allow multiple stream-returning sbot function calls is to simply instantiate additional connection to the sbot (reminder that `sbot::init()` still needs to change to `sbot::connect()`). The `ssb-stream-example.rs` can be updated to use this pattern. The examples are really nice to get others up-and-running quickly.
Another thing I saw (when running `cargo nightly clippy`) but didn't mention in the comments:
```shell
warning: avoid using `collect()` when not needed
--> src/sbot.rs:187:84
|
187 | let messages: Vec<SsbMessageValue> = kvts.into_iter().map(|kvt| kvt.value).collect();
| ^^^^^^^
...
192 | let mut about_messages: Vec<SsbMessageValue> = messages
| ________________________________________________________-
193 | | .into_iter()
| |________________________- the iterator could be used here instead
|
= note: `#[warn(clippy::needless_collect)]` on by default
= help: for further information visit https://rust-lang.github.io/rust-clippy/master/index.html#needless_collect
help: use the original Iterator instead of collecting it and then producing a new one
|
187 ~
188 | // TODO: after fixing sbot regression,
189 | // change this subset query to filter by type about in addition to author
190 | // and remove this filter section
191 | // filter down to about messages
192 ~ let mut about_messages: Vec<SsbMessageValue> = kvts.into_iter().map(|kvt| kvt.value)
```
I'd prefer if everything was documented, even if it seems self evident from the type definition.
This one might be something like:
/// Scuttlebutt message content `type` variants.
I'd prefer if everything was documented, even if it seems self evident from the type definition.
This one might be something like:
```rust
/// Scuttlebutt message content `type` variants.
```
I added this about structs because according to the current lint, you don't just need the message you provided, but I believe you need a comment for each type variant, or each struct field, in addition to the overall struct.
I added this about structs because according to the current lint, you don't just need the message you provided, but I believe you need a comment for each type variant, or each struct field, in addition to the overall struct.
I suggest moving these up to where the other kuska_ dependencies are and grouping RequestNo into the existing kuska_ssb grouping. It seems like a minor difference but I've found this kind of organisation to help me a lot when navigating code I haven't written.
std dependencies
other dependencies (usually grouped in alphabetical order)
local dependencies (ie. crate::)
I suggest moving these up to where the other `kuska_` dependencies are and grouping `RequestNo` into the existing `kuska_ssb` grouping. It seems like a minor difference but I've found this kind of organisation to help me a lot when navigating code I haven't written.
1. `std` dependencies
2. other dependencies (usually grouped in alphabetical order)
3. local dependencies (ie. `crate::`)
I was able to move this into `src/utils.rs` by using the following function signature:
```rust
pub fn get_source_stream<'a, F, T>(
sbot_client: &'a mut Sbot,
req_no: RequestNo,
f: F,
) -> impl Stream<Item = Result<T, GolgiError>> + 'a
where
F: Fn(&[u8]) -> Result<T, GolgiError> + 'a,
T: Debug + serde::Deserialize<'a> + 'a,
{
```
Then L263 becomes:
```rust
let (id, msg) = sbot_client.rpc_reader.recv().await?;
```
And L273 requires a reference to `body`:
```rust
let item = f(&body)?;
```
Then in `src/sbot.rs`:
The `rpc_reader` field on `Sbot` needs to be public:
```rust
pub rpc_reader: RpcReader<TcpStream>,
```
And we replace `self.` with `utils::` when calling `get_source_stream`:
```rust
let history_stream = utils::get_source_stream(self, req_id, utils::ssb_message_res_parse);
```
@glyph I incoporated your comments, and this is now a fully functional PR and ready for official review to merge.
can call create_history_stream multiple times (ssb-stream-example is fully working)
moved get_source_stream to utils.rs and trimmed down the lifetimes
added get_sbot_connection function and made create_history_stream make use of it. I also still kept one sbot_connection as a field for Sbot which can be used by non-stream methods if desired (I don't know if this is a major optimization, but that's what I was thinking with that).
unfortunately cargo fmt seems to have added a bunch of whitespace to the PR which makes it a bit hard to read. will try to avoid that in the future!
the one thing I didn't do,
warning: avoid using `collect()` when not needed
... I would like to address this in a separate PR, where I update the about functions to use the stream code as well.
@glyph I incoporated your comments, and this is now a fully functional PR and ready for official review to merge.
- can call create_history_stream multiple times (ssb-stream-example is fully working)
- moved get_source_stream to utils.rs and trimmed down the lifetimes
- added get_sbot_connection function and made create_history_stream make use of it. I also still kept one sbot_connection as a field for Sbot which can be used by non-stream methods if desired (I don't know if this is a major optimization, but that's what I was thinking with that).
unfortunately cargo fmt seems to have added a bunch of whitespace to the PR which makes it a bit hard to read. will try to avoid that in the future!
the one thing I didn't do,
```
warning: avoid using `collect()` when not needed
```
... I would like to address this in a separate PR, where I update the about functions to use the stream code as well.
Awesome, this look amazing. Thanks for the clear summary of the changes. Great work :)
One tiny thing I'm curious about is the underscore in front of this function name:
asyncfn_get_sbot_connection_helper(
But I don't want that to delay merging of this PR. Perhaps the underscore can be removed in one of your next PR's.
Merge at your leisure 🟢
Awesome, this look amazing. Thanks for the clear summary of the changes. Great work :)
One tiny thing I'm curious about is the underscore in front of this function name:
```rust
async fn _get_sbot_connection_helper(
```
But I don't want that to delay merging of this PR. Perhaps the underscore can be removed in one of your next PR's.
Merge at your leisure 🟢
@glyphasync fn _get_sbot_connection_helper is the one method I didn't declare as public, because I thought its just a helper and not really part of the exposed API.
Some languages have a convention to indicate private methods and variables by prefixing with an underscore, but its just stylistic, I'm happy to remove it in the next PR if you prefer that.
@glyph `async fn _get_sbot_connection_helper` is the one method I didn't declare as public, because I thought its just a helper and not really part of the exposed API.
Some languages have a convention to indicate private methods and variables by prefixing with an underscore, but its just stylistic, I'm happy to remove it in the next PR if you prefer that.
Some languages have a convention to indicate private methods and variables by prefixing with an underscore, but its just stylistic, I'm happy to remove it in the next PR if you prefer that.
Oh OK, that makes sense to me. I don't mind it staying in.
> Some languages have a convention to indicate private methods and variables by prefixing with an underscore, but its just stylistic, I'm happy to remove it in the next PR if you prefer that.
@notplants
Oh OK, that makes sense to me. I don't mind it staying in.
Blocking a user prevents them from interacting with repositories, such as opening or commenting on pull requests or issues. Learn more about blocking a user.
notplants
glyph
This PR shows a working example of changing create_history_stream to produce a
Stream<Item = Result<SsbMessageValue, GolgiError>>instead of aVec<SsbMessageValue>Examples of how this can be used are included in
examples/ssb-stream-example.rs.Currently, two of the examples in that file are commented out, because I am running into an issue with the borrow checker, where after calling sbot_client.create_history_stream once, something is borrowed, and I cannot call it a second time. Aka the commented out examples also work, if uncommented, but you can only uncomment one at a time, currently.
I am hopeful that this is fixable without a major change to the way this PR handles streams. Similarly, I currently included get_source_stream as a method in sbot.rs instead of as a function in utils.rs next to the other similar functions, because it was easier for me to reason about borrowing working in sbot.rs, but I imagine we can also change this - just need to figure out correct way to pass rpc_reader to the function. I wanted to focus on getting the stream code working first.
One of the important commented out examples is like this:
It shows how in one line we can convert this stream into a
Vec<SsbMessageValue>using try_collect. This shows that if we change our base functions to return streams, its purely 'additive' on the previous functionality. We can easily convert into Vec and work like that if we want to, and we can keep things as Stream and work like that where it would be advantageous.I ended up using the stream! macro from the async-stream crate, as suggested on the rust forum: https://users.rust-lang.org/t/how-to-create-async-std-stream-which-calls-async-function-in-poll-next/69760. As described in my question, I was having issues figuring out how to produce a stream while calling an async function. Maybe there are other ways, but I didn't figure it out. This was a pretty good related article https://gendignoux.com/blog/2021/04/01/rust-async-streams-futures-part1.html, but still couldn't get something matching exactly what we need from this. But its working with the macro so maybe thats fine.
So outstanding TODOS:
If you want to try experiment with any of the above three points, please feel free and let me know. Or if you have other comments etc., but hopefully having this working example is a good starting point for any further improvements
@@ -63,3 +63,3 @@/// Helper function which returns true if this message is of the given type,/// and false if the type does not match or is not foundpub fn is_message_type(&self, message_type: SsbMessageContentType) -> bool {pub fn is_message_type(&self, _message_type: SsbMessageContentType) -> bool {I think this was introduced by cargo fix, but I'm not sure why
Oh, interesting. I never use
cargo fix.This looks great overall! It's a major plus to have the ability of returning a
Vecor aStream; super nice to have that level of flexibility at this early stage of the library development.I've left a number of comments, including a way of moving
get_source_streaminto utils.As mentioned on
cabal, I believe the best way to allow multiple stream-returning sbot function calls is to simply instantiate additional connection to the sbot (reminder thatsbot::init()still needs to change tosbot::connect()). Thessb-stream-example.rscan be updated to use this pattern. The examples are really nice to get others up-and-running quickly.Another thing I saw (when running
cargo nightly clippy) but didn't mention in the comments:@@ -16,2 +16,4 @@serde = { version = "1", features = ["derive"] }serde_json = "1"async-stream = "0.3.2"rand = "0.8.4"I think this
randdependency must have slipped in during your async stream experimentation.@@ -31,6 +30,7 @@ pub struct SsbMessageValue {// Enum representing the different possible message content types#[derive(Debug)]#[allow(missing_docs)]I'd prefer if everything was documented, even if it seems self evident from the type definition.
This one might be something like:
I added this about structs because according to the current lint, you don't just need the message you provided, but I believe you need a comment for each type variant, or each struct field, in addition to the overall struct.
@@ -1,5 +1,8 @@//! Sbot type and associated methods.use std::fmt::Debug;use async_std::net::TcpStream;Could rather be combined:
@@ -19,6 +22,7 @@ use crate::utils;// re-export types from kuskapub use kuska_ssb::api::dto::content::SubsetQuery;use kuska_ssb::rpc::RequestNo;I suggest moving these up to where the other
kuska_dependencies are and groupingRequestNointo the existingkuska_ssbgrouping. It seems like a minor difference but I've found this kind of organisation to help me a lot when navigating code I haven't written.stddependenciescrate::)@@ -229,1 +230,3 @@&mut self,/// Call the `createHistoryStream` RPC method/// and return a Stream of Result<SsbMessageValue, GolgiError>pub async fn create_history_stream<'a>(I believe it's only necessary to add an explicit lifetime bound for return type (this compiles for me):
@@ -235,0 +241,4 @@/// 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 calledSmall typo:
handling functions->handling function@@ -235,0 +248,4 @@////// * `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,Love this explanation!
@@ -235,0 +250,4 @@/// * `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 fn get_source_stream<'a, F, T>(&'a mut self, req_no: RequestNo, f: F) -> impl Stream<Item = Result<T, GolgiError>> + 'aI was able to move this into
src/utils.rsby using the following function signature:Then L263 becomes:
And L273 requires a reference to
body:Then in
src/sbot.rs:The
rpc_readerfield onSbotneeds to be public:And we replace
self.withutils::when callingget_source_stream:@@ -235,0 +258,4 @@// 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/69760let source_stream = stream! {loop {There's a formatting issue here.
cargo fmtshould do the trick (I have this setup to run every time I save).@glyph I incoporated your comments, and this is now a fully functional PR and ready for official review to merge.
unfortunately cargo fmt seems to have added a bunch of whitespace to the PR which makes it a bit hard to read. will try to avoid that in the future!
the one thing I didn't do,
... I would like to address this in a separate PR, where I update the about functions to use the stream code as well.
Awesome, this look amazing. Thanks for the clear summary of the changes. Great work :)
One tiny thing I'm curious about is the underscore in front of this function name:
But I don't want that to delay merging of this PR. Perhaps the underscore can be removed in one of your next PR's.
Merge at your leisure 🟢
@glyph
async fn _get_sbot_connection_helperis the one method I didn't declare as public, because I thought its just a helper and not really part of the exposed API.Some languages have a convention to indicate private methods and variables by prefixing with an underscore, but its just stylistic, I'm happy to remove it in the next PR if you prefer that.
@notplants
Oh OK, that makes sense to me. I don't mind it staying in.