Merge pull request #134 from kaleido-io/async-event-enrichment

Async event enrichment

Author: matthew1001

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,15 +137,22 @@ 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) {
+      try {
+        const msg = await nextProm;
+        if (msg !== undefined) {
+          messages.push(msg);
         }
+      } catch (err) {
+        this.logger.error(`Error processing event: ${err}`);
       }
     }
     const message: WebSocketMessageWithId = {

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/blockchain.service.ts

@@ -1,4 +1,4 @@
-// Copyright © 2022 Kaleido, Inc.
+// Copyright © 2023 Kaleido, Inc.
 //
 // SPDX-License-Identifier: Apache-2.0
 //
@@ -103,7 +103,7 @@ export class BlockchainConnectorService {
   // Check if retry condition matches the err that's been hit
   private matchesRetryCondition(err: any): boolean {
     return (
-      this.retryConfiguration.retryCondition != '' &&
+      this.retryConfiguration.retryCondition !== '' &&
       `${err}`.match(this.retryConfiguration.retryCondition) !== null
     );
   }
@@ -127,8 +127,8 @@ export class BlockchainConnectorService {
     let retries = 0;
     for (
       ;
-      this.retryConfiguration.retriesMax == -1 || retries <= this.retryConfiguration.retriesMax;
-      this.retryConfiguration.retriesMax == -1 || retries++ // Don't inc 'retries' if 'retriesMax' if set to -1 (infinite retries)
+      this.retryConfiguration.retriesMax === -1 || retries <= this.retryConfiguration.retriesMax;
+      this.retryConfiguration.retriesMax === -1 || retries++ // Don't inc 'retries' if 'retriesMax' if set to -1 (infinite retries)
     ) {
       try {
         return await blockchainFunction();

src/tokens/tokens.listener.ts

@@ -1,4 +1,4 @@
-// Copyright © 2022 Kaleido, Inc.
+// Copyright © 2023 Kaleido, Inc.
 //
 // SPDX-License-Identifier: Apache-2.0
 //
@@ -16,7 +16,7 @@
 
 import { Logger } from '@nestjs/common';
 import { Event } from '../event-stream/event-stream.interfaces';
-import { EventListener, EventProcessor } from '../eventstream-proxy/eventstream-proxy.interfaces';
+import { EventListener } from '../eventstream-proxy/eventstream-proxy.interfaces';
 import { WebSocketMessage } from '../websocket-events/websocket-events.base';
 import { Context, newContext } from '../request-context/request-context.decorator';
 import {
@@ -41,7 +41,6 @@ import {
   unpackSubscriptionName,
   validatePoolLocator,
 } from './tokens.util';
-import { TokensService } from './tokens.service';
 import { AbiMapperService } from './abimapper.service';
 import { BlockchainConnectorService } from './blockchain.service';
 import { TokenURI as ERC721URI } from './erc721';
@@ -58,21 +57,17 @@ export class TokenListener implements EventListener {
 
   constructor(private mapper: AbiMapperService, 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(subName, event));
-        break;
+        return this.transformTokenPoolCreationEvent(subName, event);
       case transferEventSignature:
-        process(await this.transformTransferEvent(subName, event));
-        break;
+        return 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}`);
     }

src/tokens/tokens.service.spec.ts

@@ -1,4 +1,4 @@
-// Copyright © 2022 Kaleido, Inc.
+// Copyright © 2023 Kaleido, Inc.
 //
 // SPDX-License-Identifier: Apache-2.0
 //
@@ -240,7 +240,7 @@ describe('TokensService', () => {
       .useValue(eventstream)
       .compile();
 
-    let blockchainRetryCfg: RetryConfiguration = {
+    const blockchainRetryCfg: RetryConfiguration = {
       retryBackOffFactor: 2,
       retryBackOffLimit: 500,
       retryBackOffInitial: 50,

test/app.e2e-context.ts

@@ -69,7 +69,7 @@ export class TestContext {
     this.app.use(requestIDMiddleware);
     await this.app.init();
 
-    let blockchainRetryCfg: RetryConfiguration = {
+    const blockchainRetryCfg: RetryConfiguration = {
       retryBackOffFactor: 2,
       retryBackOffLimit: 500,
       retryBackOffInitial: 50,