golgi-ssb
/
golgi
4
5
Fork 0
Code Issues 11 Pull Requests 1 Projects 1 Releases Wiki Activity

Chang get_subset_stream to return SsbMessageValue instead of SsbMessageKVT

This commit is contained in:
notplants 2022-01-12 15:22:26 -05:00
parent 0addd8dc47
commit 1879341499
1 changed files with 8 additions and 11 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

19
src/sbot.rs
View File

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