Implement subscriptions based on ethereumjs-blockstream

1 files changed, 35 insertions, 53 deletions

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;