diff --git a/.changeset/pending-chat-sends.md b/.changeset/pending-chat-sends.md new file mode 100644 index 000000000..fc64b2bbc --- /dev/null +++ b/.changeset/pending-chat-sends.md @@ -0,0 +1,5 @@ +--- +'@livekit/components-core': patch +--- + +Keep chat and data-channel sending state true until all overlapping sends have finished. diff --git a/packages/core/src/components/chat.test.ts b/packages/core/src/components/chat.test.ts new file mode 100644 index 000000000..e3519dac3 --- /dev/null +++ b/packages/core/src/components/chat.test.ts @@ -0,0 +1,108 @@ +import { Encryption_Type } from '@livekit/protocol'; +import { Room, RoomEvent, type TextStreamInfo } from 'livekit-client'; +import { afterEach, describe, expect, it, vi } from 'vitest'; +import { Future } from '../helper/future'; +import { setupChat } from './chat'; + +describe('chat sending state', () => { + const rooms: Room[] = []; + const streamInfo: TextStreamInfo = { + id: 'message', + topic: 'lk.chat', + timestamp: 1, + mimeType: 'text/plain', + encryptionType: Encryption_Type.NONE, + }; + + afterEach(() => { + rooms.forEach((room) => room.emit(RoomEvent.Disconnected)); + rooms.length = 0; + vi.restoreAllMocks(); + }); + + it.each(['resolve', 'reject'] as const)( + 'keeps sending state true when an overlapping text send finishes with %s', + async (outcome) => { + const room = new Room(); + rooms.push(room); + const firstText = new Future(); + const secondText = new Future(); + vi.spyOn(room.localParticipant, 'sendText') + .mockReturnValueOnce(firstText.promise) + .mockReturnValueOnce(secondText.promise); + vi.spyOn(room.localParticipant, 'publishData').mockResolvedValue(); + const { send, isSendingObservable } = setupChat(room); + const states: boolean[] = []; + const subscription = isSendingObservable.subscribe((value) => states.push(value)); + const first = send('first'); + const second = send('second'); + const error = new Error('text send failed'); + + try { + expect(states.at(-1)).toBe(true); + if (outcome === 'resolve') { + firstText.resolve?.({ ...streamInfo, id: 'first' }); + await expect(first).resolves.toMatchObject({ id: 'first', message: 'first' }); + } else { + const rejected = expect(first).rejects.toBe(error); + firstText.reject?.(error); + await rejected; + } + expect(states.at(-1)).toBe(true); + const late = vi.fn(); + const lateSubscription = isSendingObservable.subscribe(late); + expect(late).toHaveBeenCalledExactlyOnceWith(true); + lateSubscription.unsubscribe(); + + secondText.resolve?.({ ...streamInfo, id: 'second' }); + await expect(second).resolves.toMatchObject({ id: 'second', message: 'second' }); + expect(states.at(-1)).toBe(false); + } finally { + firstText.resolve?.(streamInfo); + secondText.resolve?.(streamInfo); + await Promise.allSettled([first, second]); + subscription.unsubscribe(); + } + }, + ); + + it.each(['resolve', 'reject'] as const)( + 'waits for all overlapping legacy publishes when the first finishes with %s', + async (outcome) => { + const room = new Room(); + rooms.push(room); + vi.spyOn(room.localParticipant, 'sendText').mockResolvedValue(streamInfo); + const firstPublish = new Future(); + const secondPublish = new Future(); + const publish = vi + .spyOn(room.localParticipant, 'publishData') + .mockReturnValueOnce(firstPublish.promise) + .mockReturnValueOnce(secondPublish.promise); + const { send, isSendingObservable } = setupChat(room); + const states: boolean[] = []; + const subscription = isSendingObservable.subscribe((value) => states.push(value)); + const first = send('first'); + const second = send('second'); + + try { + await vi.waitFor(() => expect(publish).toHaveBeenCalledTimes(2)); + if (outcome === 'resolve') { + firstPublish.resolve?.(); + } else { + firstPublish.reject?.(new Error('legacy publish failed')); + } + await expect(first).resolves.toMatchObject({ message: 'first' }); + expect(states.at(-1)).toBe(true); + + secondPublish.resolve?.(); + await expect(second).resolves.toMatchObject({ message: 'second' }); + expect(states.at(-1)).toBe(false); + } finally { + firstPublish.resolve?.(); + secondPublish.resolve?.(); + await Promise.allSettled([first, second]); + subscription.unsubscribe(); + } + }, + ); +}); diff --git a/packages/core/src/components/chat.ts b/packages/core/src/components/chat.ts index 125a3648c..b177c9285 100644 --- a/packages/core/src/components/chat.ts +++ b/packages/core/src/components/chat.ts @@ -268,6 +268,7 @@ export function setupChat(room: Room, options?: ChatOptions) { ); const isSending$ = new BehaviorSubject(false); + let pendingSends = 0; const finalMessageEncoder = options?.messageEncoder ?? encodeLegacyMsg; const send = async (message: string, options?: SendTextOptions) => { @@ -275,6 +276,7 @@ export function setupChat(room: Room, options?: ChatOptions) { options = {}; } options.topic ??= topic; + pendingSends += 1; isSending$.next(true); try { @@ -316,7 +318,8 @@ export function setupChat(room: Room, options?: ChatOptions) { return receivedChatMsg; } finally { - isSending$.next(false); + pendingSends -= 1; + isSending$.next(pendingSends > 0); } }; diff --git a/packages/core/src/observables/dataChannel.test.ts b/packages/core/src/observables/dataChannel.test.ts index 16d43aad9..228d734ab 100644 --- a/packages/core/src/observables/dataChannel.test.ts +++ b/packages/core/src/observables/dataChannel.test.ts @@ -1,10 +1,55 @@ import { Room } from 'livekit-client'; import { afterEach, describe, expect, it, vi } from 'vitest'; import { setupDataMessageHandler } from './dataChannel'; +import { Future } from '../helper/future'; describe('data channel sending state', () => { afterEach(() => vi.restoreAllMocks()); + it.each(['resolve', 'reject'] as const)( + 'keeps sending state true when an overlapping publish finishes with %s', + async (outcome) => { + const room = new Room(); + const firstPublish = new Future(); + const secondPublish = new Future(); + vi.spyOn(room.localParticipant, 'publishData') + .mockReturnValueOnce(firstPublish.promise) + .mockReturnValueOnce(secondPublish.promise); + const { send, isSendingObservable } = setupDataMessageHandler(room, 'chat'); + const states: boolean[] = []; + const subscription = isSendingObservable.subscribe((value) => states.push(value)); + const first = send(new Uint8Array([1])); + const second = send(new Uint8Array([2])); + const error = new Error('publish failed'); + + try { + expect(states.at(-1)).toBe(true); + if (outcome === 'resolve') { + firstPublish.resolve?.(); + await expect(first).resolves.toBeUndefined(); + } else { + const rejected = expect(first).rejects.toBe(error); + firstPublish.reject?.(error); + await rejected; + } + expect(states.at(-1)).toBe(true); + const late = vi.fn(); + const lateSubscription = isSendingObservable.subscribe(late); + expect(late).toHaveBeenCalledExactlyOnceWith(true); + lateSubscription.unsubscribe(); + + secondPublish.resolve?.(); + await expect(second).resolves.toBeUndefined(); + expect(states.at(-1)).toBe(false); + } finally { + firstPublish.resolve?.(); + secondPublish.resolve?.(); + await Promise.allSettled([first, second]); + subscription.unsubscribe(); + } + }, + ); + it('sends without subscribing to the sending-state observable', async () => { const room = new Room(); const publish = vi.spyOn(room.localParticipant, 'publishData').mockResolvedValue(); diff --git a/packages/core/src/observables/dataChannel.ts b/packages/core/src/observables/dataChannel.ts index 23bff3be8..292ee5805 100644 --- a/packages/core/src/observables/dataChannel.ts +++ b/packages/core/src/observables/dataChannel.ts @@ -73,13 +73,16 @@ export function setupDataMessageHandler( ); const isSending$ = new BehaviorSubject(false); + let pendingSends = 0; const send = async (payload: Uint8Array, options: DataPublishOptions = {}) => { + pendingSends += 1; isSending$.next(true); try { await sendMessage(room.localParticipant, payload, { topic: topics[0], ...options }); } finally { - isSending$.next(false); + pendingSends -= 1; + isSending$.next(pendingSends > 0); } };