crowetic/protocol
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #209 from 0xProject/fixUnhandledPromiseBug

Browse Source 
Fix unhandled promise rejection error on subscriptions
This commit is contained in:
Brandon Millman
2017-11-12 14:06:39 -05:00
committed by GitHub
parent 76b66872d8 dae5a063cf
commit 719c51f61a
8 changed files with 154 additions and 58 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
circle.yml
View File

@@ -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:

52
src/contract_wrappers/contract_wrapper.ts
View File

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

11
src/contract_wrappers/exchange_wrapper.ts
View File

@@ -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;

11
src/contract_wrappers/token_wrapper.ts
View File

@@ -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 = {};

4
src/types.ts
View File

@@ -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: {

23
test/exchange_wrapper_test.ts
View File

@@ -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);

95
test/subscription_test.ts Normal file
View File

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

14
test/token_wrapper_test.ts
View File

@@ -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(