Ad hoc/fix thread messages stream updates on

.pubnub.yml

@@ -1,11 +1,16 @@
 ---
 name: pubnub-js-chat
-version: v0.5.1
+version: v0.5.2
 scm: github.com/pubnub/js-chat
 schema: 1
 files:
   - lib/dist/index.js
 changelog:
+  - date: 2024-01-16
+    version: v0.5.2
+    changes:
+      - type: feature
+        text: "Make ThreadMessage.streamUpdatesOn return ThreadMessage[] instead of Message[]."
   - date: 2023-12-18
     version: v0.5.1
     changes:

lib/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@pubnub/chat",
-  "version": "0.5.1",
+  "version": "0.5.2",
   "description": "PubNub JavaScript Chat SDK",
   "author": "PubNub <[email protected]>",
   "license": "SEE LICENSE IN LICENSE",

lib/src/entities/message.ts

@@ -111,14 +111,14 @@ export class Message {
   }
 
   /** @internal */
-  private clone(params: Partial<MessageFields>) {
+  protected clone(params: Partial<MessageFields>) {
     const { timetoken, content, channelId, userId, actions, meta } = this
     const data = Object.assign({}, { timetoken, content, channelId, userId, actions, meta }, params)
     return new Message(this.chat, data)
   }
 
   /** @internal */
-  private assignAction(action: PubNub.MessageAction) {
+  protected assignAction(action: PubNub.MessageAction) {
     const { actionTimetoken, type, value, uuid } = action
     const newActions = this.actions || {}
     newActions[type] ||= {}
@@ -131,7 +131,7 @@ export class Message {
   }
 
   /** @internal */
-  private filterAction(action: PubNub.MessageAction) {
+  protected filterAction(action: PubNub.MessageAction) {
     const { actionTimetoken, type, value, uuid } = action
     const newActions = this.actions || {}
     newActions[type] ||= {}

lib/src/entities/thread-message.ts

@@ -3,6 +3,7 @@ import { Chat } from "./chat"
 import { ThreadMessageDTOParams } from "../types"
 import { Channel } from "./channel"
 import { getErrorProxiedEntity } from "../error-logging"
+import PubNub from "pubnub"
 
 export class ThreadMessage extends Message {
   readonly parentChannelId: string
@@ -29,6 +30,51 @@ export class ThreadMessage extends Message {
     return getErrorProxiedEntity(new ThreadMessage(chat, data), chat.errorLogger)
   }
 
+  /** @internal */
+  protected clone(params: Partial<MessageFields>) {
+    const { timetoken, content, channelId, userId, actions, meta, parentChannelId } = this
+    const data = Object.assign(
+      {},
+      { parentChannelId, timetoken, content, channelId, userId, actions, meta },
+      params
+    )
+    return new ThreadMessage(this.chat, data)
+  }
+
+  static streamUpdatesOn(
+    threadMessages: ThreadMessage[],
+    callback: (threadMessages: ThreadMessage[]) => unknown
+  ) {
+    if (!threadMessages.length) throw "Cannot stream message updates on an empty list"
+    const listener = {
+      messageAction: (event: PubNub.MessageActionEvent) => {
+        const threadMessage = threadMessages.find(
+          (msg) => msg.timetoken === event.data.messageTimetoken
+        )
+        if (!threadMessage) return
+        if (threadMessage.channelId !== event.channel) return
+        let actions
+        if (event.event === "added") actions = threadMessage.assignAction(event.data)
+        if (event.event === "removed") actions = threadMessage.filterAction(event.data)
+        const newMessage = threadMessage.clone({ actions })
+        const newMessages = threadMessages.map((msg) =>
+          msg.timetoken === newMessage.timetoken ? newMessage : msg
+        )
+        callback(newMessages)
+      },
+    }
+    const { chat } = threadMessages[0]
+    const removeListener = chat.addListener(listener)
+    const subscriptions = threadMessages
+      .filter((m1, i) => threadMessages.findIndex((m2) => m1.channelId === m2.channelId) === i)
+      .map((message) => chat.subscribe(message.channelId))
+
+    return () => {
+      removeListener()
+      subscriptions.map((unsub) => unsub())
+    }
+  }
+
   async pinToParentChannel() {
     const parentChannel = await this.chat.getChannel(this.parentChannelId)
     if (!parentChannel) {