adds streaming support for requests

Author: jowparks

README.md

@@ -21,7 +21,13 @@ A tool to fetch blocks from rpc and save in local db for better performance duri
 
 ![basic arch](assets/arch_v2.png)
 
-## Docker Local Development
+## Docker
+
+Build
+
+```bash
+docker build -t oreowallet .
+```
 
 Run node:
 

crates/networking/src/lib.rs

@@ -5,6 +5,7 @@ pub mod rpc_handler;
 pub mod server_handler;
 pub mod socket_message;
 pub mod web_abi;
+pub mod stream;
 
 use db_handler::{DBTransaction, InnerBlock, Json};
 use rpc_abi::RpcBlock;

crates/networking/src/rpc_abi.rs

@@ -1,7 +1,7 @@
 use axum::{response::IntoResponse, Json};
 use constants::IRON_NATIVE_ASSET;
-use serde::{Deserialize, Serialize};
-use ureq::json;
+use serde::{ Deserialize, Serialize};
+use ureq::{json};
 
 use crate::orescriptions::{get_ores, is_ores_local, Ores};
 
@@ -11,6 +11,14 @@ pub struct RpcResponse<T> {
     pub data: T,
 }
 
+#[derive(Debug, Deserialize, Serialize)]
+pub struct RpcResponseStream<T> {
+    pub data: T
+}
+
+
+
+
 impl<T: Serialize> IntoResponse for RpcResponse<T> {
     fn into_response(self) -> axum::response::Response {
         Json(json!({"code": 200, "data": self.data})).into_response()

crates/networking/src/rpc_handler/handler.rs

@@ -8,17 +8,9 @@ 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
     },
-    rpc_handler::RpcError,
+    rpc_handler::RpcError, stream::RequestExt,
 };
 
 #[derive(Debug, Clone)]
@@ -137,32 +129,24 @@ impl RpcHandler {
         let resp = self.agent.clone().post(&path).send_json(&request);
         handle_response(resp)
     }
-
+    
     pub fn get_transactions(
         &self,
         request: RpcGetTransactionsRequest,
     ) -> 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 mut buffer = Vec::new();
-                response.into_reader().read_to_end(&mut buffer).map_err(|e| OreoError::InternalRpcError(e.to_string()))?;
-                let transactions_response: RpcResponse<RpcGetTransactionsResponse> = if buffer.is_empty() {
-                    RpcResponse {
-                        data: RpcGetTransactionsResponse {
-                            transactions: Vec::new(),
-                        },
-                        status: 200,
-                    }
-                } else {
-                    serde_json::from_slice(&buffer)
-                        .map_err(|e| OreoError::InternalRpcError(e.to_string()))?
-                };
-                Ok(transactions_response)
-            }
-            Err(e) => handle_response(Err(e)),
+                let transactions = response.collect_stream();
+                Ok(RpcResponse {
+                    status: 200,
+                    data: RpcGetTransactionsResponse {
+                        transactions: transactions?,
+                    },
+                })
+            },
+            Err(e) => Err(OreoError::InternalRpcError(e.to_string())),
         }
     }
 

crates/networking/src/stream.rs

@@ -0,0 +1,33 @@
+use std::io::Read;
+
+use oreo_errors::OreoError;
+use serde::de::DeserializeOwned;
+use ureq::Response;
+
+use crate::rpc_abi::RpcResponseStream;
+
+pub trait RequestExt {
+  fn collect_stream<T: DeserializeOwned>(self) -> Result<Vec<T>, OreoError>;
+}
+
+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)
+  }
+}