1 files changed, 40 insertions, 10 deletions
diff --git a/packages/pipeline/src/data_sources/contract-wrappers/exchange_events.ts b/packages/pipeline/src/data_sources/contract-wrappers/exchange_events.ts
index e25c6a731..d09f59535 100644
--- a/packages/pipeline/src/data_sources/contract-wrappers/exchange_events.ts
+++ b/packages/pipeline/src/data_sources/contract-wrappers/exchange_events.ts
@@ -1,4 +1,12 @@
-import { ContractWrappers, ExchangeEvents, ExchangeFillEventArgs, ExchangeWrapper } from '@0x/contract-wrappers';
+import {
+    ContractWrappers,
+    ExchangeCancelEventArgs,
+    ExchangeCancelUpToEventArgs,
+    ExchangeEventArgs,
+    ExchangeEvents,
+    ExchangeFillEventArgs,
+    ExchangeWrapper,
+} from '@0x/contract-wrappers';
 import { Web3ProviderEngine } from '@0x/subproviders';
 import { Web3Wrapper } from '@0x/web3-wrapper';
 import { LogWithDecodedArgs } from 'ethereum-types';
@@ -16,20 +24,41 @@ export class ExchangeEventsSource {
         this._exchangeWrapper = contractWrappers.exchange;
     }
 
-    // TODO(albrow): Get Cancel and CancelUpTo events.
-
     public async getFillEventsAsync(
-        fromBlock: number = EXCHANGE_START_BLOCK,
+        fromBlock?: number,
         toBlock?: number,
     ): Promise<Array<LogWithDecodedArgs<ExchangeFillEventArgs>>> {
+        return this._getEventsAsync<ExchangeFillEventArgs>(ExchangeEvents.Fill, fromBlock, toBlock);
+    }
+
+    public async getCancelEventsAsync(
+        fromBlock?: number,
+        toBlock?: number,
+    ): Promise<Array<LogWithDecodedArgs<ExchangeCancelEventArgs>>> {
+        return this._getEventsAsync<ExchangeCancelEventArgs>(ExchangeEvents.Cancel, fromBlock, toBlock);
+    }
+
+    public async getCancelUpToEventsAsync(
+        fromBlock?: number,
+        toBlock?: number,
+    ): Promise<Array<LogWithDecodedArgs<ExchangeCancelUpToEventArgs>>> {
+        return this._getEventsAsync<ExchangeCancelUpToEventArgs>(ExchangeEvents.CancelUpTo, fromBlock, toBlock);
+    }
+
+    private async _getEventsAsync<ArgsType extends ExchangeEventArgs>(
+        eventName: ExchangeEvents,
+        fromBlock: number = EXCHANGE_START_BLOCK,
+        toBlock?: number,
+    ): Promise<Array<LogWithDecodedArgs<ArgsType>>> {
         const calculatedToBlock =
             toBlock === undefined
                 ? (await this._web3Wrapper.getBlockNumberAsync()) - BLOCK_FINALITY_THRESHOLD
                 : toBlock;
-        let events: Array<LogWithDecodedArgs<ExchangeFillEventArgs>> = [];
+        let events: Array<LogWithDecodedArgs<ArgsType>> = [];
         for (let currFromBlock = fromBlock; currFromBlock <= calculatedToBlock; currFromBlock += NUM_BLOCKS_PER_QUERY) {
             events = events.concat(
-                await this._getFillEventsForRangeAsync(
+                await this._getEventsForRangeAsync<ArgsType>(
+                    eventName,
                     currFromBlock,
                     Math.min(currFromBlock + NUM_BLOCKS_PER_QUERY - 1, calculatedToBlock),
                 ),
@@ -38,12 +67,13 @@ export class ExchangeEventsSource {
         return events;
     }
 
-    private async _getFillEventsForRangeAsync(
+    private async _getEventsForRangeAsync<ArgsType extends ExchangeEventArgs>(
+        eventName: ExchangeEvents,
         fromBlock: number,
         toBlock: number,
-    ): Promise<Array<LogWithDecodedArgs<ExchangeFillEventArgs>>> {
-        return this._exchangeWrapper.getLogsAsync<ExchangeFillEventArgs>(
-            ExchangeEvents.Fill,
+    ): Promise<Array<LogWithDecodedArgs<ArgsType>>> {
+        return this._exchangeWrapper.getLogsAsync<ArgsType>(
+            eventName,
             {
                 fromBlock,
                 toBlock,