Make some callbacks failable and add error handling

6 files changed, 61 insertions, 38 deletions

diff --git a/packages/0x.js/src/0x.ts b/packages/0x.js/src/0x.ts
index 244b77a85..e1b0ef08e 100644
--- a/packages/0x.js/src/0x.ts
+++ b/packages/0x.js/src/0x.ts
@@ -302,26 +302,33 @@ export class ZeroEx {
 
         const txReceiptPromise = new Promise(
             (resolve: (receipt: TransactionReceiptWithDecodedLogs) => void, reject) => {
-                const intervalId = intervalUtils.setAsyncExcludingInterval(async () => {
-                    if (timeoutExceeded) {
-                        intervalUtils.clearAsyncExcludingInterval(intervalId);
-                        return reject(ZeroExError.TransactionMiningTimeout);
-                    }
+                const intervalId = intervalUtils.setAsyncExcludingInterval(
+                    async () => {
+                        if (timeoutExceeded) {
+                            intervalUtils.clearAsyncExcludingInterval(intervalId);
+                            return reject(ZeroExError.TransactionMiningTimeout);
+                        }
 
-                    const transactionReceipt = await this._web3Wrapper.getTransactionReceiptAsync(txHash);
-                    if (!_.isNull(transactionReceipt)) {
+                        const transactionReceipt = await this._web3Wrapper.getTransactionReceiptAsync(txHash);
+                        if (!_.isNull(transactionReceipt)) {
+                            intervalUtils.clearAsyncExcludingInterval(intervalId);
+                            const logsWithDecodedArgs = _.map(
+                                transactionReceipt.logs,
+                                this._abiDecoder.tryToDecodeLogOrNoop.bind(this._abiDecoder),
+                            );
+                            const transactionReceiptWithDecodedLogArgs: TransactionReceiptWithDecodedLogs = {
+                                ...transactionReceipt,
+                                logs: logsWithDecodedArgs,
+                            };
+                            resolve(transactionReceiptWithDecodedLogArgs);
+                        }
+                    },
+                    pollingIntervalMs,
+                    (err: Error) => {
                         intervalUtils.clearAsyncExcludingInterval(intervalId);
-                        const logsWithDecodedArgs = _.map(
-                            transactionReceipt.logs,
-                            this._abiDecoder.tryToDecodeLogOrNoop.bind(this._abiDecoder),
-                        );
-                        const transactionReceiptWithDecodedLogArgs: TransactionReceiptWithDecodedLogs = {
-                            ...transactionReceipt,
-                            logs: logsWithDecodedArgs,
-                        };
-                        resolve(transactionReceiptWithDecodedLogArgs);
-                    }
-                }, pollingIntervalMs);
+                        reject(err);
+                    },
+                );
             },
         );
 
diff --git a/packages/0x.js/src/contract_wrappers/contract_wrapper.ts b/packages/0x.js/src/contract_wrappers/contract_wrapper.ts
index 9c4e5dfd3..27551c01d 100644
--- a/packages/0x.js/src/contract_wrappers/contract_wrapper.ts
+++ b/packages/0x.js/src/contract_wrappers/contract_wrapper.ts
@@ -167,6 +167,7 @@ export class ContractWrapper {
         this._blockAndLogStreamInterval = intervalUtils.setAsyncExcludingInterval(
             this._reconcileBlockAsync.bind(this),
             constants.DEFAULT_BLOCK_POLLING_INTERVAL,
+            this._onReconcileBlockError.bind(this),
         );
         let isRemoved = false;
         this._onLogAddedSubscriptionToken = this._blockAndLogStreamerIfExists.subscribeToOnLogAdded(
@@ -177,6 +178,12 @@ export class ContractWrapper {
             this._onLogStateChanged.bind(this, isRemoved),
         );
     }
+    private _onReconcileBlockError(err: Error): void {
+        const filterTokens = _.keys(this._filterCallbacks);
+        _.each(filterTokens, filterToken => {
+            this._unsubscribe(filterToken, err);
+        });
+    }
     private _setNetworkId(networkId: number): void {
         this._networkId = networkId;
     }
@@ -190,18 +197,11 @@ export class ContractWrapper {
         delete this._blockAndLogStreamerIfExists;
     }
     private async _reconcileBlockAsync(): Promise<void> {
-        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._blockAndLogStreamerIfExists)) {
-                // If we clear the interval while fetching the block - this._blockAndLogStreamer will be undefined
-                await this._blockAndLogStreamerIfExists.reconcileNewBlock((latestBlock as any) as Block);
-            }
-        } catch (err) {
-            const filterTokens = _.keys(this._filterCallbacks);
-            _.each(filterTokens, filterToken => {
-                this._unsubscribe(filterToken, err);
-            });
+        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._blockAndLogStreamerIfExists)) {
+            // If we clear the interval while fetching the block - this._blockAndLogStreamer will be undefined
+            await this._blockAndLogStreamerIfExists.reconcileNewBlock((latestBlock as any) as Block);
         }
     }
 }
