Add DataStream support

.changeset/nine-lies-invent.md

@@ -0,0 +1,5 @@
+---
+"livekit-client": minor
+---
+
+Add DataStream support

examples/demo/demo.ts

@@ -35,7 +35,7 @@ import {
   supportsAV1,
   supportsVP9,
 } from '../../src/index';
-import { isSVCCodec } from '../../src/room/utils';
+import { isSVCCodec, sleep } from '../../src/room/utils';
 
 setLogLevel(LogLevel.debug);
 
@@ -48,6 +48,7 @@ const state = {
   defaultDevices: new Map<MediaDeviceKind, string>([['audioinput', 'default']]),
   bitrateInterval: undefined as any,
   e2eeKeyProvider: new ExternalE2EEKeyProvider(),
+  chatMessages: new Map<string, { text: string; participant?: Participant }>(),
 };
 let currentRoom: Room | undefined;
 
@@ -72,6 +73,15 @@ function updateSearchParams(url: string, token: string, key: string) {
 
 // handles actions from the HTML
 const appActions = {
+  sendFile: async () => {
+    console.log('start sending');
+    const file = ($('file') as HTMLInputElement).files?.[0]!;
+    currentRoom?.localParticipant.sendFile(file, {
+      mimeType: file.type,
+      topic: 'welcome',
+      onProgress: (progress) => console.log('sending file, progress', Math.ceil(progress * 100)),
+    });
+  },
   connectWithFormInput: async () => {
     const url = (<HTMLInputElement>$('url')).value;
     const token = (<HTMLInputElement>$('token')).value;
@@ -234,6 +244,50 @@ const appActions = {
         );
       });
 
+    room.setTextStreamHandler(async (reader, participant) => {
+      const info = reader.info;
+      if (info.size) {
+        handleChatMessage(
+          {
+            id: info.id,
+            timestamp: info.timestamp,
+            message: await reader.readAll(),
+          },
+          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),
+          );
+        }
+        appendLog('text stream finished');
+      }
+      console.log('final info including close extensions', reader.info);
+    }, 'chat');
+
+    room.setByteStreamHandler(async (reader, participant) => {
+      const info = reader.info;
+
+      appendLog(`started to receive a file called "${info.name}" from ${participant?.identity}`);
+      reader.onProgress = (progress) => {
+        console.log(`"progress ${progress ? (progress * 100).toFixed(0) : 'undefined'}%`);
+      };
+      const result = new Blob(await reader.readAll(), { type: info.mimeType });
+      appendLog(`completely received file called "${info.name}" from ${participant?.identity}`);
+      const downloadLink = URL.createObjectURL(result);
+      const linkEl = document.createElement('a');
+      linkEl.href = downloadLink;
+      linkEl.innerText = info.name;
+      linkEl.setAttribute('download', info.name);
+      document.body.append(linkEl);
+    }, 'welcome');
+
     try {
       // read and set current key from input
       const cryptoKey = (<HTMLSelectElement>$('crypto-key')).value;
@@ -295,32 +349,32 @@ const appActions = {
         const cssRules = [...styleSheet.cssRules].map((rule) => rule.cssText).join('');
         const style = document.createElement('style');
         style.textContent = cssRules;
-        pipWindow.document.head.appendChild(style);
+        pipWindow?.document.head.appendChild(style);
       } catch (e) {
         const link = document.createElement('link');
         link.rel = 'stylesheet';
         link.type = styleSheet.type;
         link.media = styleSheet.media;
-        link.href = styleSheet.href;
-        pipWindow.document.head.appendChild(link);
+        link.href = styleSheet.href!;
+        pipWindow?.document.head.appendChild(link);
       }
     });
     // Move participant videos to the Picture-in-Picture window
     const participantsArea = $('participants-area');
     const pipParticipantsArea = document.createElement('div');
     pipParticipantsArea.id = 'participants-area';
-    pipWindow.document.body.append(pipParticipantsArea);
+    pipWindow?.document.body.append(pipParticipantsArea);
     [...participantsArea.children].forEach((child) => pipParticipantsArea.append(child));
 
     // Move participant videos back when the Picture-in-Picture window closes.
