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

4 files changed, 38 insertions, 40 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);
+            });
         }
     }
 }
diff --git a/src/contract_wrappers/exchange_wrapper.ts b/src/contract_wrappers/exchange_wrapper.ts
index f68e80b5d..fe0c5bc00 100644
--- a/src/contract_wrappers/exchange_wrapper.ts
+++ b/src/contract_wrappers/exchange_wrapper.ts
@@ -48,7 +48,6 @@ const SHOULD_VALIDATE_BY_DEFAULT = true;
  */
 export class ExchangeWrapper extends ContractWrapper {
     private _exchangeContractIfExists?: ExchangeContract;
-    private _activeSubscriptions: string[];
     private _orderValidationUtils: OrderValidationUtils;
     private _tokenWrapper: TokenWrapper;
     private _exchangeContractErrCodesToMsg = {
@@ -83,7 +82,6 @@ export class ExchangeWrapper extends ContractWrapper {
         super(web3Wrapper, abiDecoder);
         this._tokenWrapper = tokenWrapper;
         this._orderValidationUtils = new OrderValidationUtils(tokenWrapper, this);
-        this._activeSubscriptions = [];
         this._contractAddressIfExists = contractAddressIfExists;
     }
     /**
@@ -665,7 +663,6 @@ export class ExchangeWrapper extends ContractWrapper {
         const subscriptionToken = this._subscribe<ArgsType>(
             exchangeContractAddress, eventName, indexFilterValues, artifacts.ExchangeArtifact.abi, callback,
         );
-        this._activeSubscriptions.push(subscriptionToken);
         return subscriptionToken;
     }
     /**
@@ -673,7 +670,6 @@ export class ExchangeWrapper extends ContractWrapper {
      * @param   subscriptionToken Subscription token returned by `subscribe()`
      */
     public unsubscribe(subscriptionToken: string): void {
-        _.pull(this._activeSubscriptions, subscriptionToken);
         this._unsubscribe(subscriptionToken);
     }
     /**
@@ -824,13 +820,6 @@ export class ExchangeWrapper extends ContractWrapper {
         const ZRXtokenAddress = await exchangeInstance.ZRX_TOKEN_CONTRACT.callAsync();
         return ZRXtokenAddress;
     }
-    /**
-     * Cancels all existing subscriptions
-     */
-    public unsubscribeAll(): void {
-        _.forEach(this._activeSubscriptions, this._unsubscribe.bind(this));
-        this._activeSubscriptions = [];
-    }
     private async _invalidateContractInstancesAsync(): Promise<void> {
         this.unsubscribeAll();
         delete this._exchangeContractIfExists;
diff --git a/src/contract_wrappers/token_wrapper.ts b/src/contract_wrappers/token_wrapper.ts
index dd87ebc9f..614ac19d4 100644
--- a/src/contract_wrappers/token_wrapper.ts
+++ b/src/contract_wrappers/token_wrapper.ts
@@ -29,13 +29,11 @@ const ALLOWANCE_TO_ZERO_GAS_AMOUNT = 47275;
 export class TokenWrapper extends ContractWrapper {
     public UNLIMITED_ALLOWANCE_IN_BASE_UNITS = constants.UNLIMITED_ALLOWANCE_IN_BASE_UNITS;
     private _tokenContractsByAddress: {[address: string]: TokenContract};
-    private _activeSubscriptions: string[];
     private _tokenTransferProxyContractAddressFetcher: () => Promise<string>;
     constructor(web3Wrapper: Web3Wrapper, abiDecoder: AbiDecoder,
                 tokenTransferProxyContractAddressFetcher: () => Promise<string>) {
         super(web3Wrapper, abiDecoder);
         this._tokenContractsByAddress = {};
-        this._activeSubscriptions = [];
         this._tokenTransferProxyContractAddressFetcher = tokenTransferProxyContractAddressFetcher;
     }
     /**
@@ -262,7 +260,6 @@ export class TokenWrapper extends ContractWrapper {
         const subscriptionToken = this._subscribe<ArgsType>(
             tokenAddress, eventName, indexFilterValues, artifacts.TokenArtifact.abi, callback,
         );
-        this._activeSubscriptions.push(subscriptionToken);
         return subscriptionToken;
     }
     /**
@@ -270,7 +267,6 @@ export class TokenWrapper extends ContractWrapper {
      * @param   subscriptionToken Subscription token returned by `subscribe()`
      */
     public unsubscribe(subscriptionToken: string): void {
-        _.pull(this._activeSubscriptions, subscriptionToken);
         this._unsubscribe(subscriptionToken);
     }
     /**
@@ -294,13 +290,6 @@ export class TokenWrapper extends ContractWrapper {
         );
         return logs;
     }
-    /**
-     * Cancels all existing subscriptions
-     */
-    public unsubscribeAll(): void {
-        _.forEach(this._activeSubscriptions, this._unsubscribe.bind(this));
-        this._activeSubscriptions = [];
-    }
     private _invalidateContractInstancesAsync(): void {
         this.unsubscribeAll();
         this._tokenContractsByAddress = {};
diff --git a/src/types.ts b/src/types.ts
index a366fc31e..09b611019 100644
--- a/src/types.ts
+++ b/src/types.ts
@@ -42,8 +42,8 @@ export type OrderValues = [BigNumber, BigNumber, BigNumber,
 export type LogEvent = Web3.LogEntryEvent;
 export type DecodedLogEvent<ArgsType> = Web3.DecodedLogEntryEvent<ArgsType>;
 
-export type EventCallbackAsync<ArgsType> = (log: DecodedLogEvent<ArgsType>) => Promise<void>;
-export type EventCallbackSync<ArgsType> = (log: DecodedLogEvent<ArgsType>) => void;
+export type EventCallbackAsync<ArgsType> = (err: null|Error, log?: DecodedLogEvent<ArgsType>) => Promise<void>;
+export type EventCallbackSync<ArgsType> = (err: null|Error, log?: DecodedLogEvent<ArgsType>) => void;
 export type EventCallback<ArgsType> = EventCallbackSync<ArgsType>|EventCallbackAsync<ArgsType>;
 
 export type EventWatcherCallbackSync = (log: LogEvent) => void;