Merge branch 'development' into orderWatcher

* development:
  0.23.0
  Update CHANGELOG
  Fix amounts in tests one last time. Now that we updated the testRPC snapshot, this should no longer be mismatched between CI and locally
  Update testRPC snapshot used by CircleCi
  Push unsubscribe to the base class rather than super
  Check for null rather than undefined
  Removed nits
  Test case was error then unsubscribe
  Clean up subscription state.
  Fix unhandled promise rejection error on subscriptions

# Conflicts:
#	src/types.ts
#	test/exchange_wrapper_test.ts
#	test/token_wrapper_test.ts

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);
+            });
         }
     }
 }