fix: alter mediation recipient websocket transport priority (#434)

* Squashed commit of the following: commit 41646d8d93cf1e16cad2033aeace7949864246ac Author: Adam Burdett <[email protected]> Date: Mon Aug 16 11:07:24 2021 -0600 Update packages/core/src/agent/MessageSender.ts Co-authored-by: Timo Glastra <[email protected]> Signed-off-by: Adam Burdett <[email protected]> commit de9d5d3 Author: Adam Burdett <[email protected]> Date: Mon Aug 16 11:06:28 2021 -0600 Update packages/core/src/agent/MessageSender.ts Co-authored-by: Timo Glastra <[email protected]> commit d9b9164 Author: James Ebert <[email protected]> Date: Fri Aug 13 18:07:37 2021 -0600 chore: formatted Signed-off-by: James Ebert <[email protected]> commit 335286a Author: James Ebert <[email protected]> Date: Fri Aug 13 18:03:17 2021 -0600 chore: formatted changes Signed-off-by: James Ebert <[email protected]> commit a639d1d Author: James Ebert <[email protected]> Date: Fri Aug 13 17:59:42 2021 -0600 fix: added connection request sending and added logs Signed-off-by: James Ebert <[email protected]> commit 2800d92 Author: Adam Burdett <[email protected]> Date: Fri Aug 13 16:28:26 2021 -0600 even better provision code Signed-off-by: Adam Burdett <[email protected]> commit 2a153e0 Author: Adam Burdett <[email protected]> Date: Fri Aug 13 16:11:30 2021 -0600 throw errer Signed-off-by: Adam Burdett <[email protected]> commit 8d62fcb Author: Adam Burdett <[email protected]> Date: Fri Aug 13 16:07:58 2021 -0600 better provision code Signed-off-by: Adam Burdett <[email protected]> commit 47db23e Author: Adam Burdett <[email protected]> Date: Fri Aug 13 13:35:37 2021 -0600 sort instead of filter cat Signed-off-by: Adam Burdett <[email protected]> commit 343ff78 Author: Adam Burdett <[email protected]> Date: Fri Aug 13 12:45:49 2021 -0600 update test for error message Signed-off-by: Adam Burdett <[email protected]> commit d7dff12 Author: Adam Burdett <[email protected]> Date: Fri Aug 13 12:29:36 2021 -0600 remove duplicate services Signed-off-by: Adam Burdett <[email protected]> commit 1f204d8 Author: Adam Burdett <[email protected]> Date: Fri Aug 13 12:15:45 2021 -0600 removed dead code Signed-off-by: Adam Burdett <[email protected]> commit 3961197 Author: Adam Burdett <[email protected]> Date: Fri Aug 13 12:09:08 2021 -0600 less flexible code Signed-off-by: Adam Burdett <[email protected]> commit 81e3477 Author: Adam Burdett <[email protected]> Date: Fri Aug 13 11:30:16 2021 -0600 remove un-needed changes Signed-off-by: Adam Burdett <[email protected]> commit 361c28a Author: Adam Burdett <[email protected]> Date: Fri Aug 13 09:44:32 2021 -0600 remove throw new error Signed-off-by: Adam Burdett <[email protected]> commit 203824c Author: Adam Burdett <[email protected]> Date: Thu Aug 12 18:00:52 2021 -0600 update send message in RM Signed-off-by: Adam Burdett <[email protected]> commit abea05a Author: Adam Burdett <[email protected]> Date: Thu Aug 12 17:49:25 2021 -0600 handle empty supported Protocols in TS Signed-off-by: Adam Burdett <[email protected]> commit 9b3a78e Author: Adam Burdett <[email protected]> Date: Thu Aug 12 17:33:18 2021 -0600 default declared Signed-off-by: Adam Burdett <[email protected]> commit 9da1fd3 Author: Adam Burdett <[email protected]> Date: Thu Aug 12 17:25:13 2021 -0600 requested changes Signed-off-by: Adam Burdett <[email protected]> commit 356287d Author: Adam Burdett <[email protected]> Date: Tue Aug 10 00:04:18 2021 -0600 formatting Signed-off-by: Adam Burdett <[email protected]> commit 9cefd01 Author: Adam Burdett <[email protected]> Date: Mon Aug 9 23:59:08 2021 -0600 e2e updates Signed-off-by: Adam Burdett <[email protected]> commit 812c60f Author: Adam Burdett <[email protected]> Date: Mon Aug 9 12:44:53 2021 -0600 added schemes on subjectOutBoundTransport Signed-off-by: Adam Burdett <[email protected]> commit 87bdd8f Author: Adam Burdett <[email protected]> Date: Fri Aug 6 13:49:29 2021 -0600 formatting Signed-off-by: Adam Burdett <[email protected]> commit 5e44efd Merge: dd1f1f9 f0cf209 Author: Adam Burdett <[email protected]> Date: Fri Aug 6 13:28:08 2021 -0600 Merge branch 'main' of github.com:hyperledger/aries-framework-javascript into fix/mediator-transports commit dd1f1f9 Author: Adam Burdett <[email protected]> Date: Tue Aug 3 18:23:28 2021 -0600 ignore unresolved reduxjs Signed-off-by: Adam Burdett <[email protected]> commit 77fe594 Author: Adam Burdett <[email protected]> Date: Tue Aug 3 17:39:47 2021 -0600 fix for mediator Signed-off-by: Adam Burdett <[email protected]> commit 27b4818 Author: Adam Burdett <[email protected]> Date: Wed Jul 28 15:55:36 2021 -0600 multiple socket timeout support Signed-off-by: Adam Burdett <[email protected]> commit 3e6640f Author: Adam Burdett <[email protected]> Date: Wed Jul 28 15:40:00 2021 -0600 propper recursive backoff Signed-off-by: Adam Burdett <[email protected]> commit 6814600 Author: Adam Burdett <[email protected]> Date: Wed Jul 28 14:57:36 2021 -0600 logger Signed-off-by: Adam Burdett <[email protected]> commit 8bcd966 Author: Adam Burdett <[email protected]> Date: Wed Jul 28 14:55:27 2021 -0600 timeout time Signed-off-by: Adam Burdett <[email protected]> commit 12de34e Author: Adam Burdett <[email protected]> Date: Wed Jul 28 14:54:32 2021 -0600 timeout Signed-off-by: Adam Burdett <[email protected]> commit dccb596 Author: Adam Burdett <[email protected]> Date: Wed Jul 28 14:50:19 2021 -0600 recursive backoff Signed-off-by: Adam Burdett <[email protected]> commit bc4762d Author: Adam Burdett <[email protected]> Date: Wed Jul 28 13:42:02 2021 -0600 keep sockets with mediators open Signed-off-by: Adam Burdett <[email protected]> commit d67710c Author: Adam Burdett <[email protected]> Date: Wed Jul 28 12:50:35 2021 -0600 id check Signed-off-by: Adam Burdett <[email protected]> commit 23fe187 Author: Adam Burdett <[email protected]> Date: Wed Jul 28 10:58:18 2021 -0600 check invite during init Signed-off-by: Adam Burdett <[email protected]> commit cfd7ece Author: Adam Burdett <[email protected]> Date: Wed Jul 28 09:59:14 2021 -0600 correct protocal type Signed-off-by: Adam Burdett <[email protected]> commit 178bd6c Author: Adam Burdett <[email protected]> Date: Wed Jul 28 09:55:35 2021 -0600 indy wallet friendly bits Signed-off-by: Adam Burdett <[email protected]> commit 4db4da6 Author: Adam Burdett <[email protected]> Date: Tue Jul 27 18:31:42 2021 -0600 utf-8 decode ws event.data Signed-off-by: Adam Burdett <[email protected]> commit 9d732c8 Author: Adam Burdett <[email protected]> Date: Tue Jul 27 18:09:47 2021 -0600 if(0) does not work. Signed-off-by: Adam Burdett <[email protected]> commit a8a3fb3 Author: Adam Burdett <[email protected]> Date: Mon Jul 26 16:39:24 2021 -0600 lint import ordering Signed-off-by: Adam Burdett <[email protected]> commit af6431d Author: Adam Burdett <[email protected]> Date: Mon Jul 26 14:22:42 2021 -0600 structure fix Signed-off-by: Adam Burdett <[email protected]> commit 20ba447 Author: Adam Burdett <[email protected]> Date: Fri Jul 23 18:40:35 2021 -0600 broken message sender tests Signed-off-by: Adam Burdett <[email protected]> commit 3316e0d Author: Adam Burdett <[email protected]> Date: Wed Jul 21 18:34:58 2021 -0600 sorted transports Signed-off-by: Adam Burdett <[email protected]> Signed-off-by: Adam Burdett <[email protected]> * formatting Signed-off-by: Adam Burdett <[email protected]> * Update packages/core/src/modules/routing/RecipientModule.ts Signed-off-by: Adam Burdett <[email protected]> Co-authored-by: Timo Glastra <[email protected]> * Update packages/core/src/modules/routing/RecipientModule.ts Signed-off-by: Adam Burdett <[email protected]> Co-authored-by: Timo Glastra <[email protected]> * Update packages/core/src/modules/routing/RecipientModule.ts Signed-off-by: Adam Burdett <[email protected]> Co-authored-by: Timo Glastra <[email protected]> * feat: created service retrieval with priority sorting Signed-off-by: James Ebert <[email protected]> * add get protocal scheme on service class Signed-off-by: Adam Burdett <[email protected]> * priority is ws, wss for mediation Signed-off-by: Adam Burdett <[email protected]> * feat: altered mediation recipient sending logic Signed-off-by: James Ebert <[email protected]> * chore: formatting Signed-off-by: James Ebert <[email protected]> * fix: add buffer as react native dependency for ws communciation Signed-off-by: James Ebert <[email protected]> * chore: altered buffer import Signed-off-by: James Ebert <[email protected]> * chore: altered buffer import location Signed-off-by: James Ebert <[email protected]> * chore: fixed core package version Signed-off-by: James Ebert <[email protected]> * chore: fix buffer dependency Signed-off-by: James Ebert <[email protected]> * chore: import formatting Signed-off-by: James Ebert <[email protected]> * fix: pass connectionId to allow for mediator websocket reopening Signed-off-by: James Ebert <[email protected]> * chore: reverted openMediationWebsocket sending functionality, added try catch for transport sending errors Signed-off-by: James Ebert <[email protected]> * chore: formatting Signed-off-by: James Ebert <[email protected]> * fix: failing message sender tests Signed-off-by: James Ebert <[email protected]> Co-authored-by: Adam Burdett <[email protected]> Co-authored-by: Timo Glastra <[email protected]>
openwallet-foundation · Sep 17, 2021 · 52c7897 · 52c7897
1 parent 7b9df65
commit 52c7897
Show file tree

Hide file tree

Showing 5 changed files with 55 additions and 21 deletions.
diff --git a/docs/setup-react-native.md b/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:

diff --git a/packages/core/src/agent/MessageSender.ts b/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)

diff --git a/packages/core/src/agent/__tests__/MessageSender.test.ts b/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,

diff --git a/packages/core/src/modules/routing/RecipientModule.ts b/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

diff --git a/packages/core/src/transport/WsOutboundTransport.ts b/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)