[SDK-3937] Create a tree-shakable module for realtime publishing

modules.d.ts

@@ -162,6 +162,20 @@ export declare const FetchRequest: unknown;
  */
 export declare const MessageInteractions: unknown;
 
+/**
+ * Provides a {@link BaseRealtime} instance with the ability to publish messages.
+ *
+ * To create a client that includes this module, include it in the `ModulesMap` that you pass to the {@link BaseRealtime.constructor}:
+ *
+ * ```javascript
+ * import { BaseRealtime, WebSocketTransport, FetchRequest, RealtimePublishing } from 'ably/modules';
+ * const realtime = new BaseRealtime(options, { WebSocketTransport, FetchRequest, RealtimePublishing });
+ * ```
+ *
+ * If this module is not provided, then calling {@link Types.RealtimeChannel.publish} on a channel will cause a runtime error.
+ */
+export declare const RealtimePublishing: unknown;
+
 /**
  * Pass a `ModulesMap` to { @link BaseRest.constructor | the constructor of BaseRest } or {@link BaseRealtime.constructor | that of BaseRealtime} to specify which functionality should be made available to that client.
  */
@@ -215,6 +229,11 @@ export interface ModulesMap {
    * See {@link MessageInteractions | documentation for the `MessageInteractions` module}.
    */
   MessageInteractions?: typeof MessageInteractions;
+
+  /**
+   * See {@link RealtimePublishing | documentation for the `RealtimePublishing` module}.
+   */
+  RealtimePublishing?: typeof RealtimePublishing;
 }
 
 /**

scripts/moduleReport.js

@@ -0,0 +1,108 @@
+const esbuild = require('esbuild');
+
+// List of all modules accepted in ModulesMap
+const moduleNames = [
+  'Rest',
+  'Crypto',
+  'MsgPack',
+  'RealtimePresence',
+  'XHRPolling',
+  'XHRStreaming',
+  'WebSocketTransport',
+  'XHRRequest',
+  'FetchRequest',
+  'MessageInteractions',
+  'RealtimePublishing',
+];
+
+// List of all free-standing functions exported by the library along with the
+// ModulesMap entries that we expect them to transitively import
+const functions = [
+  { name: 'generateRandomKey', transitiveImports: ['Crypto'] },
+  { name: 'getDefaultCryptoParams', transitiveImports: ['Crypto'] },
+  { name: 'decodeMessage', transitiveImports: [] },
+  { name: 'decodeEncryptedMessage', transitiveImports: ['Crypto'] },
+  { name: 'decodeMessages', transitiveImports: [] },
+  { name: 'decodeEncryptedMessages', transitiveImports: ['Crypto'] },
+  { name: 'decodePresenceMessage', transitiveImports: [] },
+  { name: 'decodePresenceMessages', transitiveImports: [] },
+  { name: 'constructPresenceMessage', transitiveImports: [] },
+];
+
+function formatBytes(bytes) {
+  const kibibytes = bytes / 1024;
+  const formatted = kibibytes.toFixed(2);
+  return `${formatted} KiB`;
+}
+
+// Gets the bundled size in bytes of an array of named exports from 'ably/modules'
+function getImportSize(modules) {
+  const outfile = modules.join('');
+  const result = esbuild.buildSync({
+    stdin: {
+      contents: `export { ${modules.join(', ')} } from './build/modules'`,
+      resolveDir: '.',
+    },
+    metafile: true,
+    minify: true,
+    bundle: true,
+    outfile,
+    write: false,
+  });
+
+  return result.metafile.outputs[outfile].bytes;
+}
+
+const errors = [];
+
+['BaseRest', 'BaseRealtime'].forEach((baseClient) => {
+  const baseClientSize = getImportSize([baseClient]);
+
+  // First display the size of the base client
+  console.log(`${baseClient}: ${formatBytes(baseClientSize)}`);
+
+  // Then display the size of each export together with the base client
+  [...moduleNames, ...Object.values(functions).map((functionData) => functionData.name)].forEach((exportName) => {
+    const size = getImportSize([baseClient, exportName]);
+    console.log(`${baseClient} + ${exportName}: ${formatBytes(size)}`);
+
+    if (!(baseClientSize < size) && !(baseClient === 'BaseRest' && exportName === 'Rest')) {
+      // Emit an error if adding the module does not increase the bundle size
+      // (this means that the module is not being tree-shaken correctly).
+      errors.push(new Error(`Adding ${exportName} to ${baseClient} does not increase the bundle size.`));
+    }
+  });
+});
+
+for (const functionData of functions) {
+  const { name: functionName, transitiveImports } = functionData;
+
+  // First display the size of the function
+  const standaloneSize = getImportSize([functionName]);
+  console.log(`${functionName}: ${formatBytes(standaloneSize)}`);
+
+  // Then display the size of the function together with the modules we expect
+  // it to transitively import
+  if (transitiveImports.length > 0) {
+    const withTransitiveImportsSize = getImportSize([functionName, ...transitiveImports]);
+    console.log(`${functionName} + ${transitiveImports.join(' + ')}: ${formatBytes(withTransitiveImportsSize)}`);
+
+    if (withTransitiveImportsSize > standaloneSize) {
+      // Emit an error if the bundle size is increased by adding the modules
+      // that we expect this function to have transitively imported anyway.
+      // This seemed like a useful sense check, but it might need tweaking in
+      // the future if we make future optimisations that mean that the
+      // standalone functions don’t necessarily import the whole module.
+      errors.push(
+        new Error(`Adding ${transitiveImports.join(' + ')} to ${functionName} unexpectedly increases the bundle size.`)
+      );
+    }
+  }
+}
+
+if (errors.length > 0) {
+  for (const error of errors) {
+    console.log(error.message);
+  }
+  process.exit(1);
+}

scripts/moduleReport.ts

@@ -17,6 +17,7 @@ const moduleNames = [
   'XHRRequest',
   'FetchRequest',
   'MessageInteractions',
+  'RealtimePublishing',
 ];
 
 // List of all free-standing functions exported by the library along with the

src/common/lib/client/acks.ts

@@ -0,0 +1,110 @@
+import ErrorInfo from '../types/errorinfo';
+import MessageQueue from '../transport/messagequeue';
+import Protocol, { PendingMessage } from '../transport/protocol';
+import Transport from '../transport/transport';
+import Logger from '../util/logger';
+import * as Utils from '../util/utils';
+import { ErrCallback } from '../../types/utils';
+import ConnectionManager from '../transport/connectionmanager';
+import Platform from 'common/platform';
+
+export class Acks {
+  messageQueue: MessageQueue;
+
+  constructor(transport: Transport) {
+    this.messageQueue = new MessageQueue();
+    transport.on('ack', (serial: number, count: number) => {
+      this.onAck(serial, count);
+    });
+    transport.on('nack', (serial: number, count: number, err: ErrorInfo) => {
+      this.onNack(serial, count, err);
+    });
+  }
+
+  onAck(serial: number, count: number): void {
+    Logger.logAction(Logger.LOG_MICRO, 'Protocol.onAck()', 'serial = ' + serial + '; count = ' + count);
+    this.messageQueue.completeMessages(serial, count);
+  }
+
+  onNack(serial: number, count: number, err: ErrorInfo): void {
+    Logger.logAction(
+      Logger.LOG_ERROR,
+      'Protocol.onNack()',
+      'serial = ' + serial + '; count = ' + count + '; err = ' + Utils.inspectError(err)
+    );
+    if (!err) {
+      err = new ErrorInfo('Unable to send message; channel not responding', 50001, 500);
+    }
+    this.messageQueue.completeMessages(serial, count, err);
+  }
+
+  onceIdle(listener: ErrCallback): void {
+    const messageQueue = this.messageQueue;
+    if (messageQueue.count() === 0) {
+      listener();
+      return;
+    }
+    messageQueue.once('idle', listener);
+  }
+
+  onProtocolWillSend(pendingMessage: PendingMessage) {
+    if (pendingMessage.ackRequired) {
+      this.messageQueue.push(pendingMessage);
+    }
+  }
+
+  getPendingMessages(): PendingMessage[] {
+    return this.messageQueue.copyAll();
+  }
+
+  clearPendingMessages(): void {
+    return this.messageQueue.clear();
+  }
+
+  thingMovedFromActivateTransport(existingActiveProtocol: Protocol, transport: Transport) {
+    if (this.messageQueue.count() > 0) {
+      /* We could just requeue pending messages on the new transport, but
+       * actually this should never happen: transports should only take over
+       * from other active transports when upgrading, and upgrading waits for
+       * the old transport to be idle. So log an error. */
+      Logger.logAction(
+        Logger.LOG_ERROR,
+        'ConnectionManager.activateTransport()',
+        'Previous active protocol (for transport ' +
+          existingActiveProtocol.transport.shortName +
+          ', new one is ' +
+          transport.shortName +
+          ') finishing with ' +
+          existingActiveProtocol.acks!.messageQueue.count() +
+          ' messages still pending'
+      );
+    }
+  }
+
+  thingMovedFromDeactivateTransport(connectionManager: ConnectionManager, currentProtocol: Protocol) {
+    Logger.logAction(
+      Logger.LOG_MICRO,
+      'ConnectionManager.deactivateTransport()',
+      'Getting, clearing, and requeuing ' + currentProtocol.acks!.messageQueue.count() + ' pending messages'
+    );
+    this.queuePendingMessages(connectionManager, this.getPendingMessages());
+    /* Clear any messages we requeue to allow the protocol to become idle.
+     * In case of an upgrade, this will trigger an immediate activation of
+     * the upgrade transport, so delay a tick so this transport can finish
+     * deactivating */
+    Platform.Config.nextTick(() => {
+      this.clearPendingMessages();
+    });
+  }
+
+  queuePendingMessages(connectionManager: ConnectionManager, pendingMessages: Array<PendingMessage>): void {
+    if (pendingMessages && pendingMessages.length) {
+      Logger.logAction(
+        Logger.LOG_MICRO,
+        'ConnectionManager.queuePendingMessages()',
+        'queueing ' + pendingMessages.length + ' pending messages'
+      );
+      connectionManager.queuedMessages.prepend(pendingMessages);
+    }
+  }
+}