diff --git a/packages/0x.js/src/order_watcher/event_watcher.ts b/packages/0x.js/src/order_watcher/event_watcher.ts
index fc0b9264c..43a60957b 100644
--- a/packages/0x.js/src/order_watcher/event_watcher.ts
+++ b/packages/0x.js/src/order_watcher/event_watcher.ts
@@ -36,6 +36,10 @@ export class EventWatcher {
         this._intervalIdIfExists = intervalUtils.setAsyncExcludingInterval(
             this._pollForBlockchainEventsAsync.bind(this, callback),
             this._pollingIntervalMs,
+            (err: Error) => {
+                this.unsubscribe();
+                callback(err);
+            },
         );
     }
     public unsubscribe(): void {
@@ -78,7 +82,7 @@ export class EventWatcher {
                 ...log,
             };
             if (!_.isUndefined(this._intervalIdIfExists)) {
-                callback(logEvent);
+                callback(null, logEvent);
             }
         }
     }
diff --git a/packages/0x.js/src/order_watcher/expiration_watcher.ts b/packages/0x.js/src/order_watcher/expiration_watcher.ts
index e7d085fc8..00b62162d 100644
--- a/packages/0x.js/src/order_watcher/expiration_watcher.ts
+++ b/packages/0x.js/src/order_watcher/expiration_watcher.ts
@@ -30,16 +30,17 @@ export class ExpirationWatcher {
         if (!_.isUndefined(this._orderExpirationCheckingIntervalIdIfExists)) {
             throw new Error(ZeroExError.SubscriptionAlreadyPresent);
         }
-        this._orderExpirationCheckingIntervalIdIfExists = intervalUtils.setAsyncExcludingInterval(
+        this._orderExpirationCheckingIntervalIdIfExists = intervalUtils.setInterval(
             this._pruneExpiredOrders.bind(this, callback),
             this._orderExpirationCheckingIntervalMs,
+            _.noop, // _pruneExpiredOrders never throws
         );
     }
     public unsubscribe(): void {
         if (_.isUndefined(this._orderExpirationCheckingIntervalIdIfExists)) {
             throw new Error(ZeroExError.SubscriptionNotFound);
         }
-        intervalUtils.clearAsyncExcludingInterval(this._orderExpirationCheckingIntervalIdIfExists);
+        intervalUtils.clearInterval(this._orderExpirationCheckingIntervalIdIfExists);
         delete this._orderExpirationCheckingIntervalIdIfExists;
     }
     public addOrder(orderHash: string, expirationUnixTimestampMs: BigNumber): void {
diff --git a/packages/0x.js/src/order_watcher/order_state_watcher.ts b/packages/0x.js/src/order_watcher/order_state_watcher.ts
index 9d7a733d8..3543480f8 100644
--- a/packages/0x.js/src/order_watcher/order_state_watcher.ts
+++ b/packages/0x.js/src/order_watcher/order_state_watcher.ts
@@ -155,6 +155,10 @@ export class OrderStateWatcher {
         this._cleanupJobIntervalIdIfExists = intervalUtils.setAsyncExcludingInterval(
             this._cleanupAsync.bind(this),
             this._cleanupJobInterval,
+            (err: Error) => {
+                this.unsubscribe();
+                callback(err);
+            },
         );
     }
     /**
@@ -207,11 +211,18 @@ export class OrderStateWatcher {
         if (!_.isUndefined(this._orderByOrderHash[orderHash])) {
             this.removeOrder(orderHash);
             if (!_.isUndefined(this._callbackIfExists)) {
-                this._callbackIfExists(orderState);
+                this._callbackIfExists(null, orderState);
             }
         }
     }
-    private async _onEventWatcherCallbackAsync(log: LogEvent): Promise<void> {
+    private async _onEventWatcherCallbackAsync(err: Error | null, logIfExists?: LogEvent): Promise<void> {
+        if (!_.isNull(err)) {
+            if (!_.isUndefined(this._callbackIfExists)) {
+                this._callbackIfExists(err);
+                this.unsubscribe();
+            }
+        }
+        const log = logIfExists as LogEvent;
         const maybeDecodedLog = this._abiDecoder.tryToDecodeLogOrNoop(log);
         const isLogDecoded = !_.isUndefined((maybeDecodedLog as LogWithDecodedArgs<any>).event);
         if (!isLogDecoded) {
@@ -332,7 +343,7 @@ export class OrderStateWatcher {
             } else {
                 this._orderStateByOrderHashCache[orderHash] = orderState;
             }
-            this._callbackIfExists(orderState);
+            this._callbackIfExists(null, orderState);
         }
     }
     private _addToDependentOrderHashes(signedOrder: SignedOrder, orderHash: string): void {
diff --git a/packages/0x.js/src/types.ts b/packages/0x.js/src/types.ts
index 44f8aa223..2decd92ba 100644
--- a/packages/0x.js/src/types.ts
+++ b/packages/0x.js/src/types.ts
@@ -51,7 +51,7 @@ export interface DecodedLogEvent<ArgsType> {
 }
 
 export type EventCallback<ArgsType> = (err: null | Error, log?: DecodedLogEvent<ArgsType>) => void;
-export type EventWatcherCallback = (log: LogEvent) => void;
+export type EventWatcherCallback = (err: null | Error, log?: LogEvent) => void;
 
 export enum SolidityTypes {
     Address = 'address',
@@ -406,5 +406,5 @@ export interface OrderStateInvalid {
 
 export type OrderState = OrderStateValid | OrderStateInvalid;
 
-export type OnOrderStateChangeCallback = (orderState: OrderState) => void;
+export type OnOrderStateChangeCallback = (err: Error | null, orderState?: OrderState) => void;
 // tslint:disable:max-file-line-count