5 files changed, 120 insertions, 29 deletions
diff --git a/src/mempool/order_state_watcher.ts b/src/mempool/order_state_watcher.ts
new file mode 100644
index 000000000..89f84647d
--- /dev/null
+++ b/src/mempool/order_state_watcher.ts
@@ -0,0 +1,70 @@
+import * as _ from 'lodash';
+import {schemas} from '0x-json-schemas';
+import {ZeroEx} from '../';
+import {EventWatcher} from './event_watcher';
+import {assert} from '../utils/assert';
+import {artifacts} from '../artifacts';
+import {AbiDecoder} from '../utils/abi_decoder';
+import {orderWatcherConfigSchema} from '../schemas/order_watcher_config_schema';
+import {
+    LogEvent,
+    SignedOrder,
+    Web3Provider,
+    LogWithDecodedArgs,
+    OrderWatcherConfig,
+    OnOrderStateChangeCallback,
+} from '../types';
+import {Web3Wrapper} from '../web3_wrapper';
+
+export class OrderStateWatcher {
+    private _orders = new Map<string, SignedOrder>();
+    private _web3Wrapper: Web3Wrapper;
+    private _config: OrderWatcherConfig;
+    private _callback?: OnOrderStateChangeCallback;
+    private _eventWatcher?: EventWatcher;
+    private _abiDecoder: AbiDecoder;
+    constructor(provider: Web3Provider, config?: OrderWatcherConfig) {
+        assert.isWeb3Provider('provider', provider);
+        if (!_.isUndefined(config)) {
+            assert.doesConformToSchema('config', config, orderWatcherConfigSchema);
+        }
+        this._web3Wrapper = new Web3Wrapper(provider);
+        this._config = config || {};
+        const artifactJSONs = _.values(artifacts);
+        const abiArrays = _.map(artifactJSONs, artifact => artifact.abi);
+        this._abiDecoder = new AbiDecoder(abiArrays);
+    }
+    public addOrder(signedOrder: SignedOrder): void {
+        assert.doesConformToSchema('signedOrder', signedOrder, schemas.signedOrderSchema);
+        const orderHash = ZeroEx.getOrderHashHex(signedOrder);
+        this._orders.set(orderHash, signedOrder);
+    }
+    public removeOrder(signedOrder: SignedOrder): void {
+        assert.doesConformToSchema('signedOrder', signedOrder, schemas.signedOrderSchema);
+        const orderHash = ZeroEx.getOrderHashHex(signedOrder);
+        this._orders.delete(orderHash);
+    }
+    public subscribe(callback: OnOrderStateChangeCallback): void {
+        assert.isFunction('callback', callback);
+        this._callback = callback;
+        this._eventWatcher = new EventWatcher(
+            this._web3Wrapper, this._config.mempoolPollingIntervalMs,
+        );
+        this._eventWatcher.subscribe(this._onMempoolEventCallbackAsync.bind(this));
+    }
+    public unsubscribe(): void {
+        delete this._callback;
+        if (!_.isUndefined(this._eventWatcher)) {
+            this._eventWatcher.unsubscribe();
+        }
+    }
+    private async _onMempoolEventCallbackAsync(log: LogEvent): Promise<void> {
+        const maybeDecodedLog = this._abiDecoder.tryToDecodeLogOrNoop(log);
+        if (!_.isUndefined((maybeDecodedLog as LogWithDecodedArgs<any>).event)) {
+            await this._revalidateOrdersAsync();
+        }
+    }
+    private async _revalidateOrdersAsync(): Promise<void> {
+        _.noop();
+    }
+}
diff --git a/src/mempool/order_watcher.ts b/src/mempool/order_watcher.ts
deleted file mode 100644
index b2c8598e7..000000000
--- a/src/mempool/order_watcher.ts
+++ /dev/null
@@ -1,28 +0,0 @@
-import * as _ from 'lodash';
-import {ZeroEx} from '../';
-import {assert} from '../utils/assert';
-import {Web3Provider, SignedOrder, OnOrderFillabilityStateChangeCallback} from '../types';
-import {Web3Wrapper} from '../web3_wrapper';
-
-export class OrderWatcher {
-    private _orders = new Map<string, SignedOrder>();
-    private _web3Wrapper: Web3Wrapper;
-    constructor(provider: Web3Provider) {
-        assert.isWeb3Provider('provider', provider);
-        this._web3Wrapper = new Web3Wrapper(provider);
-    }
-    public addOrder(signedOrder: SignedOrder): void {
-        const orderHash = ZeroEx.getOrderHashHex(signedOrder);
-        this._orders.set(orderHash, signedOrder);
-    }
-    public removeOrder(signedOrder: SignedOrder): void {
-        const orderHash = ZeroEx.getOrderHashHex(signedOrder);
-        this._orders.delete(orderHash);
-    }
-    public subscribe(callback: OnOrderFillabilityStateChangeCallback): void {
-        //
-    }
-    public unsubscribe(): void {
-        //
-    }
-}
diff --git a/src/schemas/order_watcher_config_schema.ts b/src/schemas/order_watcher_config_schema.ts
new file mode 100644
index 000000000..a88d2ecfd
--- /dev/null
+++ b/src/schemas/order_watcher_config_schema.ts
@@ -0,0 +1,7 @@
+export const orderWatcherConfigSchema = {
+    id: '/OrderWatcherConfig',
+    properties: {
+        mempoolPollingIntervalMs: {$ref: '/Number'},
+    },
+    type: 'object',
+};
diff --git a/src/types.ts b/src/types.ts
index 766bf01b3..52b22516b 100644
--- a/src/types.ts
+++ b/src/types.ts
@@ -497,6 +497,6 @@ export interface OrderStateInvalid {
     error: ExchangeContractErrs;
 }
 
