From a076e4741f79641edd73f2fc4359fa1aed704a6b Mon Sep 17 00:00:00 2001
From: notplants <max@mfowler.info>
Date: Wed, 5 Jan 2022 08:50:57 -0500
Subject: [PATCH] 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<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
     /// 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<SsbMessageValue, GolgiError>
-    pub async fn create_history_stream<'a>(
-        &'a mut self,
+    pub async fn create_history_stream(
+        &mut self,
         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 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<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>> + '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
-        F: Fn(&[u8]) -> Result<T, GolgiError> + 'a,
-        T: Debug + serde::Deserialize<'a> + 'a,
+        F: Fn(&[u8]) -> Result<T, GolgiError>,
+        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)
                     }