Add text auto chunking to text streams

.changeset/empty-buttons-brake.md

@@ -0,0 +1,5 @@
+---
+"livekit-client": patch
+---
+
+Add auto chunking to text streams

examples/demo/demo.ts

@@ -264,16 +264,15 @@ const appActions = {
           room.getParticipantByIdentity(participant?.identity),
         );
       } else {
-        for await (const msg of reader) {
-          handleChatMessage(
-            {
-              id: info.id,
-              timestamp: info.timestamp,
-              message: msg.collected,
-            },
-            room.getParticipantByIdentity(participant?.identity),
-          );
-        }
+        handleChatMessage(
+          {
+            id: info.id,
+            timestamp: info.timestamp,
+            message: await reader.readAll(),
+          },
+          room.getParticipantByIdentity(participant?.identity),
+        );
+
         appendLog('text stream finished');
       }
       console.log('final info including close extensions', reader.info);
@@ -515,7 +514,7 @@ const appActions = {
     if (!currentRoom) return;
     const textField = <HTMLInputElement>$('entry');
     if (textField.value) {
-      currentRoom.localParticipant.sendText(textField.value, { topic: 'chat' });
+      currentRoom.localParticipant.sendText(textField.value, { topic: 'lk.chat' });
       textField.value = '';
     }
   },
