Make RealtimeChannel#processMessage async

Preparation for #1293 (making ICipher.decrypt asynchronous).
ably · Jun 5, 2023 · 346e5eb · 346e5eb
1 parent ec8a1b1
commit 346e5eb
Show file tree

Hide file tree

Showing 5 changed files with 128 additions and 97 deletions.
diff --git a/src/common/lib/client/realtime.ts b/src/common/lib/client/realtime.ts
@@ -103,7 +103,7 @@ class Channels extends EventEmitter {
       );
       return;
     }
-    channel.processMessage(msg);
+    await channel.processMessage(msg);
   }
 
   /* called when a transport becomes connected; reattempt attach/detach

diff --git a/src/common/lib/client/realtimechannel.ts b/src/common/lib/client/realtimechannel.ts
@@ -582,7 +582,8 @@ class RealtimeChannel extends Channel {
     this.sendMessage(msg, callback);
   }
 
-  processMessage(message: ProtocolMessage): void {
+  // Access to this method is synchronised by ConnectionManager#processChannelMessage, in order to synchronise access to the state stored in _decodingContext.
+  async processMessage(message: ProtocolMessage): Promise<void> {
     if (
       message.action === actions.ATTACHED ||
       message.action === actions.MESSAGE ||

diff --git a/test/realtime/failure.test.js b/test/realtime/failure.test.js
@@ -329,11 +329,11 @@ define(['ably', 'shared_helper', 'async', 'chai'], function (Ably, helper, async
         channel = realtime.channels.get('failed_attach'),
         originalProcessMessage = channel.processMessage.bind(channel);
 
-      channel.processMessage = function (message) {
+      channel.processMessage = async function (message) {
         if (message.action === 11) {
           return;
         }
-        originalProcessMessage(message);
+        await originalProcessMessage(message);
       };
 
       realtime.connection.once('connected', function () {
@@ -371,12 +371,12 @@ define(['ably', 'shared_helper', 'async', 'chai'], function (Ably, helper, async
 
         var performance = isBrowser ? window.performance : require('perf_hooks').performance;
 
-        channel.processMessage = function (message) {
+        channel.processMessage = async function (message) {
           // Ignore ATTACHED messages
           if (message.action === 11) {
             return;
           }
-          originalProcessMessage(message);
+          await originalProcessMessage(message);
         };
 
         realtime.connection.on('connected', function () {

diff --git a/test/realtime/presence.test.js b/test/realtime/presence.test.js
@@ -1732,18 +1732,27 @@ define(['ably', 'shared_helper', 'async', 'chai'], function (Ably, helper, async
           },
           function (cb) {
             /* Inject an additional member locally */
-            channel.processMessage({
-              action: 14,
-              id: 'messageid:0',
-              connectionId: 'connid',
-              timestamp: utils.now(),
-              presence: [
-                {
-                  clientId: goneClientId,
-                  action: 'enter',
-                },
-              ],
-            });
+            channel
+              .processMessage({
+                action: 14,
+                id: 'messageid:0',
+                connectionId: 'connid',
+                timestamp: utils.now(),
+                presence: [
+                  {
+                    clientId: goneClientId,
+                    action: 'enter',
+                  },
+                ],
+              })
+              .then(function () {
+                cb(null);
+              })
+              .catch(function (err) {
+                cb(err);
+              });
+          },
+          function (cb) {
             channel.presence.get(function (err, members) {
               try {
                 expect(members && members.length).to.equal(2, 'Check two members present');
@@ -1812,18 +1821,27 @@ define(['ably', 'shared_helper', 'async', 'chai'], function (Ably, helper, async
           },
           function (cb) {
             /* Inject a member locally */
-            channel.processMessage({
-              action: 14,
-              id: 'messageid:0',
-              connectionId: 'connid',
-              timestamp: utils.now(),
-              presence: [
-                {
-                  clientId: fakeClientId,
-                  action: 'enter',
-                },
-              ],
-            });
+            channel
+              .processMessage({
+                action: 14,
+                id: 'messageid:0',
+                connectionId: 'connid',
+                timestamp: utils.now(),
+                presence: [
+                  {
+                    clientId: fakeClientId,
+                    action: 'enter',
+                  },
+                ],
+              })
+              .then(function () {
+                cb();
+              })
+              .catch(function () {
+                cb(err);
+              });
+          },
+          function (cb) {
             channel.presence.get(function (err, members) {
               try {
                 expect(members && members.length).to.equal(1, 'Check one member present');