diff --git a/.changeset/schema-storage-api.md b/.changeset/schema-storage-api.md new file mode 100644 index 0000000000..da9f1628b6 --- /dev/null +++ b/.changeset/schema-storage-api.md @@ -0,0 +1,5 @@ +--- +'livekit-client': minor +--- + +Added `defineSchema` and `getSchema` to `LocalParticipant` for storing and retrieving data track schema definitions diff --git a/src/api/SignalClient.test.ts b/src/api/SignalClient.test.ts index 647b713318..bb80bf3da8 100644 --- a/src/api/SignalClient.test.ts +++ b/src/api/SignalClient.test.ts @@ -1,12 +1,16 @@ import { ClientInfo_Capability, + DataBlob, + DataBlobKey, DisconnectReason, + GetDataBlobResponse, JoinRequest, JoinResponse, LeaveRequest, ReconnectResponse, SignalRequest, SignalResponse, + StoreDataBlobResponse, WrappedJoinRequest, WrappedJoinRequest_Compression, } from '@livekit/protocol'; @@ -33,7 +37,8 @@ function createJoinResponse() { } function createSignalResponse( - messageCase: 'join' | 'reconnect' | 'leave' | 'update', + messageCase: + 'join' | 'reconnect' | 'leave' | 'update' | 'storeDataBlobResponse' | 'getDataBlobResponse', value: any, ): SignalResponse { return new SignalResponse({ @@ -573,6 +578,88 @@ describe('SignalClient.connect', () => { }); }); +describe('SignalClient data blobs', () => { + let signalClient: SignalClient; + + beforeEach(() => { + vi.clearAllMocks(); + signalClient = new SignalClient(false); + }); + + function clientInternals() { + return signalClient as unknown as { + state: SignalConnectionState; + streamWriter: { write: (chunk: ArrayBuffer) => Promise } | undefined; + handleSignalResponse: (res: SignalResponse) => void; + }; + } + + it('dispatches store data blob responses', () => { + const received: StoreDataBlobResponse[] = []; + signalClient.onStoreDataBlobResponse = (res) => received.push(res); + + clientInternals().handleSignalResponse( + createSignalResponse('storeDataBlobResponse', new StoreDataBlobResponse({ requestId: 42 })), + ); + + expect(received).toHaveLength(1); + expect(received[0].requestId).toBe(42); + }); + + it('dispatches get data blob responses', () => { + const received: GetDataBlobResponse[] = []; + signalClient.onGetDataBlobResponse = (res) => received.push(res); + + clientInternals().handleSignalResponse( + createSignalResponse( + 'getDataBlobResponse', + new GetDataBlobResponse({ + requestId: 43, + blob: new DataBlob({ contents: new TextEncoder().encode('definition') }), + }), + ), + ); + + expect(received).toHaveLength(1); + expect(received[0].requestId).toBe(43); + expect(new TextDecoder().decode(received[0].blob?.contents)).toBe('definition'); + }); + + it('sends blob requests with increasing request ids', async () => { + const written: SignalRequest[] = []; + const internals = clientInternals(); + internals.state = SignalConnectionState.CONNECTED; + internals.streamWriter = { + write: async (chunk) => { + written.push(SignalRequest.fromBinary(new Uint8Array(chunk))); + }, + }; + + const key = new DataBlobKey({ key: { case: 'generic', value: 'my-key' } }); + const storeRequestId = await signalClient.sendStoreDataBlobRequest( + new DataBlob({ key, contents: new TextEncoder().encode('contents') }), + ); + const getRequestId = await signalClient.sendGetDataBlobRequest(key, 'publisher-identity'); + + expect(getRequestId).toBe(storeRequestId + 1); + expect(written).toHaveLength(2); + + const storeMessage = written[0].message; + expect(storeMessage.case).toStrictEqual('storeDataBlobRequest'); + if (storeMessage.case !== 'storeDataBlobRequest') throw new Error('unreachable'); + expect(storeMessage.value.requestId).toBe(storeRequestId); + expect(storeMessage.value.blob?.key?.key).toStrictEqual({ case: 'generic', value: 'my-key' }); + expect(new TextDecoder().decode(storeMessage.value.blob?.contents)).toBe('contents'); + + const getMessage = written[1].message; + expect(getMessage.case).toStrictEqual('getDataBlobRequest'); + if (getMessage.case !== 'getDataBlobRequest') throw new Error('unreachable'); + expect(getMessage.value.requestId).toBe(getRequestId); + expect(getMessage.value.participantIdentity).toStrictEqual('publisher-identity'); + expect(getMessage.value.key?.key).toStrictEqual({ case: 'generic', value: 'my-key' }); + }); +}); + describe('SignalClient utility functions', () => { describe('toProtoSessionDescription', () => { it('should convert RTCSessionDescriptionInit to proto SessionDescription', async () => { diff --git a/src/api/SignalClient.ts b/src/api/SignalClient.ts index 746560f854..7778eb1146 100644 --- a/src/api/SignalClient.ts +++ b/src/api/SignalClient.ts @@ -6,9 +6,13 @@ import { ClientInfo_Capability, ConnectionQualityUpdate, ConnectionSettings, + DataBlob, + DataBlobKey, DataTrackSubscriberHandles, DisconnectReason, Encryption_Type, + GetDataBlobRequest, + GetDataBlobResponse, JoinRequest, JoinResponse, LeaveRequest, @@ -30,6 +34,8 @@ import { SignalTarget, SimulateScenario, SpeakerInfo, + StoreDataBlobRequest, + StoreDataBlobResponse, StreamStateUpdate, SubscribedQualityUpdate, SubscriptionPermission, @@ -203,6 +209,10 @@ export class SignalClient { onDataTrackSubscriberHandles?: (event: DataTrackSubscriberHandles) => void; + onStoreDataBlobResponse?: (res: StoreDataBlobResponse) => void; + + onGetDataBlobResponse?: (res: GetDataBlobResponse) => void; + onJoined?: (event: JoinResponse) => void; connectOptions?: ConnectOpts; @@ -774,6 +784,24 @@ export class SignalClient { }); } + async sendStoreDataBlobRequest(blob: DataBlob) { + const requestId = this.getNextRequestId(); + await this.sendRequest({ + case: 'storeDataBlobRequest', + value: new StoreDataBlobRequest({ requestId, blob }), + }); + return requestId; + } + + async sendGetDataBlobRequest(key: DataBlobKey, participantIdentity: string) { + const requestId = this.getNextRequestId(); + await this.sendRequest({ + case: 'getDataBlobRequest', + value: new GetDataBlobRequest({ requestId, participantIdentity, key }), + }); + return requestId; + } + sendUpdateDataSubscription(sid: DataTrackSid, subscribe: boolean) { return this.sendRequest({ case: 'updateDataSubscription', @@ -940,6 +968,14 @@ export class SignalClient { if (this.onDataTrackSubscriberHandles) { this.onDataTrackSubscriberHandles(msg.value); } + } else if (msg.case === 'storeDataBlobResponse') { + if (this.onStoreDataBlobResponse) { + this.onStoreDataBlobResponse(msg.value); + } + } else if (msg.case === 'getDataBlobResponse') { + if (this.onGetDataBlobResponse) { + this.onGetDataBlobResponse(msg.value); + } } else { this.log.debug('unsupported message', { msgCase: msg.case }); } diff --git a/src/room/RTCEngine.ts b/src/room/RTCEngine.ts index 00a95a9247..92ab0ac988 100644 --- a/src/room/RTCEngine.ts +++ b/src/room/RTCEngine.ts @@ -13,6 +13,7 @@ import { EncryptedPacket, EncryptedPacketPayload, Encryption_Type, + type GetDataBlobResponse, type JoinResponse, type LeaveRequest, LeaveRequest_Action, @@ -31,6 +32,7 @@ import { SessionDescription, SignalTarget, SpeakerInfo, + type StoreDataBlobResponse, type StreamStateUpdate, SubscribedQualityUpdate, type SubscriptionPermissionUpdate, @@ -735,6 +737,14 @@ export default class RTCEngine extends (EventEmitter as new () => TypedEventEmit this.emit(EngineEvent.DataTrackSubscriberHandles, event); }; + this.client.onStoreDataBlobResponse = (res: StoreDataBlobResponse) => { + this.emit(EngineEvent.StoreDataBlobResponse, res); + }; + + this.client.onGetDataBlobResponse = (res: GetDataBlobResponse) => { + this.emit(EngineEvent.GetDataBlobResponse, res); + }; + this.client.onClose = () => { this.handleDisconnect('signal', ReconnectReason.RR_SIGNAL_DISCONNECTED); }; @@ -2054,6 +2064,8 @@ export type EngineEventCallbacks = { unPublishDataTrackResponse: (event: UnpublishDataTrackResponse) => void; dataTrackSubscriberHandles: (event: DataTrackSubscriberHandles) => void; dataTrackPacketReceived: (packet: Uint8Array) => void; + storeDataBlobResponse: (res: StoreDataBlobResponse) => void; + getDataBlobResponse: (res: GetDataBlobResponse) => void; joined: (joinResponse: JoinResponse) => void; tokenRefreshed: (token: string) => void; serverRegionsReported: (regions: RegionSettings) => void; diff --git a/src/room/data-track/schema-storage.ts b/src/room/data-track/schema-storage.ts new file mode 100644 index 0000000000..924a751de1 --- /dev/null +++ b/src/room/data-track/schema-storage.ts @@ -0,0 +1,86 @@ +import { RequestResponse_Reason } from '@livekit/protocol'; +import { LivekitReasonedError } from '../errors'; + +export enum DataTrackSchemaStorageErrorReason { + /** Request to store or retrieve a schema definition timed-out */ + Timeout = 0, + /** The participant has not defined a schema with this ID */ + NotFound = 1, + /** The server rejected the request */ + RequestFailed = 2, + /** The server response is malformed */ + MalformedResponse = 3, + /** The retrieved schema definition is not valid UTF-8 */ + InvalidDefinition = 4, + /** Request cancelled by caller */ + Cancelled = 5, + /** Cannot store or retrieve a schema definition when disconnected */ + Disconnected = 6, +} + +export class DataTrackSchemaStorageError< + Reason extends DataTrackSchemaStorageErrorReason = DataTrackSchemaStorageErrorReason, +> extends LivekitReasonedError { + readonly name = 'DataTrackSchemaStorageError'; + + reason: Reason; + + reasonName: string; + + constructor(message: string, reason: Reason, options?: { cause?: unknown }) { + super(24, message, options); + this.reason = reason; + this.reasonName = DataTrackSchemaStorageErrorReason[reason]; + } + + static timeout() { + return new DataTrackSchemaStorageError( + 'Schema storage request timed out', + DataTrackSchemaStorageErrorReason.Timeout, + ); + } + + static notFound(message: string) { + return new DataTrackSchemaStorageError( + message || 'The participant has not defined a schema with this ID', + DataTrackSchemaStorageErrorReason.NotFound, + ); + } + + static requestFailed(reason: RequestResponse_Reason, message: string) { + return new DataTrackSchemaStorageError( + `Schema storage request failed (${RequestResponse_Reason[reason]}): ${message}`, + DataTrackSchemaStorageErrorReason.RequestFailed, + ); + } + + static malformedResponse() { + return new DataTrackSchemaStorageError( + 'Schema storage response is malformed', + DataTrackSchemaStorageErrorReason.MalformedResponse, + ); + } + + static invalidDefinition(options?: { cause?: unknown }) { + return new DataTrackSchemaStorageError( + 'Schema definition is not valid UTF-8', + DataTrackSchemaStorageErrorReason.InvalidDefinition, + options, + ); + } + + // NOTE: this was introduced by web / there isn't a corresponding case in the rust version. + static cancelled() { + return new DataTrackSchemaStorageError( + 'Schema storage request cancelled by caller', + DataTrackSchemaStorageErrorReason.Cancelled, + ); + } + + static disconnected() { + return new DataTrackSchemaStorageError( + 'Cannot store or retrieve a schema definition when disconnected', + DataTrackSchemaStorageErrorReason.Disconnected, + ); + } +} diff --git a/src/room/data-track/schema.test.ts b/src/room/data-track/schema.test.ts index 4374a19dd9..7b8cdac7d8 100644 --- a/src/room/data-track/schema.test.ts +++ b/src/room/data-track/schema.test.ts @@ -1,4 +1,5 @@ import { + DataBlobKey, DataTrackFrameEncoding as ProtocolDataTrackFrameEncoding, DataTrackSchemaEncoding as ProtocolDataTrackSchemaEncoding, DataTrackSchemaId as ProtocolDataTrackSchemaId, @@ -112,6 +113,19 @@ describe('DataTrackSchemaId', () => { const protobuf = new ProtocolDataTrackSchemaId({ name: 'rgb' }); expect(DataTrackSchemaId.from(protobuf)).toStrictEqual({ name: 'rgb', encoding: 'other' }); }); + + it.each([ + { title: 'well-known encoding', encoding: 'jsonSchema' as DataTrackSchemaEncoding }, + { title: 'custom encoding', encoding: { custom: 'my_encoding' } }, + ])('converts to a data blob key ($title)', ({ encoding }) => { + const schemaId: DataTrackSchemaId = { name: 'rgb', encoding }; + const key = DataTrackSchemaId.toDataBlobKey(schemaId); + expect(key).toBeInstanceOf(DataBlobKey); + expect(key.key.case).toStrictEqual('schemaId'); + expect(DataTrackSchemaId.from(key.key.value as ProtocolDataTrackSchemaId)).toStrictEqual( + schemaId, + ); + }); }); describe('validateSchemaMetadata', () => { diff --git a/src/room/data-track/schema.ts b/src/room/data-track/schema.ts index 4a233edcad..a956f977b0 100644 --- a/src/room/data-track/schema.ts +++ b/src/room/data-track/schema.ts @@ -1,4 +1,5 @@ import { + DataBlobKey, DataTrackFrameEncoding as ProtocolDataTrackFrameEncoding, DataTrackSchemaEncoding as ProtocolDataTrackSchemaEncoding, DataTrackSchemaId as ProtocolDataTrackSchemaId, @@ -232,6 +233,12 @@ export const DataTrackSchemaId = { encoding: DataTrackSchemaEncoding.toProtobuf(schemaId.encoding), }); }, + /** Key under which the schema's definition is stored as a data blob. */ + toDataBlobKey(schemaId: DataTrackSchemaId): DataBlobKey { + return new DataBlobKey({ + key: { case: 'schemaId', value: DataTrackSchemaId.toProtobuf(schemaId) }, + }); + }, }; export enum DataTrackSchemaErrorReason { diff --git a/src/room/events.ts b/src/room/events.ts index 3f61fff1a7..d40d40738c 100644 --- a/src/room/events.ts +++ b/src/room/events.ts @@ -630,6 +630,8 @@ export enum EngineEvent { UnPublishDataTrackResponse = 'unPublishDataTrackResponse', DataTrackSubscriberHandles = 'dataTrackSubscriberHandles', DataTrackPacketReceived = 'dataTrackPacketReceived', + StoreDataBlobResponse = 'storeDataBlobResponse', + GetDataBlobResponse = 'getDataBlobResponse', Joined = 'joined', TokenRefreshed = 'tokenRefreshed', ServerRegionsReported = 'serverRegionsReported', diff --git a/src/room/participant/LocalParticipant.test.ts b/src/room/participant/LocalParticipant.test.ts index cc8adc04cf..dced600651 100644 --- a/src/room/participant/LocalParticipant.test.ts +++ b/src/room/participant/LocalParticipant.test.ts @@ -1,5 +1,22 @@ -import { PacketTrailerFeature } from '@livekit/protocol'; -import { describe, expect, it, vi } from 'vitest'; +import { + DataBlob, + type DataBlobKey, + GetDataBlobResponse, + PacketTrailerFeature, + RequestResponse, + RequestResponse_Reason, + StoreDataBlobResponse, +} from '@livekit/protocol'; +import { EventEmitter } from 'events'; +import { afterEach, describe, expect, it, vi } from 'vitest'; +import type { InternalRoomOptions } from '../../options'; +import type RTCEngine from '../RTCEngine'; +import { + DataTrackSchemaStorageError, + DataTrackSchemaStorageErrorReason, +} from '../data-track/schema-storage'; +import { DataTrackSchemaId } from '../data-track/types'; +import { EngineEvent } from '../events'; import type LocalTrack from '../track/LocalTrack'; import { Track } from '../track/Track'; import type { TrackPublishOptions } from '../track/options'; @@ -79,3 +96,246 @@ describe('LocalParticipant frame metadata publish options', () => { expect(participant.log.warn).toHaveBeenCalledOnce(); }); }); + +describe('LocalParticipant schema storage', () => { + const schemaId: DataTrackSchemaId = { name: 'rgb', encoding: 'jsonSchema' }; + + type MockEngine = RTCEngine & { + client: { + sendStoreDataBlobRequest: ReturnType; + sendGetDataBlobRequest: ReturnType; + }; + }; + + function makeEngine(requestId: number): MockEngine { + const engine = new EventEmitter() as unknown as MockEngine; + engine.client = { + sendStoreDataBlobRequest: vi.fn(async () => requestId), + sendGetDataBlobRequest: vi.fn(async () => requestId), + }; + return engine; + } + + function createParticipant(engine: MockEngine) { + return new LocalParticipant( + 'participant-sid', + 'test-identity', + engine, + {} as InternalRoomOptions, + undefined as any, + undefined as any, + undefined as any, + undefined as any, + ); + } + + /** Waits for a request to be sent and its pending future to be registered. */ + function flush() { + return new Promise((resolve) => setTimeout(resolve, 0)); + } + + function pendingRequests(participant: LocalParticipant) { + return participant as unknown as { + pendingStoreDataBlobRequests: Map; + pendingGetDataBlobRequests: Map; + }; + } + + afterEach(() => { + vi.useRealTimers(); + }); + + it('defines a schema by storing its definition as a data blob', async () => { + const engine = makeEngine(7); + const participant = createParticipant(engine); + + const promise = participant.defineSchema(schemaId, '{"type":"object"}'); + await flush(); + + expect(engine.client.sendStoreDataBlobRequest).toHaveBeenCalledOnce(); + const blob = engine.client.sendStoreDataBlobRequest.mock.calls[0][0] as DataBlob; + expect(blob.key?.key.case).toStrictEqual('schemaId'); + expect(DataTrackSchemaId.from(blob.key!.key.value! as never)).toStrictEqual(schemaId); + expect(new TextDecoder().decode(blob.contents)).toStrictEqual('{"type":"object"}'); + + engine.emit(EngineEvent.StoreDataBlobResponse, new StoreDataBlobResponse({ requestId: 7 })); + await expect(promise).resolves.toBeUndefined(); + expect(pendingRequests(participant).pendingStoreDataBlobRequests.size).toBe(0); + }); + + it('ignores a store response with a mismatched request id', async () => { + const engine = makeEngine(7); + const participant = createParticipant(engine); + + const promise = participant.defineSchema(schemaId, 'definition'); + await flush(); + + engine.emit(EngineEvent.StoreDataBlobResponse, new StoreDataBlobResponse({ requestId: 8 })); + await flush(); + expect(pendingRequests(participant).pendingStoreDataBlobRequests.size).toBe(1); + + engine.emit(EngineEvent.StoreDataBlobResponse, new StoreDataBlobResponse({ requestId: 7 })); + await expect(promise).resolves.toBeUndefined(); + }); + + it('rejects defining a schema when the server reports an error', async () => { + const engine = makeEngine(7); + const participant = createParticipant(engine); + + const promise = participant.defineSchema(schemaId, 'definition'); + await flush(); + + engine.emit( + EngineEvent.SignalRequestResponse, + new RequestResponse({ + requestId: 7, + reason: RequestResponse_Reason.INVALID_REQUEST, + message: 'schema already defined', + }), + ); + await expect(promise).rejects.toStrictEqual( + DataTrackSchemaStorageError.requestFailed( + RequestResponse_Reason.INVALID_REQUEST, + 'schema already defined', + ), + ); + expect(pendingRequests(participant).pendingStoreDataBlobRequests.size).toBe(0); + }); + + it('ignores an OK request response for a pending blob request', async () => { + const engine = makeEngine(7); + const participant = createParticipant(engine); + + const promise = participant.defineSchema(schemaId, 'definition'); + await flush(); + + engine.emit( + EngineEvent.SignalRequestResponse, + new RequestResponse({ requestId: 7, reason: RequestResponse_Reason.OK }), + ); + await flush(); + expect(pendingRequests(participant).pendingStoreDataBlobRequests.size).toBe(1); + + engine.emit(EngineEvent.StoreDataBlobResponse, new StoreDataBlobResponse({ requestId: 7 })); + await expect(promise).resolves.toBeUndefined(); + }); + + it('retrieves a schema definition', async () => { + const engine = makeEngine(3); + const participant = createParticipant(engine); + + const promise = participant.getSchema(schemaId, 'publisher-identity'); + await flush(); + + expect(engine.client.sendGetDataBlobRequest).toHaveBeenCalledOnce(); + const [key, identity] = engine.client.sendGetDataBlobRequest.mock.calls[0] as [ + DataBlobKey, + string, + ]; + expect(key.key.case).toStrictEqual('schemaId'); + expect(identity).toStrictEqual('publisher-identity'); + + engine.emit( + EngineEvent.GetDataBlobResponse, + new GetDataBlobResponse({ + requestId: 3, + blob: new DataBlob({ contents: new TextEncoder().encode('{"type":"object"}') }), + }), + ); + await expect(promise).resolves.toStrictEqual('{"type":"object"}'); + expect(pendingRequests(participant).pendingGetDataBlobRequests.size).toBe(0); + }); + + it('rejects retrieving an undefined schema', async () => { + const engine = makeEngine(3); + const participant = createParticipant(engine); + + const promise = participant.getSchema(schemaId, 'publisher-identity'); + await flush(); + + engine.emit( + EngineEvent.SignalRequestResponse, + new RequestResponse({ + requestId: 3, + reason: RequestResponse_Reason.NOT_FOUND, + message: 'blob not found', + }), + ); + await expect(promise).rejects.toStrictEqual( + DataTrackSchemaStorageError.notFound('blob not found'), + ); + expect(pendingRequests(participant).pendingGetDataBlobRequests.size).toBe(0); + }); + + it('rejects a malformed get response missing the blob', async () => { + const engine = makeEngine(3); + const participant = createParticipant(engine); + + const promise = participant.getSchema(schemaId, 'publisher-identity'); + await flush(); + + engine.emit(EngineEvent.GetDataBlobResponse, new GetDataBlobResponse({ requestId: 3 })); + await expect(promise).rejects.toStrictEqual(DataTrackSchemaStorageError.malformedResponse()); + }); + + it('rejects a schema definition that is not valid UTF-8', async () => { + const engine = makeEngine(3); + const participant = createParticipant(engine); + + const promise = participant.getSchema(schemaId, 'publisher-identity'); + await flush(); + + engine.emit( + EngineEvent.GetDataBlobResponse, + new GetDataBlobResponse({ + requestId: 3, + blob: new DataBlob({ contents: new Uint8Array([0xff, 0xfe, 0xfd]) }), + }), + ); + await expect(promise).rejects.toMatchObject({ + reason: DataTrackSchemaStorageErrorReason.InvalidDefinition, + }); + }); + + it('rejects when the request times out', async () => { + vi.useFakeTimers(); + const engine = makeEngine(7); + const participant = createParticipant(engine); + + const promise = participant.defineSchema(schemaId, 'definition'); + const expectation = expect(promise).rejects.toStrictEqual( + DataTrackSchemaStorageError.timeout(), + ); + await vi.advanceTimersByTimeAsync(0); + expect(engine.client.sendStoreDataBlobRequest).toHaveBeenCalledOnce(); + + await vi.advanceTimersByTimeAsync(5_000); + await expectation; + expect(pendingRequests(participant).pendingStoreDataBlobRequests.size).toBe(0); + }); + + it('rejects when the caller aborts the request', async () => { + const engine = makeEngine(3); + const participant = createParticipant(engine); + const controller = new AbortController(); + + const promise = participant.getSchema(schemaId, 'publisher-identity', controller.signal); + await flush(); + + controller.abort(); + await expect(promise).rejects.toStrictEqual(DataTrackSchemaStorageError.cancelled()); + expect(pendingRequests(participant).pendingGetDataBlobRequests.size).toBe(0); + }); + + it('rejects pending requests when the engine closes', async () => { + const engine = makeEngine(7); + const participant = createParticipant(engine); + + const storePromise = participant.defineSchema(schemaId, 'definition'); + await flush(); + + engine.emit(EngineEvent.Closing); + await expect(storePromise).rejects.toStrictEqual(DataTrackSchemaStorageError.disconnected()); + expect(pendingRequests(participant).pendingStoreDataBlobRequests.size).toBe(0); + }); +}); diff --git a/src/room/participant/LocalParticipant.ts b/src/room/participant/LocalParticipant.ts index 1993a1d154..36bc8c6a09 100644 --- a/src/room/participant/LocalParticipant.ts +++ b/src/room/participant/LocalParticipant.ts @@ -5,9 +5,12 @@ import { BackupCodecPolicy, ChatMessage as ChatMessageModel, Codec, + DataBlob, + type DataBlobKey, DataPacket, DataPacket_Kind, Encryption_Type, + type GetDataBlobResponse, JoinResponse, PacketTrailerFeature, ParticipantInfo, @@ -15,12 +18,14 @@ import { RequestResponse_Reason, SimulcastCodec, SipDTMF, + type StoreDataBlobResponse, SubscribedQualityUpdate, TrackInfo, TrackUnpublishedResponse, UserPacket, protoInt64, } from '@livekit/protocol'; +import type { Throws } from '@livekit/throws-transformer/throws'; import { SignalConnectionState } from '../../api/SignalClient'; import { getFrameMetadataFeatures, @@ -31,6 +36,7 @@ import { import type { InternalRoomOptions } from '../../options'; import type { NonSharedUint8Array } from '../../type-polyfills/non-shared-typed-arrays'; import TypedPromise from '../../utils/TypedPromise'; +import { abortSignalAny, abortSignalTimeout } from '../../utils/abort-signal-polyfill'; import { PCTransportState } from '../PCTransportManager'; import type RTCEngine from '../RTCEngine'; import { DataChannelKind } from '../RTCEngine'; @@ -40,6 +46,8 @@ import LocalDataTrack from '../data-track/LocalDataTrack'; import type OutgoingDataTrackManager from '../data-track/outgoing/OutgoingDataTrackManager'; import { DataTrackPublishError } from '../data-track/outgoing/errors'; import type { DataTrackOptions } from '../data-track/outgoing/types'; +import { DataTrackSchemaStorageError } from '../data-track/schema-storage'; +import { DataTrackSchemaId } from '../data-track/types'; import { defaultVideoCodec } from '../defaults'; import { DeviceUnsupportedError, @@ -119,6 +127,9 @@ import { getDefaultDegradationPreference, } from './publishUtils'; +/** How long to wait for the server to respond to a data blob request. */ +const DATA_BLOB_REQUEST_TIMEOUT_MILLISECONDS = 5_000; + export default class LocalParticipant extends Participant { audioTrackPublications: Map; @@ -179,6 +190,16 @@ export default class LocalParticipant extends Participant { } >; + private pendingStoreDataBlobRequests: Map< + number, + Future + >; + + private pendingGetDataBlobRequests: Map< + number, + Future + >; + private enabledPublishVideoCodecs: Codec[] = []; /** @internal */ @@ -208,6 +229,8 @@ export default class LocalParticipant extends Participant { ['audiooutput', 'default'], ]); this.pendingSignalRequests = new Map(); + this.pendingStoreDataBlobRequests = new Map(); + this.pendingGetDataBlobRequests = new Map(); this.roomOutgoingDataStreamManager = roomOutgoingDataStreamManager; this.roomOutgoingDataTrackManager = roomOutgoingDataTrackManager; this.rpcClientManager = rpcClientManager; @@ -271,7 +294,9 @@ export default class LocalParticipant extends Participant { .on(EngineEvent.LocalTrackUnpublished, this.handleLocalTrackUnpublished) .on(EngineEvent.SubscribedQualityUpdate, this.handleSubscribedQualityUpdate) .on(EngineEvent.Closing, this.handleClosing) - .on(EngineEvent.SignalRequestResponse, this.handleSignalRequestResponse); + .on(EngineEvent.SignalRequestResponse, this.handleSignalRequestResponse) + .on(EngineEvent.StoreDataBlobResponse, this.handleStoreDataBlobResponse) + .on(EngineEvent.GetDataBlobResponse, this.handleGetDataBlobResponse); } private handleReconnecting = () => { @@ -302,6 +327,17 @@ export default class LocalParticipant extends Participant { this.activeAgentFuture?.reject?.(new Error('Got disconnected without active agent present')); this.activeAgentFuture = undefined; this.firstActiveAgent = undefined; + + // In-flight data blob requests are orphaned on close — the server will never respond, + // so fail them fast instead of letting them run into their timeout. + for (const future of this.pendingStoreDataBlobRequests.values()) { + future.reject?.(DataTrackSchemaStorageError.disconnected()); + } + this.pendingStoreDataBlobRequests.clear(); + for (const future of this.pendingGetDataBlobRequests.values()) { + future.reject?.(DataTrackSchemaStorageError.disconnected()); + } + this.pendingGetDataBlobRequests.clear(); }; private handleSignalConnected = (joinResponse: JoinResponse) => { @@ -325,6 +361,21 @@ export default class LocalParticipant extends Participant { this.pendingSignalRequests.delete(requestId); } + // Data blob requests report success via their own response messages and errors via + // `RequestResponse`; both carry the request id, so both paths are correlated by it. + const pendingDataBlobRequest = + this.pendingStoreDataBlobRequests.get(requestId) ?? + this.pendingGetDataBlobRequests.get(requestId); + if (pendingDataBlobRequest && reason !== RequestResponse_Reason.OK) { + pendingDataBlobRequest.reject?.( + reason === RequestResponse_Reason.NOT_FOUND + ? DataTrackSchemaStorageError.notFound(message) + : DataTrackSchemaStorageError.requestFailed(reason, message), + ); + this.pendingStoreDataBlobRequests.delete(requestId); + this.pendingGetDataBlobRequests.delete(requestId); + } + switch (response.request.case) { case 'publishDataTrack': { let error; @@ -355,6 +406,22 @@ export default class LocalParticipant extends Participant { } }; + private handleStoreDataBlobResponse = (response: StoreDataBlobResponse) => { + const pendingRequest = this.pendingStoreDataBlobRequests.get(response.requestId); + if (pendingRequest) { + this.pendingStoreDataBlobRequests.delete(response.requestId); + pendingRequest.resolve?.(response); + } + }; + + private handleGetDataBlobResponse = (response: GetDataBlobResponse) => { + const pendingRequest = this.pendingGetDataBlobRequests.get(response.requestId); + if (pendingRequest) { + this.pendingGetDataBlobRequests.delete(response.requestId); + pendingRequest.resolve?.(response); + } + }; + /** * Sets and updates the metadata of the local participant. * Note: this requires `canUpdateOwnMetadata` permission. @@ -2187,4 +2254,133 @@ export default class LocalParticipant extends Participant { return track; } + + /** + * Stores the definition of a data track schema. + * + * Called by a publisher to make a schema available to subscribers, who can later look + * up its definition via {@link getSchema}. Define a schema before publishing any data + * track that references it, so that subscribers can resolve the schema by its ID. + * + * A schema can only be defined once. Attempting to redefine an existing schema results + * in an error. + * + * @param id Identifies the schema; the same ID is provided when publishing a data track + * that uses it. + * @param definition The schema definition, stored as-is. It is neither parsed nor + * validated against its {@link DataTrackSchemaId.encoding | encoding}, so the caller + * is responsible for ensuring it is well-formed. + * @param signal Optional abort signal to cancel the request. + */ + async defineSchema( + id: DataTrackSchemaId, + definition: string, + signal?: AbortSignal, + ): Promise> { + await this.storeDataBlob( + DataTrackSchemaId.toDataBlobKey(id), + new TextEncoder().encode(definition), + signal, + ); + } + + /** + * Retrieves the definition for a data track schema. + * + * Called by a subscriber that wants to inspect the schema a participant + * {@link defineSchema | defined} for a data track it is publishing. Results in an error + * if the participant has not defined a schema with this ID. + * + * @param id Identifies the schema to retrieve. + * @param participantIdentity Identity of the participant that defined the schema. + * @param signal Optional abort signal to cancel the request. + */ + async getSchema( + id: DataTrackSchemaId, + participantIdentity: string, + signal?: AbortSignal, + ): Promise> { + const contents = await this.getDataBlob( + DataTrackSchemaId.toDataBlobKey(id), + participantIdentity, + signal, + ); + + try { + return new TextDecoder('utf-8', { fatal: true }).decode(contents); + } catch (error) { + throw DataTrackSchemaStorageError.invalidDefinition({ cause: error }); + } + } + + /** Stores an arbitrary blob of data on the server, keyed by `key`. */ + private async storeDataBlob( + key: DataBlobKey, + contents: Uint8Array, + signal?: AbortSignal, + ): Promise> { + const requestId = await this.engine.client.sendStoreDataBlobRequest( + new DataBlob({ key, contents }), + ); + + // The response's key field is ignored; its arrival alone indicates success. + await this.waitForDataBlobResponse(this.pendingStoreDataBlobRequests, requestId, signal); + } + + /** Retrieves a blob of data previously stored by `participantIdentity` under `key`. */ + private async getDataBlob( + key: DataBlobKey, + participantIdentity: string, + signal?: AbortSignal, + ): Promise> { + const requestId = await this.engine.client.sendGetDataBlobRequest(key, participantIdentity); + + const response = await this.waitForDataBlobResponse( + this.pendingGetDataBlobRequests, + requestId, + signal, + ); + if (!response.blob) { + throw DataTrackSchemaStorageError.malformedResponse(); + } + return response.blob.contents; + } + + /** + * Waits for the response to a data blob request, correlated by request id. + * + * The registered future is resolved by the matching response handler or rejected via + * `RequestResponse`; this adds the timeout and caller-cancellation paths and guarantees + * the pending entry is removed however the wait ends. + */ + private async waitForDataBlobResponse( + pendingRequests: Map>, + requestId: number, + signal?: AbortSignal, + ): Promise> { + const future = new Future(); + pendingRequests.set(requestId, future); + + const timeoutSignal = abortSignalTimeout(DATA_BLOB_REQUEST_TIMEOUT_MILLISECONDS); + const combinedSignal = signal ? abortSignalAny([signal, timeoutSignal]) : timeoutSignal; + const onAbort = () => { + future.reject?.( + timeoutSignal.aborted + ? DataTrackSchemaStorageError.timeout() + : DataTrackSchemaStorageError.cancelled(), + ); + }; + if (combinedSignal.aborted) { + onAbort(); + } else { + combinedSignal.addEventListener('abort', onAbort); + } + + try { + return await future.promise; + } finally { + pendingRequests.delete(requestId); + combinedSignal.removeEventListener('abort', onAbort); + } + } }