update transaction stream method

crates/networking/src/rpc_handler/handler.rs

@@ -8,7 +8,7 @@ use ureq::{Agent, AgentBuilder, Error, Response};
 
 use crate::{
     rpc_abi::{
-        RpcBroadcastTxRequest, RpcBroadcastTxResponse, RpcCreateTxRequest, RpcCreateTxResponse, RpcExportAccountResponse, RpcGetAccountStatusRequest, RpcGetAccountStatusResponse, RpcGetAccountTransactionRequest, RpcGetAccountTransactionResponse, RpcGetBalancesRequest, RpcGetBalancesResponse, RpcGetBlockRequest, RpcGetBlockResponse, RpcGetBlocksRequest, RpcGetBlocksResponse, RpcGetLatestBlockResponse, RpcGetTransactionsRequest, RpcGetTransactionsResponse, RpcImportAccountRequest, RpcImportAccountResponse, RpcRemoveAccountRequest, RpcRemoveAccountResponse, RpcResetAccountRequest, RpcResponse, RpcSetAccountHeadRequest, RpcSetScanningRequest, SendTransactionRequest, SendTransactionResponse
+        RpcBroadcastTxRequest, RpcBroadcastTxResponse, RpcCreateTxRequest, RpcCreateTxResponse, RpcExportAccountResponse, RpcGetAccountStatusRequest, RpcGetAccountStatusResponse, RpcGetAccountTransactionRequest, RpcGetAccountTransactionResponse, RpcGetBalancesRequest, RpcGetBalancesResponse, RpcGetBlockRequest, RpcGetBlockResponse, RpcGetBlocksRequest, RpcGetBlocksResponse, RpcGetLatestBlockResponse, RpcGetTransactionsRequest, RpcGetTransactionsResponse, RpcImportAccountRequest, RpcImportAccountResponse, RpcRemoveAccountRequest, RpcRemoveAccountResponse, RpcResetAccountRequest, RpcResponse, RpcSetAccountHeadRequest, RpcSetScanningRequest, SendTransactionRequest, SendTransactionResponse, TransactionStatus
     },
     rpc_handler::RpcError, stream::RequestExt,
 };
@@ -136,16 +136,20 @@ impl RpcHandler {
     ) -> Result<RpcResponse<RpcGetTransactionsResponse>, OreoError> {
         let path = format!("http://{}/wallet/getAccountTransactions", self.endpoint);
         let resp = self.agent.clone().post(&path).send_json(&request);
+
         match resp {
             Ok(response) => {
-                let transactions = response.collect_stream();
+                let transactions: Result<Vec<_>, OreoError> = response
+                    .into_stream::<TransactionStatus>()
+                    .collect();
+
                 Ok(RpcResponse {
                     status: 200,
                     data: RpcGetTransactionsResponse {
                         transactions: transactions?,
                     },
                 })
-            },
+            }
             Err(e) => Err(OreoError::InternalRpcError(e.to_string())),
         }
     }

crates/networking/src/stream.rs

@@ -1,33 +1,63 @@
-use std::io::Read;
+use std::io::{BufRead, BufReader, Read};
+use std::marker::PhantomData;
 
 use oreo_errors::OreoError;
 use serde::de::DeserializeOwned;
 use ureq::Response;
 
 use crate::rpc_abi::RpcResponseStream;
 
+pub struct StreamReader<T> {
+    reader: BufReader<Box<dyn Read>>,
+    _marker: PhantomData<T>,
+}
+
+impl<T> StreamReader<T> {
+    pub fn new(reader: Box<dyn Read>) -> Self {
+        Self {
+            reader: BufReader::new(reader),
+            _marker: PhantomData,
+        }
+    }
+}
+
+impl<T> Iterator for StreamReader<T>
+where
+    T: DeserializeOwned,
+{
+    type Item = Result<T, OreoError>;
+
+    fn next(&mut self) -> Option<Self::Item> {
+        let mut chunk = Vec::new();
+        match self.reader.read_until(b'\x0c', &mut chunk) {
+            Ok(0) => None, // EOF reached
+            Ok(_) => {
+                if chunk.ends_with(&[b'\x0c']) {
+                    chunk.pop();
+                }
+
+                if chunk.is_empty() {
+                    self.next()
+                } else {
+                    // Deserialize the chunk
+                    let item_result = serde_json::from_slice::<RpcResponseStream<T>>(&chunk)
+                        .map(|item| item.data)
+                        .map_err(|e| OreoError::InternalRpcError(e.to_string()));
+                    Some(item_result)
+                }
+            }
+            Err(e) => Some(Err(OreoError::InternalRpcError(e.to_string()))),
+        }
+    }
+}
+
 pub trait RequestExt {
-  fn collect_stream<T: DeserializeOwned>(self) -> Result<Vec<T>, OreoError>;
+    fn into_stream<T: DeserializeOwned>(self) -> StreamReader<T>;
 }
 
 impl RequestExt for Response {
-  fn collect_stream<T: DeserializeOwned>(self) -> Result<Vec<T>, OreoError> {
-      let reader = self.into_reader();
-      let mut buffered = std::io::BufReader::new(reader);
-      let mut items = Vec::new();
-      let mut response_str = String::new();
-      buffered.read_to_string(&mut response_str).map_err(|e| OreoError::InternalRpcError(e.to_string()))?;
-      let lines = response_str.split('\x0c').collect::<Vec<&str>>();
-
-      // Get rid of status code
-      for line in lines[0..lines.len()-1].into_iter() {
-        let line = *line; // Dereference to get &str
-          if !line.trim().is_empty() {
-              let item: RpcResponseStream<T> = serde_json::from_str(line)
-                  .map_err(|e| OreoError::InternalRpcError(e.to_string()))?;
-              items.push(item.data);
-          }
-      }
-      Ok(items)
-  }
-}
+    fn into_stream<T: DeserializeOwned>(self) -> StreamReader<T> {
+        let reader = self.into_reader();
+        StreamReader::new(Box::new(reader))
+    }
+}