Data track schema metadata

package.json

@@ -71,7 +71,7 @@
   },
   "dependencies": {
     "@livekit/mutex": "1.1.1",
-    "@livekit/protocol": "1.46.6",
+    "@livekit/protocol": "1.48.2",
     "events": "^3.3.0",
     "jose": "^6.1.0",
     "loglevel": "^1.9.2",

src/api/SignalClient.ts

@@ -56,7 +56,11 @@ import {
 } from '@livekit/protocol';
 import log, { LoggerNames, getLogger } from '../logger';
 import type { DataTrackHandle } from '../room/data-track/handle';
-import { type DataTrackSid } from '../room/data-track/types';
+import {
+  DataTrackFrameEncoding,
+  DataTrackSchemaId,
+  type DataTrackSid,
+} from '../room/data-track/types';
 import { ConnectionError } from '../room/errors';
 import CriticalTimers from '../room/timers';
 import type { LoggerOptions } from '../room/types';
@@ -711,13 +715,21 @@ export class SignalClient {
     });
   }
 
-  sendPublishDataTrackRequest(handle: DataTrackHandle, name: string, usesE2ee: boolean) {
+  sendPublishDataTrackRequest(
+    handle: DataTrackHandle,
+    name: string,
+    usesE2ee: boolean,
+    schema?: DataTrackSchemaId,
+    frameEncoding?: DataTrackFrameEncoding,
+  ) {
     return this.sendRequest({
       case: 'publishDataTrackRequest',
       value: new PublishDataTrackRequest({
         pubHandle: handle,
         name: name,
         encryption: usesE2ee ? Encryption_Type.GCM : Encryption_Type.NONE,
+        schema: schema ? DataTrackSchemaId.toProtobuf(schema) : undefined,
+        frameEncoding: frameEncoding ? DataTrackFrameEncoding.toProtobuf(frameEncoding) : undefined,
       }),
     });
   }

src/index.ts

@@ -178,6 +178,11 @@ export type {
   DataTrackSubscribeOptions,
   RemoteDataTrackPipelineOptions,
 };
+export type {
+  DataTrackSchemaId,
+  DataTrackSchemaEncoding,
+  DataTrackFrameEncoding,
+} from './room/data-track/types';
 export { DataTrackPacket, type DataTrackPacketHeader } from './room/data-track/packet';
 export {
   type DataTrackExtensions,

src/room/Room.ts

@@ -294,7 +294,13 @@ class Room extends (EventEmitter as new () => TypedEmitter<RoomEventCallbacks>)
     this.outgoingDataTrackManager = new OutgoingDataTrackManager({ e2eeManager: this.e2eeManager });
     this.outgoingDataTrackManager
       .on('sfuPublishRequest', (event) => {
-        this.engine.client.sendPublishDataTrackRequest(event.handle, event.name, event.usesE2ee);
+        this.engine.client.sendPublishDataTrackRequest(
+          event.handle,
+          event.name,
+          event.usesE2ee,
+          event.schema,
+          event.frameEncoding,
+        );
       })
       .on('sfuUnpublishRequest', (event) => {
         this.engine.client.sendUnPublishDataTrackRequest(event.handle);

src/room/data-track/outgoing/OutgoingDataTrackManager.test.ts

@@ -251,6 +251,56 @@ describe('DataTrackOutgoingManager', () => {
     expect(handle).not.toStrictEqual(handle2);
   });
 
+  it.each([
+    {
+      title: 'well-known encodings',
+      schema: { name: 'my_schema', encoding: 'jsonSchema' },
+      frameEncoding: 'json',
+    },
+    {
+      title: 'custom encodings',
+      schema: { name: 'my_schema', encoding: { custom: 'a' } },
+      frameEncoding: { custom: 'b' },
+    },
+  ] as const)(
+    'should forward schema and frame encoding on publish ($title)',
+    async ({ schema, frameEncoding }) => {
+      const manager = new OutgoingDataTrackManager();
+      const managerEvents = subscribeToEvents<DataTrackOutgoingManagerCallbacks>(manager, [
+        'sfuPublishRequest',
+      ]);
+
+      const localDataTrack = new LocalDataTrack({ name: 'test', schema, frameEncoding }, manager);
+
+      // 1. Publish a data track with schema metadata
+      const publishRequestPromise = localDataTrack.publish();
+
+      // 2. The publish request sent to the SFU carries the schema and frame encoding
+      const sfuPublishEvent = await managerEvents.waitFor('sfuPublishRequest');
+      expect(sfuPublishEvent.schema).toStrictEqual(schema);
+      expect(sfuPublishEvent.frameEncoding).toStrictEqual(frameEncoding);
+      const handle = sfuPublishEvent.handle;
+
+      // 3. Respond as the SFU would, echoing the metadata back on the DataTrackInfo
+      manager.receivedSfuPublishResponse(handle, {
+        type: 'ok',
+        data: {
+          sid: 'bogus-sid',
+          pubHandle: handle,
+          name: 'test',
+          usesE2ee: false,
+          schema,
+          frameEncoding,
+        },
+      });
+      await publishRequestPromise;
+
+      // 4. The metadata is reflected on the local track's info
+      expect(localDataTrack.info?.schema).toStrictEqual(schema);
+      expect(localDataTrack.info?.frameEncoding).toStrictEqual(frameEncoding);
+    },
+  );
+
   it.each([
     // Single packet payload case
     [

src/room/data-track/outgoing/OutgoingDataTrackManager.ts

@@ -263,6 +263,8 @@ export default class OutgoingDataTrackManager extends (EventEmitter as new () =>
       handle,
       name: options.name,
       usesE2ee: this.e2eeManager !== null,
+      schema: options.schema,
+      frameEncoding: options.frameEncoding,
     });
 
     await descriptor.completionFuture.promise;
@@ -395,6 +397,8 @@ export default class OutgoingDataTrackManager extends (EventEmitter as new () =>
             handle: descriptor.info.pubHandle,
             name: descriptor.info.name,
             usesE2ee: descriptor.info.usesE2ee,
+            schema: descriptor.info.schema,
+            frameEncoding: descriptor.info.frameEncoding,
           });
       }
     }

src/room/data-track/outgoing/types.ts

@@ -1,11 +1,22 @@
 import type LocalDataTrack from '../LocalDataTrack';
 import { type DataTrackHandle } from '../handle';
-import { type DataTrackInfo, type DataTrackSid } from '../types';
+import {
+  type DataTrackFrameEncoding,
+  type DataTrackInfo,
+  type DataTrackSchemaId,
+  type DataTrackSid,
+} from '../types';
 import { type DataTrackPublishError, type DataTrackPublishErrorReason } from './errors';
 
 /** Options for publishing a data track. */
 export type DataTrackOptions = {
   name: string;
+
+  /** Schema describing frames sent on the track. */
+  schema?: DataTrackSchemaId;
+
+  /** Encoding of frames sent on the track. */
+  frameEncoding?: DataTrackFrameEncoding;
 };
 
 /** Encodes whether a data track publish request to the SFU has been successful or not. */
@@ -26,6 +37,8 @@ export type EventSfuPublishRequest = {
   handle: DataTrackHandle;
   name: string;
   usesE2ee: boolean;
+  schema?: DataTrackSchemaId;
+  frameEncoding?: DataTrackFrameEncoding;
 };
 
 /** Request sent to the SFU to unpublish a track. */

src/room/data-track/schema.test.ts

@@ -0,0 +1,108 @@
+import {
+  DataTrackFrameEncoding as ProtocolDataTrackFrameEncoding,
+  DataTrackSchemaEncoding as ProtocolDataTrackSchemaEncoding,
+  DataTrackSchemaId as ProtocolDataTrackSchemaId,
+  DataTrackFrameEncoding_WellKnownFrameEncoding as ProtocolWellKnownFrameEncoding,
+  DataTrackSchemaEncoding_WellKnownSchemaEncoding as ProtocolWellKnownSchemaEncoding,
+} from '@livekit/protocol';
+import { describe, expect, it } from 'vitest';
+import { DataTrackFrameEncoding, DataTrackSchemaEncoding, DataTrackSchemaId } from './schema';
+
+describe('DataTrackSchemaEncoding', () => {
+  const wellKnown: Array<DataTrackSchemaEncoding> = [
+    'protobuf',
+    'flatbuffer',
+    'ros1Msg',
+    'ros2Msg',
+    'ros2Idl',
+    'omgIdl',
+    'jsonSchema',
+  ];
+
+  it.each(wellKnown)('round-trips well-known encoding %s', (encoding) => {
+    const protobuf = DataTrackSchemaEncoding.toProtobuf(encoding);
+    expect(protobuf.value.case).toStrictEqual('wellKnown');
+    expect(DataTrackSchemaEncoding.from(protobuf)).toStrictEqual(encoding);
+  });
+
+  it('round-trips a custom encoding', () => {
+    const encoding: DataTrackSchemaEncoding = { custom: 'my_encoding' };
+    const protobuf = DataTrackSchemaEncoding.toProtobuf(encoding);
+    expect(protobuf.value).toStrictEqual({ case: 'custom', value: 'my_encoding' });
+    expect(DataTrackSchemaEncoding.from(protobuf)).toStrictEqual(encoding);
+  });
+
+  it('maps an unspecified well-known value to "other"', () => {
+    const protobuf = new ProtocolDataTrackSchemaEncoding({
+      value: { case: 'wellKnown', value: ProtocolWellKnownSchemaEncoding.UNSPECIFIED },
+    });
+    expect(DataTrackSchemaEncoding.from(protobuf)).toStrictEqual('other');
+  });
+
+  it('maps a well-known value introduced after this version to "other"', () => {
+    const protobuf = new ProtocolDataTrackSchemaEncoding({
+      value: { case: 'wellKnown', value: 999 as ProtocolWellKnownSchemaEncoding },
+    });
+    expect(DataTrackSchemaEncoding.from(protobuf)).toStrictEqual('other');
+  });
+
+  it('maps an absent oneof to "other"', () => {
+    expect(DataTrackSchemaEncoding.from(new ProtocolDataTrackSchemaEncoding())).toStrictEqual(
+      'other',
+    );
+  });
+});
+
+describe('DataTrackFrameEncoding', () => {
+  const wellKnown: Array<DataTrackFrameEncoding> = [
+    'ros1',
+    'cdr',
+    'protobuf',
+    'flatbuffer',
+    'cbor',
+    'msgpack',
+    'json',
+  ];
+
+  it.each(wellKnown)('round-trips well-known encoding %s', (encoding) => {
+    const protobuf = DataTrackFrameEncoding.toProtobuf(encoding);
+    expect(protobuf.value.case).toStrictEqual('wellKnown');
+    expect(DataTrackFrameEncoding.from(protobuf)).toStrictEqual(encoding);
+  });
+
+  it('round-trips a custom encoding', () => {
+    const encoding: DataTrackFrameEncoding = { custom: 'my_encoding' };
+    const protobuf = DataTrackFrameEncoding.toProtobuf(encoding);
+    expect(protobuf.value).toStrictEqual({ case: 'custom', value: 'my_encoding' });
+    expect(DataTrackFrameEncoding.from(protobuf)).toStrictEqual(encoding);
+  });
+
+  it('maps an unspecified well-known value to "other"', () => {
+    const protobuf = new ProtocolDataTrackFrameEncoding({
+      value: { case: 'wellKnown', value: ProtocolWellKnownFrameEncoding.UNSPECIFIED },
+    });
+    expect(DataTrackFrameEncoding.from(protobuf)).toStrictEqual('other');
+  });
+
+  it('maps a well-known value introduced after this version to "other"', () => {
+    const protobuf = new ProtocolDataTrackFrameEncoding({
+      value: { case: 'wellKnown', value: 999 as ProtocolWellKnownFrameEncoding },
+    });
+    expect(DataTrackFrameEncoding.from(protobuf)).toStrictEqual('other');
+  });
+});
+
+describe('DataTrackSchemaId', () => {
+  it('round-trips name and encoding', () => {
+    const schemaId: DataTrackSchemaId = { name: 'rgb', encoding: 'protobuf' };
+    const protobuf = DataTrackSchemaId.toProtobuf(schemaId);
+    expect(protobuf).toBeInstanceOf(ProtocolDataTrackSchemaId);
+    expect(protobuf.name).toStrictEqual('rgb');
+    expect(DataTrackSchemaId.from(protobuf)).toStrictEqual(schemaId);
+  });
+
+  it('defaults encoding to "other" when the protobuf encoding is absent', () => {
+    const protobuf = new ProtocolDataTrackSchemaId({ name: 'rgb' });
+    expect(DataTrackSchemaId.from(protobuf)).toStrictEqual({ name: 'rgb', encoding: 'other' });
+  });
+});