Multiple RPC reader
This commit is contained in:
parent
db146446d0
commit
a076e4741f
57
src/sbot.rs
57
src/sbot.rs
|
@ -92,6 +92,43 @@ impl Sbot {
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
|
pub async fn get_rpc_reader(&self, ip_port: Option<String>, net_id: Option<String>) -> Result<RpcReader<TcpStream>, 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
|
/// Call the `partialReplication getSubset` RPC method and return a vector
|
||||||
/// of messages as KVTs (key, value, timestamp).
|
/// of messages as KVTs (key, value, timestamp).
|
||||||
// TODO: add args for `descending` and `page` (max number of msgs in response)
|
// TODO: add args for `descending` and `page` (max number of msgs in response)
|
||||||
|
@ -229,13 +266,14 @@ impl Sbot {
|
||||||
|
|
||||||
/// Call the `createHistoryStream` RPC method
|
/// Call the `createHistoryStream` RPC method
|
||||||
/// and return a Stream of Result<SsbMessageValue, GolgiError>
|
/// and return a Stream of Result<SsbMessageValue, GolgiError>
|
||||||
pub async fn create_history_stream<'a>(
|
pub async fn create_history_stream(
|
||||||
&'a mut self,
|
&mut self,
|
||||||
id: String,
|
id: String,
|
||||||
) -> Result<impl Stream<Item = Result<SsbMessageValue, GolgiError>> + 'a, GolgiError> {
|
) -> Result<impl Stream<Item = Result<SsbMessageValue, GolgiError>>, GolgiError> {
|
||||||
let args = CreateHistoryStreamIn::new(id);
|
let args = CreateHistoryStreamIn::new(id);
|
||||||
let req_id = self.client.create_history_stream_req_send(&args).await?;
|
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)
|
Ok(history_stream)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ -250,17 +288,18 @@ impl Sbot {
|
||||||
/// * `f` - A function which takes in an array of u8 and returns a Result<T, GolgiError>.
|
/// * `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,
|
/// 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.)
|
/// 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>> + 'a
|
pub async fn get_source_stream<'a, F, T>(mut rpc_reader: RpcReader<TcpStream>, req_no: RequestNo, f: F) -> impl Stream<Item = Result<T, GolgiError>>
|
||||||
where
|
where
|
||||||
F: Fn(&[u8]) -> Result<T, GolgiError> + 'a,
|
F: Fn(&[u8]) -> Result<T, GolgiError>,
|
||||||
T: Debug + serde::Deserialize<'a> + 'a,
|
T: Debug + serde::Deserialize<'a>,
|
||||||
{
|
{
|
||||||
// we use the async_stream::stream macro to allow for creating a stream which calls async functions
|
// 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
|
// 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! {
|
let source_stream = stream! {
|
||||||
loop {
|
loop {
|
||||||
// get the next message from the rpc_reader
|
// 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();
|
let x : i32 = id.clone();
|
||||||
// check if the next message from rpc_reader matches the req_no we are looking for
|
// 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
|
// if it matches, then this rpc response is for the given request
|
||||||
|
@ -270,7 +309,7 @@ impl Sbot {
|
||||||
RecvMsg::RpcResponse(_type, body) => {
|
RecvMsg::RpcResponse(_type, body) => {
|
||||||
// parse an item of type T from the message body using the provided
|
// parse an item of type T from the message body using the provided
|
||||||
// function for parsing
|
// function for parsing
|
||||||
let item = f(body)?;
|
let item = f(&body)?;
|
||||||
// return Ok(item) as the next value in the stream
|
// return Ok(item) as the next value in the stream
|
||||||
yield Ok(item)
|
yield Ok(item)
|
||||||
}
|
}
|
||||||
|
|
Loading…
Reference in New Issue