From 4ae9482d506ccdf028fcd39bbaf652bbcbef52e8 Mon Sep 17 00:00:00 2001 From: Jacob Evans Date: Sat, 11 Nov 2017 11:01:59 -0500 Subject: Clean up subscription state. In the case of an exception, keep the state correct between contract wrapper, exchange wrapper and token wrapper. --- src/contract_wrappers/contract_wrapper.ts | 6 ++ src/contract_wrappers/exchange_wrapper.ts | 18 +++--- src/contract_wrappers/token_wrapper.ts | 7 +-- test/subscription_test.ts | 93 +++++++++++++++++++++++++++++++ 4 files changed, 107 insertions(+), 17 deletions(-) create mode 100644 test/subscription_test.ts diff --git a/src/contract_wrappers/contract_wrapper.ts b/src/contract_wrappers/contract_wrapper.ts index a02465982..5bdc66b3b 100644 --- a/src/contract_wrappers/contract_wrapper.ts +++ b/src/contract_wrappers/contract_wrapper.ts @@ -50,6 +50,12 @@ export class ContractWrapper { this._filterCallbacks[filterToken] = callback; return filterToken; } + protected 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); diff --git a/src/contract_wrappers/exchange_wrapper.ts b/src/contract_wrappers/exchange_wrapper.ts index b8704e72c..4fd2b23a8 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( exchangeContractAddress, eventName, indexFilterValues, artifacts.ExchangeArtifact.abi, callback, ); - this._activeSubscriptions.push(subscriptionToken); return subscriptionToken; } /** @@ -674,9 +671,14 @@ export class ExchangeWrapper extends ContractWrapper { * @param subscriptionToken Subscription token returned by `subscribe()` */ public unsubscribe(subscriptionToken: string): void { - _.pull(this._activeSubscriptions, subscriptionToken); this._unsubscribe(subscriptionToken); } + /** + * Cancels all existing subscriptions + */ + public unsubscribeAll(): void { + super.unsubscribeAll(); + } /** * Gets historical logs without creating a subscription * @param eventName The exchange contract event you would like to subscribe to. @@ -825,13 +827,7 @@ 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 { this.unsubscribeAll(); delete this._exchangeContractIfExists; diff --git a/src/contract_wrappers/token_wrapper.ts b/src/contract_wrappers/token_wrapper.ts index 5d6d61cef..fbe7354dc 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; constructor(web3Wrapper: Web3Wrapper, abiDecoder: AbiDecoder, tokenTransferProxyContractAddressFetcher: () => Promise) { super(web3Wrapper, abiDecoder); this._tokenContractsByAddress = {}; - this._activeSubscriptions = []; this._tokenTransferProxyContractAddressFetcher = tokenTransferProxyContractAddressFetcher; } /** @@ -262,7 +260,6 @@ export class TokenWrapper extends ContractWrapper { const subscriptionToken = this._subscribe( 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); } /** @@ -298,8 +294,7 @@ export class TokenWrapper extends ContractWrapper { * Cancels all existing subscriptions */ public unsubscribeAll(): void { - _.forEach(this._activeSubscriptions, this._unsubscribe.bind(this)); - this._activeSubscriptions = []; + super.unsubscribeAll(); } private _invalidateContractInstancesAsync(): void { this.unsubscribeAll(); diff --git a/test/subscription_test.ts b/test/subscription_test.ts new file mode 100644 index 000000000..d328084af --- /dev/null +++ b/test/subscription_test.ts @@ -0,0 +1,93 @@ +import 'mocha'; +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, + SubscriptionOpts, + TokenEvents, + ContractEvent, + TransferContractEventArgs, + ApprovalContractEventArgs, + TokenContractEventArgs, + LogWithDecodedArgs, + 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); + before(() => { + const token = tokens[0]; + tokenAddress = token.address; + }); + afterEach(() => { + zeroEx.token.unsubscribeAll(); + }); + it('Should receive the Error when an error occurs', (done: DoneCallback) => { + (async () => { + const callback = (err: Error, logEvent: LogEvent) => { + expect(err).to.not.be.undefined(); + expect(logEvent).to.be.undefined(); + done(); + }; + 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 multiple times', (done: DoneCallback) => { + (async () => { + const callback = (err: Error, logEvent: LogEvent) => { }; + zeroEx.token.subscribe( + tokenAddress, TokenEvents.Approval, indexFilterValues, callback); + await zeroEx.token.setAllowanceAsync(tokenAddress, coinbase, addressWithoutFunds, allowanceAmount); + zeroEx.token.unsubscribeAll(); + zeroEx.token.unsubscribeAll(); + done(); + })().catch(done); + }); + }) + }) \ No newline at end of file -- cgit