channelgroup: emit channel assigned/active events

ably.d.ts

@@ -106,6 +106,11 @@ export type ChannelEvent =
   | ChannelEvents.DETACHING
   | ChannelEvents.UPDATE;
 
+/**
+ * Describes the events emitted by a {@link ChannelGroup} object.
+ */
+export type ChannelGroupEvent = 'active.updated' | 'assigned.updated';
+
 /**
  * The `ConnectionStates` namespace describes the possible values of the {@link ConnectionState} type.
  */
@@ -1489,6 +1494,12 @@ export type channelAndMessageCallback<T> = (channel: string, message: T) => void
  * @param changeStateChange - The state change that occurred.
  */
 export type channelEventCallback = (changeStateChange: ChannelStateChange) => void;
+/**
+ * The callback used for the events emitted by {@link ChannelGroup}.
+ *
+ * @param channels - The set of channels.
+ */
+export type channelGroupCallback = (channels: string[]) => void;
 /**
  * The callback used for the events emitted by {@link Connection}.
  *
@@ -2020,7 +2031,7 @@ export declare interface Channel {
  *
  * Enables messages to be subscribed to on a group of channels.
  */
-export declare interface ChannelGroup {
+export declare interface ChannelGroup extends EventEmitter<channelGroupCallback, string[], ChannelGroupEvent> {
   /**
    * Registers a listener for messages on this channel group. The caller supplies a listener function, which is called each time one or more messages arrives on the channels in the group.
    *

src/common/lib/client/baserealtime.ts

@@ -235,7 +235,7 @@ class ConsumerGroup extends EventEmitter {
   }
 }
 
-class ChannelGroup {
+class ChannelGroup extends EventEmitter {
   activeChannels: string[] = [];
   assignedChannels: string[] = [];
   active: RealtimeChannel;
@@ -246,6 +246,7 @@ class ChannelGroup {
   consumerGroup: ConsumerGroup;
 
   constructor(readonly channels: Channels, readonly filter: string, readonly options?: API.ChannelGroupOptions) {
+    super();
     this.subscriptions = new EventEmitter();
     this.active = channels.get(this.safeChannelName(options?.activeChannel || '$ably:active'));
     this.consumerGroup = new ConsumerGroup(channels, options?.consumerGroup?.name);
@@ -266,6 +267,7 @@ class ChannelGroup {
     await this.active.setOptions({ params: { rewind: '1' } });
     await this.active.subscribe((msg: any) => {
       this.activeChannels = msg.data.active;
+      this.emit('active.updated', this.activeChannels);
       this.updateAssignedChannels();
     });
   }
@@ -275,6 +277,7 @@ class ChannelGroup {
     await this.active.detach();
     await this.consumerGroup.leave();
     this.assignedChannels = [];
+    this.emit('assigned.updated', this.assignedChannels);
     this.removeSubscriptions(Object.keys(this.subscribedChannels));
   }
 
@@ -310,6 +313,7 @@ class ChannelGroup {
       'ChannelGroups.updateAssignedChannels',
       'assignedChannels=' + this.assignedChannels + ' consumerId=' + this.consumerGroup.consumerId,
     );
+    this.emit('assigned.updated', this.assignedChannels);
   }
 
   private unsubscribeTimeout(channel: string) {