Fix eth_subscribe to logs, add remaining chains to cli tool (#256)

dshackle-cli/src/cli.js

@@ -3,13 +3,21 @@ import arg from 'arg';
 import clc from "cli-color";
 import util from "util";
 
-const chains = {
+const path = require('path')
-    CHAIN_BSC__MAINNET: 1006,
+const protoLoader = require("@grpc/proto-loader");
-    CHAIN_OPTIMISM__MAINNET: 1005,
+const grpc = require("@grpc/grpc-js");
-    CHAIN_ARBITRUM__MAINNET: 1004,
+
-    CHAIN_POLYGON_POS__MAINNET: 1002,
+const options = {
-    CHAIN_ETHEREUM__MAINNET: 100
+    keepCase: true,
-}
+    longs: String,
+    enums: String,
+    defaults: true,
+    oneofs: true,
+};
+
+const PROTO_PATH = path.join(__dirname, "../../emerald-grpc/proto/blockchain.proto");
+const packageDefinition = protoLoader.loadSync(PROTO_PATH, options);
+const emerald = grpc.loadPackageDefinition(packageDefinition).emerald
 
 export function cli(args) {
     let opts = parseArgumentsIntoOptions(args);
@@ -18,7 +26,9 @@ export function cli(args) {
         return
     }
 
-    const client = connect(opts.url, opts.ca, opts.cert, opts.key)
+    const chains = mapChains()
+
+    const client = connect(opts.url, opts.ca, opts.cert, opts.key, emerald)
     describe(client, (error, response) => {
         if (error) {
             console.error(clc.red('Connection to ' + opts.url + ' failed:  ' + error.message));
@@ -28,11 +38,11 @@ export function cli(args) {
         if (opts.print) {
             console.log(util.inspect(response, false, null, true /* enable colors */));
         }
-        processDescribe(client, response, opts.testRun)
+        processDescribe(client, response, opts.testRun, chains)
     })
 }
 
-function processDescribe(client, response, testRun) {
+function processDescribe(client, response, testRun, chains) {
     let promises = []
     let statuses = new Map()
 
@@ -58,7 +68,7 @@ function processDescribe(client, response, testRun) {
         }
 
         if (state === 'AVAIL_OK') {
-            promises.push(nativeCall(client, chains[chain], chain))
+            promises.push(nativeCall(client, chains.get(chain), chain))
         } else {
             status.failed = true
         }
@@ -95,6 +105,14 @@ function processDescribe(client, response, testRun) {
     })
 }
 
+function mapChains() {
+    return new Map(
+        emerald.ChainRef.type.value.map(obj => {
+            return [obj.name, obj.number]
+        })
+    )
+}
+
 function printState(chain, status) {
     console.log(chain + ' -> ' + "state: " + clc.bold(status.state) + " gRPC: " + clc.bold(status.grpc))
 }

dshackle-cli/src/grpc-clent.js

@@ -1,25 +1,9 @@
 const grpc = require("@grpc/grpc-js");
-const path = require('path')
-const protoLoader = require("@grpc/proto-loader");
 const fs = require('fs');
 
-const PROTO_PATH = path.join(__dirname, "../../emerald-grpc/proto/blockchain.proto");
-
-const options = {
-    keepCase: true,
-    longs: String,
-    enums: String,
-    defaults: true,
-    oneofs: true,
-};
-
-const packageDefinition = protoLoader.loadSync(PROTO_PATH, options);
-const emerald = grpc.loadPackageDefinition(packageDefinition).emerald
-
-
 var id = 100
 
-export function connect(url, ca, cert, key) {
+export function connect(url, ca, cert, key, emerald) {
     let credentials = grpc.credentials.createInsecure()
     if (ca || cert || key) {
         console.log("Using TLS")

src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeSubscribe.kt

@@ -97,7 +97,11 @@ open class NativeSubscribe(
     }
 
     open fun subscribe(chain: Chain, method: String, params: Any?, matcher: Selector.Matcher): Flux<out Any> =
-        getUpstream(chain).getEgressSubscription().subscribe(method, params, matcher)
+        getUpstream(chain).getEgressSubscription()
+            .subscribe(method, params, matcher)
+            .doOnError {
+                log.error("Error during subscription to $method, chain $chain, params $params", it)
+            }
 
     private fun getUpstream(chain: Chain): EthereumLikeMultistream =
         multistreamHolder.getUpstream(chain).let { it as EthereumLikeMultistream }

src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumEgressSubscription.kt

@@ -93,6 +93,7 @@ open class EthereumEgressSubscription(
                         null
                     }
                 }
+                null -> emptyList()
                 else -> throw IllegalArgumentException("Invalid type of address field. Must be string or list of strings")
             }
         } else {
@@ -114,6 +115,7 @@ open class EthereumEgressSubscription(
                         null
                     }
                 }
+                null -> emptyList()
                 else -> throw IllegalArgumentException("Invalid type of topics field. Must be string or list of strings")
             }
         } else {