fix: add processMessagesConcurrently config for processing message co…

…ncurrently Signed-off-by: Pritam Singh <[email protected]>
openwallet-foundation · Sep 13, 2024 · e24b92f · e24b92f
1 parent de7d359
commit e24b92f
Show file tree

Hide file tree

Showing 3 changed files with 17 additions and 11 deletions.
diff --git a/packages/core/src/agent/Agent.ts b/packages/core/src/agent/Agent.ts
@@ -8,7 +8,7 @@ import type { InitConfig } from '../types'
 import type { Subscription } from 'rxjs'
 
 import { Subject } from 'rxjs'
-import { concatMap, takeUntil } from 'rxjs/operators'
+import { mergeMap, takeUntil } from 'rxjs/operators'
 
 import { InjectionSymbols } from '../constants'
 import { SigningProviderToken } from '../crypto'
@@ -152,16 +152,17 @@ export class Agent<AgentModules extends AgentModulesInput = any> extends BaseAge
       .observable<AgentMessageReceivedEvent>(AgentEventTypes.AgentMessageReceived)
       .pipe(
         takeUntil(stop$),
-        concatMap((e) =>
-          this.messageReceiver
-            .receiveMessage(e.payload.message, {
-              connection: e.payload.connection,
-              contextCorrelationId: e.payload.contextCorrelationId,
-              receivedAt: e.payload.receivedAt,
-            })
-            .catch((error) => {
-              this.logger.error('Failed to process message', { error })
-            })
+        mergeMap(
+          (e) =>
+            this.messageReceiver
+              .receiveMessage(e.payload.message, {
+                connection: e.payload.connection,
+                contextCorrelationId: e.payload.contextCorrelationId,
+              })
+              .catch((error) => {
+                this.logger.error('Failed to process message', { error })
+              }),
+          this.agentConfig.processMessagesConcurrently ? undefined : 1
         )
       )
       .subscribe()

diff --git a/packages/core/src/agent/AgentConfig.ts b/packages/core/src/agent/AgentConfig.ts
@@ -75,6 +75,10 @@ export class AgentConfig {
     return this.initConfig.backupBeforeStorageUpdate ?? true
   }
 
+  public get processMessagesConcurrently() {
+    return this.initConfig.processMessagesConcurrently ?? false
+  }
+
   public extend(config: Partial<InitConfig>): AgentConfig {
     return new AgentConfig(
       { ...this.initConfig, logger: this.logger, label: this.label, ...config },

diff --git a/packages/core/src/types.ts b/packages/core/src/types.ts
@@ -83,6 +83,7 @@ export interface InitConfig {
   connectionImageUrl?: string
   autoUpdateStorageOnStartup?: boolean
   backupBeforeStorageUpdate?: boolean
+  processMessagesConcurrently?: boolean
 }
 
 export type ProtocolVersion = `${number}.${number}`