Address Service: Removed event listeners prior to stopping

lib/services/address/index.js

@@ -41,8 +41,10 @@ var AddressService = function(options) {
   this.subscriptions['address/transaction'] = {};
   this.subscriptions['address/balance'] = {};
 
-  this.node.services.bitcoind.on('tx', this.transactionHandler.bind(this));
-  this.node.services.bitcoind.on('txleave', this.transactionLeaveHandler.bind(this));
+  this._bitcoindTransactionListener = this.transactionHandler.bind(this);
+  this._bitcoindTransactionLeaveListener = this.transactionLeaveHandler.bind(this);
+  this.node.services.bitcoind.on('tx', this._bitcoindTransactionListener);
+  this.node.services.bitcoind.on('txleave', this._bitcoindTransactionLeaveListener);
 
   this.maxInputsQueryLength = options.maxInputsQueryLength || constants.MAX_INPUTS_QUERY_LENGTH;
   this.maxOutputsQueryLength = options.maxOutputsQueryLength || constants.MAX_OUTPUTS_QUERY_LENGTH;
@@ -103,6 +105,8 @@ AddressService.prototype.start = function(callback) {
 
 AddressService.prototype.stop = function(callback) {
   // TODO Keep track of ongoing db requests before shutting down
+  this.node.services.bitcoind.removeListener('tx', this._bitcoindTransactionListener);
+  this.node.services.bitcoind.removeListener('txleave', this._bitcoindTransactionLeaveListener);
   this.mempoolIndex.close(callback);
 };
 
@@ -227,6 +231,10 @@ AddressService.prototype.transactionLeaveHandler = function(txInfo) {
 AddressService.prototype.transactionHandler = function(txInfo, callback) {
   var self = this;
 
+  if (this.node.stopping) {
+    return callback();
+  }
+
   // Basic transaction format is handled by the daemon
   // and we can safely assume the buffer is properly formatted.
   var tx = bitcore.Transaction().fromBuffer(txInfo.buffer);
@@ -760,11 +768,6 @@ AddressService.prototype.createInputsStream = function(addressStr, options) {
       inputStream.end();
     }).pipe(inputStream);
 
-
-  inputStream.on('end', function() {
-    stream.end();
-  });
-
   return stream;
 
 };
@@ -967,7 +970,6 @@ AddressService.prototype._getSpentMempool = function(txidBuffer, outputIndex, ca
 };
 
 AddressService.prototype.createOutputsStream = function(addressStr, options) {
-
   var outputStream = new OutputsTransformStream({
     address: new Address(addressStr, this.node.network),
     tipHeight: this.node.services.db.tip.__height
@@ -981,10 +983,6 @@ AddressService.prototype.createOutputsStream = function(addressStr, options) {
     })
     .pipe(outputStream);
 
-  outputStream.on('end', function() {
-    stream.end();
-  });
-
   return stream;
 
 };

test/services/address/index.unit.js

@@ -192,14 +192,26 @@ describe('Address Service', function() {
 
   describe('#stop', function() {
     it('will close mempool levelup', function(done) {
+      var testnode = {
+        network: Networks.testnet,
+        datadir: 'testdir',
+        db: mockdb,
+        services: {
+          bitcoind: {
+            on: sinon.stub(),
+            removeListener: sinon.stub()
+          }
+        }
+      };
       var am = new AddressService({
         mempoolMemoryIndex: true,
-        node: mocknode
+        node: testnode
       });
       am.mempoolIndex = {};
       am.mempoolIndex.close = sinon.stub().callsArg(0);
       am.stop(function() {
         am.mempoolIndex.close.callCount.should.equal(1);
+        am.node.services.bitcoind.removeListener.callCount.should.equal(2);
         done();
       });
     });