Merge pull request #119 from eosnetworkfoundation/kayan_ws_dest

taokayan · web-flow · commit 61085a8b54e8 · 2023-11-24T15:10:07.000+08:00
route websocket proxy traffics to different destinations
diff --git a/peripherals/eos-evm-ws-proxy/block-monitor.js b/peripherals/eos-evm-ws-proxy/block-monitor.js
@@ -130,9 +130,10 @@ class BlockMonitor extends EventEmitter {
         }
       }
 
-      if( found_next_block == true ) {
+      if( found_next_block == true && this.reversible_blocks.length > 180 + 6) { // reduce frequency of calling get_evm_lib
         const evm_lib = await this.get_evm_lib();
-        while(this.reversible_blocks.length > 0 && this.reversible_blocks.peek().number < evm_lib) {
+        // keep at least 180 blocks in case evm-node is out of sync with leap
+        while(this.reversible_blocks.length > 180 && this.reversible_blocks.peek().number < evm_lib) {
           this.remove_front_block();
         }
       }
diff --git a/peripherals/eos-evm-ws-proxy/config.js b/peripherals/eos-evm-ws-proxy/config.js
@@ -1,8 +1,11 @@
 require('dotenv').config();
 const ws_listening_port = parseInt(process.env.WS_LISTENING_PORT, 10) || 3333;
 const ws_listening_host = process.env.WS_LISTENING_HOST || "localhost";
-const web3_rpc_endpoint = process.env.WEB3_RPC_ENDPOINT || "http://localhost:5000/";
+const web3_rpc_endpoint = process.env.WEB3_RPC_ENDPOINT || "http://127.0.0.1:8881/";
+const web3_rpc_test_endpoint = process.env.WEB3_RPC_TEST_ENDPOINT || "http://127.0.0.1:8882/";
 const nodeos_rpc_endpoint = process.env.NODEOS_RPC_ENDPOINT || "http://127.0.0.1:8888/";
+const miner_rpc_endpoint = process.env.MINER_RPC_ENDPOINT || "http://127.0.0.1:18888/";
+const whitelist_methods = process.env.WHITELIST_METHODS || "net_version,eth_blockNumber,eth_chainId,eth_protocolVersion,eth_getBlockByHash,eth_getBlockByNumber,eth_getBlockTransactionCountByHash,eth_getBlockTransactionCountByNumber,eth_getUncleByBlockHashAndIndex,eth_getUncleByBlockNumberAndIndex,eth_getUncleCountByBlockHash,eth_getUncleCountByBlockNumber,eth_getTransactionByHash,eth_getRawTransactionByHash,eth_getTransactionByBlockHashAndIndex,eth_getRawTransactionByBlockHashAndIndex,eth_getTransactionByBlockNumberAndIndex,eth_getRawTransactionByBlockNumberAndIndex,eth_getTransactionReceipt,eth_getBlockReceipts,eth_estimateGas,eth_getBalance,eth_getCode,eth_getTransactionCount,eth_getStorageAt,eth_call,eth_callBundle,eth_createAccessList";
 const poll_interval = parseInt(process.env.POLL_INTERVAL, 10) || 1000;
 const max_logs_subs_per_connection = parseInt(process.env.MAX_LOGS_SUBS_PER_CONNECTION, 10) || 1;
 const max_minedtx_subs_per_connection = parseInt(process.env.MAX_MINEDTX_SUBS_PER_CONNECTION, 10) || 1;
@@ -13,10 +16,13 @@ module.exports = {
   ws_listening_port,
   ws_listening_host,
   web3_rpc_endpoint,
+  web3_rpc_test_endpoint,
   nodeos_rpc_endpoint,
+  miner_rpc_endpoint,
   poll_interval,
   max_logs_subs_per_connection,
   max_minedtx_subs_per_connection,
   log_level,
+  whitelist_methods,
   genesis_json
 };
diff --git a/peripherals/eos-evm-ws-proxy/subscription-server.js b/peripherals/eos-evm-ws-proxy/subscription-server.js
@@ -5,17 +5,17 @@ const {Web3} = require('web3');
 const {bigint_replacer, load_json_file} = require('./utils');
 class SubscriptionServer extends EventEmitter {
 
-  constructor({web3_rpc_endpoint, nodeos_rpc_endpoint, ws_listening_host, ws_listening_port, poll_interval, max_logs_subs_per_connection, max_minedtx_subs_per_connection, genesis_json, logger}) {
+  constructor({web3_rpc_endpoint, web3_rpc_test_endpoint, nodeos_rpc_endpoint, miner_rpc_endpoint, ws_listening_host, ws_listening_port, poll_interval, max_logs_subs_per_connection, max_minedtx_subs_per_connection, logger, whitelist_methods, genesis_json}) {
     super();
 
     const genesis = load_json_file(genesis_json);
 
     this.block_monitor = new BlockMonitor({web3_rpc_endpoint, nodeos_rpc_endpoint, poll_interval, genesis, logger});
-    this.web_socket_handler = new WebSocketHandler({ws_listening_host, ws_listening_port, web3_rpc_endpoint, logger});
+    this.web_socket_handler = new WebSocketHandler({ws_listening_host, ws_listening_port, web3_rpc_endpoint, web3_rpc_test_endpoint, miner_rpc_endpoint, logger, whitelist_methods});
     this.max_logs_subs_per_connection = max_logs_subs_per_connection;
     this.max_minedtx_subs_per_connection = max_minedtx_subs_per_connection;
     this.web3 = new Web3(web3_rpc_endpoint);
-    this.logger = logger
+    this.logger = logger;
     this.genesis = genesis;
 
     this.new_head_subs = new Map();
diff --git a/peripherals/eos-evm-ws-proxy/websocket-handler.js b/peripherals/eos-evm-ws-proxy/websocket-handler.js
@@ -28,13 +28,21 @@ const { is_plain_object } = require('./utils');
 
 class WebSocketHandler extends EventEmitter {
 
-  constructor({ ws_listening_host, ws_listening_port, web3_rpc_endpoint, logger }) {
+  constructor({ ws_listening_host, ws_listening_port, web3_rpc_endpoint, web3_rpc_test_endpoint, miner_rpc_endpoint, logger, whitelist_methods }) {
     super();
 
     this.host = ws_listening_host;
     this.port = ws_listening_port;
     this.web3_rpc_endpoint = web3_rpc_endpoint;
+    this.web3_rpc_test_endpoint = web3_rpc_test_endpoint;
+    this.miner_rpc_endpoint = miner_rpc_endpoint;
     this.logger = logger;
+    this.whitelist_methods = whitelist_methods.split(",");
+
+    console.log("web3_rpc_endpoint is:", this.web3_rpc_endpoint);
+    console.log("web3_rpc_test_endpoint is:", this.web3_rpc_test_endpoint);
+    console.log("miner_rpc_endpoint is:", this.miner_rpc_endpoint);
+    console.log("whitelist_methods are:", this.whitelist_methods);
 
     this.server = http.createServer((req, res) => {
       this.handle_http_request(req, res);
@@ -62,96 +70,94 @@ class WebSocketHandler extends EventEmitter {
     });
   }
   
-  async handle_eth_subscribe(ws, data) {
+  handle_eth_subscribe(ws, data) {
 
-    try {
-      if (!data.hasOwnProperty('params') || !Array.isArray(data.params) || data.params.length == 0) {
-        throw new Error("No params");
-      }
+    if (!data.hasOwnProperty('params') || !Array.isArray(data.params) || data.params.length == 0) {
+      throw new Error("No params");
+    }
 
-      const subscription_type = data.params[0];
-      if( subscription_type == "newHeads") {
-        const subid = uuidv4();
-        this.emit('newHeads', {subid, ws});
-        ws.send(JSON.stringify({ jsonrpc: "2.0", result: subid, id: data.id }));
-      } else if (subscription_type == "logs") {
-        const subid = uuidv4();
-        let filter = {};
-        if(data.params.length > 1) {
-          filter = data.params[1];
-          if (!(is_plain_object(filter) && filter != null)) {
-            throw new Error("Invalid filter");
-          }
-          if (filter.hasOwnProperty('address') && !(typeof filter.address === 'string' || Array.isArray(filter.address))) {
-            throw new Error("Invalid address filter");
-          }
-          if (filter.hasOwnProperty('topics') && !Array.isArray(filter.topics)) {
-            throw new Error("Invalid topics filter");
-          }
+    const subscription_type = data.params[0];
+    if( subscription_type == "newHeads") {
+      const subid = uuidv4();
+      this.emit('newHeads', {subid, ws});
+      return { jsonrpc: "2.0", result: subid, id: data.id };
+    } else if (subscription_type == "logs") {
+      const subid = uuidv4();
+      let filter = {};
+      if(data.params.length > 1) {
+        filter = data.params[1];
+        if (!(is_plain_object(filter) && filter != null)) {
+          throw new Error("Invalid filter");
+        }
+        if (filter.hasOwnProperty('address') && !(typeof filter.address === 'string' || Array.isArray(filter.address))) {
+          throw new Error("Invalid address filter");
+        }
+        if (filter.hasOwnProperty('topics') && !Array.isArray(filter.topics)) {
+          throw new Error("Invalid topics filter");
+        }
+      }
+      this.emit('logs', {subid, ws, filter});
+      return { jsonrpc: "2.0", result: subid, id: data.id };
+    } else if (subscription_type == "minedTransactions") {
+      const subid = uuidv4();
+      let filter = {};
+      if(data.params.length > 1) {
+        filter = data.params[1];
+        if (!(is_plain_object(filter) && filter != null)) {
+          throw new Error("Invalid filter");
         }
-        this.emit('logs', {subid, ws, filter});
-        ws.send(JSON.stringify({ jsonrpc: "2.0", result: subid, id: data.id }));
-      } else if (subscription_type == "minedTransactions") {
-        const subid = uuidv4();
-        let filter = {};
-        if(data.params.length > 1) {
-          filter = data.params[1];
-          if (!(is_plain_object(filter) && filter != null)) {
-            throw new Error("Invalid filter");
+        if (filter.hasOwnProperty('addresses')) {
+          if(!Array.isArray(filter.addresses)) {
+            throw new Error("Invalid addresses filter");
           }
-          if (filter.hasOwnProperty('addresses')) {
-            if(!Array.isArray(filter.addresses)) {
-              throw new Error("Invalid addresses filter");
+          for(const ofilter of filter.addresses) {
+            this.logger.debug("ELEMENT: ", ofilter);
+            if(!is_plain_object(ofilter) || (typeof(ofilter.to) != 'string' && typeof(ofilter.from) != 'string') ) {
+              throw new Error("Invalid addresses filter element");
             }
-            for(const ofilter of filter.addresses) {
-              this.logger.debug("ELEMENT: ", ofilter);
-              if(!is_plain_object(ofilter) || (typeof(ofilter.to) != 'string' && typeof(ofilter.from) != 'string') ) {
-                throw new Error("Invalid addresses filter element");
-              }
-              if(typeof(ofilter.to) == 'string') {
-                ofilter.to = ofilter.to.toLowerCase();
-              }
-              if(typeof(ofilter.from) == 'string') {
-                ofilter.from = ofilter.from.toLowerCase();
-              }
+            if(typeof(ofilter.to) == 'string') {
+              ofilter.to = ofilter.to.toLowerCase();
+            }
+            if(typeof(ofilter.from) == 'string') {
+              ofilter.from = ofilter.from.toLowerCase();
             }
-          }
-          if (filter.hasOwnProperty('includeRemoved') && typeof(filter.includeRemoved) != 'boolean') {
-            throw new Error("Invalid includeRemoved filter");
-          }
-          if (filter.hasOwnProperty('hashesOnly') && typeof(filter.hashesOnly) != 'boolean') {
-            throw new Error("Invalid hashesOnly filter");
           }
         }
-        this.emit('minedTransactions', {subid, ws, filter});
-        ws.send(JSON.stringify({ jsonrpc: "2.0", result: subid, id: data.id }));
-      } else {
-        throw new Error(`${data.params[0]} not supported`);
+        if (filter.hasOwnProperty('includeRemoved') && typeof(filter.includeRemoved) != 'boolean') {
+          throw new Error("Invalid includeRemoved filter");
+        }
+        if (filter.hasOwnProperty('hashesOnly') && typeof(filter.hashesOnly) != 'boolean') {
+          throw new Error("Invalid hashesOnly filter");
+        }
       }
-    } catch (error) {
-      this.send_json_rpc_error(ws, data.id, -32000, error.message);
+      this.emit('minedTransactions', {subid, ws, filter});
+      return { jsonrpc: "2.0", result: subid, id: data.id };
+    } else {
+      throw new Error(`${data.params[0]} not supported`);
     }
   }
  
-  async handle_eth_unsubscribe(ws, data) {
-    try {
-      if (!data.hasOwnProperty('params') || !Array.isArray(data.params) || data.params.length == 0) {
-        throw new Error("Invalid params");
-      }
-      const subid = data.params[0];
-      this.emit('unsubscribe', {subid, ws});
-      ws.send(JSON.stringify({ jsonrpc: "2.0", result: true, id: data.id }));
-    } catch (error) {
-      this.send_json_rpc_error(ws, data.id, -32000, error.message);
+  handle_eth_unsubscribe(ws, data) {
+    if (!data.hasOwnProperty('params') || !Array.isArray(data.params) || data.params.length == 0) {
+      throw new Error("Invalid params");
     }
+    const subid = data.params[0];
+    this.emit('unsubscribe', {subid, ws});
+    return { jsonrpc: "2.0", result: true, id: data.id };
   }
 
-  async handle_other_methods(ws, data) {
-    try {
+  async handle_miner_methods(data) {
+    const response = await axios.post(this.miner_rpc_endpoint, data);
+    return response.data;
+  }
+
+  async handle_other_methods(data, use_test_rpc) {
+    if (use_test_rpc) {
+      const response = await axios.post(this.web3_rpc_test_endpoint, data);
+      return response.data;
+    } else {
       const response = await axios.post(this.web3_rpc_endpoint, data);
-      ws.send(JSON.stringify(response.data));
-    } catch (error) {
-      this.send_json_rpc_error(ws, data.id, -32000, "Sever Error");
+      return response.data;
     }
   }
 
@@ -166,27 +172,139 @@ class WebSocketHandler extends EventEmitter {
   async handle_message(ws, message) {
 
     let data;
+    let use_test_rpc = false;
     try {
         data = JSON.parse(message);
     } catch (e) {
         ws.send(JSON.stringify({ error: "Invalid JSON" }));
         return;
     }
 
-    switch (data.method) {
+    if (Array.isArray(data)) {
+      let data2 = [];
+      for (let i = 0; i < data.length; ++i) {
+        switch(data[i].method) {
+          case 'eth_subscribe':
+          case 'eth_unsubscribe':
+          case 'eth_gasPrice':
+          case 'eth_sendRawTransaction':
+            break;
+          default:
+            if ((this.whitelist_methods.indexOf(data[i].method) < 0)) {
+              use_test_rpc = true;
+            }
+            data2.push(data[i]);
+        }
+      }
+      let rpc_response_data = [];
+      let rpc_error_message = null;
+      if (data2.length > 0) {
+        try {
+          const rpc_response = await this.handle_other_methods(data2, use_test_rpc);
+          if (Array.isArray(rpc_response)) {
+            rpc_response_data = rpc_response;
+          } else {
+            rpc_error_message = "RPC Server Error:" + JSON.stringify({message: rpc_response});
+          }
+        } catch (error) {
+          rpc_error_message = "RPC Server Error:" + error.message;
+        }
+      }
+      let rpc_index = 0;
+      let batch_response = [];
+      for (let i = 0; i < data.length; ++i) {
+        switch(data[i].method) {
+          case 'eth_subscribe':
+            try {
+              batch_response.push(this.handle_eth_subscribe(ws, data[i]));
+            } catch (error) {
+              batch_response.push({ 
+                id      : data[i].id,
+                jsonrpc : "2.0",
+                error   : { code: -32000, message: error.message }
+              });
+            }
+            break;
+          case 'eth_unsubscribe':
+            try {
+              batch_response.push(this.handle_eth_unsubscribe(ws, data[i]));
+            } catch (error) {
+              batch_response.push({ 
+                id      : data[i].id,
+                jsonrpc : "2.0",
+                error   : { code: -32000, message: error.message }
+              });
+            }
+            break;
+          case 'eth_gasPrice':
+          case 'eth_sendRawTransaction':
+            try {
+              const response_json = await this.handle_miner_methods(data[i]);
+              batch_response.push(response_json);
+            } catch (error) {
+              batch_response.push({ 
+                id      : data[i].id,
+                jsonrpc : "2.0",
+                error   : { code: -32000, message: error.message }
+              });
+            }
+            break;
+          default:
+            if (rpc_index < rpc_response_data.length) {
+              batch_response.push(rpc_response_data[rpc_index]);
+              rpc_index++;
+            } else {
+              batch_response.push({ 
+                id      : data[i].id,
+                jsonrpc : "2.0",
+                error   : { code: -32000, message: rpc_error_message }
+              });
+            }
+        }
+      }
+      ws.send(JSON.stringify(batch_response));
+    }
+    else {
+      switch (data.method) {
         case 'eth_subscribe':
-          await this.handle_eth_subscribe(ws, data);
+          try {
+            const response_json = this.handle_eth_subscribe(ws, data);
+            ws.send(JSON.stringify(response_json));
+          } catch (error) {
+            this.send_json_rpc_error(ws, data.id, -32000, error.message);
+          }
           break;
 
         case 'eth_unsubscribe':
-          await this.handle_eth_unsubscribe(ws, data);  
+          try {
+            const response_json = this.handle_eth_unsubscribe(ws, data);  
+            ws.send(JSON.stringify(response_json));
+          } catch (error) {
+            this.send_json_rpc_error(ws, data.id, -32000, error.message);
+          }
+          break;
+        case 'eth_gasPrice':
+        case 'eth_sendRawTransaction':
+          try {
+            const response_json = await this.handle_miner_methods(data);
+            ws.send(JSON.stringify(response_json));
+          } catch (error) {
+            this.send_json_rpc_error(ws, data.id, -32000, "RPC Miner Error:" + error.message);
+          }
           break;
-
         default:
-          await this.handle_other_methods(ws, data);
+          try {
+            if ((this.whitelist_methods.indexOf(data.method) < 0)) {
+              use_test_rpc = true;
+            }
+            const response_json = await this.handle_other_methods(data, use_test_rpc);
+            ws.send(JSON.stringify(response_json));
+          } catch (error) {
+            this.send_json_rpc_error(ws, data.id, -32000, "RPC Server Error:" + error.message);
+          }
           break;
+      }
     }
-
   }
 
   handle_close(ws) {

Original file line number	Diff line number	Diff line change
`@@ -130,9 +130,10 @@ class BlockMonitor extends EventEmitter {`
`130`	`130`	`}`
`131`	`131`	`}`
`132`	`132`
`133`		`- if( found_next_block == true ) {`
	`133`	`+ if( found_next_block == true && this.reversible_blocks.length > 180 + 6) { // reduce frequency of calling get_evm_lib`
`134`	`134`	`const evm_lib = await this.get_evm_lib();`
`135`		`- while(this.reversible_blocks.length > 0 && this.reversible_blocks.peek().number < evm_lib) {`
	`135`	`+ // keep at least 180 blocks in case evm-node is out of sync with leap`
	`136`	`+ while(this.reversible_blocks.length > 180 && this.reversible_blocks.peek().number < evm_lib) {`
`136`	`137`	`this.remove_front_block();`
`137`	`138`	`}`
`138`	`139`	`}`