-export type OnOrderFillabilityStateChangeCallback = (
+export type OnOrderStateChangeCallback = (
     orderState: OrderStateValid|OrderStateInvalid,
 ) => void;
diff --git a/test/order_watcher_test.ts b/test/order_watcher_test.ts
new file mode 100644
index 000000000..f273a1d84
--- /dev/null
+++ b/test/order_watcher_test.ts
@@ -0,0 +1,42 @@
+import 'mocha';
+import * as chai from 'chai';
+import * as _ from 'lodash';
+import * as Sinon from 'sinon';
+import * as Web3 from 'web3';
+import BigNumber from 'bignumber.js';
+import {chaiSetup} from './utils/chai_setup';
+import {web3Factory} from './utils/web3_factory';
+import {Web3Wrapper} from '../src/web3_wrapper';
+import {OrderStateWatcher} from '../src/mempool/order_state_watcher';
+import {
+    ZeroEx,
+    LogEvent,
+    DecodedLogEvent,
+} from '../src';
+import {DoneCallback} from '../src/types';
+
+chaiSetup.configure();
+const expect = chai.expect;
+
+describe('EventWatcher', () => {
+    let web3: Web3;
+    let stubs: Sinon.SinonStub[] = [];
+    let orderStateWatcher: OrderStateWatcher;
+    before(async () => {
+        web3 = web3Factory.create();
+        const mempoolPollingIntervalMs = 10;
+        const orderStateWatcherConfig = {
+            mempoolPollingIntervalMs,
+        };
+        orderStateWatcher = new OrderStateWatcher(web3.currentProvider, orderStateWatcherConfig);
+    });
+    afterEach(() => {
+        // clean up any stubs after the test has completed
+        _.each(stubs, s => s.restore());
+        stubs = [];
+        orderStateWatcher.unsubscribe();
+    });
+    it.only('', (done: DoneCallback) => {
+        orderStateWatcher.subscribe(console.log);
+    }).timeout(1000000000000);
+});