Merge pull request #1347 from ably/has-backlog

feat: add `ChannelStateChange.hasBacklog` and return state change to attach promise/callback

ably.d.ts

@@ -1149,6 +1149,10 @@ declare namespace Types {
      * Indicates whether message continuity on this channel is preserved, see [Nonfatal channel errors](https://ably.com/docs/realtime/channels#nonfatal-errors) for more info.
      */
     resumed: boolean;
+    /**
+     * Indicates whether the client can expect a backlog of messages from a rewind or resume.
+     */
+    hasBacklog?: boolean;
   }
 
   /**
@@ -2556,9 +2560,9 @@ declare namespace Types {
     /**
      * Attach to this channel ensuring the channel is created in the Ably system and all messages published on the channel are received by any channel listeners registered using {@link RealtimeChannelCallbacks.subscribe | `subscribe()`}. Any resulting channel state change will be emitted to any listeners registered using the {@link EventEmitter.on | `on()`} or {@link EventEmitter.once | `once()`} methods. As a convenience, `attach()` is called implicitly if {@link RealtimeChannelCallbacks.subscribe | `subscribe()`} for the channel is called, or {@link RealtimePresenceCallbacks.enter | `enter()`} or {@link RealtimePresenceCallbacks.subscribe | `subscribe()`} are called on the {@link RealtimePresenceCallbacks} object for this channel.
      *
-     * @param callback - A function which will be called upon completion of the operation. If the operation succeeded, then the function will be called with `null`. If it failed, the function will be called with information about the error.
+     * @param callback - A function which will be called upon completion of the operation. If the operation succeeded and the channel became attached, then the function will be called with a {@link ChannelStateChange} object. If the channel was already attached the function will be called with `null`. If it failed, the function will be called with information about the error.
      */
-    attach(callback?: errorCallback): void;
+    attach(callback?: StandardCallback<ChannelStateChange | null>): void;
     /**
      * Detach from this channel. Any resulting channel state change is emitted to any listeners registered using the {@link EventEmitter.on | `on()`} or {@link EventEmitter.once | `once()`} methods. Once all clients globally have detached from the channel, the channel will be released in the Ably service within two minutes.
      *
@@ -2590,32 +2594,44 @@ declare namespace Types {
      *
      * @param event - The event name.
      * @param listener - An event listener function.
-     * @param callbackWhenAttached - A function which will be called upon completion of the channel {@link RealtimeChannelCallbacks.attach | `attach()`} operation. If the operation succeeded, then the function will be called with `null`. If it failed, the function will be called with information about the error.
+     * @param callbackWhenAttached - A function which will be called upon completion of the channel {@link RealtimeChannelCallbacks.attach | `attach()`} operation. If the operation succeeded and the channel became attached, then the function will be called with a {@link ChannelStateChange} object. If the channel was already attached the function will be called with `null`. If it failed, the function will be called with information about the error.
      */
-    subscribe(event: string, listener?: messageCallback<Message>, callbackWhenAttached?: errorCallback): void;
+    subscribe(
+      event: string,
+      listener?: messageCallback<Message>,
+      callbackWhenAttached?: StandardCallback<ChannelStateChange | null>
+    ): void;
     /**
      * Registers a listener for messages on this channel for multiple event name values.
      *
      * @param events - An array of event names.
      * @param listener - An event listener function.
-     * @param callbackWhenAttached - A function which will be called upon completion of the channel {@link RealtimeChannelCallbacks.attach | `attach()`} operation. If the operation succeeded, then the function will be called with `null`. If it failed, the function will be called with information about the error.
+     * @param callbackWhenAttached - A function which will be called upon completion of the channel {@link RealtimeChannelCallbacks.attach | `attach()`} operation. If the operation succeeded and the channel became attached, then the function will be called with a {@link ChannelStateChange} object. If the channel was already attached the function will be called with `null`. If it failed, the function will be called with information about the error.
      */
-    subscribe(events: Array<string>, listener?: messageCallback<Message>, callbackWhenAttached?: errorCallback): void;
+    subscribe(
+      events: Array<string>,
+      listener?: messageCallback<Message>,
+      callbackWhenAttached?: StandardCallback<ChannelStateChange | null>
+    ): void;
     /**
      * Registers a listener for messages on this channel that match the supplied filter.
      *
      * @param filter - A {@link MessageFilter}.
      * @param listener - An event listener function.
-     * @param callbackWhenAttached - A function which will be called upon completion of the channel {@link RealtimeChannelCallbacks.attach | `attach()`} operation. If the operation succeeded, then the function will be called with `null`. If it failed, the function will be called with information about the error.
+     * @param callbackWhenAttached - A function which will be called upon completion of the channel {@link RealtimeChannelCallbacks.attach | `attach()`} operation. If the operation succeeded and the channel became attached, then the function will be called with a {@link ChannelStateChange} object. If the channel was already attached the function will be called with `null`. If it failed, the function will be called with information about the error.
      */
-    subscribe(filter: MessageFilter, listener?: messageCallback<Message>, callbackWhenAttached?: errorCallback): void;
+    subscribe(
+      filter: MessageFilter,
+      listener?: messageCallback<Message>,
+      callbackWhenAttached?: StandardCallback<ChannelStateChange | null>
+    ): void;
     /**
      * Registers a listener for messages on this channel. The caller supplies a listener function, which is called each time one or more messages arrives on the channel.
      *
      * @param listener - An event listener function.
-     * @param callbackWhenAttached - A function which will be called upon completion of the channel {@link RealtimeChannelCallbacks.attach | `attach()`} operation. If the operation succeeded, then the function will be called with `null`. If it failed, the function will be called with information about the error.
+     * @param callbackWhenAttached - A function which will be called upon completion of the channel {@link RealtimeChannelCallbacks.attach | `attach()`} operation. If the operation succeeded and the channel became attached, then the function will be called with a {@link ChannelStateChange} object. If the channel was already attached the function will be called with `null`. If it failed, the function will be called with information about the error.
      */
-    subscribe(listener: messageCallback<Message>, callbackWhenAttached?: errorCallback): void;
+    subscribe(listener: messageCallback<Message>, callbackWhenAttached?: StandardCallback<ChannelStateChange>): void;
     /**
      * Publishes a single message to the channel with the given event name and payload. When publish is called with this client library, it won't attempt to implicitly attach to the channel, so long as [transient publishing](https://ably.com/docs/realtime/channels#transient-publish) is available in the library. Otherwise, the client will implicitly attach.
      *
@@ -2666,9 +2682,9 @@ declare namespace Types {
     /**
      * Attach to this channel ensuring the channel is created in the Ably system and all messages published on the channel are received by any channel listeners registered using {@link RealtimeChannelPromise.subscribe | `subscribe()`}. Any resulting channel state change will be emitted to any listeners registered using the {@link EventEmitter.on | `on()`} or {@link EventEmitter.once | `once()`} methods. As a convenience, `attach()` is called implicitly if {@link RealtimeChannelPromise.subscribe | `subscribe()`} for the channel is called, or {@link RealtimePresencePromise.enter | `enter()`} or {@link RealtimePresencePromise.subscribe | `subscribe()`} are called on the {@link RealtimePresencePromise} object for this channel.
      *
-     * @returns A promise which resolves upon success of the operation and rejects with an {@link ErrorInfo} object upon its failure.
+     * @returns A promise which, upon success, if the channel became attached will be fulfilled with a {@link ChannelStateChange} object. If the channel was already attached the promise will be fulfilled with `null`. Upon failure, the promise will be rejected with an {@link ErrorInfo} object.
      */
-    attach(): Promise<void>;
+    attach(): Promise<ChannelStateChange | null>;
     /**
      * Detach from this channel. Any resulting channel state change is emitted to any listeners registered using the {@link EventEmitter.on | `on()`} or {@link EventEmitter.once | `once()`} methods. Once all clients globally have detached from the channel, the channel will be released in the Ably service within two minutes.
      *
@@ -2694,32 +2710,32 @@ declare namespace Types {
      *
      * @param event - The event name.
      * @param listener - An event listener function.
-     * @returns A promise which resolves upon success of the channel {@link RealtimeChannelPromise.attach | `attach()`} operation and rejects with an {@link ErrorInfo} object upon its failure.
+     * @returns A promise which, upon successful attachment to the channel, will be fulfilled with a {@link ChannelStateChange} object. If the channel was already attached the promise will be resolved with `null`. Upon failure, the promise will be rejected with an {@link ErrorInfo} object.
      */
-    subscribe(event: string, listener?: messageCallback<Message>): Promise<void>;
+    subscribe(event: string, listener?: messageCallback<Message>): Promise<ChannelStateChange | null>;
     /**
      * Registers a listener for messages on this channel for multiple event name values.
      *
      * @param events - An array of event names.
      * @param listener - An event listener function.
-     * @returns A promise which resolves upon success of the channel {@link RealtimeChannelPromise.attach | `attach()`} operation and rejects with an {@link ErrorInfo} object upon its failure.
+     * @returns A promise which, upon successful attachment to the channel, will be fulfilled with a {@link ChannelStateChange} object. If the channel was already attached the promise will be resolved with `null`. Upon failure, the promise will be rejected with an {@link ErrorInfo} object.
      */
-    subscribe(events: Array<string>, listener?: messageCallback<Message>): Promise<void>;
+    subscribe(events: Array<string>, listener?: messageCallback<Message>): Promise<ChannelStateChange | null>;
     /**
      * Registers a listener for messages on this channel that match the supplied filter.
      *
      * @param filter - A {@link MessageFilter}.
      * @param listener - An event listener function.
-     * @returns A promise which resolves upon success of the channel {@link RealtimeChannelPromise.attach | `attach()`} operation and rejects with an {@link ErrorInfo} object upon its failure.
+     * @returns A promise which, upon successful attachment to the channel, will be fulfilled with a {@link ChannelStateChange} object. If the channel was already attached the promise will be resolved with `null`. Upon failure, the promise will be rejected with an {@link ErrorInfo} object.
      */
-    subscribe(filter: MessageFilter, listener?: messageCallback<Message>): Promise<void>;
+    subscribe(filter: MessageFilter, listener?: messageCallback<Message>): Promise<ChannelStateChange | null>;
     /**
      * Registers a listener for messages on this channel. The caller supplies a listener function, which is called each time one or more messages arrives on the channel.
      *
      * @param callback - An event listener function.
-     * @returns A promise which resolves upon success of the channel {@link RealtimeChannelPromise.attach | `attach()`} operation and rejects with an {@link ErrorInfo} object upon its failure.
+     * @returns A promise which, upon successful attachment to the channel, will be fulfilled with a {@link ChannelStateChange} object. If the channel was already attached the promise will be resolved with `null`. Upon failure, the promise will be rejected with an {@link ErrorInfo} object.
      */
-    subscribe(callback: messageCallback<Message>): Promise<void>;
+    subscribe(callback: messageCallback<Message>): Promise<ChannelStateChange | null>;
     /**
      * Publishes a single message to the channel with the given event name and payload. When publish is called with this client library, it won't attempt to implicitly attach to the channel, so long as [transient publishing](https://ably.com/docs/realtime/channels#transient-publish) is available in the library. Otherwise, the client will implicitly attach.
      *

src/common/lib/client/channelstatechange.ts

@@ -5,11 +5,21 @@ class ChannelStateChange {
   current: string;
   resumed?: boolean;
   reason?: string | Error | ErrorInfo;
+  hasBacklog?: boolean;
 
-  constructor(previous: string, current: string, resumed?: boolean, reason?: string | Error | ErrorInfo | null) {
+  constructor(
+    previous: string,
+    current: string,
+    resumed?: boolean,
+    hasBacklog?: boolean,
+    reason?: string | Error | ErrorInfo | null
+  ) {
     this.previous = previous;
     this.current = current;
-    if (current === 'attached') this.resumed = resumed;
+    if (current === 'attached') {
+      this.resumed = resumed;
+      this.hasBacklog = hasBacklog;
+    }
     if (reason) this.reason = reason;
   }
 }

src/common/lib/client/realtimechannel.ts

@@ -12,7 +12,7 @@ import ConnectionErrors from '../transport/connectionerrors';
 import * as API from '../../../../ably';
 import ConnectionManager from '../transport/connectionmanager';
 import ConnectionStateChange from './connectionstatechange';
-import { ErrCallback, PaginatedResultCallback } from '../../types/utils';
+import { ErrCallback, PaginatedResultCallback, StandardCallback } from '../../types/utils';
 import Realtime from './realtime';
 
 interface RealtimeHistoryParams {
@@ -272,7 +272,10 @@ class RealtimeChannel extends Channel {
     }
   }
 
-  attach(flags?: API.Types.ChannelMode[] | ErrCallback, callback?: ErrCallback): void | Promise<void> {
+  attach(
+    flags?: API.Types.ChannelMode[] | ErrCallback,
+    callback?: StandardCallback<ChannelStateChange | null>
+  ): void | Promise<ChannelStateChange> {
     let _flags: API.Types.ChannelMode[] | null | undefined;
     if (typeof flags === 'function') {
       callback = flags;
@@ -296,14 +299,18 @@ class RealtimeChannel extends Channel {
        * current mode differs from requested mode */
       this._requestedFlags = _flags as API.Types.ChannelMode[];
     } else if (this.state === 'attached') {
-      callback();
+      callback(null, null);
       return;
     }
 
     this._attach(false, null, callback);
   }
 
