fix: add config to process messages concurrently

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,18 @@ 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,
+                session: e.payload.session,
+              })
+              .catch((error) => {
+                this.logger.error('Failed to process message', { error })
+              }),
+          this.agentConfig.processDidCommMessagesConcurrently ? undefined : 1
         )
       )
       .subscribe()

packages/core/src/agent/AgentConfig.ts

@@ -75,6 +75,10 @@ export class AgentConfig {
     return this.initConfig.backupBeforeStorageUpdate ?? true
   }
 
+  public get processDidCommMessagesConcurrently() {
+    return this.initConfig.processDidCommMessagesConcurrently ?? false
+  }
+
   public extend(config: Partial<InitConfig>): AgentConfig {
     return new AgentConfig(
       { ...this.initConfig, logger: this.logger, label: this.label, ...config },

packages/core/src/agent/Events.ts

@@ -1,4 +1,5 @@
 import type { AgentMessage } from './AgentMessage'
+import type { TransportSession } from './TransportService'
 import type { OutboundMessageContext, OutboundMessageSendStatus } from './models'
 import type { ConnectionRecord } from '../modules/connections'
 import type { Observable } from 'rxjs'
@@ -34,6 +35,7 @@ export interface AgentMessageReceivedEvent extends BaseEvent {
     connection?: ConnectionRecord
     contextCorrelationId?: string
     receivedAt?: Date
+    session?: TransportSession
   }
 }
 

packages/core/src/types.ts

@@ -83,6 +83,7 @@ export interface InitConfig {
   connectionImageUrl?: string
   autoUpdateStorageOnStartup?: boolean
   backupBeforeStorageUpdate?: boolean
+  processDidCommMessagesConcurrently?: boolean
 }
 
 export type ProtocolVersion = `${number}.${number}`

packages/node/package.json

@@ -31,6 +31,7 @@
     "@credo-ts/core": "workspace:*",
     "@types/express": "^4.17.15",
     "express": "^4.17.1",
+    "rxjs": "^7.8.0",
     "ws": "^8.13.0"
   },
   "devDependencies": {

packages/node/src/transport/HttpInboundTransport.ts

@@ -1,9 +1,18 @@
-import type { InboundTransport, Agent, TransportSession, EncryptedMessage, AgentContext } from '@credo-ts/core'
+import type {
+  InboundTransport,
+  Agent,
+  TransportSession,
+  EncryptedMessage,
+  AgentContext,
+  AgentMessageReceivedEvent,
+  AgentMessageProcessedEvent,
+} from '@credo-ts/core'
 import type { Express, Request, Response } from 'express'
 import type { Server } from 'http'
 
-import { DidCommMimeType, CredoError, TransportService, utils, MessageReceiver } from '@credo-ts/core'
+import { DidCommMimeType, CredoError, TransportService, utils, AgentEventTypes } from '@credo-ts/core'
 import express, { text } from 'express'
+import { first, firstValueFrom, ReplaySubject, timeout } from 'rxjs'
 
 const supportedContentTypes: string[] = [DidCommMimeType.V0, DidCommMimeType.V1]
 
@@ -29,7 +38,6 @@ export class HttpInboundTransport implements InboundTransport {
 
   public async start(agent: Agent) {
     const transportService = agent.dependencyManager.resolve(TransportService)
-    const messageReceiver = agent.dependencyManager.resolve(MessageReceiver)
 
     agent.config.logger.debug(`Starting HTTP inbound transport`, {
       port: this.port,
@@ -51,11 +59,32 @@ export class HttpInboundTransport implements InboundTransport {
 
       try {
         const message = req.body
-        const encryptedMessage = JSON.parse(message)
-        await messageReceiver.receiveMessage(encryptedMessage, {
-          session,
+        const encryptedMessage = JSON.parse(message) as EncryptedMessage
+
+        const observable = agent.events.observable<AgentMessageProcessedEvent>(AgentEventTypes.AgentMessageProcessed)
+        const subject = new ReplaySubject(1)
+
+        observable
+          .pipe(
+            first((e) => e.type === AgentEventTypes.AgentMessageProcessed),
+            timeout({
+              first: 10000, // timeout after 10 seconds
+              meta: 'HttpInboundTransport.start',
+            })
+          )
+          .subscribe(subject)
+
+        agent.events.emit<AgentMessageReceivedEvent>(agent.context, {
+          type: AgentEventTypes.AgentMessageReceived,
+          payload: {
+            message: encryptedMessage,
+            session: session,
+          },
         })
 
+        // Wait for message to be processed
+        await firstValueFrom(subject)
+
         // If agent did not use session when processing message we need to send response here.
         if (!res.headersSent) {
           res.status(200).end()

packages/node/src/transport/WsInboundTransport.ts

@@ -1,6 +1,16 @@
-import type { Agent, InboundTransport, Logger, TransportSession, EncryptedMessage, AgentContext } from '@credo-ts/core'
-
-import { CredoError, TransportService, utils, MessageReceiver } from '@credo-ts/core'
+import type {
+  Agent,
+  InboundTransport,
+  Logger,
+  TransportSession,
+  EncryptedMessage,
+  AgentContext,
+  AgentMessageReceivedEvent,
+  AgentMessageProcessedEvent,
+} from '@credo-ts/core'
+
+import { CredoError, TransportService, utils, AgentEventTypes } from '@credo-ts/core'
+import { first, firstValueFrom, ReplaySubject, timeout } from 'rxjs'
 // eslint-disable-next-line import/no-named-as-default
 import WebSocket, { Server } from 'ws'
 
@@ -58,13 +68,35 @@ export class WsInboundTransport implements InboundTransport {
   }
 
   private listenOnWebSocketMessages(agent: Agent, socket: WebSocket, session: TransportSession) {
-    const messageReceiver = agent.dependencyManager.resolve(MessageReceiver)
-
     // eslint-disable-next-line @typescript-eslint/no-explicit-any
     socket.addEventListener('message', async (event: any) => {
       this.logger.debug('WebSocket message event received.', { url: event.target.url })
       try {
-        await messageReceiver.receiveMessage(JSON.parse(event.data), { session })
+        const encryptedMessage = JSON.parse(event.data) as EncryptedMessage
+
+        const observable = agent.events.observable<AgentMessageProcessedEvent>(AgentEventTypes.AgentMessageProcessed)
+        const subject = new ReplaySubject(1)
+
+        observable
+          .pipe(
+            first((e) => e.type === AgentEventTypes.AgentMessageProcessed),
+            timeout({
+              first: 10000, // timeout after 10 seconds
+              meta: 'WsInboundTransport.listenOnWebSocketMessages',
+            })
+          )
+          .subscribe(subject)
+
+        agent.events.emit<AgentMessageReceivedEvent>(agent.context, {
+          type: AgentEventTypes.AgentMessageReceived,
+          payload: {
+            message: encryptedMessage,
+            session: session,
+          },
+        })
+
+        // Wait for message to be processed
+        await firstValueFrom(subject)
       } catch (error) {
         this.logger.error(`Error processing message: ${error}`)
       }