[DTP-1118] Apply ConnectionDetails.maxMessageSize limit to state messages

src/common/lib/client/defaultrealtime.ts

@@ -19,6 +19,7 @@ import {
 import { Http } from 'common/types/http';
 import Defaults from '../util/defaults';
 import Logger from '../util/logger';
+import { MessageEncoding } from '../types/message';
 
 /**
  `DefaultRealtime` is the class that the non tree-shakable version of the SDK exports as `Realtime`. It ensures that this version of the SDK includes all of the functionality which is optionally available in the tree-shakable version.
@@ -71,4 +72,5 @@ export class DefaultRealtime extends BaseRealtime {
   // Used by tests
   static _Http = Http;
   static _PresenceMap = PresenceMap;
+  static _MessageEncoding = MessageEncoding;
 }

src/common/lib/client/realtimechannel.ts

@@ -265,11 +265,7 @@ class RealtimeChannel extends EventEmitter {
     const size = getMessagesSize(messages);
     if (size > maxMessageSize) {
       throw new ErrorInfo(
-        'Maximum size of messages that can be published at once exceeded ( was ' +
-          size +
-          ' bytes; limit is ' +
-          maxMessageSize +
-          ' bytes)',
+        `Maximum size of messages that can be published at once exceeded (was ${size} bytes; limit is ${maxMessageSize} bytes)`,
         40009,
         400,
       );

src/common/lib/client/restchannel.ts

@@ -117,11 +117,7 @@ class RestChannel {
       maxMessageSize = options.maxMessageSize;
     if (size > maxMessageSize) {
       throw new ErrorInfo(
-        'Maximum size of messages that can be published at once exceeded ( was ' +
-          size +
-          ' bytes; limit is ' +
-          maxMessageSize +
-          ' bytes)',
+        `Maximum size of messages that can be published at once exceeded (was ${size} bytes; limit is ${maxMessageSize} bytes)`,
         40009,
         400,
       );

src/common/lib/util/utils.ts

@@ -1,4 +1,4 @@
-import Platform from 'common/platform';
+import Platform, { Bufferlike } from 'common/platform';
 import ErrorInfo, { PartialErrorInfo } from 'common/lib/types/errorinfo';
 import { ModularPlugins } from '../client/modularplugins';
 import { MsgPack } from 'common/types/msgpack';
@@ -279,15 +279,23 @@ export function inspectBody(body: unknown): string {
   }
 }
 
-/* Data is assumed to be either a string or a buffer. */
-export function dataSizeBytes(data: string | Buffer): number {
+/* Data is assumed to be either a string, a number, a boolean or a buffer. */
+export function dataSizeBytes(data: string | number | boolean | Bufferlike): number {
   if (Platform.BufferUtils.isBuffer(data)) {
     return Platform.BufferUtils.byteLength(data);
   }
   if (typeof data === 'string') {
     return Platform.Config.stringByteSize(data);
   }
-  throw new Error('Expected input of Utils.dataSizeBytes to be a buffer or string, but was: ' + typeof data);
+  if (typeof data === 'number') {
+    return 8;
+  }
+  if (typeof data === 'boolean') {
+    return 1;
+  }
+  throw new Error(
+    `Expected input of Utils.dataSizeBytes to be a string, a number, a boolean or a buffer, but was: ${typeof data}`,
+  );
 }
 
 export function cheapRandStr(): string {

src/plugins/liveobjects/liveobjects.ts

@@ -250,6 +250,15 @@ export class LiveObjects {
     }
 
     stateMessages.forEach((x) => StateMessage.encode(x, this._client.MessageEncoding));
+    const maxMessageSize = this._client.options.maxMessageSize;
+    const size = stateMessages.reduce((acc, msg) => acc + msg.getMessageSize(), 0);
+    if (size > maxMessageSize) {
+      throw new this._client.ErrorInfo(
+        `Maximum size of state messages that can be published at once exceeded (was ${size} bytes; limit is ${maxMessageSize} bytes)`,
+        40009,
+        400,
+      );
+    }
 
     return this._channel.sendState(stateMessages);
   }

src/plugins/liveobjects/statemessage.ts

@@ -445,4 +445,121 @@ export class StateMessage {
 
     return result;
   }
+
+  getMessageSize(): number {
+    let size = 0;
+
+    size += this.clientId?.length ?? 0;
+    if (this.operation) {
+      size += this._getStateOperationSize(this.operation);
+    }
+    if (this.object) {
+      size += this._getStateObjectSize(this.object);
+    }
+    if (this.extras) {
+      size += JSON.stringify(this.extras).length;
+    }
+
+    return size;
+  }
+
+  private _getStateOperationSize(operation: StateOperation): number {
+    let size = 0;
+
+    if (operation.mapOp) {
+      size += this._getStateMapOpSize(operation.mapOp);
+    }
+    if (operation.counterOp) {
+      size += this._getStateCounterOpSize(operation.counterOp);
+    }
+    if (operation.map) {
+      size += this._getStateMapSize(operation.map);
+    }
+    if (operation.counter) {
+      size += this._getStateCounterSize(operation.counter);
+    }
+
+    return size;
+  }
+
+  private _getStateObjectSize(obj: StateObject): number {
+    let size = 0;
+
+    if (obj.map) {
+      size += this._getStateMapSize(obj.map);
+    }
+    if (obj.counter) {
+      size += this._getStateCounterSize(obj.counter);
+    }
+    if (obj.createOp) {
+      size += this._getStateOperationSize(obj.createOp);
+    }
+
+    return size;
+  }
+
+  private _getStateMapSize(map: StateMap): number {
+    let size = 0;
+
+    Object.entries(map.entries ?? {}).forEach(([key, entry]) => {
+      size += key?.length ?? 0;
+      if (entry) {
+        size += this._getStateMapEntrySize(entry);
+      }
+    });
+
+    return size;
+  }
+
+  private _getStateCounterSize(counter: StateCounter): number {
+    if (counter.count == null) {
+      return 0;
+    }
+
+    return 8;
+  }
+
+  private _getStateMapEntrySize(entry: StateMapEntry): number {
+    let size = 0;
+
+    if (entry.data) {
+      size += this._getStateDataSize(entry.data);
+    }
+
+    return size;
+  }
+
+  private _getStateMapOpSize(mapOp: StateMapOp): number {
+    let size = 0;
+
+    size += mapOp.key?.length ?? 0;
+
+    if (mapOp.data) {
+      size += this._getStateDataSize(mapOp.data);
+    }
+
+    return size;
+  }
+
+  private _getStateCounterOpSize(operation: StateCounterOp): number {
+    if (operation.amount == null) {
+      return 0;
+    }
+
+    return 8;
+  }
+
+  private _getStateDataSize(data: StateData): number {
+    let size = 0;
+
+    if (data.value) {
+      size += this._getStateValueSize(data.value);
+    }
+
+    return size;
+  }
+
+  private _getStateValueSize(value: StateValue): number {
+    return this._utils.dataSizeBytes(value);
+  }
 }

test/common/modules/private_api_recorder.js

@@ -16,6 +16,7 @@ define(['test/support/output_directory_paths'], function (outputDirectoryPaths)
     'call.Defaults.getPort',
     'call.Defaults.normaliseOptions',
     'call.EventEmitter.emit',
+    'call.LiveObject.getObjectId',
     'call.LiveObject.isTombstoned',
     'call.LiveObjects._liveObjectsPool._onGCInterval',
     'call.LiveObjects._liveObjectsPool.get',
@@ -25,7 +26,11 @@ define(['test/support/output_directory_paths'], function (outputDirectoryPaths)
     'call.Platform.nextTick',
     'call.PresenceMessage.fromValues',
     'call.ProtocolMessage.setFlag',
+    'call.StateMessage.encode',
+    'call.StateMessage.fromValues',
+    'call.StateMessage.getMessageSize',
     'call.Utils.copy',
+    'call.Utils.dataSizeBytes',
     'call.Utils.getRetryTime',
     'call.Utils.inspectError',
     'call.Utils.keysArray',
@@ -47,7 +52,6 @@ define(['test/support/output_directory_paths'], function (outputDirectoryPaths)
     'call.http._getHosts',
     'call.http.checkConnectivity',
     'call.http.doUri',
-    'call.LiveObject.getObjectId',
     'call.msgpack.decode',
     'call.msgpack.encode',
     'call.presence._myMembers.put',