Move event watcher to util (#262)

packages/util/src/events.ts

@@ -23,7 +23,6 @@ export const BlockProgressEvent = 'block-progress-event';
 export class EventWatcher {
   _ethClient: EthClient
   _indexer: IndexerInterface
-  _subscription?: ZenObservable.Subscription
   _pubsub: PubSub
   _jobQueue: JobQueue
 
@@ -42,11 +41,22 @@ export class EventWatcher {
     return this._pubsub.asyncIterator([BlockProgressEvent]);
   }
 
-  async stop (): Promise<void> {
-    if (this._subscription) {
-      log('Stopped watching upstream blocks');
-      this._subscription.unsubscribe();
-    }
+  async start (): Promise<void> {
+    await this.initBlockProcessingOnCompleteHandler();
+    await this.initEventProcessingOnCompleteHandler();
+    this.startBlockProcessing();
+  }
+
+  async initBlockProcessingOnCompleteHandler (): Promise<void> {
+    this._jobQueue.onComplete(QUEUE_BLOCK_PROCESSING, async (job) => {
+      await this.blockProcessingCompleteHandler(job);
+    });
+  }
+
+  async initEventProcessingOnCompleteHandler (): Promise<void> {
+    await this._jobQueue.onComplete(QUEUE_EVENT_PROCESSING, async (job) => {
+      await this.eventProcessingCompleteHandler(job);
+    });
   }
 
   async startBlockProcessing (): Promise<void> {

packages/util/src/fill.ts

@@ -5,17 +5,18 @@
 import debug from 'debug';
 
 import { JobQueue } from './job-queue';
-import { EventWatcherInterface, IndexerInterface } from './types';
+import { IndexerInterface } from './types';
 import { wait } from './misc';
 import { processBlockByNumber } from './common';
 import { DEFAULT_PREFETCH_BATCH_SIZE } from './constants';
+import { EventWatcher } from './events';
 
 const log = debug('vulcanize:fill');
 
 export const fillBlocks = async (
   jobQueue: JobQueue,
   indexer: IndexerInterface,
-  eventWatcher: EventWatcherInterface,
+  eventWatcher: EventWatcher,
   blockDelayInMilliSecs: number,
   argv: {
     startBlock: number,

packages/util/src/types.ts

@@ -140,13 +140,6 @@ export interface IndexerInterface {
   getResultEvent (event: EventInterface): any
 }
 
-export interface EventWatcherInterface {
-  start (): Promise<void>
-  getBlockProgressEventIterator (): AsyncIterator<any>
-  initBlockProcessingOnCompleteHandler (): Promise<void>
-  initEventProcessingOnCompleteHandler (): Promise<void>
-}
-
 export interface DatabaseInterface {
   _conn: Connection;
   readonly baseDatabase: Database