src/common/lib/client/baserealtime.ts

@@ -12,12 +12,16 @@ import * as API from '../../../../ably';
 import { ModulesMap, RealtimePresenceModule } from './modulesmap';
 import { TransportNames } from 'common/constants/TransportName';
 import { TransportImplementations } from 'common/platform';
+import { RealtimePublishing } from './realtimepublishing';
+import { Acks } from './acks';
 
 /**
  `BaseRealtime` is an export of the tree-shakable version of the SDK, and acts as the base class for the `DefaultRealtime` class exported by the non tree-shakable version.
  */
 class BaseRealtime extends BaseClient {
   readonly _RealtimePresence: RealtimePresenceModule | null;
+  readonly _Acks: typeof Acks | null;
+  readonly __RealtimePublishing: typeof RealtimePublishing | null;
   // Extra transport implementations available to this client, in addition to those in Platform.Transports.bundledImplementations
   readonly _additionalTransportImplementations: TransportImplementations;
   _channels: any;
@@ -28,6 +32,8 @@ class BaseRealtime extends BaseClient {
     Logger.logAction(Logger.LOG_MINOR, 'Realtime()', '');
     this._additionalTransportImplementations = BaseRealtime.transportImplementationsFromModules(modules);
     this._RealtimePresence = modules.RealtimePresence ?? null;
+    this._Acks = modules.RealtimePublishing?.Acks ?? modules.RealtimePresence?.Acks ?? null;
+    this.__RealtimePublishing = modules.RealtimePublishing ?? null;
     this.connection = new Connection(this, this.options);
     this._channels = new Channels(this);
     if (options.autoConnect !== false) this.connect();
@@ -53,6 +59,13 @@ class BaseRealtime extends BaseClient {
     return this._channels;
   }
 
+  get _RealtimePublishing(): typeof RealtimePublishing {
+    if (!this.__RealtimePublishing) {
+      Utils.throwMissingModuleError('RealtimePublishing');
+    }
+    return this.__RealtimePublishing;
+  }
+
   connect(): void {
     Logger.logAction(Logger.LOG_MINOR, 'Realtime.connect()', '');
     this.connection.connect();

src/common/lib/client/defaultrealtime.ts

@@ -15,6 +15,8 @@ import {
   fromValues as presenceMessageFromValues,
   fromValuesArray as presenceMessagesFromValuesArray,
 } from '../types/presencemessage';
+import { RealtimePublishing } from './realtimepublishing';
+import { Acks } from './acks';
 
 /**
  `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.
@@ -34,9 +36,11 @@ export class DefaultRealtime extends BaseRealtime {
         RealtimePresence,
         presenceMessageFromValues,
         presenceMessagesFromValuesArray,
+        Acks,
       },
       WebSocketTransport: initialiseWebSocketTransport,
       MessageInteractions: FilteredSubscriptions,
+      RealtimePublishing: RealtimePublishing,
     });
   }
 

src/common/lib/client/modulesmap.ts

@@ -10,6 +10,8 @@ import {
   fromValues as presenceMessageFromValues,
   fromValuesArray as presenceMessagesFromValuesArray,
 } from '../types/presencemessage';
+import { RealtimePublishing } from './realtimepublishing';
+import { Acks } from './acks';
 
 export interface PresenceMessageModule {
   presenceMessageFromValues: typeof presenceMessageFromValues;
@@ -18,6 +20,7 @@ export interface PresenceMessageModule {
 
 export type RealtimePresenceModule = PresenceMessageModule & {
   RealtimePresence: typeof RealtimePresence;
+  Acks: typeof Acks;
 };
 
 export interface ModulesMap {
@@ -31,6 +34,7 @@ export interface ModulesMap {
   XHRRequest?: typeof XHRRequest;
   FetchRequest?: typeof fetchRequest;
   MessageInteractions?: typeof FilteredSubscriptions;
+  RealtimePublishing?: typeof RealtimePublishing;
 }
 
 export const allCommonModules: ModulesMap = { Rest };