11/* eslint-disable camelcase */
2- import type { Participant , Room , ChatMessage } from 'livekit-client' ;
2+ import type { Participant , Room , ChatMessage , SendTextOptions } from 'livekit-client' ;
33import { compareVersions , RoomEvent } from 'livekit-client' ;
4- import { BehaviorSubject , Subject , scan , map , takeUntil , merge } from 'rxjs' ;
4+ import { BehaviorSubject , Subject , scan , map , takeUntil , from , filter } from 'rxjs' ;
55import {
66 DataTopic ,
7+ LegacyDataTopic ,
78 sendMessage ,
8- setupChatMessageHandler ,
99 setupDataMessageHandler ,
1010} from '../observables/dataChannel' ;
1111
@@ -18,11 +18,11 @@ export interface ReceivedChatMessage extends ChatMessage {
1818}
1919
2020export interface LegacyChatMessage extends ChatMessage {
21- ignore ?: boolean ;
21+ ignoreLegacy ?: boolean ;
2222}
2323
2424export interface LegacyReceivedChatMessage extends ReceivedChatMessage {
25- ignore ?: boolean ;
25+ ignoreLegacy ?: boolean ;
2626}
2727
2828/**
@@ -41,83 +41,86 @@ export type ChatOptions = {
4141 messageEncoder ?: ( message : LegacyChatMessage ) => Uint8Array ;
4242 /** @deprecated the new chat API doesn't rely on encoders and decoders anymore and uses a dedicated chat API instead */
4343 messageDecoder ?: ( message : Uint8Array ) => LegacyReceivedChatMessage ;
44- /** @deprecated the new chat API doesn't rely on topics anymore and uses a dedicated chat API instead */
4544 channelTopic ?: string ;
46- /** @deprecated the new chat API doesn't rely on topics anymore and uses a dedicated chat API instead */
45+ /** @deprecated the new chat API doesn't rely on update topics anymore and uses a dedicated chat API instead */
4746 updateChannelTopic ?: string ;
4847} ;
4948
50- type RawMessage = {
51- payload : Uint8Array ;
52- topic : string | undefined ;
53- from : Participant | undefined ;
54- } ;
55-
56- const encoder = new TextEncoder ( ) ;
57- const decoder = new TextDecoder ( ) ;
49+ const topicSubjectMap : WeakMap < Room , Map < string , Subject < ReceivedChatMessage > > > = new WeakMap ( ) ;
5850
59- const topicSubjectMap : Map < Room , Map < string , Subject < RawMessage > > > = new Map ( ) ;
51+ function isIgnorableChatMessage ( msg : ReceivedChatMessage | LegacyReceivedChatMessage ) {
52+ return ( msg as LegacyChatMessage ) . ignoreLegacy == true ;
53+ }
6054
61- const encode = ( message : LegacyReceivedChatMessage ) => encoder . encode ( JSON . stringify ( message ) ) ;
55+ const decodeLegacyMsg = ( message : Uint8Array ) =>
56+ JSON . parse ( new TextDecoder ( ) . decode ( message ) ) as LegacyReceivedChatMessage | ReceivedChatMessage ;
6257
63- const decode = ( message : Uint8Array ) =>
64- JSON . parse ( decoder . decode ( message ) ) as LegacyReceivedChatMessage | ReceivedChatMessage ;
58+ const encodeLegacyMsg = ( message : LegacyReceivedChatMessage ) =>
59+ new TextEncoder ( ) . encode ( JSON . stringify ( message ) ) ;
6560
6661export function setupChat ( room : Room , options ?: ChatOptions ) {
67- const onDestroyObservable = new Subject < void > ( ) ;
68-
69- const serverSupportsChatApi = ( ) =>
62+ const serverSupportsDataStreams = ( ) =>
7063 room . serverInfo ?. edition === 1 ||
71- ( ! ! room . serverInfo ?. version && compareVersions ( room . serverInfo ?. version , '1.17.2' ) > 0 ) ;
72-
73- const { messageDecoder, messageEncoder, channelTopic, updateChannelTopic } = options ?? { } ;
64+ ( ! ! room . serverInfo ?. version && compareVersions ( room . serverInfo ?. version , '1.8.2' ) > 0 ) ;
7465
75- const topic = channelTopic ?? DataTopic . CHAT ;
66+ const onDestroyObservable = new Subject < void > ( ) ;
7667
77- const updateTopic = updateChannelTopic ?? DataTopic . CHAT_UPDATE ;
68+ const topic = options ?. channelTopic ?? DataTopic . CHAT ;
69+ const legacyTopic = options ?. channelTopic ?? LegacyDataTopic . CHAT ;
7870
7971 let needsSetup = false ;
8072 if ( ! topicSubjectMap . has ( room ) ) {
8173 needsSetup = true ;
8274 }
83- const topicMap = topicSubjectMap . get ( room ) ?? new Map < string , Subject < RawMessage > > ( ) ;
84- const messageSubject = topicMap . get ( topic ) ?? new Subject < RawMessage > ( ) ;
75+ const topicMap = topicSubjectMap . get ( room ) ?? new Map < string , Subject < ReceivedChatMessage > > ( ) ;
76+ const messageSubject = topicMap . get ( topic ) ?? new Subject < ReceivedChatMessage > ( ) ;
8577 topicMap . set ( topic , messageSubject ) ;
8678 topicSubjectMap . set ( room , topicMap ) ;
8779
80+ const finalMessageDecoder = options ?. messageDecoder ?? decodeLegacyMsg ;
8881 if ( needsSetup ) {
89- /** Subscribe to all appropriate messages sent over the wire. */
90- const { messageObservable } = setupDataMessageHandler ( room , [ topic , updateTopic ] ) ;
91- messageObservable . pipe ( takeUntil ( onDestroyObservable ) ) . subscribe ( messageSubject ) ;
82+ room . registerTextStreamHandler ( topic , async ( reader , participantInfo ) => {
83+ const { id, timestamp } = reader . info ;
84+ const streamObservable = from ( reader ) . pipe (
85+ scan ( ( acc : string , chunk : string ) => {
86+ return acc + chunk ;
87+ } ) ,
88+ map ( ( chunk : string ) => {
89+ return {
90+ id,
91+ timestamp,
92+ message : chunk ,
93+ from : room . getParticipantByIdentity ( participantInfo . identity ) ,
94+ // editTimestamp: type === 'update' ? timestamp : undefined,
95+ } as ReceivedChatMessage ;
96+ } ) ,
97+ ) ;
98+ streamObservable . subscribe ( {
99+ next : ( value ) => messageSubject . next ( value ) ,
100+ } ) ;
101+ } ) ;
102+
103+ /** legacy chat protocol handling */
104+ const { messageObservable } = setupDataMessageHandler ( room , [ legacyTopic ] ) ;
105+ messageObservable
106+ . pipe (
107+ map ( ( msg ) => {
108+ const parsedMessage = finalMessageDecoder ( msg . payload ) ;
109+ if ( isIgnorableChatMessage ( parsedMessage ) ) {
110+ return undefined ;
111+ }
112+ const newMessage : ReceivedChatMessage = { ...parsedMessage , from : msg . from } ;
113+ return newMessage ;
114+ } ) ,
115+ filter ( ( msg ) => ! ! msg ) ,
116+ takeUntil ( onDestroyObservable ) ,
117+ )
118+ . subscribe ( messageSubject ) ;
92119 }
93- const { chatObservable, send : sendChatMessage } = setupChatMessageHandler ( room ) ;
94-
95- const finalMessageDecoder = messageDecoder ?? decode ;
96120
97121 /** Build up the message array over time. */
98- const messagesObservable = merge (
99- messageSubject . pipe (
100- map ( ( msg ) => {
101- const parsedMessage = finalMessageDecoder ( msg . payload ) ;
102- const newMessage = { ...parsedMessage , from : msg . from } ;
103- if ( isIgnorableChatMessage ( newMessage ) ) {
104- return undefined ;
105- }
106- return newMessage ;
107- } ) ,
108- ) ,
109- chatObservable . pipe (
110- map ( ( [ msg , participant ] ) => {
111- return { ...msg , from : participant } ;
112- } ) ,
113- ) ,
114- ) . pipe (
115- scan < ReceivedChatMessage | undefined , ReceivedChatMessage [ ] > ( ( acc , value ) => {
116- // ignore legacy message updates
117- if ( ! value ) {
118- return acc ;
119- }
120- // handle message updates
122+ const messagesObservable = messageSubject . pipe (
123+ scan < ReceivedChatMessage , ReceivedChatMessage [ ] > ( ( acc , value ) => {
121124 if (
122125 'id' in value &&
123126 acc . find ( ( msg ) => msg . from ?. identity === value . from ?. identity && msg . id === value . id )
@@ -128,10 +131,9 @@ export function setupChat(room: Room, options?: ChatOptions) {
128131 acc [ replaceIndex ] = {
129132 ...value ,
130133 timestamp : originalMsg . timestamp ,
131- editTimestamp : value . editTimestamp ?? value . timestamp ,
134+ editTimestamp : value . timestamp ,
132135 } ;
133136 }
134-
135137 return [ ...acc ] ;
136138 }
137139 return [ ...acc , value ] ;
@@ -140,42 +142,34 @@ export function setupChat(room: Room, options?: ChatOptions) {
140142 ) ;
141143
142144 const isSending$ = new BehaviorSubject < boolean > ( false ) ;
145+ const finalMessageEncoder = options ?. messageEncoder ?? encodeLegacyMsg ;
143146
144- const finalMessageEncoder = messageEncoder ?? encode ;
145-
146- const send = async ( message : string ) => {
147+ const send = async ( message : string , options ?: SendTextOptions ) => {
148+ if ( ! options ) {
149+ options = { } ;
150+ }
151+ options . topic ??= topic ;
147152 isSending$ . next ( true ) ;
153+
148154 try {
149- const chatMessage = await sendChatMessage ( message ) ;
155+ const info = await room . localParticipant . sendText ( message , options ) ;
156+ const chatMsg : ReceivedChatMessage = {
157+ id : info . id ,
158+ timestamp : Date . now ( ) ,
159+ message,
160+ from : room . localParticipant ,
161+ attachedFiles : options . attachments ,
162+ } ;
163+ messageSubject . next ( chatMsg ) ;
150164 const encodedLegacyMsg = finalMessageEncoder ( {
151- ...chatMessage ,
152- ignore : serverSupportsChatApi ( ) ,
165+ ...chatMsg ,
166+ ignoreLegacy : serverSupportsDataStreams ( ) ,
153167 } ) ;
154168 await sendMessage ( room . localParticipant , encodedLegacyMsg , {
155169 reliable : true ,
156- topic,
170+ topic : legacyTopic ,
157171 } ) ;
158- return chatMessage ;
159- } finally {
160- isSending$ . next ( false ) ;
161- }
162- } ;
163-
164- const update = async ( message : string , originalMessageOrId : string | ChatMessage ) => {
165- const timestamp = Date . now ( ) ;
166- const originalMessage : ChatMessage =
167- typeof originalMessageOrId === 'string'
168- ? { id : originalMessageOrId , message : '' , timestamp }
169- : originalMessageOrId ;
170- isSending$ . next ( true ) ;
171- try {
172- const editedMessage = await room . localParticipant . editChatMessage ( message , originalMessage ) ;
173- const encodedLegacyMessage = finalMessageEncoder ( editedMessage ) ;
174- await sendMessage ( room . localParticipant , encodedLegacyMessage , {
175- topic : updateTopic ,
176- reliable : true ,
177- } ) ;
178- return editedMessage ;
172+ return chatMsg ;
179173 } finally {
180174 isSending$ . next ( false ) ;
181175 }
@@ -184,20 +178,15 @@ export function setupChat(room: Room, options?: ChatOptions) {
184178 function destroy ( ) {
185179 onDestroyObservable . next ( ) ;
186180 onDestroyObservable . complete ( ) ;
181+ messageSubject . complete ( ) ;
187182 topicSubjectMap . delete ( room ) ;
183+ room . unregisterTextStreamHandler ( topic ) ;
188184 }
189185 room . once ( RoomEvent . Disconnected , destroy ) ;
190186
191187 return {
192188 messageObservable : messagesObservable ,
193189 isSendingObservable : isSending$ ,
194190 send,
195- update,
196191 } ;
197192}
198-
199- function isIgnorableChatMessage (
200- msg : ReceivedChatMessage | LegacyReceivedChatMessage ,
201- ) : msg is ReceivedChatMessage {
202- return ( msg as LegacyChatMessage ) . ignore == true ;
203- }
0 commit comments