Update blockstream to latest version, propogate errors to top-level callers

1 files changed, 11 insertions, 2 deletions

diff --git a/packages/contract-wrappers/src/contract_wrappers/contract_wrapper.ts b/packages/contract-wrappers/src/contract_wrappers/contract_wrapper.ts
index 04f69bc3d..9cc661080 100644
--- a/packages/contract-wrappers/src/contract_wrappers/contract_wrapper.ts
+++ b/packages/contract-wrappers/src/contract_wrappers/contract_wrapper.ts
@@ -9,7 +9,7 @@ import {
 } from '@0xproject/types';
 import { intervalUtils } from '@0xproject/utils';
 import { Web3Wrapper } from '@0xproject/web3-wrapper';
-import { Block, BlockAndLogStreamer } from 'ethereumjs-blockstream';
+import { Block, BlockAndLogStreamer, Log } from 'ethereumjs-blockstream';
 import * as _ from 'lodash';
 
 import {
@@ -39,7 +39,7 @@ export abstract class ContractWrapper {
     public abstract abi: ContractAbi;
     protected _web3Wrapper: Web3Wrapper;
     protected _networkId: number;
-    private _blockAndLogStreamerIfExists?: BlockAndLogStreamer;
+    private _blockAndLogStreamerIfExists: BlockAndLogStreamer<Block, Log> | undefined;
     private _blockAndLogStreamIntervalIfExists?: NodeJS.Timer;
     private _filters: { [filterToken: string]: FilterObject };
     private _filterCallbacks: {
@@ -163,6 +163,7 @@ export abstract class ContractWrapper {
         this._blockAndLogStreamerIfExists = new BlockAndLogStreamer(
             this._web3Wrapper.getBlockAsync.bind(this._web3Wrapper),
             this._web3Wrapper.getLogsAsync.bind(this._web3Wrapper),
+            this._onBlockAndLogStreamerError.bind(this),
         );
         const catchAllLogFilter = {};
         this._blockAndLogStreamerIfExists.addLogFilter(catchAllLogFilter);
@@ -180,6 +181,14 @@ export abstract class ContractWrapper {
             this._onLogStateChanged.bind(this, isRemoved),
         );
     }
+    private _onBlockAndLogStreamerError(err: Error): void {
+        // Propogate all Blockstream subscriber errors to all
+        // top-level subscriptions
+        const filterCallbacks = _.values(this._filterCallbacks);
+        _.each(filterCallbacks, filterCallback => {
+            filterCallback(err);
+        });
+    }
     private _onReconcileBlockError(err: Error): void {
         const filterTokens = _.keys(this._filterCallbacks);
         _.each(filterTokens, filterToken => {