feat: upgrade to ZMQ 6.0 using dashd-zmq client

.travis.yml

@@ -1,17 +1,17 @@
 language: node_js
 sudo: false
 node_js:
-  - '8'
-  - '10'
+  - '12'
+  - '14'
 env:
-  - CXX=g++-4.8 CC=gcc-4.8
+  - CXX=g++-8 CC=gcc-8
 addons:
   apt:
     sources:
       - ubuntu-toolchain-r-test
     packages:
-      - g++-4.8
-      - gcc-4.8
+      - g++-8
+      - gcc-8
       - libzmq3-dev
 
 script:

lib/services/dashd.js

@@ -6,9 +6,9 @@ var spawn = require('child_process').spawn;
 var util = require('util');
 var mkdirp = require('mkdirp');
 var dashcore = require('@dashevo/dashcore-lib');
-var zmq = require('zeromq');
 var async = require('async');
 var LRU = require('lru-cache');
+var DashdZMQ = require('@dashevo/dashd-zmq');
 var DashdRPC = require('@dashevo/dashd-rpc');
 var $ = dashcore.util.preconditions;
 var _  = dashcore.deps._;
@@ -776,46 +776,47 @@ Dash.prototype._checkSyncedAndSubscribeZmqEvents = function(node) {
 
 Dash.prototype._subscribeZmqEvents = function(node) {
   var self = this;
-  node.zmqSubSocket.subscribe('hashblock');
-  node.zmqSubSocket.subscribe('rawtx');
-  node.zmqSubSocket.subscribe('rawtxlock');
-  node.zmqSubSocket.on('message', function(topic, message) {
-    var topicString = topic.toString('utf8');
-    if (topicString === 'rawtxlock') {
-      self._zmqTransactionLockHandler(node, message);
-    } else if (topicString === 'rawtx') {
-      self._zmqTransactionHandler(node, message);
-    } else if (topicString === 'hashblock') {
-      self._zmqBlockHandler(node, message);
-    }
+  node.zmqSubSocket.subscribe(DashdZMQ.TOPICS.hashblock);
+  node.zmqSubSocket.on(DashdZMQ.TOPICS.hashblock, (message) => {
+    self._zmqBlockHandler(node, Buffer.from(message, 'hex'));
+  });
+
+  node.zmqSubSocket.subscribe(DashdZMQ.TOPICS.rawtx);
+  node.zmqSubSocket.on(DashdZMQ.TOPICS.rawtx, (message) => {
+    self._zmqTransactionHandler(node, Buffer.from(message, 'hex'));
+  });
+
+  node.zmqSubSocket.subscribe(DashdZMQ.TOPICS.rawtxlock);
+  node.zmqSubSocket.on(DashdZMQ.TOPICS.rawtxlock, (message) => {
+    self._zmqTransactionLockHandler(node, Buffer.from(message, 'hex'));
   });
 };
 
 Dash.prototype._initZmqSubSocket = function(node, zmqUrl) {
-  var self = this;
-  node.zmqSubSocket = zmq.socket('sub');
 
-  node.zmqSubSocket.on('connect', function(fd, endPoint) {
-    log.info('ZMQ connected to:', endPoint);
-  });
+  return new Promise((resolve) => {
+    node.zmqSubSocket = new DashdZMQ({
+      uri: zmqUrl
+    });
 
-  node.zmqSubSocket.on('connect_delay', function(fd, endPoint) {
-    log.warn('ZMQ connection delay:', endPoint);
-  });
+    node.zmqSubSocket.on('connect', function (info) {
+      log.info('ZMQ connected to:', info.address);
+    });
 
-  node.zmqSubSocket.on('disconnect', function(fd, endPoint) {
-    log.warn('ZMQ disconnect:', endPoint);
-  });
+    node.zmqSubSocket.on('connect:delay', function(info) {
+      log.warn('ZMQ connection delay:', info.address);
+    });
 
-  node.zmqSubSocket.on('monitor_error', function(err) {
-    log.error('Error in monitoring: %s, will restart monitoring in 5 seconds', err);
-    setTimeout(function() {
-      self.zmqSubSocket.monitor(500, 0);
-    }, 5000);
-  });
+    node.zmqSubSocket.on('disconnect', function(info) {
+      log.warn('ZMQ disconnect:', info.address);
+    });
+
+    const socketOptions = { reconnectInterval: 1500 };
 
-  node.zmqSubSocket.monitor(500, 0);
-  node.zmqSubSocket.connect(zmqUrl);
+    node.zmqSubSocket
+        .connect({ socketOptions })
+        .then(resolve);
+  });
 };
 
 Dash.prototype._checkReindex = function(node, callback) {
@@ -980,16 +981,19 @@ Dash.prototype._spawnChildProcess = function(callback) {
         return callback(new Error('Stopping while trying to spawn dashd.'));
       }
 
-      self._initZmqSubSocket(node, self.spawn.config.zmqpubrawtx);
+      self._initZmqSubSocket(node, self.spawn.config.zmqpubrawtx)
+          .then(() => {
+              self._checkReindex(node, function (err) {
 
-      self._checkReindex(node, function(err) {
-        if (err) {
-          return callback(err);
-        }
-        self._checkSyncedAndSubscribeZmqEvents(node);
-        callback(null, node);
-      });
+              if (err) {
+                return callback(err);
+              }
+
+              self._checkSyncedAndSubscribeZmqEvents(node);
 
+              callback(null, node);
+          });
+      });
     });
 
   });
@@ -1026,8 +1030,9 @@ Dash.prototype._connectProcess = function(config, callback) {
       return callback(new Error('Stopping while trying to connect to dashd.'));
     }
 
-    self._initZmqSubSocket(node, config.zmqpubrawtx);
-    self._subscribeZmqEvents(node);
+    self._initZmqSubSocket(node, config.zmqpubrawtx).then(() => {
+            self._subscribeZmqEvents(node);
+        });
 
     callback(null, node);
   });