Merge pull request #3752 from ethcore/jg-subscription-check

Allow autoRemove from api.subscribe based on callback return values

js/src/api/contract/contract.js

@@ -240,8 +240,8 @@ export default class Contract {
   }
 
   _bindEvent = (event) => {
-    event.subscribe = (options = {}, callback) => {
-      return this._subscribe(event, options, callback);
+    event.subscribe = (options = {}, callback, autoRemove) => {
+      return this._subscribe(event, options, callback, autoRemove);
     };
 
     event.unsubscribe = (subscriptionId) => {
@@ -306,16 +306,31 @@ export default class Contract {
     return this._api.eth.newFilter(options);
   }
 
-  subscribe (eventName = null, options = {}, callback) {
+  subscribe (eventName = null, options = {}, callback, autoRemove) {
     try {
       const event = this._findEvent(eventName);
-      return this._subscribe(event, options, callback);
+      return this._subscribe(event, options, callback, autoRemove);
     } catch (e) {
       return Promise.reject(e);
     }
   }
 
-  _subscribe (event = null, _options, callback) {
+  _sendData (subscriptionId, error, logs) {
+    const { autoRemove, callback } = this._subscriptions[subscriptionId];
+    let result = true;
+
+    try {
+      result = callback(error, logs);
+    } catch (error) {
+      console.warn('_sendData', subscriptionId, error);
+    }
+
+    if (autoRemove && result && typeof result === 'boolean') {
+      this.unsubscribe(subscriptionId);
+    }
+  }
+
+  _subscribe (event = null, _options, callback, autoRemove = false) {
     const subscriptionId = nextSubscriptionId++;
     const { skipInitFetch } = _options;
     delete _options['skipInitFetch'];
@@ -325,6 +340,7 @@ export default class Contract {
       .then((filterId) => {
         this._subscriptions[subscriptionId] = {
           options: _options,
+          autoRemove,
           callback,
           filterId
         };
@@ -337,8 +353,7 @@ export default class Contract {
         return this._api.eth
           .getFilterLogs(filterId)
           .then((logs) => {
-            callback(null, this.parseEventLogs(logs));
-
+            this._sendData(subscriptionId, null, this.parseEventLogs(logs));
             this._subscribeToChanges();
             return subscriptionId;
           });
@@ -437,13 +452,13 @@ export default class Contract {
         })
       )
       .then((logsArray) => {
-        logsArray.forEach((logs, idx) => {
+        logsArray.forEach((logs, subscriptionId) => {
           if (!logs || !logs.length) {
             return;
           }
 
           try {
-            subscriptions[idx].callback(null, this.parseEventLogs(logs));
+            this.sendData(subscriptionId, null, this.parseEventLogs(logs));
           } catch (error) {
             console.error('_sendSubscriptionChanges', error);
           }

js/src/api/subscriptions/manager.js

@@ -59,7 +59,7 @@ export default class Manager {
     return subscription;
   }
 
-  subscribe (subscriptionName, callback) {
+  subscribe (subscriptionName, callback, autoRemove = false) {
     return new Promise((resolve, reject) => {
       const subscription = this._validateType(subscriptionName);
 
@@ -75,6 +75,7 @@ export default class Manager {
       this.subscriptions[subscriptionId] = {
         name: subscriptionName,
         id: subscriptionId,
+        autoRemove,
         callback
       };
 
@@ -101,13 +102,18 @@ export default class Manager {
   }
 
   _sendData (subscriptionId, error, data) {
-    const { callback } = this.subscriptions[subscriptionId];
+    const { autoRemove, callback } = this.subscriptions[subscriptionId];
+    let result = true;
 
     try {
-      callback(error, data);
+      result = callback(error, data);
     } catch (error) {
       console.error(`Unable to update callback for subscriptionId ${subscriptionId}`, error);
     }
+
+    if (autoRemove && result && typeof result === 'boolean') {
+      this.unsubscribe(subscriptionId);
+    }
   }
 
   _updateSubscriptions = (subscriptionName, error, data) => {