-  _attach(forceReattach: boolean, attachReason: ErrorInfo | null, callback?: ErrCallback): void {
+  _attach(
+    forceReattach: boolean,
+    attachReason: ErrorInfo | null,
+    callback?: StandardCallback<ChannelStateChange>
+  ): void {
     if (!callback) {
       callback = function (err?: ErrorInfo | null) {
         if (err) {
@@ -325,7 +332,7 @@ class RealtimeChannel extends Channel {
     this.once(function (this: { event: string }, stateChange: ChannelStateChange) {
       switch (this.event) {
         case 'attached':
-          callback?.();
+          callback?.(null, stateChange);
           break;
         case 'detached':
         case 'suspended':
@@ -422,7 +429,7 @@ class RealtimeChannel extends Channel {
     this.sendMessage(msg, callback || noop);
   }
 
-  subscribe(...args: unknown[] /* [event], listener, [callback] */): void | Promise<void> {
+  subscribe(...args: unknown[] /* [event], listener, [callback] */): void | Promise<ChannelStateChange> {
     const [event, listener, callback] = RealtimeChannel.processListenerArgs(args);
 
     if (!callback && this.realtime.options.promises) {
@@ -615,12 +622,13 @@ class RealtimeChannel extends Channel {
         this.modes = (modesFromFlags && Utils.allToLowerCase(modesFromFlags)) || undefined;
         const resumed = message.hasFlag('RESUMED');
         const hasPresence = message.hasFlag('HAS_PRESENCE');
+        const hasBacklog = message.hasFlag('HAS_BACKLOG');
         if (this.state === 'attached') {
           if (!resumed) {
             /* On a loss of continuity, the presence set needs to be re-synced */
             this.presence.onAttached(hasPresence);
           }
-          const change = new ChannelStateChange(this.state, this.state, resumed, message.error);
+          const change = new ChannelStateChange(this.state, this.state, resumed, hasBacklog, message.error);
           this._allChannelChanges.emit('update', change);
           if (!resumed || this.channelOptions.updateOnAttached) {
             this.emit('update', change);
@@ -629,7 +637,7 @@ class RealtimeChannel extends Channel {
           /* RTL5i: re-send DETACH and remain in the 'detaching' state */
           this.checkPendingState();
         } else {
-          this.notifyState('attached', message.error, resumed, hasPresence);
+          this.notifyState('attached', message.error, resumed, hasPresence, hasBacklog);
         }
         break;
       }
@@ -790,7 +798,8 @@ class RealtimeChannel extends Channel {
     state: API.Types.ChannelState,
     reason?: ErrorInfo | null,
     resumed?: boolean,
-    hasPresence?: boolean
+    hasPresence?: boolean,
+    hasBacklog?: boolean
   ): void {
     Logger.logAction(
       Logger.LOG_MICRO,
@@ -816,7 +825,7 @@ class RealtimeChannel extends Channel {
     if (reason) {
       this.errorReason = reason;
     }
-    const change = new ChannelStateChange(this.state, state, resumed, reason);
+    const change = new ChannelStateChange(this.state, state, resumed, hasBacklog, reason);
     const logLevel = state === 'failed' ? Logger.LOG_ERROR : Logger.LOG_MAJOR;
     Logger.logAction(
       logLevel,

src/common/lib/client/realtimepresence.ts

@@ -477,7 +477,7 @@ class RealtimePresence extends Presence {
           const msg = 'Presence auto-re-enter failed: ' + err.toString();
           const wrappedErr = new ErrorInfo(msg, 91004, 400);
           Logger.logAction(Logger.LOG_ERROR, 'RealtimePresence._ensureMyMembersPresent()', msg);
-          const change = new ChannelStateChange(this.channel.state, this.channel.state, true, wrappedErr);
+          const change = new ChannelStateChange(this.channel.state, this.channel.state, true, false, wrappedErr);
           this.channel.emit('update', change);
         }
       };