Support annotations

.eslintrc.js

@@ -21,6 +21,13 @@ module.exports = {
     // TODO remove this once we start using the full @typescript-eslint/recommended ruleset in #958
     'no-redeclare': 'off',
     '@typescript-eslint/no-redeclare': 'error',
+    'jsdoc/multiline-blocks': [
+      'warn',
+      {
+        noZeroLineText: false,
+        noFinalLineText: false,
+      },
+    ],
   },
   overrides: [
     {

modular.d.ts

@@ -20,6 +20,8 @@
  * | `PresenceMessageStatic.fromEncodedArray()` | [`decodeEncryptedPresenceMessages()`](../functions/modular.decodeEncryptedPresenceMessages.html)     |
  * | `PresenceMessageStatic.fromEncodedArray()` | [`decodePresenceMessages()`](../functions/modular.decodePresenceMessages.html)     |
  * | `PresenceMessageStatic.fromValues()`       | [`constructPresenceMessage()`](../functions/modular.constructPresenceMessage.html) |
+ * | `AnnotationStatic.fromEncoded()`           | [`decodeAnnotation()`](../functions/modular.decodeAnnotation.html)                 |
+ * | `AnnotationStatic.fromEncodedArray()`      | [`decodeAnnotations()`](../functions/modular.decodeAnnotations.html)               |
  *
  * @module
  */
@@ -31,6 +33,7 @@ import {
   Crypto as CryptoClass,
   MessageStatic,
   PresenceMessageStatic,
+  AnnotationStatic,
   RealtimeClient,
   Auth,
   Channels,
@@ -65,6 +68,8 @@ export declare const decodePresenceMessages: PresenceMessageStatic['fromEncodedA
 export declare const decodeEncryptedPresenceMessage: PresenceMessageStatic['fromEncoded'];
 export declare const decodeEncryptedPresenceMessages: PresenceMessageStatic['fromEncodedArray'];
 export declare const constructPresenceMessage: PresenceMessageStatic['fromValues'];
+export declare const decodeAnnotation: AnnotationStatic['fromEncoded'];
+export declare const decodeAnnotations: AnnotationStatic['fromEncodedArray'];
 
 /**
  * Provides REST-related functionality to a {@link BaseRealtime} client.
@@ -135,6 +140,21 @@ export declare const MsgPack: unknown;
  */
 export declare const RealtimePresence: unknown;
 
+/**
+ * Provides a {@link BaseRealtime} instance with the ability to interact with message
+ * annotations.
+ *
+ * To create a client that includes this plugin, include it in the client options that you pass to the {@link BaseRealtime.constructor}:
+ *
+ * ```javascript
+ * import { BaseRealtime, WebSocketTransport, FetchRequest, Annotations } from 'ably/modular';
+ * const realtime = new BaseRealtime({ ...options, plugins: { WebSocketTransport, FetchRequest, Annotations } });
+ * ```
+ *
+ * If you do not provide this plugin, then attempting to access a channel’s {@link ably!RealtimeChannel.annotations} property will cause a runtime error.
+ */
+export declare const Annotations: unknown;
+
 /**
  * Provides a {@link BaseRealtime} instance with the ability to establish a connection with the Ably realtime service using a [WebSocket](https://developer.mozilla.org/en-US/docs/Web/API/WebSockets_API) connection.
  *
@@ -230,6 +250,11 @@ export interface ModularPlugins {
    */
   RealtimePresence?: typeof RealtimePresence;
 
+  /**
+   * See {@link Annotations | documentation for the `Annotations` plugin}.
+   */
+  Annotations?: typeof Annotations;
+
   /**
    * See {@link WebSocketTransport | documentation for the `WebSocketTransport` plugin}.
    */

src/common/lib/client/baseclient.ts

@@ -12,6 +12,7 @@ import * as Utils from '../util/utils';
 import Platform from '../../platform';
 import { Rest } from './rest';
 import { IUntypedCryptoStatic } from 'common/types/ICryptoStatic';
+import { AnnotationsPlugin } from './modularplugins';
 import { throwMissingPluginError } from '../util/utils';
 import { MsgPack } from 'common/types/msgpack';
 import { HTTPRequestImplementations } from 'platform/web/lib/http/http';
@@ -46,6 +47,7 @@ class BaseClient {
   // Extra HTTP request implementations available to this client, in addition to those in web’s Http.bundledRequestImplementations
   readonly _additionalHTTPRequestImplementations: HTTPRequestImplementations | null;
   private readonly __FilteredSubscriptions: typeof FilteredSubscriptions | null;
+  readonly _Annotations: AnnotationsPlugin | null;
   readonly logger: Logger;
   _device?: LocalDevice;
 
@@ -98,6 +100,7 @@ class BaseClient {
     this._rest = options.plugins?.Rest ? new options.plugins.Rest(this) : null;
     this._Crypto = options.plugins?.Crypto ?? null;
     this.__FilteredSubscriptions = options.plugins?.MessageInteractions ?? null;
+    this._Annotations = options.plugins?.Annotations ?? null;
   }
 
   get rest(): Rest {

src/common/lib/client/defaultrealtime.ts

@@ -9,10 +9,14 @@ import { DefaultMessage } from '../types/defaultmessage';
 import { MsgPack } from 'common/types/msgpack';
 import RealtimePresence from './realtimepresence';
 import { DefaultPresenceMessage } from '../types/defaultpresencemessage';
+import { DefaultAnnotation } from '../types/defaultannotation';
 import WebSocketTransport from '../transport/websockettransport';
 import { FilteredSubscriptions } from './filteredsubscriptions';
 import { PresenceMap } from './presencemap';
 import PresenceMessage, { WirePresenceMessage } from '../types/presencemessage';
+import RealtimeAnnotations from './realtimeannotations';
+import RestAnnotations from './restannotations';
+import Annotation, { WireAnnotation } from '../types/annotation';
 import { Http } from 'common/types/http';
 import Defaults from '../util/defaults';
 import Logger from '../util/logger';
@@ -38,6 +42,12 @@ export class DefaultRealtime extends BaseRealtime {
           PresenceMessage,
           WirePresenceMessage,
         },
+        Annotations: {
+          Annotation,
+          WireAnnotation,
+          RealtimeAnnotations,
+          RestAnnotations,
+        },
         WebSocketTransport,
         MessageInteractions: FilteredSubscriptions,
       }),
@@ -62,6 +72,7 @@ export class DefaultRealtime extends BaseRealtime {
 
   static Message = DefaultMessage;
   static PresenceMessage = DefaultPresenceMessage;
+  static Annotation = DefaultAnnotation;
 
   static _MsgPack: MsgPack | null = null;
 

src/common/lib/client/defaultrest.ts

@@ -5,7 +5,11 @@ import Platform from 'common/platform';
 import { DefaultMessage } from '../types/defaultmessage';
 import { MsgPack } from 'common/types/msgpack';
 import { DefaultPresenceMessage } from '../types/defaultpresencemessage';
+import { DefaultAnnotation } from '../types/defaultannotation';
 import { Http } from 'common/types/http';
+import RealtimeAnnotations from './realtimeannotations';
+import RestAnnotations from './restannotations';
+import Annotation, { WireAnnotation } from '../types/annotation';
 import Defaults from '../util/defaults';
 import Logger from '../util/logger';
 
@@ -25,6 +29,12 @@ export class DefaultRest extends BaseRest {
         ...allCommonModularPlugins,
         Crypto: DefaultRest.Crypto ?? undefined,
         MsgPack: DefaultRest._MsgPack ?? undefined,
+        Annotations: {
+          Annotation,
+          WireAnnotation,
+          RealtimeAnnotations,
+          RestAnnotations,
+        },
       }),
     );
   }
@@ -43,6 +53,7 @@ export class DefaultRest extends BaseRest {
 
   static Message = DefaultMessage;
   static PresenceMessage = DefaultPresenceMessage;
+  static Annotation = DefaultAnnotation;
 
   static _MsgPack: MsgPack | null = null;
 

src/common/lib/client/modularplugins.ts

@@ -2,10 +2,13 @@ import { Rest } from './rest';
 import { IUntypedCryptoStatic } from '../../types/ICryptoStatic';
 import { MsgPack } from 'common/types/msgpack';
 import RealtimePresence from './realtimepresence';
+import RealtimeAnnotations from './realtimeannotations';
+import RestAnnotations from './restannotations';
 import XHRRequest from 'platform/web/lib/http/request/xhrrequest';
 import fetchRequest from 'platform/web/lib/http/request/fetchrequest';
 import { FilteredSubscriptions } from './filteredsubscriptions';
 import PresenceMessage, { WirePresenceMessage } from '../types/presencemessage';
+import Annotation, { WireAnnotation } from '../types/annotation';
 import { TransportCtor } from '../transport/transport';
 import * as PushPlugin from 'plugins/push';
 
@@ -18,11 +21,19 @@ export type RealtimePresencePlugin = PresenceMessagePlugin & {
   RealtimePresence: typeof RealtimePresence;
 };
 
+export type AnnotationsPlugin = {
+  Annotation: typeof Annotation;
+  WireAnnotation: typeof WireAnnotation;
+  RealtimeAnnotations: typeof RealtimeAnnotations;
+  RestAnnotations: typeof RestAnnotations;
+};
+
 export interface ModularPlugins {
   Rest?: typeof Rest;
   Crypto?: IUntypedCryptoStatic;
   MsgPack?: MsgPack;
   RealtimePresence?: RealtimePresencePlugin;
+  Annotations?: AnnotationsPlugin;
   WebSocketTransport?: TransportCtor;
   XHRPolling?: TransportCtor;
   XHRRequest?: typeof XHRRequest;

src/common/lib/client/realtimeannotations.ts

@@ -0,0 +1,103 @@
+import EventEmitter from '../util/eventemitter';
+import Logger from '../util/logger';
+import Annotation from '../types/annotation';
+import { actions, flags } from '../types/protocolmessagecommon';
+import { fromValues as protocolMessageFromValues } from '../types/protocolmessage';
+import ErrorInfo from '../types/errorinfo';
+import RealtimeChannel from './realtimechannel';
+import RestAnnotations, { RestGetAnnotationsParams, constructValidateAnnotation } from './restannotations';
+import type { PaginatedResult } from './paginatedresource';
+import type Message from '../types/message';
+import type { Properties } from '../util/utils';
+
+class RealtimeAnnotations {
+  private channel: RealtimeChannel;
+  private logger: Logger;
+  private subscriptions: EventEmitter;
+
+  constructor(channel: RealtimeChannel) {
+    this.channel = channel;
+    this.logger = channel.logger;
+    this.subscriptions = new EventEmitter(this.logger);
+  }
+
+  async publish(msgOrSerial: string | Message, annotationValues: Partial<Properties<Annotation>>): Promise<void> {
+    const channelName = this.channel.name;
+    const annotation = constructValidateAnnotation(msgOrSerial, annotationValues);
+    const wireAnnotation = await annotation.encode();
+
+    this.channel._throwIfUnpublishableState();
+
+    Logger.logAction(
+      this.logger,
+      Logger.LOG_MICRO,
+      'RealtimeAnnotations.publish()',
+      'channelName = ' +
+        channelName +
+        ', sending annotation with messageSerial = ' +
+        annotation.messageSerial +
+        ', type = ' +
+        annotation.type,
+    );
+
+    const pm = protocolMessageFromValues({
+      action: actions.ANNOTATION,
+      channel: channelName,
+      annotations: [wireAnnotation],
+    });
+    return this.channel.sendMessage(pm);
+  }
+
+  async delete(msgOrSerial: string | Message, annotationValues: Partial<Properties<Annotation>>): Promise<void> {
+    annotationValues.action = 'annotation.delete';
+    return this.publish(msgOrSerial, annotationValues);
+  }
+
+  async subscribe(..._args: unknown[] /* [type], listener */): Promise<void> {
+    const args = RealtimeChannel.processListenerArgs(_args);
+    const event = args[0];
+    const listener = args[1];
+    const channel = this.channel;
+
+    if (channel.state === 'failed') {
+      throw ErrorInfo.fromValues(channel.invalidStateError());
+    }
+
+    this.subscriptions.on(event, listener);
+
+    if (this.channel.channelOptions.attachOnSubscribe !== false) {
+      await channel.attach();
+    }
+
+    // explicit check for attach state in caes attachOnSubscribe=false
+    if ((this.channel.state === 'attached' && this.channel._mode & flags.ANNOTATION_SUBSCRIBE) === 0) {
+      throw new ErrorInfo(
+        "You are trying to add an annotation listener, but you haven't requested the annotation_subscribe channel mode in ChannelOptions, so this won't do anything (we only deliver annotations to clients who have explicitly requested them)",
+        93001,
+        400,
+      );
+    }
+  }
+
+  unsubscribe(..._args: unknown[] /* [event], listener */): void {
+    const args = RealtimeChannel.processListenerArgs(_args);
+    const event = args[0];
+    const listener = args[1];
+    this.subscriptions.off(event, listener);
+  }
+
+  _processIncoming(annotations: Annotation[]): void {
+    for (const annotation of annotations) {
+      this.subscriptions.emit(annotation.type || '', annotation);
+    }
+  }
+
+  async get(
+    msgOrSerial: string | Message,
+    params: RestGetAnnotationsParams | null,
+  ): Promise<PaginatedResult<Annotation>> {
+    return RestAnnotations.prototype.get.call(this, msgOrSerial, params);
+  }
+}
+
+export default RealtimeAnnotations;