Implement subscriptions based on ethereumjs-blockstream

1 files changed, 34 insertions, 48 deletions

diff --git a/src/contract_wrappers/exchange_wrapper.ts b/src/contract_wrappers/exchange_wrapper.ts
index 32eaa590c..e5f190864 100644
--- a/src/contract_wrappers/exchange_wrapper.ts
+++ b/src/contract_wrappers/exchange_wrapper.ts
@@ -16,11 +16,8 @@ import {
     SignedOrder,
     ContractEvent,
     ExchangeEvents,
-    ContractEventEmitter,
     SubscriptionOpts,
     IndexedFilterValues,
-    CreateContractEvent,
-    ContractEventObj,
     OrderCancellationRequest,
     OrderFillRequest,
     LogErrorContractEventArgs,
@@ -31,10 +28,10 @@ import {
     ValidateOrderFillableOpts,
     OrderTransactionOpts,
     RawLog,
+    EventCallback,
 } from '../types';
 import {assert} from '../utils/assert';
 import {utils} from '../utils/utils';
-import {eventUtils} from '../utils/event_utils';
 import {OrderValidationUtils} from '../utils/order_validation_utils';
 import {ContractWrapper} from './contract_wrapper';
 import {constants} from '../utils/constants';
@@ -51,7 +48,7 @@ const SHOULD_VALIDATE_BY_DEFAULT = true;
  */
 export class ExchangeWrapper extends ContractWrapper {
     private _exchangeContractIfExists?: ExchangeContract;
-    private _exchangeLogEventEmitters: ContractEventEmitter[];
+    private _activeSubscriptions: string[];
     private _orderValidationUtils: OrderValidationUtils;
     private _tokenWrapper: TokenWrapper;
     private _exchangeContractErrCodesToMsg = {
@@ -86,7 +83,7 @@ export class ExchangeWrapper extends ContractWrapper {
         super(web3Wrapper, abiDecoder);
         this._tokenWrapper = tokenWrapper;
         this._orderValidationUtils = new OrderValidationUtils(tokenWrapper, this);
-        this._exchangeLogEventEmitters = [];
+        this._activeSubscriptions = [];
         this._contractAddressIfExists = contractAddressIfExists;
     }
     /**
@@ -622,41 +619,32 @@ export class ExchangeWrapper extends ContractWrapper {
         return txHash;
     }
     /**
-     * Subscribe to an event type emitted by the Exchange smart contract
-     * @param   eventName               The exchange contract event you would like to subscribe to.
-     * @param   subscriptionOpts        Subscriptions options that let you configure the subscription.
-     * @param   indexFilterValues       An object where the keys are indexed args returned by the event and
-     *                                  the value is the value you are interested in. E.g `{maker: aUserAddressHex}`
-     * @param   exchangeContractAddress The hex encoded address of the Exchange contract to call.
-     * @return                      ContractEventEmitter object
+     * Subscribe to an event type emitted by the Exchange contract.
+     * @param   tokenAddress        The hex encoded address where the ERC20 token is deployed.
+     * @param   eventName           The exchange contract event you would like to subscribe to.
+     * @param   indexFilterValues   An object where the keys are indexed args returned by the event and
+     *                              the value is the value you are interested in. E.g `{maker: aUserAddressHex}`
+     * @param   callback            Callback that gets called when a log is added/removed
+     * @return ContractEventEmitter object
      */
-    public async subscribeAsync(eventName: ExchangeEvents, subscriptionOpts: SubscriptionOpts,
-                                indexFilterValues: IndexedFilterValues, exchangeContractAddress: string):
-                                Promise<ContractEventEmitter> {
-        assert.isETHAddressHex('exchangeContractAddress', exchangeContractAddress);
+    public async subscribeAsync(eventName: ExchangeEvents, indexFilterValues: IndexedFilterValues,
+                                callback: EventCallback): Promise<string> {
         assert.doesBelongToStringEnum('eventName', eventName, ExchangeEvents);
-        assert.doesConformToSchema('subscriptionOpts', subscriptionOpts, schemas.subscriptionOptsSchema);
         assert.doesConformToSchema('indexFilterValues', indexFilterValues, schemas.indexFilterValuesSchema);
-        const exchangeContract = await this._getExchangeContractAsync();
-        let createLogEvent: CreateContractEvent;
-        switch (eventName) {
-            case ExchangeEvents.LogFill:
-                createLogEvent = exchangeContract.LogFill;
-                break;
-            case ExchangeEvents.LogError:
-                createLogEvent = exchangeContract.LogError;
-                break;
-            case ExchangeEvents.LogCancel:
-                createLogEvent = exchangeContract.LogCancel;
-                break;
-            default:
-                throw utils.spawnSwitchErr('ExchangeEvents', eventName);
-        }
-
-        const logEventObj: ContractEventObj = createLogEvent(indexFilterValues, subscriptionOpts);
-        const eventEmitter = eventUtils.wrapEventEmitter(logEventObj);
-        this._exchangeLogEventEmitters.push(eventEmitter);
-        return eventEmitter;
+        const exchangeContractAddress = await this.getContractAddressAsync();
+        const subscriptionToken = this._subscribe(
+            exchangeContractAddress, eventName, indexFilterValues, artifacts.ExchangeArtifact.abi, callback,
+        );
+        this._activeSubscriptions.push(subscriptionToken);
+        return subscriptionToken;
+    }
+    /**
+     * Cancel a subscription
+     * @param   subscriptionToken Subscription token returned by `subscribe()`
+     */
+    public unsubscribe(subscriptionToken: string): void {
+        _.pull(this._activeSubscriptions, subscriptionToken);
+        this._unsubscribe(subscriptionToken);
     }
     /**
      * Gets historical logs without creating a subscription
@@ -678,15 +666,6 @@ export class ExchangeWrapper extends ContractWrapper {
         return logs;
     }
     /**
-     * Stops watching for all exchange events
-     */
-    public async stopWatchingAllEventsAsync(): Promise<void> {
-        const stopWatchingPromises = _.map(this._exchangeLogEventEmitters,
-                                           logEventObj => logEventObj.stopWatchingAsync());
-        await Promise.all(stopWatchingPromises);
-        this._exchangeLogEventEmitters = [];
-    }
-    /**
      * Retrieves the Ethereum address of the Exchange contract deployed on the network
      * that the user-passed web3 provider is connected to.
      * @returns The Ethereum address of the Exchange contract being used.
@@ -809,8 +788,15 @@ 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> {
-        await this.stopWatchingAllEventsAsync();
+        this.unsubscribeAll();
         delete this._exchangeContractIfExists;
     }
     private async _isValidSignatureUsingContractCallAsync(dataHex: string, ecSignature: ECSignature,