feat: stop polling when chat relat is present. Process incoming messages from chat relay

Signed-off-by: Dorra Jaouad <[email protected]>

Author: DorraJaouad

src/composables/useGetMessages.ts

@@ -74,6 +74,8 @@ export function useGetMessagesProvider() {
 	const loadingNewMessages = ref(false)
 	const isInitialisingMessages = ref(true)
 	const stopFetchingOldMessages = ref(false)
+	let chatRelayEnabled = false
+	let chatRelaySupported = false
 
 	/**
 	 * Returns whether the current participant is a participant of current conversation.
@@ -144,6 +146,7 @@ export function useGetMessagesProvider() {
 			}
 			if (oldToken && oldToken !== newToken) {
 				store.dispatch('cancelPollNewMessages', { requestId: oldToken })
+				stopChatRelay()
 			}
 
 			if (newToken && canGetMessages) {
@@ -162,6 +165,7 @@ export function useGetMessagesProvider() {
 	subscribe('networkOnline', handleNetworkOnline)
 	EventBus.on('route-change', onRouteChange)
 	EventBus.on('set-context-id-to-bottom', setContextIdToBottom)
+	EventBus.on('signaling-supported-features', checkChatRelaySupport)
 
 	/** Every 30 seconds we remove expired messages from the store */
 	expirationInterval = setInterval(() => {
@@ -173,6 +177,8 @@ export function useGetMessagesProvider() {
 		unsubscribe('networkOnline', handleNetworkOnline)
 		EventBus.off('route-change', onRouteChange)
 		EventBus.off('set-context-id-to-bottom', setContextIdToBottom)
+		EventBus.off('chat-relay-message-received', addMessageFromChatRelay)
+		EventBus.off('signaling-supported-features', checkChatRelaySupport)
 
 		store.dispatch('cancelPollNewMessages', { requestId: currentToken.value })
 		clearInterval(pollingTimeout)
@@ -482,6 +488,10 @@ export function useGetMessagesProvider() {
 	 * @param token token of conversation where a method was called
 	 */
 	async function pollNewMessages(token: string) {
+		if (chatRelayEnabled) {
+			// Stop polling if chat relay is supported
+			return
+		}
 		// Check that the token has not changed
 		if (currentToken.value !== token) {
 			console.debug(`token has changed to ${currentToken.value}, breaking the loop for ${token}`)
@@ -499,6 +509,7 @@ export function useGetMessagesProvider() {
 				requestId: token,
 			})
 			debugTimer.end(`${token} | long polling`, 'status 200')
+			tryChatRelay()
 		} catch (exception) {
 			if (Axios.isCancel(exception)) {
 				debugTimer.end(`${token} | long polling`, 'cancelled')
@@ -512,6 +523,7 @@ export function useGetMessagesProvider() {
 				// This is not an error, so reset error timeout and poll again
 				pollingErrorTimeout = 1_000
 				clearTimeout(pollingTimeout)
+				tryChatRelay()
 				pollingTimeout = setTimeout(() => {
 					pollNewMessages(token)
 				}, 500)
@@ -539,6 +551,66 @@ export function useGetMessagesProvider() {
 		}, 500)
 	}
 
+	/**
+	 * Try to start chat relay
+	 */
+	function tryChatRelay() {
+		if (chatRelaySupported && isChatEndReached.value) {
+			startChatRelay()
+		}
+	}
+
+	/**
+	 * Check whether chat relay is supported
+	 *
+	 * @param features
+	 */
+	function checkChatRelaySupport(features: string[]) {
+		if (features.includes('chat-relay')) {
+			chatRelaySupported = true
+		}
+	}
+
+	/**
+	 * Initialize chat relay support by stopping polling and listening to chat relay messages
+	 */
+	function startChatRelay() {
+		if (currentToken.value) {
+			// it might have been set already, ensure we cancel it
+			store.dispatch('cancelPollNewMessages', { requestId: currentToken.value })
+		}
+		chatRelayEnabled = true
+		EventBus.on('chat-relay-message-received', addMessageFromChatRelay)
+	}
+
+	/**
+	 *
+	 * Chat relay sends one message at a time, we update our stores directly
+	 *
+	 * @param payload
+	 * @param payload.token
+	 * @param payload.message
+	 */
+	function addMessageFromChatRelay(payload: { token: string, message: ChatMessage }) {
+		const { token, message } = payload
+		if (token !== currentToken.value) {
+			// Guard: Message is for another conversation
+			// e.g., user switched conversation while messages were in-flight
+			return
+		}
+
+		chatStore.processChatBlocks(token, [message], { mergeBy: conversationLastMessageId.value })
+		store.dispatch('processMessage', { token, message })
+	}
+
+	/**
+	 * Stop chat relay and remove listener
+	 */
+	function stopChatRelay() {
+		chatRelayEnabled = false
+		EventBus.off('chat-relay-message-received', addMessageFromChatRelay)
+	}
+
 	provide(GET_MESSAGES_CONTEXT_KEY, {
 		contextMessageId,
 		loadingOldMessages,

src/services/EventBus.ts

@@ -21,6 +21,8 @@ import mitt from 'mitt'
 export type Events = {
 	[key: EventType]: unknown
 	'audio-player-ended': number
+	'chat-relay-message-received': { token: string, message: ChatMessage }
+	'chat-relay-supported': void
 	'conversations-received': { singleConversation?: Conversation, fromBrowserStorage?: boolean }
 	'session-conflict-confirmation': string
 	'deleted-session-detected': void
@@ -39,7 +41,6 @@ export type Events = {
 	'route-change': { from: RouteLocation, to: RouteLocation }
 	'set-context-id-to-bottom': void
 	'scroll-chat-to-bottom': { smooth?: boolean, force?: boolean }
-	'should-refresh-chat-messages': void
 	'should-refresh-conversations': { token: string, properties: Partial<Conversation> } | { all: true } | void
 	'signaling-join-call': [string, number]
 	'signaling-join-call-failed': [string, { meta: components['schemas']['OCSMeta'], data: { error: string } }]
@@ -53,6 +54,7 @@ export type Events = {
 	'signaling-users-joined': [StandaloneSignalingJoinSession[]]
 	'signaling-users-left': [string[]]
 	'signaling-all-users-changed-in-call-to-disconnected': void
+	'signaling-supported-features': string[]
 	'smart-picker-open': void
 	'switch-to-conversation': { token: string }
 	'talk:poll-added': { token: string, message: ChatMessage }

src/utils/signaling.js

@@ -1113,6 +1113,7 @@ Signaling.Standalone.prototype.helloResponseReceived = function(data) {
 		for (i = 0; i < features.length; i++) {
 			this.features[features[i]] = true
 		}
+		this._trigger('supportedFeatures', Object.keys(this.features))
 	}
 
 	if (!this.settings.helloAuthParams.internal
@@ -1430,8 +1431,12 @@ Signaling.Standalone.prototype.processRoomEvent = function(data) {
 Signaling.Standalone.prototype.processRoomMessageEvent = function(token, data) {
 	switch (data.type) {
 		case 'chat':
-		// FIXME this is not listened to
-			EventBus.emit('should-refresh-chat-messages')
+			if ('comment' in data.chat) {
+				EventBus.emit('chat-relay-message-received', { token, message: data.chat.comment })
+			} else {
+				// FIXME this is not listened to
+				EventBus.emit('should-refresh-chat-messages')
+			}
 			break
 		case 'recording':
 			EventBus.emit('signaling-recording-status-changed', [token, data.recording.status])