Add numConfirmations arg so that caller can decide on numConfirmations at which they want to watch orders

4 files changed, 27 insertions, 13 deletions

diff --git a/src/mempool/event_watcher.ts b/src/mempool/event_watcher.ts
index cb8921cfd..3f40606e7 100644
--- a/src/mempool/event_watcher.ts
+++ b/src/mempool/event_watcher.ts
@@ -19,10 +19,10 @@ export class EventWatcher {
                                   DEFAULT_MEMPOOL_POLLING_INTERVAL :
                                   pollingIntervalMs;
     }
-    public subscribe(callback: MempoolEventCallback): void {
+    public subscribe(callback: MempoolEventCallback, numConfirmations: number): void {
         this._callbackAsync = callback;
         this._intervalId = intervalUtils.setAsyncExcludingInterval(
-            this._pollForMempoolEventsAsync.bind(this), this._pollingIntervalMs,
+            this._pollForMempoolEventsAsync.bind(this, numConfirmations), this._pollingIntervalMs,
         );
     }
     public unsubscribe(): void {
@@ -30,8 +30,8 @@ export class EventWatcher {
         this._lastMempoolEvents = [];
         intervalUtils.clearAsyncExcludingInterval(this._intervalId);
     }
-    private async _pollForMempoolEventsAsync(): Promise<void> {
-        const pendingEvents = await this._getMempoolEventsAsync();
+    private async _pollForMempoolEventsAsync(numConfirmations: number): Promise<void> {
+        const pendingEvents = await this._getMempoolEventsAsync(numConfirmations);
         if (pendingEvents.length === 0) {
             // HACK: Sometimes when node rebuilds the pending block we get back the empty result.
             // We don't want to emit a lot of removal events and bring them back after a couple of miliseconds,
@@ -46,11 +46,19 @@ export class EventWatcher {
         await this._emitDifferencesAsync(newEvents, isRemoved);
         this._lastMempoolEvents = pendingEvents;
     }
-    private async _getMempoolEventsAsync(): Promise<Web3.LogEntry[]> {
-        // TODO: Allow users to listen to any number of confirmations deep, not just mempool
+    private async _getMempoolEventsAsync(numConfirmations: number): Promise<Web3.LogEntry[]> {
+        let fromBlock: BlockParamLiteral|number;
+        let toBlock: BlockParamLiteral|number;
+        if (numConfirmations === 0) {
+            fromBlock = BlockParamLiteral.Pending;
+            toBlock = BlockParamLiteral.Pending;
+        } else {
+            toBlock = await this._web3Wrapper.getBlockNumberAsync();
+            fromBlock = toBlock - numConfirmations;
+        }
         const mempoolFilter = {
-            fromBlock: BlockParamLiteral.Pending,
-            toBlock: BlockParamLiteral.Pending,
+            fromBlock,
+            toBlock,
         };
         const pendingEvents = await this._web3Wrapper.getLogsAsync(mempoolFilter);
         return pendingEvents;
diff --git a/src/mempool/order_state_watcher.ts b/src/mempool/order_state_watcher.ts
index dc24d5b4a..05d77d15f 100644
--- a/src/mempool/order_state_watcher.ts
+++ b/src/mempool/order_state_watcher.ts
@@ -64,9 +64,10 @@ export class OrderStateWatcher {
         // We currently do not remove the maker/makerToken keys from the mapping when all orderHashes removed
     }
     public subscribe(callback: OnOrderStateChangeCallback): void {
+    public subscribe(callback: OnOrderStateChangeCallback, numConfirmations: number): void {
         assert.isFunction('callback', callback);
         this._callbackAsync = callback;
-        this._eventWatcher.subscribe(this._onMempoolEventCallbackAsync.bind(this));
+        this._eventWatcher.subscribe(this._onMempoolEventCallbackAsync.bind(this), numConfirmations);
     }
     public unsubscribe(): void {
         delete this._callbackAsync;
diff --git a/src/web3_wrapper.ts b/src/web3_wrapper.ts
index 01d572654..d03c102b2 100644
--- a/src/web3_wrapper.ts
+++ b/src/web3_wrapper.ts
@@ -100,6 +100,10 @@ export class Web3Wrapper {
         const signData = await promisify(this.web3.eth.sign)(address, message);
         return signData;
     }
+    public async getBlockNumberAsync(): Promise<number> {
+        const blockNumber = await promisify(this.web3.eth.getBlockNumber)();
+        return blockNumber;
+    }
     public async getBlockAsync(blockParam: string|Web3.BlockParam): Promise<Web3.BlockWithoutTransactionData> {
         const block = await promisify(this.web3.eth.getBlock)(blockParam);
         return block;
diff --git a/test/order_state_watcher_test.ts b/test/order_state_watcher_test.ts
index f1b027c40..9d0e1a625 100644
--- a/test/order_state_watcher_test.ts
+++ b/test/order_state_watcher_test.ts
@@ -41,6 +41,7 @@ describe('OrderStateWatcher', () => {
     let web3Wrapper: Web3Wrapper;
     let signedOrder: SignedOrder;
     const fillableAmount = new BigNumber(5);
+    const numConfirmations = 0;
     before(async () => {
         web3 = web3Factory.create();
         zeroEx = new ZeroEx(web3.currentProvider);
@@ -73,7 +74,7 @@ describe('OrderStateWatcher', () => {
                 expect(invalidOrderState.error).to.be.equal(ExchangeContractErrs.InsufficientMakerAllowance);
                 done();
             };
-            zeroEx.orderStateWatcher.subscribe(callback);
+            zeroEx.orderStateWatcher.subscribe(callback, numConfirmations);
             await zeroEx.token.setProxyAllowanceAsync(makerToken.address, maker, new BigNumber(0));
         })().catch(done);
     });
@@ -91,7 +92,7 @@ describe('OrderStateWatcher', () => {
                 expect(invalidOrderState.error).to.be.equal(ExchangeContractErrs.InsufficientMakerBalance);
                 done();
             };
-            zeroEx.orderStateWatcher.subscribe(callback);
+            zeroEx.orderStateWatcher.subscribe(callback, numConfirmations);
             const anyRecipient = taker;
             const makerBalance = await zeroEx.token.getBalanceAsync(makerToken.address, maker);
             await zeroEx.token.transferAsync(makerToken.address, maker, anyRecipient, makerBalance);
@@ -116,7 +117,7 @@ describe('OrderStateWatcher', () => {
                     done();
                 }
             };
-            zeroEx.orderStateWatcher.subscribe(callback);
+            zeroEx.orderStateWatcher.subscribe(callback, numConfirmations);
 
             const shouldThrowOnInsufficientBalanceOrAllowance = true;
             await zeroEx.exchange.fillOrderAsync(
@@ -150,7 +151,7 @@ describe('OrderStateWatcher', () => {
                     done();
                 }
             };
-            zeroEx.orderStateWatcher.subscribe(callback);
+            zeroEx.orderStateWatcher.subscribe(callback, numConfirmations);
             const shouldThrowOnInsufficientBalanceOrAllowance = true;
             await zeroEx.exchange.fillOrderAsync(
                 signedOrder, fillAmountInBaseUnits, shouldThrowOnInsufficientBalanceOrAllowance, taker,