Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

[SDK-3630] feat: add ChannelStateChange.hasBacklog and return state change to attach promise/callback #1347

Merged
merged 2 commits into from
Jun 29, 2023
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
56 changes: 36 additions & 20 deletions ably.d.ts
Original file line number Diff line number Diff line change
Expand Up @@ -1149,6 +1149,10 @@
* 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;
}

/**
Expand Down Expand Up @@ -2556,9 +2560,9 @@
/**
* 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.
*
Expand Down Expand Up @@ -2590,32 +2594,44 @@
*
* @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.
*
Expand Down Expand Up @@ -2666,9 +2682,9 @@
/**
* 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.
*
Expand All @@ -2694,32 +2710,32 @@
*
* @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.
*
Expand Down Expand Up @@ -2763,8 +2779,8 @@
*/
get(name: string, channelOptions?: ChannelOptions): T;
/**
* @experimental This is a preview feature and may change in a future non-major release.

Check warning on line 2782 in ably.d.ts

View workflow job for this annotation

GitHub Actions / lint

Invalid JSDoc tag name "experimental"
* This experimental method allows you to create custom realtime data feeds by selectively subscribing

Check warning on line 2783 in ably.d.ts

View workflow job for this annotation

GitHub Actions / lint

Expected no lines between tags
* to receive only part of the data from the channel.
* See the [announcement post](https://pages.ably.com/subscription-filters-preview) for more information.
*
Expand Down
14 changes: 12 additions & 2 deletions src/common/lib/client/channelstatechange.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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;
}
}
Expand Down
29 changes: 19 additions & 10 deletions src/common/lib/client/realtimechannel.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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 {
Expand Down Expand Up @@ -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;
Expand All @@ -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) {
Expand All @@ -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':
Expand Down Expand Up @@ -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) {
Expand Down Expand Up @@ -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);
Expand All @@ -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;
}
Expand Down Expand Up @@ -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,
Expand All @@ -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,
Expand Down
2 changes: 1 addition & 1 deletion src/common/lib/client/realtimepresence.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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);
}
};
Expand Down
Loading
Loading