Add support for topics on data messages (#597)

* Add support for setting topics on data messages

* make destination comment clearer

* update protocol to current main

* update REASON to RR enum

* Create .changeset/metal-otters-give.md

.changeset/metal-otters-give.md

@@ -0,0 +1,5 @@
+---
+"livekit-client": patch
+---
+
+Add support for topics on data messages

src/room/Room.ts

@@ -1002,7 +1002,7 @@ class Room extends (EventEmitter as new () => TypedEmitter<RoomEventCallbacks>)
     // find the participant
     const participant = this.participants.get(userPacket.participantSid);
 
-    this.emit(RoomEvent.DataReceived, userPacket.payload, participant, kind);
+    this.emit(RoomEvent.DataReceived, userPacket.payload, participant, kind, userPacket.topic);
 
     // also emit on the participant
     participant?.emit(ParticipantEvent.DataReceived, userPacket.payload, kind);
@@ -1463,6 +1463,7 @@ export type RoomEventCallbacks = {
     payload: Uint8Array,
     participant?: RemoteParticipant,
     kind?: DataPacket_Kind,
+    topic?: string,
   ) => void;
   connectionQualityChanged: (quality: ConnectionQuality, participant: Participant) => void;
   mediaDevicesError: (error: Error) => void;

src/room/events.ts

@@ -175,7 +175,7 @@ export enum RoomEvent {
    * Data packets provides the ability to use LiveKit to send/receive arbitrary payloads.
    * All participants in the room will receive the messages sent to the room.
    *
-   * args: (payload: Uint8Array, participant: [[Participant]], kind: [[DataPacket_Kind]])
+   * args: (payload: Uint8Array, participant: [[Participant]], kind: [[DataPacket_Kind]], topic?: string)
    */
   DataReceived = 'dataReceived',
 

src/room/participant/LocalParticipant.ts

@@ -29,6 +29,7 @@ import {
 } from '../track/options';
 import { Track } from '../track/Track';
 import { constraintsForOptions, mergeDefaultOptions } from '../track/utils';
+import type { DataPublishOptions } from '../types';
 import { isFireFox, isSafari, isWeb, supportsAV1 } from '../utils';
 import Participant from './Participant';
 import { ParticipantTrackPermission, trackPermissionToProto } from './ParticipantTrackPermission';
@@ -811,6 +812,22 @@ export default class LocalParticipant extends Participant {
     );
   }
 
+  /**
+   * Publish a new data payload to the room. Data will be forwarded to each
+   * participant in the room if the destination field in publishOptions is empty
+   *
+   * @param data Uint8Array of the payload. To send string data, use TextEncoder.encode
+   * @param kind whether to send this as reliable or lossy.
+   * For data that you need delivery guarantee (such as chat messages), use Reliable.
+   * For data that should arrive as quickly as possible, but you are ok with dropped
+   * packets, use Lossy.
+   * @param publishOptions optionally specify a `topic` and `destination`
+   */
+  async publishData(
+    data: Uint8Array,
+    kind: DataPacket_Kind,
+    publishOptions?: DataPublishOptions,
+  ): Promise<void>;
   /**
    * Publish a new data payload to the room. Data will be forwarded to each
    * participant in the room if the destination argument is empty
@@ -826,14 +843,26 @@ export default class LocalParticipant extends Participant {
     data: Uint8Array,
     kind: DataPacket_Kind,
     destination?: RemoteParticipant[] | string[],
+  ): Promise<void>;
+
+  async publishData(
+    data: Uint8Array,
+    kind: DataPacket_Kind,
+    publishOptions: DataPublishOptions | RemoteParticipant[] | string[] = {},
   ) {
-    const dest: string[] = [];
+    const destination = Array.isArray(publishOptions)
+      ? publishOptions
+      : publishOptions?.destination;
+    const destinationSids: string[] = [];
+
+    const topic = !Array.isArray(publishOptions) ? publishOptions.topic : undefined;
+
     if (destination !== undefined) {
       destination.forEach((val: any) => {
         if (val instanceof RemoteParticipant) {
-          dest.push(val.sid);
+          destinationSids.push(val.sid);
         } else {
-          dest.push(val);
+          destinationSids.push(val);
         }
       });
     }
@@ -845,7 +874,8 @@ export default class LocalParticipant extends Participant {
         user: {
           participantSid: this.sid,
           payload: data,
-          destinationSids: dest,
+          destinationSids: destinationSids,
+          topic,
         },
       },
     };

src/room/types.ts

@@ -1,3 +1,5 @@
+import type RemoteParticipant from './participant/RemoteParticipant';
+
 export type SimulationOptions = {
   publish?: {
     audio?: boolean;
@@ -11,3 +13,10 @@ export type SimulationOptions = {
     video?: boolean;
   };
 };
+
+export type DataPublishOptions = {
+  /** the participants who will receive the message, will be sent to every one if empty */
+  destination?: RemoteParticipant[] | string[];
+  /** the topic under which the message gets published */
+  topic?: string;
+};