Implement subscriptions based on ethereumjs-blockstream

2 files changed, 69 insertions, 114 deletions

diff --git a/test/exchange_wrapper_test.ts b/test/exchange_wrapper_test.ts
index 71c5713ad..51d6957d5 100644
--- a/test/exchange_wrapper_test.ts
+++ b/test/exchange_wrapper_test.ts
@@ -19,6 +19,7 @@ import {
     OrderFillRequest,
     LogFillContractEventArgs,
     OrderFillOrKillRequest,
+    LogEvent,
 } from '../src';
 import {DoneCallback} from '../src/types';
 import {FillScenarios} from './utils/fill_scenarios';
@@ -616,7 +617,7 @@ describe('ExchangeWrapper', () => {
             });
         });
     });
-    describe('#subscribeAsync', () => {
+    describe('#subscribe', () => {
         const indexFilterValues = {};
         const shouldThrowOnInsufficientBalanceOrAllowance = true;
         let makerTokenAddress: string;
@@ -626,10 +627,6 @@ describe('ExchangeWrapper', () => {
         let makerAddress: string;
         let fillableAmount: BigNumber.BigNumber;
         let signedOrder: SignedOrder;
-        const subscriptionOpts: SubscriptionOpts = {
-            fromBlock: 0,
-            toBlock: 'latest',
-        };
         const fillTakerAmountInBaseUnits = new BigNumber(1);
         const cancelTakerAmountInBaseUnits = new BigNumber(1);
         before(() => {
@@ -645,24 +642,22 @@ describe('ExchangeWrapper', () => {
             );
         });
         afterEach(async () => {
-            await zeroEx.exchange.stopWatchingAllEventsAsync();
+            zeroEx.exchange.unsubscribeAll();
         });
         // Hack: Mocha does not allow a test to be both async and have a `done` callback
-        // Since we need to await the receipt of the event in the `subscribeAsync` callback,
+        // Since we need to await the receipt of the event in the `subscribe` callback,
         // we do need both. A hack is to make the top-level a sync fn w/ a done callback and then
         // wrap the rest of the test in an async block
         // Source: https://github.com/mochajs/mocha/issues/2407
         it('Should receive the LogFill event when an order is filled', (done: DoneCallback) => {
             (async () => {
-                const zeroExEvent = await zeroEx.exchange.subscribeAsync(
-                    ExchangeEvents.LogFill, subscriptionOpts, indexFilterValues, exchangeContractAddress,
-                );
-                zeroExEvent.watch((err: Error, event: ContractEvent) => {
-                    expect(err).to.be.null();
-                    expect(event).to.not.be.undefined();
-                    expect(event.event).to.be.equal('LogFill');
+                const callback = (logEvent: LogEvent) => {
+                    expect(logEvent.event).to.be.equal(ExchangeEvents.LogFill);
                     done();
-                });
+                };
+                await zeroEx.exchange.subscribeAsync(
+                    ExchangeEvents.LogFill, indexFilterValues, callback,
+                );
                 await zeroEx.exchange.fillOrderAsync(
                     signedOrder, fillTakerAmountInBaseUnits, shouldThrowOnInsufficientBalanceOrAllowance, takerAddress,
                 );
@@ -670,75 +665,53 @@ describe('ExchangeWrapper', () => {
         });
         it('Should receive the LogCancel event when an order is cancelled', (done: DoneCallback) => {
             (async () => {
-                const zeroExEvent = await zeroEx.exchange.subscribeAsync(
-                    ExchangeEvents.LogCancel, subscriptionOpts, indexFilterValues, exchangeContractAddress,
+                const callback = (logEvent: LogEvent) => {
+                    expect(logEvent.event).to.be.equal(ExchangeEvents.LogCancel);
+                    done();
+                };
+                await zeroEx.exchange.subscribeAsync(
+                    ExchangeEvents.LogCancel, indexFilterValues, callback,
                 );
-                zeroExEvent.watch((err: Error, event: ContractEvent) => {
-                        expect(err).to.be.null();
-                        expect(event).to.not.be.undefined();
-                        expect(event.event).to.be.equal('LogCancel');
-                        done();
-                });
                 await zeroEx.exchange.cancelOrderAsync(signedOrder, cancelTakerAmountInBaseUnits);
             })().catch(done);
         });
         it('Outstanding subscriptions are cancelled when zeroEx.setProviderAsync called', (done: DoneCallback) => {
             (async () => {
-                const eventSubscriptionToBeCancelled = await zeroEx.exchange.subscribeAsync(
-                    ExchangeEvents.LogFill, subscriptionOpts, indexFilterValues, exchangeContractAddress,
-                );
-                eventSubscriptionToBeCancelled.watch((err: Error, event: ContractEvent) => {
+                const callbackNeverToBeCalled = (logEvent: LogEvent) => {
                     done(new Error('Expected this subscription to have been cancelled'));
-                });
+                };
+                await zeroEx.exchange.subscribeAsync(
+                    ExchangeEvents.LogFill, indexFilterValues, callbackNeverToBeCalled,
+                );
 
                 const newProvider = web3Factory.getRpcProvider();
                 await zeroEx.setProviderAsync(newProvider);
 
-                const eventSubscriptionToStay = await zeroEx.exchange.subscribeAsync(
-                    ExchangeEvents.LogFill, subscriptionOpts, indexFilterValues, exchangeContractAddress,
-                );
-                eventSubscriptionToStay.watch((err: Error, event: ContractEvent) => {
-                    expect(err).to.be.null();
-                    expect(event).to.not.be.undefined();
-                    expect(event.event).to.be.equal('LogFill');
+                const callback = (logEvent: LogEvent) => {
+                    expect(logEvent.event).to.be.equal(ExchangeEvents.LogFill);
                     done();
-                });
-                await zeroEx.exchange.fillOrderAsync(
-                    signedOrder, fillTakerAmountInBaseUnits, shouldThrowOnInsufficientBalanceOrAllowance, takerAddress,
-                );
-            })().catch(done);
-        });
-        it('Should stop watch for events when stopWatchingAsync called on the eventEmitter', (done: DoneCallback) => {
-            (async () => {
-                const eventSubscriptionToBeStopped = await zeroEx.exchange.subscribeAsync(
-                    ExchangeEvents.LogFill, subscriptionOpts, indexFilterValues, exchangeContractAddress,
+                };
+                await zeroEx.exchange.subscribeAsync(
+                    ExchangeEvents.LogFill, indexFilterValues, callback,
                 );
-                eventSubscriptionToBeStopped.watch((err: Error, event: ContractEvent) => {
-                    done(new Error('Expected this subscription to have been stopped'));
-                });
-                await eventSubscriptionToBeStopped.stopWatchingAsync();
                 await zeroEx.exchange.fillOrderAsync(
                     signedOrder, fillTakerAmountInBaseUnits, shouldThrowOnInsufficientBalanceOrAllowance, takerAddress,
                 );
-                done();
             })().catch(done);
         });
-        it('Should wrap all event args BigNumber instances in a newer version of BigNumber', (done: DoneCallback) => {
+        it('Should cancel subscription when unsubscribe called', (done: DoneCallback) => {
             (async () => {
-                const zeroExEvent = await zeroEx.exchange.subscribeAsync(
-                    ExchangeEvents.LogFill, subscriptionOpts, indexFilterValues, exchangeContractAddress,
+                const callbackNeverToBeCalled = (logEvent: LogEvent) => {
+                    done(new Error('Expected this subscription to have been cancelled'));
+                };
+                const subscriptionToken = await zeroEx.exchange.subscribeAsync(
+                    ExchangeEvents.LogFill, indexFilterValues, callbackNeverToBeCalled,
                 );
-                zeroExEvent.watch((err: Error, event: ContractEvent) => {
-                    const args = event.args as LogFillContractEventArgs;
-                    expect(args.filledMakerTokenAmount.isBigNumber).to.be.true();
-                    expect(args.filledTakerTokenAmount.isBigNumber).to.be.true();
-                    expect(args.paidMakerFee.isBigNumber).to.be.true();
-                    expect(args.paidTakerFee.isBigNumber).to.be.true();
-                    done();
-                });
+                zeroEx.exchange.unsubscribe(subscriptionToken);
                 await zeroEx.exchange.fillOrderAsync(
                     signedOrder, fillTakerAmountInBaseUnits, shouldThrowOnInsufficientBalanceOrAllowance, takerAddress,
                 );
+                done();
             })().catch(done);
         });
     });
diff --git a/test/token_wrapper_test.ts b/test/token_wrapper_test.ts
index da020f714..be97496e0 100644
--- a/test/token_wrapper_test.ts
+++ b/test/token_wrapper_test.ts
@@ -15,6 +15,7 @@ import {
     TransferContractEventArgs,
     ApprovalContractEventArgs,
     LogWithDecodedArgs,
+    LogEvent,
 } from '../src';
 import {BlockchainLifecycle} from './utils/blockchain_lifecycle';
 import {TokenUtils} from './utils/token_utils';
@@ -336,22 +337,18 @@ describe('TokenWrapper', () => {
             return expect(allowance).to.be.bignumber.equal(zeroEx.token.UNLIMITED_ALLOWANCE_IN_BASE_UNITS);
         });
     });
-    describe('#subscribeAsync', () => {
+    describe('#subscribe', () => {
         const indexFilterValues = {};
         const shouldThrowOnInsufficientBalanceOrAllowance = true;
         let tokenAddress: string;
-        const subscriptionOpts: SubscriptionOpts = {
-            fromBlock: 0,
-            toBlock: 'latest',
-        };
         const transferAmount = new BigNumber(42);
         const allowanceAmount = new BigNumber(42);
         before(() => {
             const token = tokens[0];
             tokenAddress = token.address;
         });
-        afterEach(async () => {
-            await zeroEx.token.stopWatchingAllEventsAsync();
+        afterEach(() => {
+            zeroEx.token.unsubscribeAll();
         });
         // Hack: Mocha does not allow a test to be both async and have a `done` callback
         // Since we need to await the receipt of the event in the `subscribeAsync` callback,
@@ -360,81 +357,66 @@ describe('TokenWrapper', () => {
         // Source: https://github.com/mochajs/mocha/issues/2407
         it('Should receive the Transfer event when tokens are transfered', (done: DoneCallback) => {
             (async () => {
-                const zeroExEvent = await zeroEx.token.subscribeAsync(
-                    tokenAddress, TokenEvents.Transfer, subscriptionOpts, indexFilterValues);
-                zeroExEvent.watch((err: Error, event: ContractEvent) => {
-                    expect(err).to.be.null();
-                    expect(event).to.not.be.undefined();
-                    const args = event.args as TransferContractEventArgs;
+                const callback = (logEvent: LogEvent) => {
+                    expect(logEvent).to.not.be.undefined();
+                    const args = logEvent.args as any as TransferContractEventArgs;
                     expect(args._from).to.be.equal(coinbase);
                     expect(args._to).to.be.equal(addressWithoutFunds);
                     expect(args._value).to.be.bignumber.equal(transferAmount);
                     done();
-                });
+                };
+                zeroEx.token.subscribe(
+                    tokenAddress, TokenEvents.Transfer, indexFilterValues, callback);
                 await zeroEx.token.transferAsync(tokenAddress, coinbase, addressWithoutFunds, transferAmount);
             })().catch(done);
         });
         it('Should receive the Approval event when allowance is being set', (done: DoneCallback) => {
             (async () => {
-                const zeroExEvent = await zeroEx.token.subscribeAsync(
-                    tokenAddress, TokenEvents.Approval, subscriptionOpts, indexFilterValues);
-                zeroExEvent.watch((err: Error, event: ContractEvent) => {
-                    expect(err).to.be.null();
-                    expect(event).to.not.be.undefined();
-                    const args = event.args as ApprovalContractEventArgs;
+                const callback = (logEvent: LogEvent) => {
+                    expect(logEvent).to.not.be.undefined();
+                    const args = logEvent.args as any as ApprovalContractEventArgs;
                     expect(args._owner).to.be.equal(coinbase);
                     expect(args._spender).to.be.equal(addressWithoutFunds);
                     expect(args._value).to.be.bignumber.equal(allowanceAmount);
                     done();
-                });
+                };
+                zeroEx.token.subscribe(
+                    tokenAddress, TokenEvents.Approval, indexFilterValues, callback);
                 await zeroEx.token.setAllowanceAsync(tokenAddress, coinbase, addressWithoutFunds, allowanceAmount);
             })().catch(done);
         });
         it('Outstanding subscriptions are cancelled when zeroEx.setProviderAsync called', (done: DoneCallback) => {
             (async () => {
-                const eventSubscriptionToBeCancelled = await zeroEx.token.subscribeAsync(
-                    tokenAddress, TokenEvents.Transfer, subscriptionOpts, indexFilterValues);
-                eventSubscriptionToBeCancelled.watch((err: Error, event: ContractEvent) => {
+                const callbackNeverToBeCalled = (logEvent: LogEvent) => {
                     done(new Error('Expected this subscription to have been cancelled'));
-                });
-
+                };
+                zeroEx.token.subscribe(
+                    tokenAddress, TokenEvents.Transfer, indexFilterValues, callbackNeverToBeCalled,
+                );
+                const callbackToBeCalled = (logEvent: LogEvent) => {
+                    done();
+                };
                 const newProvider = web3Factory.getRpcProvider();
                 await zeroEx.setProviderAsync(newProvider);
-
-                const eventSubscriptionToStay = await zeroEx.token.subscribeAsync(
-                    tokenAddress, TokenEvents.Transfer, subscriptionOpts, indexFilterValues);
-                eventSubscriptionToStay.watch((err: Error, event: ContractEvent) => {
-                    expect(err).to.be.null();
-                    expect(event).to.not.be.undefined();
-                    done();
-                });
+                zeroEx.token.subscribe(
+                    tokenAddress, TokenEvents.Transfer, indexFilterValues, callbackToBeCalled,
+                );
                 await zeroEx.token.transferAsync(tokenAddress, coinbase, addressWithoutFunds, transferAmount);
             })().catch(done);
         });
-        it('Should stop watch for events when stopWatchingAsync called on the eventEmitter', (done: DoneCallback) => {
+        it('Should cancel subscription when unsubscribe called', (done: DoneCallback) => {
             (async () => {
-                const eventSubscriptionToBeStopped = await zeroEx.token.subscribeAsync(
-                    tokenAddress, TokenEvents.Transfer, subscriptionOpts, indexFilterValues);
-                eventSubscriptionToBeStopped.watch((err: Error, event: ContractEvent) => {
-                    done(new Error('Expected this subscription to have been stopped'));
-                });
-                await eventSubscriptionToBeStopped.stopWatchingAsync();
+                const callbackNeverToBeCalled = (logEvent: LogEvent) => {
+                    done(new Error('Expected this subscription to have been cancelled'));
+                };
+                const subscriptionToken = zeroEx.token.subscribe(
+                    tokenAddress, TokenEvents.Transfer, indexFilterValues, callbackNeverToBeCalled);
+                zeroEx.token.unsubscribe(subscriptionToken);
                 await zeroEx.token.transferAsync(tokenAddress, coinbase, addressWithoutFunds, transferAmount);
                 done();
             })().catch(done);
         });
-        it('Should wrap all event args BigNumber instances in a newer version of BigNumber', (done: DoneCallback) => {
-            (async () => {
-                const zeroExEvent = await zeroEx.token.subscribeAsync(
-                    tokenAddress, TokenEvents.Transfer, subscriptionOpts, indexFilterValues);
-                zeroExEvent.watch((err: Error, event: ContractEvent) => {
-                    const args = event.args as TransferContractEventArgs;
-                    expect(args._value.isBigNumber).to.be.true();
-                    done();
-                });
-                await zeroEx.token.transferAsync(tokenAddress, coinbase, addressWithoutFunds, transferAmount);
-            })().catch(done);
-        });
+        // TODO test block reorgs
     });
     describe('#getLogsAsync', () => {
         let tokenAddress: string;