-    pipWindow.addEventListener('pagehide', (event) => {
+    pipWindow?.addEventListener('pagehide', () => {
       setButtonState('toggle-pip-button', 'Open PiP', false);
       if (currentRoom?.state === ConnectionState.Connected)
         [...pipParticipantsArea.children].forEach((child) => participantsArea.append(child));
     });
 
     // Close PiP when room disconnects
-    currentRoom.once('disconnected', (e) => window.documentPictureInPicture?.window.close());
+    currentRoom!.once('disconnected', () => window.documentPictureInPicture?.window?.close());
   },
 
   ratchetE2EEKey: async () => {
@@ -402,7 +456,7 @@ const appActions = {
     if (!currentRoom) return;
     const textField = <HTMLInputElement>$('entry');
     if (textField.value) {
-      currentRoom.localParticipant.sendChatMessage(textField.value);
+      currentRoom.localParticipant.sendText(textField.value, { topic: 'chat' });
       textField.value = '';
     }
   },
@@ -492,13 +546,32 @@ declare global {
 window.appActions = appActions;
 
 // --------------------------- event handlers ------------------------------- //
+function handleChatMessage(msg: ChatMessage, participant?: Participant) {
+  state.chatMessages.set(msg.id, { text: msg.message, participant });
 
-function handleChatMessage(msg: ChatMessage, participant?: LocalParticipant | RemoteParticipant) {
-  (<HTMLTextAreaElement>$('chat')).value +=
-    `${participant?.identity}${participant instanceof LocalParticipant ? ' (me)' : ''}: ${msg.message}\n`;
+  const chatEl = <HTMLTextAreaElement>$('chat');
+  chatEl.value = '';
+  for (const chatMsg of state.chatMessages.values()) {
+    chatEl.value += `${chatMsg.participant?.identity}${chatMsg.participant instanceof LocalParticipant ? ' (me)' : ''}: ${chatMsg.text}\n`;
+  }
+}
+
+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',
+    destinationIdentities: [participant.identity],
+  });
+
+  for (const char of greeting) {
+    await streamWriter.write(char);
+    await sleep(20);
+  }
+  await streamWriter.close();
 }
 
-function participantConnected(participant: Participant) {
+async function participantConnected(participant: Participant) {
   appendLog('participant', participant.identity, 'connected', participant.metadata);
   participant
     .on(ParticipantEvent.TrackMuted, (pub: TrackPublication) => {
@@ -515,6 +588,8 @@ function participantConnected(participant: Participant) {
     .on(ParticipantEvent.ConnectionQualityChanged, () => {
       renderParticipant(participant);
     });
+
+  await sendGreetingTo(participant);
 }
 
 function participantDisconnected(participant: RemoteParticipant) {
@@ -803,14 +878,14 @@ function renderBitrate() {
   }
 }
 
-function getParticipantsAreaElement() {
+function getParticipantsAreaElement(): HTMLElement {
   return (
     window.documentPictureInPicture?.window?.document.querySelector('#participants-area') ||
     $('participants-area')
   );
 }
 
-function updateVideoSize(element: HTMLVideoElement, target: HTMLElement) {
+function updateVideoSize(element: HTMLVideoElement, target: Element) {
   target.innerHTML = `(${element.videoWidth}x${element.videoHeight})`;
 }
 

examples/demo/index.html

@@ -159,6 +159,8 @@ <h2>Livekit Sample App</h2>
               >
                 Ratchet Key
               </button>
+              <input id="file" type="file" />
+              <button onclick="appActions.sendFile()">Send file</button>
               <select
                 id="simulate-scenario"
                 class="custom-select"

package.json

@@ -55,7 +55,7 @@
   },
   "dependencies": {
     "@livekit/mutex": "1.1.1",
-    "@livekit/protocol": "1.30.0",
+    "@livekit/protocol": "1.32.0",
     "events": "^3.3.0",
     "loglevel": "^1.8.0",
     "sdp-transform": "^2.14.1",

src/index.ts

@@ -60,6 +60,7 @@ export type {
   SimulationScenario,
   TranscriptionSegment,
   ChatMessage,
+  SendTextOptions,
 } from './room/types';
 export * from './version';
 export {

src/room/RTCEngine.ts

@@ -1119,6 +1119,22 @@ export default class RTCEngine extends (EventEmitter as new () => TypedEventEmit
     }
   };
 
+  waitForBufferStatusLow(kind: DataPacket_Kind): Promise<void> {
+    return new Promise(async (resolve, reject) => {
+      if (this.isBufferStatusLow(kind)) {
+        resolve();
+      } else {
+        const onClosing = () => reject('Engine closed');
+        this.once(EngineEvent.Closing, onClosing);
+        while (!this.dcBufferStatus.get(kind)) {
+          await sleep(10);
+        }
+        this.off(EngineEvent.Closing, onClosing);
+        resolve();
+      }
+    });
+  }
+
   /**
    * @internal
    */