1 files changed, 36 insertions, 16 deletions
diff --git a/src/contract_wrappers/contract_wrapper.ts b/src/contract_wrappers/contract_wrapper.ts
index 19dccc6f2..7997b1647 100644
--- a/src/contract_wrappers/contract_wrapper.ts
+++ b/src/contract_wrappers/contract_wrapper.ts
@@ -38,6 +38,29 @@ export class ContractWrapper {
         this._onLogAddedSubscriptionToken = undefined;
         this._onLogRemovedSubscriptionToken = undefined;
     }
+    /**
+     * Cancels all existing subscriptions
+     */
+    public unsubscribeAll(): void {
+        const filterTokens = _.keys(this._filterCallbacks);
+        _.each(filterTokens, filterToken => {
+            this._unsubscribe(filterToken);
+        });
+    }
+    protected _unsubscribe(filterToken: string, err?: Error): void {
+        if (_.isUndefined(this._filters[filterToken])) {
+            throw new Error(ZeroExError.SubscriptionNotFound);
+        }
+        if (!_.isUndefined(err)) {
+            const callback = this._filterCallbacks[filterToken];
+            callback(err, undefined);
+        }
+        delete this._filters[filterToken];
+        delete this._filterCallbacks[filterToken];
+        if (_.isEmpty(this._filters)) {
+            this._stopBlockAndLogStream();
+        }
+    }
     protected _subscribe<ArgsType extends ContractEventArgs>(
         address: string, eventName: ContractEvents, indexFilterValues: IndexedFilterValues, abi: Web3.ContractAbi,
         callback: EventCallback<ArgsType>): string {
@@ -50,16 +73,6 @@ export class ContractWrapper {
         this._filterCallbacks[filterToken] = callback;
         return filterToken;
     }
-    protected _unsubscribe(filterToken: string): void {
-        if (_.isUndefined(this._filters[filterToken])) {
-            throw new Error(ZeroExError.SubscriptionNotFound);
-        }
-        delete this._filters[filterToken];
-        delete this._filterCallbacks[filterToken];
-        if (_.isEmpty(this._filters)) {
-            this._stopBlockAndLogStream();
-        }
-    }
     protected async _getLogsAsync<ArgsType extends ContractEventArgs>(
         address: string, eventName: ContractEvents, subscriptionOpts: SubscriptionOpts,
         indexFilterValues: IndexedFilterValues, abi: Web3.ContractAbi): Promise<Array<LogWithDecodedArgs<ArgsType>>> {
@@ -90,7 +103,7 @@ export class ContractWrapper {
                     ...decodedLog,
                     removed,
                 };
-                this._filterCallbacks[filterToken](logEvent);
+                this._filterCallbacks[filterToken](null, logEvent);
             }
         });
     }
@@ -122,11 +135,18 @@ export class ContractWrapper {
         delete this._blockAndLogStreamer;
     }
     private async _reconcileBlockAsync(): Promise<void> {
-        const latestBlock = await this._web3Wrapper.getBlockAsync(BlockParamLiteral.Latest);
-        // We need to coerce to Block type cause Web3.Block includes types for mempool blocks
-        if (!_.isUndefined(this._blockAndLogStreamer)) {
-            // If we clear the interval while fetching the block - this._blockAndLogStreamer will be undefined
-            this._blockAndLogStreamer.reconcileNewBlock(latestBlock as any as Block);
+        try {
+            const latestBlock = await this._web3Wrapper.getBlockAsync(BlockParamLiteral.Latest);
+            // We need to coerce to Block type cause Web3.Block includes types for mempool blocks
+            if (!_.isUndefined(this._blockAndLogStreamer)) {
+                // If we clear the interval while fetching the block - this._blockAndLogStreamer will be undefined
+                this._blockAndLogStreamer.reconcileNewBlock(latestBlock as any as Block);
+            }
+        } catch (err) {
+            const filterTokens = _.keys(this._filterCallbacks);
+            _.each(filterTokens, filterToken => {
+                this._unsubscribe(filterToken, err);
+            });
         }
     }
 }