address subscriptions

lib/modules/address.js

@@ -12,6 +12,9 @@ var Address = bitcore.Address;
 
 var AddressModule = function(options) {
   BaseModule.call(this, options);
+
+  this.txSubscriptions = {};
+  this.balanceSubscriptions = {};
 };
 
 inherits(AddressModule, BaseModule);
@@ -29,6 +32,13 @@ AddressModule.prototype.getAPIMethods = function() {
   ];
 };
 
+AddressModule.prototype.getEvents = function() {
+  return [
+    ['transaction', this, this.subscribeTransaction, this.unsubscribeTransaction],
+    ['balance', this, this.subscribeBalance, this.unsubscribeBalance]
+  ];
+};
+
 AddressModule.prototype.blockHandler = function(block, addOutput, callback) {
   var txs = this.db.getTransactionsFromBlock(block);
 
@@ -82,6 +92,26 @@ AddressModule.prototype.blockHandler = function(block, addOutput, callback) {
         key: [AddressModule.PREFIXES.OUTPUTS, address, timestamp, txid, outputIndex].join('-'),
         value: [output.satoshis, script, height].join(':')
       });
+
+      if(self.txSubscriptions[address]) {
+        var subscriptions = self.txSubscriptions[address];
+        for(var i = 0; i < subscriptions.length; i++) {
+          subscriptions[i].emit('transaction', tx);
+        }
+      }
+
+      if(self.balanceSubscriptions[address]) {
+        var emitters = self.balanceSubsriptions[address];
+        for(var i = 0; i < subscriptions.length; i++) {
+          self.getBalance(address, true, function(err, balance) {
+            if(err) {
+              return self.emit(err);
+            }
+
+            emitters[i].emit('balance', address, balance, block);
+          });
+        }
+      }
     }
 
     if(tx.isCoinbase()) {
@@ -95,6 +125,50 @@ AddressModule.prototype.blockHandler = function(block, addOutput, callback) {
   });
 };
 
+AddressModule.prototype.subscribeTransaction = function(emitter, addresses) {
+  for(var i = 0; i < addresses.length; i++) {
+    if(!this.txSubscriptions[addresses[i]]) {
+      this.txSubscriptions[addresses[i]] = [];
+    }
+
+    this.txSubscriptions[addresses[i]].push(emitter);
+  }
+};
+
+AddressModule.prototype.subscribeBalance = function(emitter, addresses) {
+  for(var i = 0; i < addresses.length; i++) {
+    if(!this.balanceSubscriptions[addresses[i]]) {
+      this.balanceSubscriptions[addresses[i]] = [];
+    }
+
+    this.balanceSubscriptions[addresses[i]].push(emitter);
+  }
+};
+
+AddressModule.prototype.unsubscribeTransaction = function(emitter, addresses) {
+  for(var i = 0; i < addresses.length; i++) {
+    if(this.txSubscriptions[addresses[i]]) {
+      var emitters = self.txSubscriptions[addresses[i]];
+      var index = emitters.indexOf(emitter);
+      if(index > -1) {
+        emitters.splice(index, 1);
+      }
+    }
+  }
+};
+
+AddressModule.prototype.unsubscribeBalance = function(emitter, addresses) {
+  for(var i = 0; i < addresses.length; i++) {
+    if(this.balanceSubscriptions[addresses[i]]) {
+      var emitters = self.balanceSubscriptions[addresses[i]];
+      var index = emitters.indexOf(emitter);
+      if(index > -1) {
+        emitters.splice(index, 1);
+      }
+    }
+  }
+};
+
 AddressModule.prototype.getBalance = function(address, queryMempool, callback) {
   this.getUnspentOutputs(address, queryMempool, function(err, outputs) {
     if(err) {