Merge pull request #209 from 0xProject/fixUnhandledPromiseBug

Fix unhandled promise rejection error on subscriptions

8 files changed, 154 insertions, 58 deletions

diff --git a/circle.yml b/circle.yml
index 5a8d340f0..3dc00bd03 100644
--- a/circle.yml
+++ b/circle.yml
@@ -2,7 +2,7 @@ machine:
   node:
     version: 6.5.0
   environment:
-      CONTRACTS_COMMIT_HASH: '35053f9'
+      CONTRACTS_COMMIT_HASH: '78fe8dd'
       PATH: "${PATH}:${HOME}/${CIRCLE_PROJECT_REPONAME}/node_modules/.bin"
 
 dependencies:
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 b8704e72c..b608e6931 100644
--- a/src/contract_wrappers/exchange_wrapper.ts
+++ b/src/contract_wrappers/exchange_wrapper.ts
@@ -49,7 +49,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 = {
@@ -84,7 +83,6 @@ export class ExchangeWrapper extends ContractWrapper {
         super(web3Wrapper, abiDecoder);
         this._tokenWrapper = tokenWrapper;
         this._orderValidationUtils = new OrderValidationUtils(tokenWrapper, this);
-        this._activeSubscriptions = [];
         this._contractAddressIfExists = contractAddressIfExists;
     }
     /**
@@ -666,7 +664,6 @@ export class ExchangeWrapper extends ContractWrapper {
         const subscriptionToken = this._subscribe<ArgsType>(
             exchangeContractAddress, eventName, indexFilterValues, artifacts.ExchangeArtifact.abi, callback,
         );
-        this._activeSubscriptions.push(subscriptionToken);
         return subscriptionToken;
     }
     /**
@@ -674,7 +671,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);
     }
     /**
@@ -825,13 +821,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 5d6d61cef..081113964 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 a9eac56d8..af3c36736 100644
--- a/src/types.ts
+++ b/src/types.ts
@@ -41,8 +41,8 @@ export type OrderValues = [BigNumber, BigNumber, BigNumber,
 export interface LogEvent<ArgsType> extends LogWithDecodedArgs<ArgsType> {
     removed: boolean;
 }
-export type EventCallbackAsync<ArgsType> = (log: LogEvent<ArgsType>) => Promise<void>;
-export type EventCallbackSync<ArgsType> = (log: LogEvent<ArgsType>) => void;
+export type EventCallbackAsync<ArgsType> = (err: null|Error, log?: LogEvent<ArgsType>) => Promise<void>;
+export type EventCallbackSync<ArgsType> = (err: null|Error, log?: LogEvent<ArgsType>) => void;
 export type EventCallback<ArgsType> = EventCallbackSync<ArgsType>|EventCallbackAsync<ArgsType>;
 export interface ExchangeContract extends Web3.ContractInstance {
     isValidSignature: {
diff --git a/test/exchange_wrapper_test.ts b/test/exchange_wrapper_test.ts
index 7c76499d5..654f626c6 100644
--- a/test/exchange_wrapper_test.ts
+++ b/test/exchange_wrapper_test.ts
@@ -304,11 +304,11 @@ describe('ExchangeWrapper', () => {
                     orderFillBatch = [
                         {
                             signedOrder,
-                            takerTokenFillAmount: takerTokenFillAmount,
+                            takerTokenFillAmount,
                         },
                         {
                             signedOrder: anotherSignedOrder,
-                            takerTokenFillAmount: takerTokenFillAmount,
+                            takerTokenFillAmount,
                         },
                     ];
                 });
@@ -647,7 +647,7 @@ describe('ExchangeWrapper', () => {
         // Source: https://github.com/mochajs/mocha/issues/2407
         it('Should receive the LogFill event when an order is filled', (done: DoneCallback) => {
             (async () => {
-                const callback = (logEvent: LogEvent<LogFillContractEventArgs>) => {
+                const callback = (err: Error, logEvent: LogEvent<LogFillContractEventArgs>) => {
                     expect(logEvent.event).to.be.equal(ExchangeEvents.LogFill);
                     done();
                 };
@@ -655,13 +655,14 @@ describe('ExchangeWrapper', () => {
                     ExchangeEvents.LogFill, indexFilterValues, callback,
                 );
                 await zeroEx.exchange.fillOrderAsync(
-                    signedOrder, takerTokenFillAmountInBaseUnits, shouldThrowOnInsufficientBalanceOrAllowance, takerAddress,
+                    signedOrder, takerTokenFillAmountInBaseUnits, shouldThrowOnInsufficientBalanceOrAllowance,
+                    takerAddress,
                 );
             })().catch(done);
         });
         it('Should receive the LogCancel event when an order is cancelled', (done: DoneCallback) => {
             (async () => {
-                const callback = (logEvent: LogEvent<LogCancelContractEventArgs>) => {
+                const callback = (err: Error, logEvent: LogEvent<LogCancelContractEventArgs>) => {
                     expect(logEvent.event).to.be.equal(ExchangeEvents.LogCancel);
                     done();
                 };
@@ -673,7 +674,7 @@ describe('ExchangeWrapper', () => {
         });
         it('Outstanding subscriptions are cancelled when zeroEx.setProviderAsync called', (done: DoneCallback) => {
             (async () => {
-                const callbackNeverToBeCalled = (logEvent: LogEvent<LogFillContractEventArgs>) => {
+                const callbackNeverToBeCalled = (err: Error, logEvent: LogEvent<LogFillContractEventArgs>) => {
                     done(new Error('Expected this subscription to have been cancelled'));
                 };
                 await zeroEx.exchange.subscribeAsync(
@@ -683,7 +684,7 @@ describe('ExchangeWrapper', () => {
                 const newProvider = web3Factory.getRpcProvider();
                 await zeroEx.setProviderAsync(newProvider);
 
-                const callback = (logEvent: LogEvent<LogFillContractEventArgs>) => {
+                const callback = (err: Error, logEvent: LogEvent<LogFillContractEventArgs>) => {
                     expect(logEvent.event).to.be.equal(ExchangeEvents.LogFill);
                     done();
                 };
@@ -691,13 +692,14 @@ describe('ExchangeWrapper', () => {
                     ExchangeEvents.LogFill, indexFilterValues, callback,
                 );
                 await zeroEx.exchange.fillOrderAsync(
-                    signedOrder, takerTokenFillAmountInBaseUnits, shouldThrowOnInsufficientBalanceOrAllowance, takerAddress,
+                    signedOrder, takerTokenFillAmountInBaseUnits, shouldThrowOnInsufficientBalanceOrAllowance,
+                    takerAddress,
                 );
             })().catch(done);
         });
         it('Should cancel subscription when unsubscribe called', (done: DoneCallback) => {
             (async () => {
-                const callbackNeverToBeCalled = (logEvent: LogEvent<LogFillContractEventArgs>) => {
+                const callbackNeverToBeCalled = (err: Error, logEvent: LogEvent<LogFillContractEventArgs>) => {
                     done(new Error('Expected this subscription to have been cancelled'));
                 };
                 const subscriptionToken = await zeroEx.exchange.subscribeAsync(
@@ -705,7 +707,8 @@ describe('ExchangeWrapper', () => {
                 );
                 zeroEx.exchange.unsubscribe(subscriptionToken);
                 await zeroEx.exchange.fillOrderAsync(
-                    signedOrder, takerTokenFillAmountInBaseUnits, shouldThrowOnInsufficientBalanceOrAllowance, takerAddress,
+                    signedOrder, takerTokenFillAmountInBaseUnits, shouldThrowOnInsufficientBalanceOrAllowance,
+                    takerAddress,
                 );
                 done();
             })().catch(done);
diff --git a/test/subscription_test.ts b/test/subscription_test.ts
new file mode 100644
index 000000000..985fdc1d6
--- /dev/null
+++ b/test/subscription_test.ts
@@ -0,0 +1,95 @@
+import 'mocha';
+import * as _ from 'lodash';
+import * as chai from 'chai';
+import * as Sinon from 'sinon';
+import {chaiSetup} from './utils/chai_setup';
+import * as Web3 from 'web3';
+import BigNumber from 'bignumber.js';
+import promisify = require('es6-promisify');
+import {web3Factory} from './utils/web3_factory';
+import {
+    ZeroEx,
+    ZeroExError,
+    Token,
+    ApprovalContractEventArgs,
+    TokenEvents,
+    LogEvent,
+} from '../src';
+import {BlockchainLifecycle} from './utils/blockchain_lifecycle';
+import {TokenUtils} from './utils/token_utils';
+import {DoneCallback, BlockParamLiteral} from '../src/types';
+
+chaiSetup.configure();
+const expect = chai.expect;
+const blockchainLifecycle = new BlockchainLifecycle();
+
+describe('SubscriptionTest', () => {
+    let web3: Web3;
+    let zeroEx: ZeroEx;
+    let userAddresses: string[];
+    let tokens: Token[];
+    let tokenUtils: TokenUtils;
+    let coinbase: string;
+    let addressWithoutFunds: string;
+    before(async () => {
+        web3 = web3Factory.create();
+        zeroEx = new ZeroEx(web3.currentProvider);
+        userAddresses = await zeroEx.getAvailableAddressesAsync();
+        tokens = await zeroEx.tokenRegistry.getTokensAsync();
+        tokenUtils = new TokenUtils(tokens);
+        coinbase = userAddresses[0];
+        addressWithoutFunds = userAddresses[1];
+    });
+    beforeEach(async () => {
+        await blockchainLifecycle.startAsync();
+    });
+    afterEach(async () => {
+        await blockchainLifecycle.revertAsync();
+    });
+    describe('#subscribe', () => {
+        const indexFilterValues = {};
+        const shouldThrowOnInsufficientBalanceOrAllowance = true;
+        let tokenAddress: string;
+        const transferAmount = new BigNumber(42);
+        const allowanceAmount = new BigNumber(42);
+        let stubs: Sinon.SinonStub[] = [];
+        before(() => {
+            const token = tokens[0];
+            tokenAddress = token.address;
+        });
+        afterEach(() => {
+            zeroEx.token.unsubscribeAll();
+            _.each(stubs, s => s.restore());
+            stubs = [];
+        });
+        it('Should receive the Error when an error occurs', (done: DoneCallback) => {
+            (async () => {
+                const callback = (err: Error, logEvent: LogEvent<ApprovalContractEventArgs>) => {
+                    expect(err).to.not.be.null();
+                    expect(logEvent).to.be.undefined();
+                    done();
+                };
+                stubs = [
+                  Sinon.stub((zeroEx as any)._web3Wrapper, 'getBlockAsync')
+                   .throws("JSON RPC error")
+                ]
+                zeroEx.token.subscribe(
+                    tokenAddress, TokenEvents.Approval, indexFilterValues, callback);
+                await zeroEx.token.setAllowanceAsync(tokenAddress, coinbase, addressWithoutFunds, allowanceAmount);
+            })().catch(done);
+         });
+        it('Should allow unsubscribeAll to be called successfully after an error', (done: DoneCallback) => {
+            (async () => {
+                const callback = (err: Error, logEvent: LogEvent<ApprovalContractEventArgs>) => { };
+                zeroEx.token.subscribe(
+                    tokenAddress, TokenEvents.Approval, indexFilterValues, callback);
+                stubs = [
+                  Sinon.stub((zeroEx as any)._web3Wrapper, 'getBlockAsync')
+                   .throws("JSON RPC error")
+                ]
+                zeroEx.token.unsubscribeAll();
+                done();
+            })().catch(done);
+         });
+    })
+  })
\ No newline at end of file
diff --git a/test/token_wrapper_test.ts b/test/token_wrapper_test.ts
index b35fa43f9..d9d0fb9ad 100644
--- a/test/token_wrapper_test.ts
+++ b/test/token_wrapper_test.ts
@@ -161,7 +161,7 @@ describe('TokenWrapper', () => {
                 const token = tokens[0];
                 const ownerAddress = coinbase;
                 const balance = await zeroEx.token.getBalanceAsync(token.address, ownerAddress);
-                const expectedBalance = new BigNumber('100000000000000000000000000');
+                const expectedBalance = new BigNumber('1000000000000000000000000000');
                 return expect(balance).to.be.bignumber.equal(expectedBalance);
             });
             it('should throw a CONTRACT_DOES_NOT_EXIST error for a non-existent token contract', async () => {
@@ -189,7 +189,7 @@ describe('TokenWrapper', () => {
                     const token = tokens[0];
                     const ownerAddress = coinbase;
                     const balance = await zeroExWithoutAccounts.token.getBalanceAsync(token.address, ownerAddress);
-                    const expectedBalance = new BigNumber('100000000000000000000000000');
+                    const expectedBalance = new BigNumber('1000000000000000000000000000');
                     return expect(balance).to.be.bignumber.equal(expectedBalance);
             });
         });
@@ -358,7 +358,7 @@ describe('TokenWrapper', () => {
         // Source: https://github.com/mochajs/mocha/issues/2407
         it('Should receive the Transfer event when tokens are transfered', (done: DoneCallback) => {
             (async () => {
-                const callback = (logEvent: LogEvent<TransferContractEventArgs>) => {
+                const callback = (err: Error, logEvent: LogEvent<TransferContractEventArgs>) => {
                     expect(logEvent).to.not.be.undefined();
                     const args = logEvent.args;
                     expect(args._from).to.be.equal(coinbase);
@@ -373,7 +373,7 @@ describe('TokenWrapper', () => {
         });
         it('Should receive the Approval event when allowance is being set', (done: DoneCallback) => {
             (async () => {
-                const callback = (logEvent: LogEvent<ApprovalContractEventArgs>) => {
+                const callback = (err: Error, logEvent: LogEvent<ApprovalContractEventArgs>) => {
                     expect(logEvent).to.not.be.undefined();
                     const args = logEvent.args;
                     expect(args._owner).to.be.equal(coinbase);
@@ -388,13 +388,13 @@ describe('TokenWrapper', () => {
         });
         it('Outstanding subscriptions are cancelled when zeroEx.setProviderAsync called', (done: DoneCallback) => {
             (async () => {
-                const callbackNeverToBeCalled = (logEvent: LogEvent<TransferContractEventArgs>) => {
+                const callbackNeverToBeCalled = (err: Error, logEvent: LogEvent<TransferContractEventArgs>) => {
                     done(new Error('Expected this subscription to have been cancelled'));
                 };
                 zeroEx.token.subscribe(
                     tokenAddress, TokenEvents.Transfer, indexFilterValues, callbackNeverToBeCalled,
                 );
-                const callbackToBeCalled = (logEvent: LogEvent<TransferContractEventArgs>) => {
+                const callbackToBeCalled = (err: Error, logEvent: LogEvent<TransferContractEventArgs>) => {
                     done();
                 };
                 const newProvider = web3Factory.getRpcProvider();
@@ -407,7 +407,7 @@ describe('TokenWrapper', () => {
         });
         it('Should cancel subscription when unsubscribe called', (done: DoneCallback) => {
             (async () => {
-                const callbackNeverToBeCalled = (logEvent: LogEvent<TokenContractEventArgs>) => {
+                const callbackNeverToBeCalled = (err: Error, logEvent: LogEvent<TokenContractEventArgs>) => {
                     done(new Error('Expected this subscription to have been cancelled'));
                 };
                 const subscriptionToken = zeroEx.token.subscribe(