@@ -619,7 +618,7 @@ async function sendGreetingTo(participant: Participant) {
   const greeting = `Hello new participant ${participant.identity}. This is just an progressively updating chat message from me, participant ${currentRoom?.localParticipant.identity}.`;
 
   const streamWriter = await currentRoom!.localParticipant.streamText({
-    topic: 'chat',
+    topic: 'lk.chat',
     destinationIdentities: [participant.identity],
   });
 

src/room/StreamReader.ts

@@ -1,5 +1,5 @@
 import type { DataStream_Chunk } from '@livekit/protocol';
-import type { BaseStreamInfo, ByteStreamInfo, TextStreamChunk, TextStreamInfo } from './types';
+import type { BaseStreamInfo, ByteStreamInfo, TextStreamInfo } from './types';
 import { bigIntToNumber } from './utils';
 
 abstract class BaseStreamReader<T extends BaseStreamInfo> {
@@ -124,7 +124,7 @@ export class TextStreamReader extends BaseStreamReader<TextStreamInfo> {
     const decoder = new TextDecoder();
 
     return {
-      next: async (): Promise<IteratorResult<TextStreamChunk>> => {
+      next: async (): Promise<IteratorResult<string>> => {
         try {
           const { done, value } = await reader.read();
           if (done) {
@@ -134,14 +134,7 @@ export class TextStreamReader extends BaseStreamReader<TextStreamInfo> {
 
             return {
               done: false,
-              value: {
-                index: bigIntToNumber(value.chunkIndex),
-                current: decoder.decode(value.content),
-                collected: Array.from(this.receivedChunks.values())
-                  .sort((a, b) => bigIntToNumber(a.chunkIndex) - bigIntToNumber(b.chunkIndex))
-                  .map((chunk) => decoder.decode(chunk.content))
-                  .join(''),
-              },
+              value: decoder.decode(value.content),
             };
           }
         } catch (error) {
@@ -150,19 +143,19 @@ export class TextStreamReader extends BaseStreamReader<TextStreamInfo> {
         }
       },
 
-      async return(): Promise<IteratorResult<TextStreamChunk>> {
+      async return(): Promise<IteratorResult<string>> {
         reader.releaseLock();
         return { done: true, value: undefined };
       },
     };
   }
 
   async readAll(): Promise<string> {
-    let latestString: string = '';
-    for await (const { collected } of this) {
-      latestString = collected;
+    let finalString: string = '';
+    for await (const chunk of this) {
+      finalString += chunk;
     }
-    return latestString;
+    return finalString;
   }
 }
 

src/room/StreamWriter.ts

@@ -1,23 +1,23 @@
 import type { BaseStreamInfo, ByteStreamInfo, TextStreamInfo } from './types';
 
 class BaseStreamWriter<T, InfoType extends BaseStreamInfo> {
-  protected writableStream: WritableStream<[T, number?]>;
+  protected writableStream: WritableStream<T>;
 
-  protected defaultWriter: WritableStreamDefaultWriter<[T, number?]>;
+  protected defaultWriter: WritableStreamDefaultWriter<T>;
 
   protected onClose?: () => void;
 
   readonly info: InfoType;
 
-  constructor(writableStream: WritableStream<[T, number?]>, info: InfoType, onClose?: () => void) {
+  constructor(writableStream: WritableStream<T>, info: InfoType, onClose?: () => void) {
     this.writableStream = writableStream;
     this.defaultWriter = writableStream.getWriter();
     this.onClose = onClose;
     this.info = info;
   }
 
   write(chunk: T): Promise<void> {
-    return this.defaultWriter.write([chunk]);
+    return this.defaultWriter.write(chunk);
   }
 
   async close() {

src/room/participant/LocalParticipant.ts

@@ -90,6 +90,7 @@ import {
   isWeb,
   numberToBigInt,
   sleep,
+  splitUtf8,
   supportsAV1,
   supportsVP9,
 } from '../utils';
@@ -1536,19 +1537,9 @@ export default class LocalParticipant extends Participant {
       attachedStreamIds: fileIds,
     });
 
-    const textChunkSize = Math.floor(STREAM_CHUNK_SIZE / 4); // utf8 is at most 4 bytes long, so play it safe and take a quarter of the byte size to slice the string
-    const totalTextChunks = Math.ceil(totalTextLength / textChunkSize);
-
-    for (let i = 0; i < totalTextChunks; i++) {
-      const chunkData = text.slice(
-        i * textChunkSize,
-        Math.min((i + 1) * textChunkSize, totalTextLength),
-      );
-      await this.engine.waitForBufferStatusLow(DataPacket_Kind.RELIABLE);
-      await writer.write(chunkData);
-
-      handleProgress(Math.ceil((i + 1) / totalTextChunks), 0);
-    }
+    await writer.write(text);
+    // set text part of progress to 1
+    handleProgress(1, 0);
 
     await writer.close();
 
@@ -1614,20 +1605,13 @@ export default class LocalParticipant extends Participant {
     let chunkId = 0;
     const localP = this;
 
-    const writableStream = new WritableStream<[string, number?]>({
+    const writableStream = new WritableStream<string>({
       // Implement the sink
-      write([textChunk]) {
-        const textInBytes = new TextEncoder().encode(textChunk);
-
-        if (textInBytes.byteLength > STREAM_CHUNK_SIZE) {
-          this.abort?.();
-          throw new Error('chunk size too large');
-        }
-
-        return new Promise(async (resolve) => {
+      async write(text) {
+        for (const textChunk in splitUtf8(text, STREAM_CHUNK_SIZE)) {
           await localP.engine.waitForBufferStatusLow(DataPacket_Kind.RELIABLE);
           const chunk = new DataStream_Chunk({
-            content: textInBytes,
+            content: new TextEncoder().encode(textChunk),
             streamId,
             chunkIndex: numberToBigInt(chunkId),
           });
@@ -1641,8 +1625,7 @@ export default class LocalParticipant extends Participant {
           await localP.engine.sendDataPacket(chunkPacket, DataPacket_Kind.RELIABLE);
 
           chunkId += 1;
-          resolve();
-        });
+        }
       },
       async close() {
         const trailer = new DataStream_Trailer({

src/room/types.ts

@@ -119,9 +119,3 @@ export interface ByteStreamInfo extends BaseStreamInfo {
 }
 
 export interface TextStreamInfo extends BaseStreamInfo {}
-
-export type TextStreamChunk = {
-  index: number;
-  current: string;
-  collected: string;
-};

src/room/utils.ts

@@ -625,3 +625,19 @@ export function isLocalParticipant(p: Participant): p is LocalParticipant {
 export function isRemoteParticipant(p: Participant): p is RemoteParticipant {
   return !p.isLocal;
 }
+
+export function splitUtf8(s: string, n: number): string[] {
+  // adapted from https://stackoverflow.com/a/6043797
+  const result: string[] = [];
+  while (s.length > n) {
+    let k = n;
+    // Move back to find the start of a UTF-8 character
+    while ((s.charCodeAt(k) & 0xc0) === 0x80) {
+      k--;
+    }
+    result.push(s.slice(0, k));
+    s = s.slice(k);
+  }
+  result.push(s);
+  return result;
+}