Async event enrichment

Signed-off-by: Matthew Whitehead <matthew1001@gmail.com>

src/eventstream-proxy/eventstream-proxy.base.ts

@@ -127,6 +127,7 @@ export abstract class EventStreamProxyBase extends WebSocketEventsBase {
 
   private async processEvents(batch: EventBatch) {
     const messages: WebSocketMessage[] = [];
+    const eventHandlers: Promise<WebSocketMessage | undefined>[] = [];
     for (const event of batch.events) {
       this.logger.log(`Proxying event: ${JSON.stringify(event)}`);
       const subName = await this.getSubscriptionName(newContext(), event.subId);
@@ -136,17 +137,25 @@ export abstract class EventStreamProxyBase extends WebSocketEventsBase {
       }
 
       for (const listener of this.eventListeners) {
-        try {
-          await listener.onEvent(subName, event, (msg: WebSocketMessage | undefined) => {
-            if (msg !== undefined) {
-              messages.push(msg);
-            }
-          });
-        } catch (err) {
-          this.logger.error(`Error processing event: ${err}`);
-        }
+        // Some events require enrichment that could involve a call to the blockchain,
+        // so we don't want to do those synchronously. Create a promise for each onEvent()
+        // handler and when they're all complete we'll create the batch message
+        eventHandlers.push(Promise.resolve(listener.onEvent(subName, event)));
       }
     }
+
+    // Now we need to await the promises in order so the messages stay in order
+    for (const nextProm of eventHandlers) {
+      await nextProm
+        .then((msg: WebSocketMessage | undefined) => {
+          if (msg !== undefined) {
+            messages.push(msg);
+          }
+        })
+        .catch(err => {
+          this.logger.error(`Error processing event: ${err}`);
+        });
+    }
     const message: WebSocketMessageWithId = {
       id: uuidv4(),
       event: 'batch',

src/eventstream-proxy/eventstream-proxy.interfaces.ts

@@ -26,7 +26,7 @@ export interface ConnectionListener {
 }
 
 export interface EventListener {
-  onEvent: (subName: string, event: Event, process: EventProcessor) => void | Promise<void>;
+  onEvent: (subName: string, event: Event) => undefined | Promise<WebSocketMessage | undefined>;
 }
 
 export interface WebSocketMessageWithId extends WebSocketMessage {

src/tokens/tokens.listener.ts

@@ -62,21 +62,17 @@ export class TokenListener implements EventListener {
     private blockchain: BlockchainConnectorService,
   ) {}
 
-  async onEvent(subName: string, event: Event, process: EventProcessor) {
+  async onEvent(subName: string, event: Event) {
     const signature = this.trimEventSignature(event.signature);
     switch (signature) {
       case tokenCreateEventSignature:
-        process(await this.transformTokenPoolCreationEvent(event));
-        break;
+        return await this.transformTokenPoolCreationEvent(event);
       case transferEventSignature:
-        process(await this.transformTransferEvent(subName, event));
-        break;
+        return await this.transformTransferEvent(subName, event);
       case approvalEventSignature:
-        process(this.transformApprovalEvent(subName, event));
-        break;
+        return this.transformApprovalEvent(subName, event);
       case approvalForAllEventSignature:
-        process(this.transformApprovalForAllEvent(subName, event));
-        break;
+        return this.transformApprovalForAllEvent(subName, event);
       default:
         this.logger.error(`Unknown event signature: ${event.signature}`);
     }