ADD: RPC received over time

launcher/src/backend/Monitoring.js

@@ -2802,6 +2802,7 @@ export class Monitoring {
       // if(p2pstatus.code)
       //   return p2pstatus;
       const subnetSubs = await this.getSubnetSubs();
+      const rpcReceivedData = await this.nodeConnection.sshService.getRPCReceivedData();
       return {
         code: 0,
         info: "success: data successfully retrieved",
@@ -2814,6 +2815,7 @@ export class Monitoring {
           beaconstatus: beaconstatus,
           portstatus: portstatus,
           subnetSubs: subnetSubs,
+          rpcReceivedData: rpcReceivedData,
         },
       };
     } catch (err) {

launcher/src/backend/SSHService.js

@@ -15,6 +15,7 @@ export class SSHService {
     this.connectionInfo = null;
     this.connected = false;
     this.tunnels = [];
+    this.rpcReceivedDatas = [];
     this.addingConnection = false;
     this.removeConnectionCount = 0;
     this.checkPoolPolling = setInterval(async () => {
@@ -287,13 +288,64 @@ export class SSHService {
           log.error("Tunnel SSH Connection error: ", error);
         });
 
+        server.on("connection", (connection) => {
+          // Forward the connection to the destination address and port
+          conn.forwardOut(forwardOptions.srcAddr, forwardOptions.srcPort, forwardOptions.dstAddr, forwardOptions.dstPort, (err, stream) => {
+            if (err) {
+              log.error("Forwarding error: ", err);
+              return;
+            }
+
+            // Pipe the connection to the stream and vice versa
+            connection.pipe(stream).pipe(connection);
+
+            // Listen for data on the stream
+            stream.on("data", (data) => {
+              // Call the rpcReceivedData method to handle the received data
+              this.handleReceivedData(data.length, forwardOptions.srcPort);
+            });
+          });
+        });
+
         server.on("error", function (error) {
           log.error("Tunnel connection error: ", error);
         });
       });
     });
   }
 
+  /**
+   * Handles the received data by storing it in the rpcReceivedDatas array.
+   * @param {number} dataLength - The length of the received data (byte).
+   * @param {number} dstPort - The destination port.
+   */
+  async handleReceivedData(dataLength, srcPort) {
+    try {
+      const receivedData = {
+        receivedDataLength: dataLength,
+        dstPort: srcPort,
+      };
+      this.rpcReceivedDatas.push(receivedData);
+    } catch (error) {
+      console.error("Error handling received data:", error);
+    }
+  }
+
+  /**
+   * Retrieves and clears the stored received data.
+   * @returns {Array} - The array of received data objects.
+   */
+  async getRPCReceivedData() {
+    try {
+      const dataToReturn = [...this.rpcReceivedDatas];
+      this.rpcReceivedDatas = [];
+      return dataToReturn;
+    } catch (error) {
+      console.error("Error retrieving and clearing received data:", error);
+      return [];
+    }
+  }
+
   async closeTunnels(onlySpecificPorts = []) {
     return new Promise((resolve, reject) => {
       let i = this.tunnels.length;