Feat/subscribealerts

bin/xud

@@ -42,8 +42,8 @@ const { argv } = require('yargs')
       type: 'boolean',
       default: undefined,
     },
-    regnet: {
-      describe: 'Whether to run XUD on regnet',
+    regtest: {
+      describe: 'Whether to run XUD on regtest',
       type: 'boolean',
       default: undefined,
     },

lib/Xud.ts

@@ -130,6 +130,12 @@ class Xud extends EventEmitter {
           nodeKey = await NodeKey.generate();
           await nodeKey.toFile(nodeKeyPath);
         }
+
+        // we need to initialize connext every time xud starts, even in noencrypt mode
+        // the call below is in lieu of the UnlockNode/CreateNode call flow
+        await this.swapClientManager.initConnext(
+          nodeKey.childSeed(SwapClientType.Connext),
+        );
       } else if (this.rpcServer) {
         this.rpcServer.grpcService.locked = true;
         const initService = new InitService(this.swapClientManager, nodeKeyPath, nodeKeyExists, this.config.dbpath);
@@ -194,12 +200,6 @@ class Xud extends EventEmitter {
       // wait for components to initialize in parallel
       await Promise.all(initPromises);
 
-      // We initialize Connext separately because it
-      // requires a NodeKey.
-      await this.swapClientManager.initConnext(
-        nodeKey.childSeed(SwapClientType.Connext),
-      );
-
       // initialize pool and start listening/connecting only once other components are initialized
       await this.pool.init();
 

lib/cli/command.ts

@@ -38,6 +38,18 @@ const loadXudConfig = async (argv: Arguments<any>) => {
   }
 };
 
+const getTlsCert = (certPath: string) => {
+  try {
+    return fs.readFileSync(certPath);
+  } catch (err) {
+    if (err.code === 'ENOENT') {
+      throw `tls cert could not be found at ${certPath}, it may take several seconds to be created on xud's first run`;
+    }
+
+    throw err;
+  }
+};
+
 /**
  * A generic function to instantiate an XU client.
  * @param argv the command line arguments
@@ -46,7 +58,7 @@ export const loadXudClient = async (argv: Arguments<any>) => {
   await loadXudConfig(argv);
 
   const certPath = argv.tlscertpath || path.join(argv.xudir, 'tls.cert');
-  const cert = fs.readFileSync(certPath);
+  const cert = getTlsCert(certPath);
   const credentials = grpc.credentials.createSsl(cert);
 
   return new XudClient(`${argv.rpchost}:${argv.rpcport}`, credentials);
@@ -56,7 +68,7 @@ export const loadXudInitClient = async (argv: Arguments<any>) => {
   await loadXudConfig(argv);
 
   const certPath = argv.tlscertpath || path.join(argv.xudir, 'tls.cert');
-  const cert = fs.readFileSync(certPath);
+  const cert = getTlsCert(certPath);
   const credentials = grpc.credentials.createSsl(cert);
 
   return new XudInitClient(`${argv.rpchost}:${argv.rpcport}`, credentials);

lib/cli/commands/openchannel.ts

@@ -37,6 +37,10 @@ export const builder = (argv: Argv) => argv
   .example('$0 openchannel ETH 0.5', 'deposit 0.5 into an ETH Connext channel without specifying a remote node');
 
 export const handler = async (argv: Arguments<any>) => {
+  if (isNaN(argv.amount)) {
+    throw 'amount must be a number';
+  }
+
   const request = new OpenChannelRequest();
   if (argv.node_identifier) {
     request.setNodeIdentifier(argv.node_identifier);

lib/cli/commands/removeorder.ts

@@ -1,7 +1,7 @@
 import { Arguments, Argv } from 'yargs';
-import { RemoveOrderRequest } from '../../proto/xudrpc_pb';
+import { RemoveOrderRequest, RemoveOrderResponse } from '../../proto/xudrpc_pb';
 import { callback, loadXudClient } from '../command';
-import { coinsToSats } from '../utils';
+import { coinsToSats, satsToCoinsStr } from '../utils';
 
 export const command = 'removeorder <order_id> [quantity]';
 
@@ -18,11 +18,23 @@ export const builder = (argv: Argv) => argv
   .example('$0 removeorder 79d2cd30-8a26-11ea-90cf-439fb244cf44', 'remove an order by id')
   .example('$0 removeorder 79d2cd30-8a26-11ea-90cf-439fb244cf44 0.1', 'remove 0.1 quantity from an order by id');
 
+const displayOutput = (orderId: string, removeOrderResponse: RemoveOrderResponse.AsObject) => {
+  const removedCurrency = removeOrderResponse.pairId.split('/')[0];
+  if (removeOrderResponse.quantityOnHold === 0 && removeOrderResponse.remainingQuantity === 0) {
+    console.log(`Order ${orderId} successfully removed.`);
+  } else {
+    console.log(`
+Order ${orderId} partially removed, remaining quantity: \
+${satsToCoinsStr(removeOrderResponse.remainingQuantity)} ${removedCurrency}, \
+on hold: ${satsToCoinsStr(removeOrderResponse.quantityOnHold)} ${removedCurrency}`);
+  }
+};
+
 export const handler = async (argv: Arguments<any>) => {
   const request = new RemoveOrderRequest();
   request.setOrderId(argv.order_id);
   if (argv.quantity) {
     request.setQuantity(coinsToSats(argv.quantity));
   }
-  (await loadXudClient(argv)).removeOrder(request, callback(argv));
+  (await loadXudClient(argv)).removeOrder(request, callback(argv, displayOutput.bind(undefined, argv.order_id)));
 };

lib/cli/commands/streamorders.ts

@@ -1,9 +1,8 @@
-import { ServiceError, status } from 'grpc';
 import { Arguments, Argv } from 'yargs';
 import { XudClient } from '../../proto/xudrpc_grpc_pb';
 import * as xudrpc from '../../proto/xudrpc_pb';
-import { setTimeoutPromise } from '../../utils/utils';
 import { loadXudClient } from '../command';
+import { onStreamError, waitForClient } from '../utils';
 
 export const command = 'streamorders [existing]';
 
@@ -26,20 +25,8 @@ const ensureConnection = async (argv: Arguments, printError?: boolean) => {
   if (!client) {
     client = await loadXudClient(argv);
   }
-  client.waitForReady(Date.now() + 3000, (error: Error | null) => {
-    if (error) {
-      if (error.message === 'Failed to connect before the deadline') {
-        console.error(`could not connect to xud at ${argv.rpchost}:${argv.rpcport}, is xud running?`);
-        process.exit(1);
-      }
 
-      if (printError) console.error(`${error.name}: ${error.message}`);
-      setTimeout(ensureConnection.bind(undefined, argv, printError), 3000);
-    } else {
-      console.log('Successfully connected, subscribing for orders');
-      streamOrders(argv);
-    }
-  });
+  waitForClient(client, argv, ensureConnection, streamOrders, printError);
 };
 
 const streamOrders = (argv: Arguments<any>) => {
@@ -57,15 +44,7 @@ const streamOrders = (argv: Arguments<any>) => {
   // adding end, close, error events only once,
   // since they'll be thrown for three of subscriptions in the corresponding cases, catching once is enough.
   ordersSubscription.on('end', reconnect.bind(undefined, argv));
-  ordersSubscription.on('error', async (err: ServiceError) => {
-    if (err.code === status.UNIMPLEMENTED) {
-      console.error("xud is locked, run 'xucli unlock', 'xucli create', or 'xucli restore' then try again");
-      process.exit(1);
-    }
-    console.warn(`Unexpected error occured: ${err.message}, reconnecting in 1 second`);
-    await setTimeoutPromise(1000);
-    await ensureConnection(argv);
-  });
+  ordersSubscription.on('error', onStreamError.bind(undefined, ensureConnection.bind(undefined, argv)));
 
   const swapsRequest = new xudrpc.SubscribeSwapsRequest();
   swapsRequest.setIncludeTaker(true);

lib/cli/commands/subscribealerts.ts

@@ -0,0 +1,51 @@
+import { Arguments, Argv } from 'yargs';
+import { XudClient } from '../../proto/xudrpc_grpc_pb';
+import * as xudrpc from '../../proto/xudrpc_pb';
+import { loadXudClient } from '../command';
+import { AlertType } from '../../constants/enums';
+import { onStreamError, waitForClient } from '../utils';
+
+export const command = 'subscribealerts';
+
+export const describe = 'subscribe alerts such as low balance';
+
+export const builder = (argv: Argv) => argv
+    .option('pretty', {
+      type: 'boolean',
+    })
+    .example('$0 subscribealerts -j', 'prints alert payload in a JSON structure')
+    .example('$0 subscribealerts', 'prints alert message only');
+
+export const handler = async (argv: Arguments) => {
+  await ensureConnection(argv, true);
+};
+
+let client: XudClient;
+
+const ensureConnection = async (argv: Arguments, printError?: boolean) => {
+  if (!client) {
+    client = await loadXudClient(argv);
+  }
+
+  waitForClient(client, argv, ensureConnection, subscribeAlerts, printError);
+};
+
+const subscribeAlerts = (argv: Arguments<any>) => {
+  const request = new xudrpc.SubscribeAlertsRequest();
+  const alertsSubscription = client.subscribeAlerts(request);
+
+  alertsSubscription.on('data', (alert: xudrpc.Alert) => {
+    if (argv.json) {
+      console.log(JSON.stringify(alert, undefined, 2));
+    } else {
+      console.log(`${AlertType[alert.getType()]}: ${alert.getMessage()}`);
+    }
+  });
+  alertsSubscription.on('end', reconnect.bind(undefined, argv));
+  alertsSubscription.on('error', onStreamError.bind(undefined, ensureConnection.bind(undefined, argv)));
+};
+
+const reconnect = async (argv: Arguments) => {
+  console.log('Stream has closed, trying to reconnect');
+  await ensureConnection(argv, false);
+};