fix: alter mediation recipient websocket transport priority

docs/setup-react-native.md

@@ -5,10 +5,10 @@
 To start using Aries Framework JavaScript in React Native some platform specific dependencies are required.
 
 1. Follow the [React Native Setup](https://reactnative.dev/docs/environment-setup) guide to set up your environment.
-2. Add `@aries-framework/core` and `@aries-framework/react-native` to your project.
+2. Add `@aries-framework/core`, `@aries-framework/react-native`, `react-native-fs`, and `react-native-get-random-values` to your project.
 
 ```bash
-yarn add @aries-framework/core @aries-framework/react-native
+yarn add @aries-framework/core @aries-framework/react-native react-native-fs react-native-get-random-values
 ```
 
 3. Install [Libindy](https://github.com/hyperledger/indy-sdk) for iOS and Android:

packages/core/src/agent/MessageSender.ts

@@ -183,6 +183,7 @@ export class MessageSender {
           service,
           senderKey: connection.verkey,
           returnRoute: shouldUseReturnRoute,
+          connectionId: connection.id,
         })
         return
       } catch (error) {
@@ -225,11 +226,13 @@ export class MessageSender {
     service,
     senderKey,
     returnRoute,
+    connectionId,
   }: {
     message: AgentMessage
     service: DidCommService
     senderKey: string
     returnRoute?: boolean
+    connectionId?: string
   }) {
     if (this.outboundTransports.length === 0) {
       throw new AriesFrameworkError('Agent has no outbound transport!')
@@ -250,6 +253,7 @@ export class MessageSender {
 
     const outboundPackage = await this.packMessage({ message, keys, endpoint: service.serviceEndpoint })
     outboundPackage.endpoint = service.serviceEndpoint
+    outboundPackage.connectionId = connectionId
     for (const transport of this.outboundTransports) {
       if (transport.supportedSchemes.includes(service.protocolScheme)) {
         await transport.sendMessage(outboundPackage)

packages/core/src/agent/__tests__/MessageSender.test.ts

@@ -132,6 +132,7 @@ describe('MessageSender', () => {
       await messageSender.sendMessage(outboundMessage)
 
       expect(sendMessageSpy).toHaveBeenCalledWith({
+        connectionId: 'test-123',
         payload: wireMessage,
         endpoint: firstDidCommService.serviceEndpoint,
         responseRequested: false,
@@ -149,6 +150,7 @@ describe('MessageSender', () => {
       await messageSender.sendMessage(outboundMessage)
 
       expect(sendMessageSpy).toHaveBeenCalledWith({
+        connectionId: 'test-123',
         payload: wireMessage,
         endpoint: firstDidCommService.serviceEndpoint,
         responseRequested: false,
@@ -164,6 +166,7 @@ describe('MessageSender', () => {
       await messageSender.sendMessage(outboundMessage)
 
       expect(sendMessageToServiceSpy).toHaveBeenCalledWith({
+        connectionId: 'test-123',
         message: outboundMessage.payload,
         senderKey: connection.verkey,
         service: firstDidCommService,
@@ -184,6 +187,7 @@ describe('MessageSender', () => {
       await messageSender.sendMessage(outboundMessage)
 
       expect(sendMessageToServiceSpy).toHaveBeenNthCalledWith(2, {
+        connectionId: 'test-123',
         message: outboundMessage.payload,
         senderKey: connection.verkey,
         service: secondDidCommService,

packages/core/src/modules/routing/RecipientModule.ts

@@ -1,5 +1,6 @@
 import type { Logger } from '../../logger'
 import type { OutboundWebSocketClosedEvent } from '../../transport'
+import type { OutboundMessage } from '../../types'
 import type { ConnectionRecord } from '../connections'
 import type { MediationStateChangedEvent } from './RoutingEvents'
 import type { MediationRecord } from './index'
@@ -73,6 +74,23 @@ export class RecipientModule {
     }
   }
 
+  private async sendMessage(outboundMessage: OutboundMessage) {
+    const { mediatorPickupStrategy } = this.agentConfig
+    const transportPriority =
+      mediatorPickupStrategy === MediatorPickupStrategy.Implicit
+        ? { schemes: ['wss', 'ws'], restrictive: true }
+        : undefined
+
+    await this.messageSender.sendMessage(outboundMessage, {
+      transportPriority,
+      // TODO: add keepAlive: true to enforce through the public api
+      // we need to keep the socket alive. It already works this way, but would
+      // be good to make more explicit from the public facing API.
+      // This would also make it easier to change the internal API later on.
+      // keepAlive: true,
+    })
+  }
+
   private async openMediationWebSocket(mediator: MediationRecord) {
     const { message, connectionRecord } = await this.connectionService.createTrustPing(mediator.connectionId)
 
@@ -85,17 +103,21 @@ export class RecipientModule {
       throw new AriesFrameworkError('Cannot open websocket to connection without websocket service endpoint')
     }
 
-    await this.messageSender.sendMessage(createOutboundMessage(connectionRecord, message), {
-      transportPriority: {
-        schemes: websocketSchemes,
-        restrictive: true,
-        // TODO: add keepAlive: true to enforce through the public api
-        // we need to keep the socket alive. It already works this way, but would
-        // be good to make more explicit from the public facing API.
-        // This would also make it easier to change the internal API later on.
-        // keepAlive: true,
-      },
-    })
+    try {
+      await this.messageSender.sendMessage(createOutboundMessage(connectionRecord, message), {
+        transportPriority: {
+          schemes: websocketSchemes,
+          restrictive: true,
+          // TODO: add keepAlive: true to enforce through the public api
+          // we need to keep the socket alive. It already works this way, but would
+          // be good to make more explicit from the public facing API.
+          // This would also make it easier to change the internal API later on.
+          // keepAlive: true,
+        },
+      })
+    } catch (error) {
+      this.logger.warn('Unable to open websocket connection to mediator', { error })
+    }
   }
 
   private async initiateImplicitPickup(mediator: MediationRecord) {
@@ -118,7 +140,10 @@ export class RecipientModule {
         // Wait for interval time before reconnecting
         delay(interval)
       )
-      .subscribe(() => {
+      .subscribe(async () => {
+        this.logger.warn(
+          `Websocket connection to mediator with connectionId '${mediator.connectionId}' is closed, attempting to reconnect...`
+        )
         this.openMediationWebSocket(mediator)
       })
 
@@ -163,7 +188,7 @@ export class RecipientModule {
 
     const batchPickupMessage = new BatchPickupMessage({ batchSize: 10 })
     const outboundMessage = createOutboundMessage(mediatorConnection, batchPickupMessage)
-    await this.messageSender.sendMessage(outboundMessage)
+    await this.sendMessage(outboundMessage)
   }
 
   public async setDefaultMediator(mediatorRecord: MediationRecord) {
@@ -173,15 +198,15 @@ export class RecipientModule {
   public async requestMediation(connection: ConnectionRecord): Promise<MediationRecord> {
     const { mediationRecord, message } = await this.mediationRecipientService.createRequest(connection)
     const outboundMessage = createOutboundMessage(connection, message)
-    await this.messageSender.sendMessage(outboundMessage)
+
+    await this.sendMessage(outboundMessage)
     return mediationRecord
   }
 
   public async notifyKeylistUpdate(connection: ConnectionRecord, verkey: string) {
     const message = this.mediationRecipientService.createKeylistUpdateMessage(verkey)
     const outboundMessage = createOutboundMessage(connection, message)
-    const response = await this.messageSender.sendMessage(outboundMessage)
-    return response
+    await this.sendMessage(outboundMessage)
   }
 
   public async findByConnectionId(connectionId: string) {
@@ -230,7 +255,7 @@ export class RecipientModule {
 
     // Send mediation request message
     const outboundMessage = createOutboundMessage(connection, message)
-    await this.messageSender.sendMessage(outboundMessage)
+    await this.sendMessage(outboundMessage)
 
     const event = await firstValueFrom(subject)
     return event.payload.mediationRecord

packages/core/src/transport/WsOutboundTransport.ts

@@ -8,6 +8,7 @@ import type WebSocket from 'ws'
 import { AgentConfig } from '../agent/AgentConfig'
 import { EventEmitter } from '../agent/EventEmitter'
 import { AriesFrameworkError } from '../error/AriesFrameworkError'
+import { Buffer } from '../utils/buffer'
 
 import { TransportEventTypes } from './TransportEventTypes'
 
@@ -134,8 +135,8 @@ export class WsOutboundTransport implements OutboundTransport {
         reject(error)
       }
 
-      socket.onclose = async () => {
-        this.logger.debug(`WebSocket closing to ${endpoint}`)
+      socket.onclose = async (event: WebSocket.CloseEvent) => {
+        this.logger.debug(`WebSocket closing to ${endpoint}`, { event })
         socket.removeEventListener('message', this.handleMessageEvent)
         this.transportTable.delete(socketId)
 

packages/react-native/package.json

@@ -29,19 +29,19 @@
     "events": "^3.3.0"
   },
   "devDependencies": {
-    "@types/react-native": "^0.64.10",
     "@types/indy-sdk-react-native": "npm:@types/indy-sdk@^1.16.6",
+    "@types/react-native": "^0.64.10",
+    "indy-sdk-react-native": "^0.1.13",
     "react": "17.0.1",
     "react-native": "0.64.2",
-    "rimraf": "~3.0.2",
-    "typescript": "~4.3.0",
-    "indy-sdk-react-native": "^0.1.13",
+    "react-native-fs": "^2.18.0",
     "react-native-get-random-values": "^1.7.0",
-    "react-native-fs": "^2.18.0"
+    "rimraf": "~3.0.2",
+    "typescript": "~4.3.0"
   },
   "peerDependencies": {
     "indy-sdk-react-native": "^0.1.13",
-    "react-native-get-random-values": "^1.7.0",
-    "react-native-fs": "^2.18.0"
+    "react-native-fs": "^2.18.0",
+    "react-native-get-random-values": "^1.7.0"
   }
 }