/* eslint-disable @typescript-eslint/no-unused-vars */ import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; import { subscribeToEvents } from '../../../utils/subscribeToEvents'; import { type DataTrackFrame } from '../frame'; import { DataTrackHandle, DataTrackHandleAllocator } from '../handle'; import { PrefixingEncryptionProvider } from '../outgoing/OutgoingDataTrackManager.test'; import { DataTrackPacket, DataTrackPacketHeader, FrameMarker } from '../packet'; import { DataTrackE2eeExtension, DataTrackExtensions } from '../packet/extensions'; import { DataTrackTimestamp, WrapAroundUnsignedInt } from '../utils'; import IncomingDataTrackManager, { type DataTrackIncomingManagerCallbacks, } from './IncomingDataTrackManager'; import { DataTrackSubscribeError } from './errors'; describe('DataTrackIncomingManager', () => { describe('Track publication', () => { it('should test track publication additions / removals', async () => { const manager = new IncomingDataTrackManager(); const managerEvents = subscribeToEvents(manager, [ 'sfuUpdateSubscription', 'trackPublished', 'trackUnpublished', ]); // 1. Add a track, make sure the track available event was sent await manager.receiveSfuPublicationUpdates( new Map([ [ 'identity1', [ { sid: 'sid1', pubHandle: DataTrackHandle.fromNumber(5), name: 'test', usesE2ee: false, }, ], ], ]), ); const trackPublishedEvent = await managerEvents.waitFor('trackPublished'); expect(trackPublishedEvent.track.info.sid).toStrictEqual('sid1'); expect(trackPublishedEvent.track.info.pubHandle).toStrictEqual(DataTrackHandle.fromNumber(5)); expect(trackPublishedEvent.track.info.name).toStrictEqual('test'); expect(trackPublishedEvent.track.info.usesE2ee).toStrictEqual(false); // 2. Check to make sure the publication has been noted in internal state expect((await manager.queryPublications()).map((p) => p.pubHandle)).to.deep.equal([ DataTrackHandle.fromNumber(5), ]); // 3. Remove all tracks, and make sure the internal state is cleared await manager.receiveSfuPublicationUpdates(new Map([['identity1', []]])); expect(await manager.queryPublications()).to.deep.equal([]); const trackUnpublishedEvent = await managerEvents.waitFor('trackUnpublished'); expect(trackUnpublishedEvent.sid).toStrictEqual('sid1'); expect(trackUnpublishedEvent.publisherIdentity).toStrictEqual('identity1'); }); it('should process sfu publication updates idempotently', async () => { const manager = new IncomingDataTrackManager(); // 1. Simulate three identical track publications being received for (let i = 0; i < 3; i += 1) { await manager.receiveSfuPublicationUpdates( new Map([ [ 'identity1', [ { sid: 'sid1', pubHandle: DataTrackHandle.fromNumber(5), name: 'test', usesE2ee: false, }, ], ], ]), ); } // 2. Check to make sure the publication has been noted in internal state only once expect((await manager.queryPublications()).map((p) => p.pubHandle)).to.deep.equal([ DataTrackHandle.fromNumber(5), ]); }); it('should not republish when a track sid is reassigned', async () => { const manager = new IncomingDataTrackManager(); const managerEvents = subscribeToEvents(manager, [ 'sfuUpdateSubscription', 'trackPublished', 'trackUnpublished', ]); const pubHandle = DataTrackHandle.fromNumber(5); const oldSid = 'old sid'; const newSid = 'new sid'; // Simulate track published await manager.receiveSfuPublicationUpdates( new Map([['id', [{ sid: oldSid, pubHandle, name: 'test', usesE2ee: false }]]]), ); const { track } = await managerEvents.waitFor('trackPublished'); expect(track.info.sid).toStrictEqual(oldSid); // Simulate publisher full reconnect: same track, new SID await manager.receiveSfuPublicationUpdates( new Map([['id', [{ sid: newSid, pubHandle, name: 'test', usesE2ee: false }]]]), ); // No publish/unpublish should appear expect(managerEvents.areThereBufferedEvents('trackPublished')).toStrictEqual(false); expect(managerEvents.areThereBufferedEvents('trackUnpublished')).toStrictEqual(false); expect(track.info.sid).toStrictEqual(newSid); }); }); describe('Track subscription', () => { it('should test data track subscribing (ok case)', async () => { const manager = new IncomingDataTrackManager(); const managerEvents = subscribeToEvents(manager, [ 'sfuUpdateSubscription', 'trackPublished', ]); const senderIdentity = 'identity'; const sid = 'data track sid'; const handle = DataTrackHandle.fromNumber(5); // 1. Make sure the data track publication is registered await manager.receiveSfuPublicationUpdates( new Map([[senderIdentity, [{ sid, pubHandle: handle, name: 'test', usesE2ee: false }]]]), ); await managerEvents.waitFor('trackPublished'); // 2. Create a subscription readable stream (SFU subscription starts lazily in the background) const [stream, sfuSubscriptionComplete] = manager.openSubscriptionStream(sid); const reader = stream.getReader(); // 3. This subscribe request should be sent along to the SFU const sfuUpdateSubscriptionEvent = await managerEvents.waitFor('sfuUpdateSubscription'); expect(sfuUpdateSubscriptionEvent.sid).toStrictEqual(sid); expect(sfuUpdateSubscriptionEvent.subscribe).toStrictEqual(true); // 4. Once the SFU has acknowledged the subscription, a handle is sent back representing // the subscription manager.receivedSfuSubscriberHandles(new Map([[handle, sid]])); // 5. Wait for the subscription to be fully established await sfuSubscriptionComplete; // 6. Simulate receiving a packet manager.packetReceived( new DataTrackPacket( new DataTrackPacketHeader({ extensions: new DataTrackExtensions(), frameNumber: WrapAroundUnsignedInt.u16(0), marker: FrameMarker.Single, sequence: WrapAroundUnsignedInt.u16(0), timestamp: DataTrackTimestamp.fromRtpTicks(0), trackHandle: DataTrackHandle.fromNumber(5), }), new Uint8Array([0x01, 0x02, 0x03, 0x04, 0x05]), ).toBinary(), ); // 7. Make sure that packet comes out of the ReadableStream const { value, done } = await reader.read(); expect(done).toStrictEqual(false); expect(value?.payload).toStrictEqual(new Uint8Array([0x01, 0x02, 0x03, 0x04, 0x05])); }); it('should test data track subscribing with end to end encryption (ok case)', async () => { const manager = new IncomingDataTrackManager({ e2eeManager: new PrefixingEncryptionProvider(), }); const managerEvents = subscribeToEvents(manager, [ 'sfuUpdateSubscription', 'trackPublished', ]); const senderIdentity = 'identity'; const sid = 'data track sid'; const handle = DataTrackHandle.fromNumber(5); // 1. Make sure the data track publication is registered await manager.receiveSfuPublicationUpdates( new Map([[senderIdentity, [{ sid, pubHandle: handle, name: 'test', usesE2ee: true }]]]), ); await managerEvents.waitFor('trackPublished'); // 2. Create a subscription readable stream const [stream, sfuSubscriptionComplete] = manager.openSubscriptionStream(sid); const reader = stream.getReader(); // 3. This subscribe request should be sent along to the SFU const sfuUpdateSubscriptionEvent = await managerEvents.waitFor('sfuUpdateSubscription'); expect(sfuUpdateSubscriptionEvent.sid).toStrictEqual(sid); expect(sfuUpdateSubscriptionEvent.subscribe).toStrictEqual(true); // 4. Once the SFU has acknowledged the subscription, a handle is sent back representing // the subscription manager.receivedSfuSubscriberHandles(new Map([[handle, sid]])); // 5. Wait for the subscription to be fully established await sfuSubscriptionComplete; // 6. Simulate receiving a (fake) encrypted packet manager.packetReceived( new DataTrackPacket( new DataTrackPacketHeader({ extensions: new DataTrackExtensions({ e2ee: new DataTrackE2eeExtension(0, new Uint8Array(12)), }), frameNumber: WrapAroundUnsignedInt.u16(0), marker: FrameMarker.Single, sequence: WrapAroundUnsignedInt.u16(0), timestamp: DataTrackTimestamp.fromRtpTicks(0), trackHandle: DataTrackHandle.fromNumber(5), }), new Uint8Array([ // Fake encryption bytes prefix 0xde, 0xad, 0xbe, 0xef, // Actual payload 0x01, 0x02, 0x03, 0x04, 0x05, ]), ).toBinary(), ); // 7. Make sure that packet comes out of the ReadableStream const { value, done } = await reader.read(); expect(done).toStrictEqual(false); expect(value?.payload).toStrictEqual(new Uint8Array([0x01, 0x02, 0x03, 0x04, 0x05])); }); it('should fan out received events across multiple subscriptions', async () => { const manager = new IncomingDataTrackManager(); const managerEvents = subscribeToEvents(manager, [ 'sfuUpdateSubscription', 'trackPublished', ]); const senderIdentity = 'identity'; const sid = 'data track sid'; const handleAllocator = new DataTrackHandleAllocator(); // 1. Make sure the data track publication is registered const pubHandle = handleAllocator.get()!; await manager.receiveSfuPublicationUpdates( new Map([[senderIdentity, [{ sid, pubHandle, name: 'test', usesE2ee: false }]]]), ); await managerEvents.waitFor('trackPublished'); // 2. Set up lots of subscribers const readers: Array> = []; for (let index = 0; index < 8; index += 1) { // Create a subscription readable stream const [stream, sfuSubscriptionComplete] = manager.openSubscriptionStream(sid); readers.push(stream.getReader()); // Make sure that the sfu interactions ONLY happen for the first subscription opened. if (index === 0) { // This subscribe request should be sent along to the SFU const sfuUpdateSubscriptionEvent = await managerEvents.waitFor('sfuUpdateSubscription'); expect(sfuUpdateSubscriptionEvent.sid).toStrictEqual(sid); expect(sfuUpdateSubscriptionEvent.subscribe).toStrictEqual(true); // Simulate the subscribe being acknowledged by the SFU manager.receivedSfuSubscriberHandles( new Map([[DataTrackHandle.fromNumber(1 /* publish handle */ + index), sid]]), ); } // 5. Wait for the subscription to be fully established await sfuSubscriptionComplete; } // 6. Simulate receiving a packet manager.packetReceived( new DataTrackPacket( new DataTrackPacketHeader({ extensions: new DataTrackExtensions(), frameNumber: WrapAroundUnsignedInt.u16(0), marker: FrameMarker.Single, sequence: WrapAroundUnsignedInt.u16(0), timestamp: DataTrackTimestamp.fromRtpTicks(0), trackHandle: pubHandle, }), new Uint8Array([0x01, 0x02, 0x03, 0x04, 0x05]), ).toBinary(), ); // 7. Make sure that packet comes out of all of the `ReadableStream`s const results = await Promise.all(readers.map((reader) => reader.read())); for (const { value, done } of results) { expect(done).toStrictEqual(false); expect(value?.payload).toStrictEqual(new Uint8Array([0x01, 0x02, 0x03, 0x04, 0x05])); } }); it('should be unable to subscribe to a non existing data track', async () => { const manager = new IncomingDataTrackManager(); const [, subscriptionPromise] = manager.openSubscriptionStream('does not exist'); await expect(subscriptionPromise).rejects.toThrowError('Cannot subscribe to unknown track'); }); it('should terminate the sfu subscription if the abortsignal is triggered on the only subscription', async () => { const manager = new IncomingDataTrackManager(); const managerEvents = subscribeToEvents(manager, [ 'sfuUpdateSubscription', 'trackPublished', ]); const sid = 'data track sid'; // 1. Make sure the data track publication is registered await manager.receiveSfuPublicationUpdates( new Map([ [ 'identity', [{ sid, pubHandle: DataTrackHandle.fromNumber(5), name: 'test', usesE2ee: false }], ], ]), ); await managerEvents.waitFor('trackPublished'); // 2. Subscribe to a data track const controller = new AbortController(); const [stream, sfuSubscriptionComplete] = manager.openSubscriptionStream( sid, controller.signal, ); await managerEvents.waitFor('sfuUpdateSubscription'); manager.receivedSfuSubscriberHandles(new Map([[DataTrackHandle.fromNumber(5), sid]])); // 3. Wait for the subscription to be fully established await sfuSubscriptionComplete; // 4. Start consuming the readable stream const reader = stream.getReader(); const inFlightReadPromise = reader.read(); // 5. Cancel the subscription controller.abort(); await expect(inFlightReadPromise).rejects.toThrowError( 'Subscription to data track cancelled by caller', ); // 6. Make sure the underlying sfu subscription is also terminated, since nothing needs it // anymore. const sfuUpdateSubscriptionEvent = await managerEvents.waitFor('sfuUpdateSubscription'); expect(sfuUpdateSubscriptionEvent.sid).toStrictEqual(sid); expect(sfuUpdateSubscriptionEvent.subscribe).toStrictEqual(false); // 7. Make sure shutting down the manager doesn't throw errors manager.reset(); }); it('should NOT terminate the sfu subscription if the abortsignal is triggered on one of two active subscriptions', async () => { const manager = new IncomingDataTrackManager(); const managerEvents = subscribeToEvents(manager, [ 'sfuUpdateSubscription', 'trackPublished', ]); const sid = 'data track sid'; // 1. Make sure the data track publication is registered await manager.receiveSfuPublicationUpdates( new Map([ [ 'identity', [{ sid, pubHandle: DataTrackHandle.fromNumber(5), name: 'test', usesE2ee: false }], ], ]), ); await managerEvents.waitFor('trackPublished'); // 2. Subscribe to a data track twice const controllerOne = new AbortController(); const [streamOne, sfuSubscriptionOneComplete] = manager.openSubscriptionStream( sid, controllerOne.signal, ); await managerEvents.waitFor('sfuUpdateSubscription'); // Subscription started manager.receivedSfuSubscriberHandles(new Map([[DataTrackHandle.fromNumber(5), sid]])); await sfuSubscriptionOneComplete; const controllerTwo = new AbortController(); const [streamTwo, sfuSubscriptionTwoComplete] = manager.openSubscriptionStream( sid, controllerTwo.signal, ); // NOTE: no new sfu subscription here, the first stream handled setting this up await sfuSubscriptionTwoComplete; // 3. Start consuming the both subscription's readable streams const readerOne = streamOne.getReader(); const inFlightReadOnePromise = readerOne.read(); const readerTwo = streamTwo.getReader(); const inFlightReadTwoPromise = readerTwo.read(); // 3. Cancel the first subscription, make sure JUST that subscription is cancelled controllerOne.abort(); await expect(inFlightReadOnePromise).rejects.toThrowError( 'Subscription to data track cancelled by caller', ); // 4. Make sure the other subscription is still active / was untouched by the first stream // being aborted. await expect( Promise.race([inFlightReadTwoPromise, Promise.resolve('pending')]), ).resolves.toStrictEqual('pending'); // 4. Make sure the underlying sfu subscription has not been also cancelled, there still is // one data track subscription active expect(managerEvents.areThereBufferedEvents('sfuUpdateSubscription')).toBe(false); }); it('should terminate the sfu subscription if the abortsignal is already aborted', async () => { const manager = new IncomingDataTrackManager(); const managerEvents = subscribeToEvents(manager, [ 'sfuUpdateSubscription', ]); const sid = 'data track sid'; // Make sure the data track publication is registered await manager.receiveSfuPublicationUpdates( new Map([ [ 'identity', [{ sid, pubHandle: DataTrackHandle.fromNumber(5), name: 'test', usesE2ee: false }], ], ]), ); // Subscribe to a data track const [, subscribeRequestPromise] = manager.openSubscriptionStream( sid, AbortSignal.abort(/* already aborted */), ); const start = await managerEvents.waitFor('sfuUpdateSubscription'); expect(start.subscribe).toBe(true); // Make sure cancellation is immediately bubbled up await expect(subscribeRequestPromise).rejects.toStrictEqual( DataTrackSubscribeError.cancelled(), ); // Make sure that there is immediately another "unsubscribe" sent const end = await managerEvents.waitFor('sfuUpdateSubscription'); expect(end.subscribe).toBe(false); }); it('should terminate the sfu subscription once all listeners have unsubscribed', async () => { const manager = new IncomingDataTrackManager(); const managerEvents = subscribeToEvents(manager, [ 'sfuUpdateSubscription', 'trackPublished', ]); const sid = 'data track sid'; // 1. Make sure the data track publication is registered await manager.receiveSfuPublicationUpdates( new Map([ [ 'identity', [{ sid, pubHandle: DataTrackHandle.fromNumber(5), name: 'test', usesE2ee: false }], ], ]), ); await managerEvents.waitFor('trackPublished'); // 2. Create subscription A const controllerA = new AbortController(); const [, subscribeAPromise] = manager.openSubscriptionStream(sid, controllerA.signal); const startEvent = await managerEvents.waitFor('sfuUpdateSubscription'); expect(startEvent.sid).toStrictEqual(sid); expect(startEvent.subscribe).toStrictEqual(true); // 2. Create subscription B const controllerB = new AbortController(); const [, subscribeBPromise] = manager.openSubscriptionStream(sid, controllerB.signal); expect(managerEvents.areThereBufferedEvents('sfuUpdateSubscription')).toStrictEqual(false); // 3. Cancel the subscription A controllerA.abort(); expect(managerEvents.areThereBufferedEvents('sfuUpdateSubscription')).toStrictEqual(false); // 4. Cancel the subscription B, make sure the underlying sfu subscription is disposed controllerB.abort(); const endEvent = await managerEvents.waitFor('sfuUpdateSubscription'); expect(endEvent.sid).toStrictEqual(sid); expect(endEvent.subscribe).toStrictEqual(false); await expect(subscribeAPromise).rejects.toThrow(); await expect(subscribeBPromise).rejects.toThrow(); }); it('should terminate PENDING sfu subscriptions if the participant disconnects', async () => { const manager = new IncomingDataTrackManager(); const managerEvents = subscribeToEvents(manager, [ 'sfuUpdateSubscription', 'trackPublished', ]); const senderIdentity = 'identity'; const sid = 'data track sid'; const handle = DataTrackHandle.fromNumber(5); // 1. Make sure the data track publication is registered await manager.receiveSfuPublicationUpdates( new Map([[senderIdentity, [{ sid, pubHandle: handle, name: 'test', usesE2ee: false }]]]), ); await managerEvents.waitFor('trackPublished'); // 2. Begin subscribing to a data track const [, subscriptionCompletePromise] = manager.openSubscriptionStream(sid); // 3. Simulate the remote participant disconnecting manager.handleRemoteParticipantDisconnected(senderIdentity); // 4. Make sure the pending subscribe was terminated await expect(subscriptionCompletePromise).rejects.toThrowError( 'Cannot subscribe to data track when disconnected', ); }); it('should terminate ACTIVE sfu subscriptions if the participant disconnects', async () => { const manager = new IncomingDataTrackManager(); const managerEvents = subscribeToEvents(manager, [ 'sfuUpdateSubscription', 'trackPublished', ]); const senderIdentity = 'identity'; const sid = 'data track sid'; const handle = DataTrackHandle.fromNumber(5); // 1. Make sure the data track publication is registered await manager.receiveSfuPublicationUpdates( new Map([[senderIdentity, [{ sid, pubHandle: handle, name: 'test', usesE2ee: false }]]]), ); await managerEvents.waitFor('trackPublished'); // 2. Subscribe to a data track, and send the handle back as if the SFU acknowledged it const [stream, sfuSubscriptionComplete] = manager.openSubscriptionStream(sid); const reader = stream.getReader(); const sfuUpdateSubscriptionEvent = await managerEvents.waitFor('sfuUpdateSubscription'); expect(sfuUpdateSubscriptionEvent.sid).toStrictEqual(sid); expect(sfuUpdateSubscriptionEvent.subscribe).toStrictEqual(true); manager.receivedSfuSubscriberHandles(new Map([[handle, sid]])); // 3. Start an active stream read for later await sfuSubscriptionComplete; // 4. Simulate the remote participant disconnecting manager.handleRemoteParticipantDisconnected(senderIdentity); // 5. Make sure the sfu unsubscribes const endEvent = await managerEvents.waitFor('sfuUpdateSubscription'); expect(endEvent.sid).toStrictEqual(sid); expect(endEvent.subscribe).toStrictEqual(false); // 6. Make sure the in flight stream read was closed await reader.closed; }); it('should terminate ACTIVE sfu subscriptions which have been aborted if the participant disconnects', async () => { const manager = new IncomingDataTrackManager(); const managerEvents = subscribeToEvents(manager, [ 'sfuUpdateSubscription', 'trackPublished', ]); const senderIdentity = 'identity'; const sid = 'data track sid'; const handle = DataTrackHandle.fromNumber(5); // 1. Make sure the data track publication is registered await manager.receiveSfuPublicationUpdates( new Map([[senderIdentity, [{ sid, pubHandle: handle, name: 'test', usesE2ee: false }]]]), ); await managerEvents.waitFor('trackPublished'); // 2. Subscribe to a data track, and send the handle back as if the SFU acknowledged it const controller = new AbortController(); const [stream, sfuSubscriptionComplete] = manager.openSubscriptionStream( sid, controller.signal, ); const reader = stream.getReader(); const sfuUpdateSubscriptionEvent = await managerEvents.waitFor('sfuUpdateSubscription'); expect(sfuUpdateSubscriptionEvent.sid).toStrictEqual(sid); expect(sfuUpdateSubscriptionEvent.subscribe).toStrictEqual(true); manager.receivedSfuSubscriberHandles(new Map([[handle, sid]])); // 3. Start an in flight stream read await sfuSubscriptionComplete; const inFlightReadPromise = reader.read(); // 4. Abort the abort controller, which should abort the in flight stream read controller.abort(); await expect(inFlightReadPromise).rejects.toThrowError( 'Subscription to data track cancelled by caller', ); // 4. Simulate the remote participant disconnecting manager.handleRemoteParticipantDisconnected(senderIdentity); // 5. Make sure the sfu unsubscribes const endEvent = await managerEvents.waitFor('sfuUpdateSubscription'); expect(endEvent.sid).toStrictEqual(sid); expect(endEvent.subscribe).toStrictEqual(false); }); it('should terminate ACTIVE sfu subscriptions which have been aborted if the track is unpublished', async () => { const manager = new IncomingDataTrackManager(); const managerEvents = subscribeToEvents(manager, [ 'sfuUpdateSubscription', 'trackPublished', 'trackUnpublished', ]); const senderIdentity = 'identity'; const sid = 'data track sid'; const handle = DataTrackHandle.fromNumber(5); // 1. Make sure the data track publication is registered await manager.receiveSfuPublicationUpdates( new Map([[senderIdentity, [{ sid, pubHandle: handle, name: 'test', usesE2ee: false }]]]), ); await managerEvents.waitFor('trackPublished'); // 2. Subscribe to a data track, and send the handle back as if the SFU acknowledged it const controller = new AbortController(); const [stream, sfuSubscriptionComplete] = manager.openSubscriptionStream( sid, controller.signal, ); const reader = stream.getReader(); await managerEvents.waitFor('sfuUpdateSubscription'); manager.receivedSfuSubscriberHandles(new Map([[handle, sid]])); // 3. Start an in flight stream read await sfuSubscriptionComplete; const inFlightReadPromise = reader.read(); // 4. Abort the controller - this errors the stream's underlying controller controller.abort(); await expect(inFlightReadPromise).rejects.toThrowError( 'Subscription to data track cancelled by caller', ); // 5. Unpublish the track - closeStreamControllers must tolerate the already-errored // controller (this used to crashing with "Cannot close an errored readable stream") await manager.receiveSfuPublicationUpdates(new Map([[senderIdentity, []]])); // 6. Make sure the trackUnpublished event fires const trackUnpublishedEvent = await managerEvents.waitFor('trackUnpublished'); expect(trackUnpublishedEvent.sid).toStrictEqual(sid); }); it('should not throw when shutting down with an aborted active subscription', async () => { const manager = new IncomingDataTrackManager(); const managerEvents = subscribeToEvents(manager, [ 'sfuUpdateSubscription', 'trackPublished', 'trackUnpublished', ]); const senderIdentity = 'identity'; const sid = 'data track sid'; const handle = DataTrackHandle.fromNumber(5); // 1. Make sure the data track publication is registered await manager.receiveSfuPublicationUpdates( new Map([[senderIdentity, [{ sid, pubHandle: handle, name: 'test', usesE2ee: false }]]]), ); await managerEvents.waitFor('trackPublished'); // 2. Subscribe to a data track, and send the handle back as if the SFU acknowledged it const controller = new AbortController(); const [stream, sfuSubscriptionComplete] = manager.openSubscriptionStream( sid, controller.signal, ); await managerEvents.waitFor('sfuUpdateSubscription'); manager.receivedSfuSubscriberHandles(new Map([[handle, sid]])); await sfuSubscriptionComplete; // 3. Abort the controller to error the stream's underlying controller const reader = stream.getReader(); const inFlightReadPromise = reader.read(); controller.abort(); await expect(inFlightReadPromise).rejects.toThrowError( 'Subscription to data track cancelled by caller', ); // 4. Shutdown the manager, and make sure it doesn't throw manager.reset(); // 5. Make sure the trackUnpublished event fires for the descriptor const trackUnpublishedEvent = await managerEvents.waitFor('trackUnpublished'); expect(trackUnpublishedEvent.sid).toStrictEqual(sid); }); it('should close the remaining active stream when one of two active subscriptions is aborted before disconnect', async () => { const manager = new IncomingDataTrackManager(); const managerEvents = subscribeToEvents(manager, [ 'sfuUpdateSubscription', 'trackPublished', ]); const senderIdentity = 'identity'; const sid = 'data track sid'; const handle = DataTrackHandle.fromNumber(5); // 1. Make sure the data track publication is registered await manager.receiveSfuPublicationUpdates( new Map([[senderIdentity, [{ sid, pubHandle: handle, name: 'test', usesE2ee: false }]]]), ); await managerEvents.waitFor('trackPublished'); // 2. Open two subscriptions with separate abort controllers const controllerA = new AbortController(); const [streamA, sfuSubscriptionACompletePromise] = manager.openSubscriptionStream( sid, controllerA.signal, ); await managerEvents.waitFor('sfuUpdateSubscription'); manager.receivedSfuSubscriberHandles(new Map([[handle, sid]])); await sfuSubscriptionACompletePromise; const controllerB = new AbortController(); const [streamB, sfuSubscriptionBCompletePromise] = manager.openSubscriptionStream( sid, controllerB.signal, ); await sfuSubscriptionBCompletePromise; const readerA = streamA.getReader(); const readerB = streamB.getReader(); const inFlightReadAPromise = readerA.read(); // 3. Abort only A - this errors A's controller controllerA.abort(); await expect(inFlightReadAPromise).rejects.toThrowError( 'Subscription to data track cancelled by caller', ); // 4. Disconnect the participant. closeStreamControllers must gracefully close B // without crashing on A's errored controller (which should already have been removed // from the map by onAbort). manager.handleRemoteParticipantDisconnected(senderIdentity); // 5. B's reader closes cleanly await readerB.closed; // 6. Perform a single unsubscribe event - no double-unsubscribe from A's abort + disconnect const endEvent = await managerEvents.waitFor('sfuUpdateSubscription'); expect(endEvent.sid).toStrictEqual(sid); expect(endEvent.subscribe).toStrictEqual(false); expect(managerEvents.areThereBufferedEvents('sfuUpdateSubscription')).toBe(false); }); it('should error the stream if the descriptor is unpublished between subscribe resolve and post-subscribe setup', async () => { const manager = new IncomingDataTrackManager(); const managerEvents = subscribeToEvents(manager, [ 'sfuUpdateSubscription', 'trackPublished', 'trackUnpublished', ]); const senderIdentity = 'identity'; const sid = 'data track sid'; const handle = DataTrackHandle.fromNumber(5); // 1. Make sure the data track publication is registered await manager.receiveSfuPublicationUpdates( new Map([[senderIdentity, [{ sid, pubHandle: handle, name: 'test', usesE2ee: false }]]]), ); await managerEvents.waitFor('trackPublished'); // 2. Start subscribing - the .then handler on subscribeRequest is now pending const [stream, sfuSubscriptionComplete] = manager.openSubscriptionStream(sid); const reader = stream.getReader(); await managerEvents.waitFor('sfuUpdateSubscription'); // 3. Acknowledge the SFU handle - this synchronously resolves the completionFuture // and flips the subscription state to 'active', but the .then microtask has not run yet manager.receivedSfuSubscriberHandles(new Map([[handle, sid]])); // 4. Synchronously unpublish the track before the .then microtask fires manager.handleTrackUnpublished(sid); // 5. When .then runs, the descriptor lookup returns undefined and the handler // must error the stream and reject sfuSubscriptionComplete (instead of hanging) await expect(sfuSubscriptionComplete).rejects.toStrictEqual( DataTrackSubscribeError.disconnected(), ); await expect(reader.read()).rejects.toStrictEqual(DataTrackSubscribeError.disconnected()); }); it('should not throw or emit extra events when aborting after a manager-driven close', async () => { const manager = new IncomingDataTrackManager(); const managerEvents = subscribeToEvents(manager, [ 'sfuUpdateSubscription', 'trackPublished', ]); const senderIdentity = 'identity'; const sid = 'data track sid'; const handle = DataTrackHandle.fromNumber(5); // 1. Make sure the data track publication is registered await manager.receiveSfuPublicationUpdates( new Map([[senderIdentity, [{ sid, pubHandle: handle, name: 'test', usesE2ee: false }]]]), ); await managerEvents.waitFor('trackPublished'); // 2. Subscribe to a data track, and send the handle back as if the SFU acknowledged it const controller = new AbortController(); const [stream, sfuSubscriptionComplete] = manager.openSubscriptionStream( sid, controller.signal, ); const reader = stream.getReader(); await managerEvents.waitFor('sfuUpdateSubscription'); manager.receivedSfuSubscriberHandles(new Map([[handle, sid]])); await sfuSubscriptionComplete; // 3. Manager-driven close via disconnect. detachSignal must have run so that the // user's AbortSignal no longer triggers onAbort. manager.handleRemoteParticipantDisconnected(senderIdentity); await reader.closed; // 4. Consume the unsubscribe event const endEvent = await managerEvents.waitFor('sfuUpdateSubscription'); expect(endEvent.subscribe).toBe(false); // 5. Aborting after the manager has already closed the stream must be a no-op: // no throw, and no additional sfuUpdateSubscription events expect(() => controller.abort()).not.toThrow(); expect(managerEvents.areThereBufferedEvents('sfuUpdateSubscription')).toBe(false); }); it('should terminate the sfu subscription once all downstream ReadableStreams are cancelled', async () => { const manager = new IncomingDataTrackManager(); const managerEvents = subscribeToEvents(manager, [ 'sfuUpdateSubscription', 'trackPublished', ]); const senderIdentity = 'identity'; const sid = 'data track sid'; const handle = DataTrackHandle.fromNumber(5); // 1. Make sure the data track publication is registered await manager.receiveSfuPublicationUpdates( new Map([[senderIdentity, [{ sid, pubHandle: handle, name: 'test', usesE2ee: false }]]]), ); await managerEvents.waitFor('trackPublished'); // 2. Create a subscription readable stream const [stream, sfuSubscriptionComplete] = manager.openSubscriptionStream(sid); const reader = stream.getReader(); // 3. This subscribe request should be sent along to the SFU const sfuUpdateSubscriptionInitEvent = await managerEvents.waitFor('sfuUpdateSubscription'); expect(sfuUpdateSubscriptionInitEvent.sid).toStrictEqual(sid); expect(sfuUpdateSubscriptionInitEvent.subscribe).toStrictEqual(true); // 4. Once the SFU has acknowledged the subscription, a handle is sent back representing // the subscription manager.receivedSfuSubscriberHandles(new Map([[handle, sid]])); // 5. Wait for the subscription to be fully established await sfuSubscriptionComplete; // 6. Manually cancel the readable stream await reader.cancel(); // 7. Make sure the underlying SFU subscription is terminated const sfuUpdateSubscriptionCancelEvent = await managerEvents.waitFor('sfuUpdateSubscription'); expect(sfuUpdateSubscriptionCancelEvent.sid).toStrictEqual(sid); expect(sfuUpdateSubscriptionCancelEvent.subscribe).toStrictEqual(false); // 8. Make sure the in flight stream is now complete await expect(reader.read()).resolves.toStrictEqual({ value: undefined, done: true }); }); it(`should not produce an unhandled promise rejection when RemoteDataTrack.subscribe()'s signal is aborted`, async () => { const manager = new IncomingDataTrackManager(); const managerEvents = subscribeToEvents(manager, [ 'sfuUpdateSubscription', 'trackPublished', ]); const sid = 'data track sid'; // 1. Register the data track so we can get a RemoteDataTrack via trackPublished. await manager.receiveSfuPublicationUpdates( new Map([ [ 'identity', [{ sid, pubHandle: DataTrackHandle.fromNumber(5), name: 'test', usesE2ee: false }], ], ]), ); const { track } = await managerEvents.waitFor('trackPublished'); // 2. Listen for unhandled rejections (coming from sfuSubscriptionComplete) and throw // them so the test will terminate. const onUnhandled = (reason: unknown) => { throw reason; }; process.on('unhandledRejection', onUnhandled); try { const controller = new AbortController(); const stream = track.subscribe({ signal: controller.signal }); // 3. Consume the stream the way a user would, catching the rejection. const caughtByUser: unknown[] = []; const consumerDone = (async () => { try { const reader = stream.getReader(); while (true) { const { done } = await reader.read(); if (done) { return; } } } catch (err) { caughtByUser.push(err); } })(); // Wait until subscribeRequest has kicked off so we abort during the // 'pending' state — the path that rejects sfuSubscriptionComplete. await managerEvents.waitFor('sfuUpdateSubscription'); // 4. Abort the subscription controller.abort(); await consumerDone; // Drain microtasks so any unhandledRejection has a chance to fire. await new Promise((resolve) => setTimeout(resolve, 0)); await new Promise((resolve) => setTimeout(resolve, 0)); // 5. Make sure that no `unhandledrejection`s occur and get bubbled up as user // facing errors. // But, the error should still get raised by the user so they can catch it / do with it as // they please. expect(caughtByUser).toHaveLength(1); } finally { process.off('unhandledRejection', onUnhandled); } }); it('should depacketize multiple interleaved partial frames when setMaxPartialFrames is called before subscribe', async () => { const manager = new IncomingDataTrackManager(); const managerEvents = subscribeToEvents(manager, [ 'sfuUpdateSubscription', 'trackPublished', ]); const senderIdentity = 'identity'; const sid = 'data track sid'; const handle = DataTrackHandle.fromNumber(5); await manager.receiveSfuPublicationUpdates( new Map([[senderIdentity, [{ sid, pubHandle: handle, name: 'test', usesE2ee: false }]]]), ); const trackPublishedEvent = await managerEvents.waitFor('trackPublished'); // Configure the track BEFORE any subscribe. trackPublishedEvent.track.setPipelineOptions({ maxPartialFrames: 3 }); const [stream, sfuSubscriptionComplete] = manager.openSubscriptionStream(sid); const reader = stream.getReader(); await managerEvents.waitFor('sfuUpdateSubscription'); manager.receivedSfuSubscriberHandles(new Map([[handle, sid]])); await sfuSubscriptionComplete; // Two interleaved partial frames: Start(1), Start(2), Final(1), Final(2). With the default // maxPartialFrames=1 frame 1 would be evicted by frame 2; with maxPartialFrames=3 both // frames coexist and emerge. pushInterleavedTwoFramePair(manager, handle, { frameOneNumber: 1, frameOneStartSequence: 0, frameOnePayloads: [new Uint8Array([0xa1]), new Uint8Array([0xa2])], frameTwoNumber: 2, frameTwoStartSequence: 100, frameTwoPayloads: [new Uint8Array([0xb1]), new Uint8Array([0xb2])], }); const first = await reader.read(); expect(first.done).toStrictEqual(false); expect(first.value?.payload).toStrictEqual(new Uint8Array([0xa1, 0xa2])); const second = await reader.read(); expect(second.done).toStrictEqual(false); expect(second.value?.payload).toStrictEqual(new Uint8Array([0xb1, 0xb2])); }); it('should pick up setMaxPartialFrames live on an already-active subscription', async () => { const manager = new IncomingDataTrackManager(); const managerEvents = subscribeToEvents(manager, [ 'sfuUpdateSubscription', 'trackPublished', ]); const senderIdentity = 'identity'; const sid = 'data track sid'; const handle = DataTrackHandle.fromNumber(5); await manager.receiveSfuPublicationUpdates( new Map([[senderIdentity, [{ sid, pubHandle: handle, name: 'test', usesE2ee: false }]]]), ); const trackPublishedEvent = await managerEvents.waitFor('trackPublished'); const [stream, sfuSubscriptionComplete] = manager.openSubscriptionStream(sid); const reader = stream.getReader(); await managerEvents.waitFor('sfuUpdateSubscription'); manager.receivedSfuSubscriberHandles(new Map([[handle, sid]])); await sfuSubscriptionComplete; // Subscription is now active; flip the cap on the live pipeline. trackPublishedEvent.track.setPipelineOptions({ maxPartialFrames: 3 }); pushInterleavedTwoFramePair(manager, handle, { frameOneNumber: 1, frameOneStartSequence: 0, frameOnePayloads: [new Uint8Array([0xa1]), new Uint8Array([0xa2])], frameTwoNumber: 2, frameTwoStartSequence: 100, frameTwoPayloads: [new Uint8Array([0xb1]), new Uint8Array([0xb2])], }); const first = await reader.read(); expect(first.value?.payload).toStrictEqual(new Uint8Array([0xa1, 0xa2])); const second = await reader.read(); expect(second.value?.payload).toStrictEqual(new Uint8Array([0xb1, 0xb2])); }); it('should drop the older partial frame by default (no setMaxPartialFrames call)', async () => { const manager = new IncomingDataTrackManager(); const managerEvents = subscribeToEvents(manager, [ 'sfuUpdateSubscription', 'trackPublished', ]); const senderIdentity = 'identity'; const sid = 'data track sid'; const handle = DataTrackHandle.fromNumber(5); await manager.receiveSfuPublicationUpdates( new Map([[senderIdentity, [{ sid, pubHandle: handle, name: 'test', usesE2ee: false }]]]), ); await managerEvents.waitFor('trackPublished'); const [stream, sfuSubscriptionComplete] = manager.openSubscriptionStream(sid); const reader = stream.getReader(); await managerEvents.waitFor('sfuUpdateSubscription'); manager.receivedSfuSubscriberHandles(new Map([[handle, sid]])); await sfuSubscriptionComplete; // Default cap of 1: Start(2) evicts Start(1), so Final(1) is unknown and only frame 2 // makes it through. pushInterleavedTwoFramePair(manager, handle, { frameOneNumber: 1, frameOneStartSequence: 0, frameOnePayloads: [new Uint8Array([0xa1]), new Uint8Array([0xa2])], frameTwoNumber: 2, frameTwoStartSequence: 100, frameTwoPayloads: [new Uint8Array([0xb1]), new Uint8Array([0xb2])], }); const onlyFrame = await reader.read(); expect(onlyFrame.done).toStrictEqual(false); expect(onlyFrame.value?.payload).toStrictEqual(new Uint8Array([0xb1, 0xb2])); }); it('should resubscribe an active subscription when a track sid is reassigned', async () => { const manager = new IncomingDataTrackManager(); const managerEvents = subscribeToEvents(manager, [ 'sfuUpdateSubscription', 'trackPublished', ]); const pubHandle = DataTrackHandle.fromNumber(5); const oldSid = 'old sid'; const newSid = 'new sid'; const oldSubHandle = DataTrackHandle.fromNumber(6); const newSubHandle = DataTrackHandle.fromNumber(7); // Simulate track published await manager.receiveSfuPublicationUpdates( new Map([['id', [{ sid: oldSid, pubHandle, name: 'test', usesE2ee: false }]]]), ); const { track } = await managerEvents.waitFor('trackPublished'); // Subscribe to the track const [stream, sfuSubscriptionComplete] = manager.openSubscriptionStream(oldSid); const reader = stream.getReader(); await managerEvents.waitFor('sfuUpdateSubscription'); // Simulate SFU assigning subscriber handle manager.receivedSfuSubscriberHandles(new Map([[oldSubHandle, oldSid]])); await sfuSubscriptionComplete; // Simulate publisher full reconnect: same track, new SID await manager.receiveSfuPublicationUpdates( new Map([['id', [{ sid: newSid, pubHandle, name: 'test', usesE2ee: false }]]]), ); // Manager should re-subscribe under the new SID const event = await managerEvents.waitFor('sfuUpdateSubscription'); expect(event.subscribe).toStrictEqual(true); expect(event.sid).toStrictEqual(newSid); expect(track.info.sid).toStrictEqual(newSid); // Simulate SFU assigning a new subscriber handle manager.receivedSfuSubscriberHandles(new Map([[newSubHandle, newSid]])); // Frames received on the new handle reach the existing subscriber manager.packetReceived( new DataTrackPacket( new DataTrackPacketHeader({ extensions: new DataTrackExtensions(), frameNumber: WrapAroundUnsignedInt.u16(0), marker: FrameMarker.Single, sequence: WrapAroundUnsignedInt.u16(0), timestamp: DataTrackTimestamp.fromRtpTicks(0), trackHandle: newSubHandle, }), new Uint8Array([0x01, 0x02, 0x03, 0x04, 0x05]), ).toBinary(), ); const { value, done } = await reader.read(); expect(done).toStrictEqual(false); expect(value?.payload).toStrictEqual(new Uint8Array([0x01, 0x02, 0x03, 0x04, 0x05])); }); it('should stop routing packets on the old handle once a new handle is assigned', async () => { const manager = new IncomingDataTrackManager(); const managerEvents = subscribeToEvents(manager, [ 'sfuUpdateSubscription', 'trackPublished', ]); const pubHandle = DataTrackHandle.fromNumber(5); const oldSid = 'old sid'; const newSid = 'new sid'; const oldSubHandle = DataTrackHandle.fromNumber(6); const newSubHandle = DataTrackHandle.fromNumber(7); // Simulate track published await manager.receiveSfuPublicationUpdates( new Map([['id', [{ sid: oldSid, pubHandle, name: 'test', usesE2ee: false }]]]), ); await managerEvents.waitFor('trackPublished'); // Subscribe to the track const [stream, sfuSubscriptionComplete] = manager.openSubscriptionStream(oldSid); const reader = stream.getReader(); await managerEvents.waitFor('sfuUpdateSubscription'); // Simulate SFU assigning subscriber handle manager.receivedSfuSubscriberHandles(new Map([[oldSubHandle, oldSid]])); await sfuSubscriptionComplete; // Simulate publisher full reconnect: same track, new SID await manager.receiveSfuPublicationUpdates( new Map([['id', [{ sid: newSid, pubHandle, name: 'test', usesE2ee: false }]]]), ); await managerEvents.waitFor('sfuUpdateSubscription'); // Simulate SFU assigning a new subscriber handle manager.receivedSfuSubscriberHandles(new Map([[newSubHandle, newSid]])); const buildPacket = (trackHandle: DataTrackHandle, frameNumber: number, payload: number[]) => new DataTrackPacket( new DataTrackPacketHeader({ extensions: new DataTrackExtensions(), frameNumber: WrapAroundUnsignedInt.u16(frameNumber), marker: FrameMarker.Single, sequence: WrapAroundUnsignedInt.u16(frameNumber), timestamp: DataTrackTimestamp.fromRtpTicks(0), trackHandle, }), new Uint8Array(payload), ).toBinary(); // A packet received on the old handle must be dropped, not routed to the subscriber await manager.packetReceived(buildPacket(oldSubHandle, 0, [0xaa, 0xbb])); // A packet received on the new handle reaches the subscriber await manager.packetReceived(buildPacket(newSubHandle, 1, [0x01, 0x02])); const { value, done } = await reader.read(); expect(done).toStrictEqual(false); expect(value?.payload).toStrictEqual(new Uint8Array([0x01, 0x02])); }); it('should terminate the sfu subscription when the stream is cancelled after a sid reassignment', async () => { const manager = new IncomingDataTrackManager(); const managerEvents = subscribeToEvents(manager, [ 'sfuUpdateSubscription', 'trackPublished', ]); const pubHandle = DataTrackHandle.fromNumber(5); const oldSid = 'old sid'; const newSid = 'new sid'; const oldSubHandle = DataTrackHandle.fromNumber(6); const newSubHandle = DataTrackHandle.fromNumber(7); // Simulate track published await manager.receiveSfuPublicationUpdates( new Map([['id', [{ sid: oldSid, pubHandle, name: 'test', usesE2ee: false }]]]), ); await managerEvents.waitFor('trackPublished'); // Subscribe to the track const [stream, sfuSubscriptionComplete] = manager.openSubscriptionStream(oldSid); const reader = stream.getReader(); await managerEvents.waitFor('sfuUpdateSubscription'); // Simulate SFU assigning subscriber handle manager.receivedSfuSubscriberHandles(new Map([[oldSubHandle, oldSid]])); await sfuSubscriptionComplete; // Simulate publisher full reconnect: same track, new SID await manager.receiveSfuPublicationUpdates( new Map([['id', [{ sid: newSid, pubHandle, name: 'test', usesE2ee: false }]]]), ); // Manager should re-subscribe under the new SID const resubscribeEvent = await managerEvents.waitFor('sfuUpdateSubscription'); expect(resubscribeEvent.subscribe).toStrictEqual(true); expect(resubscribeEvent.sid).toStrictEqual(newSid); // Simulate SFU assigning a new subscriber handle manager.receivedSfuSubscriberHandles(new Map([[newSubHandle, newSid]])); // Cancel the stream; the SFU unsubscribe must go out under the new SID await reader.cancel(); const endEvent = await managerEvents.waitFor('sfuUpdateSubscription'); expect(endEvent.sid).toStrictEqual(newSid); expect(endEvent.subscribe).toStrictEqual(false); }); it('should terminate the sfu subscription when the abortsignal fires after a sid reassignment', async () => { const manager = new IncomingDataTrackManager(); const managerEvents = subscribeToEvents(manager, [ 'sfuUpdateSubscription', 'trackPublished', ]); const pubHandle = DataTrackHandle.fromNumber(5); const oldSid = 'old sid'; const newSid = 'new sid'; const oldSubHandle = DataTrackHandle.fromNumber(6); const newSubHandle = DataTrackHandle.fromNumber(7); // Simulate track published await manager.receiveSfuPublicationUpdates( new Map([['id', [{ sid: oldSid, pubHandle, name: 'test', usesE2ee: false }]]]), ); await managerEvents.waitFor('trackPublished'); // Subscribe to the track const controller = new AbortController(); const [stream, sfuSubscriptionComplete] = manager.openSubscriptionStream( oldSid, controller.signal, ); const reader = stream.getReader(); await managerEvents.waitFor('sfuUpdateSubscription'); // Simulate SFU assigning subscriber handle manager.receivedSfuSubscriberHandles(new Map([[oldSubHandle, oldSid]])); await sfuSubscriptionComplete; // Simulate publisher full reconnect: same track, new SID await manager.receiveSfuPublicationUpdates( new Map([['id', [{ sid: newSid, pubHandle, name: 'test', usesE2ee: false }]]]), ); // Manager should re-subscribe under the new SID const resubscribeEvent = await managerEvents.waitFor('sfuUpdateSubscription'); expect(resubscribeEvent.subscribe).toStrictEqual(true); expect(resubscribeEvent.sid).toStrictEqual(newSid); // Simulate SFU assigning a new subscriber handle manager.receivedSfuSubscriberHandles(new Map([[newSubHandle, newSid]])); // Abort the subscription; the SFU unsubscribe must go out under the new SID const inFlightReadPromise = reader.read(); controller.abort(); await expect(inFlightReadPromise).rejects.toThrowError( 'Subscription to data track cancelled by caller', ); const endEvent = await managerEvents.waitFor('sfuUpdateSubscription'); expect(endEvent.sid).toStrictEqual(newSid); expect(endEvent.subscribe).toStrictEqual(false); }); it('should cancel a pending subscription under the new sid after a sid reassignment', async () => { const manager = new IncomingDataTrackManager(); const managerEvents = subscribeToEvents(manager, [ 'sfuUpdateSubscription', 'trackPublished', ]); const pubHandle = DataTrackHandle.fromNumber(5); const oldSid = 'old sid'; const newSid = 'new sid'; // Simulate track published await manager.receiveSfuPublicationUpdates( new Map([['id', [{ sid: oldSid, pubHandle, name: 'test', usesE2ee: false }]]]), ); await managerEvents.waitFor('trackPublished'); // Begin subscribing to the track, leaving the subscription pending const controller = new AbortController(); const [, sfuSubscriptionComplete] = manager.openSubscriptionStream(oldSid, controller.signal); const startEvent = await managerEvents.waitFor('sfuUpdateSubscription'); expect(startEvent.sid).toStrictEqual(oldSid); expect(startEvent.subscribe).toStrictEqual(true); // Simulate publisher full reconnect: same track, new SID await manager.receiveSfuPublicationUpdates( new Map([['id', [{ sid: newSid, pubHandle, name: 'test', usesE2ee: false }]]]), ); // Manager should re-subscribe under the new SID const resubscribeEvent = await managerEvents.waitFor('sfuUpdateSubscription'); expect(resubscribeEvent.subscribe).toStrictEqual(true); expect(resubscribeEvent.sid).toStrictEqual(newSid); // Abort the pending subscription; the SFU unsubscribe must go out under the new SID controller.abort(); await expect(sfuSubscriptionComplete).rejects.toStrictEqual( DataTrackSubscribeError.cancelled(), ); const endEvent = await managerEvents.waitFor('sfuUpdateSubscription'); expect(endEvent.sid).toStrictEqual(newSid); expect(endEvent.subscribe).toStrictEqual(false); }); }); }); /** Pushes Start(frame1), Start(frame2), Final(frame1), Final(frame2) packets through the manager * to exercise the depacketizer's concurrent-partial-frame handling. */ function pushInterleavedTwoFramePair( manager: IncomingDataTrackManager, trackHandle: DataTrackHandle, args: { frameOneNumber: number; frameOneStartSequence: number; frameOnePayloads: [Uint8Array, Uint8Array]; frameTwoNumber: number; frameTwoStartSequence: number; frameTwoPayloads: [Uint8Array, Uint8Array]; }, ) { const buildPacket = ( frameNumber: number, sequence: number, marker: FrameMarker, payload: Uint8Array, ) => new DataTrackPacket( new DataTrackPacketHeader({ extensions: new DataTrackExtensions(), frameNumber: WrapAroundUnsignedInt.u16(frameNumber), marker, sequence: WrapAroundUnsignedInt.u16(sequence), timestamp: DataTrackTimestamp.fromRtpTicks(0), trackHandle, }), payload, ).toBinary(); manager.packetReceived( buildPacket( args.frameOneNumber, args.frameOneStartSequence, FrameMarker.Start, args.frameOnePayloads[0], ), ); manager.packetReceived( buildPacket( args.frameTwoNumber, args.frameTwoStartSequence, FrameMarker.Start, args.frameTwoPayloads[0], ), ); manager.packetReceived( buildPacket( args.frameOneNumber, args.frameOneStartSequence + 1, FrameMarker.Final, args.frameOnePayloads[1], ), ); manager.packetReceived( buildPacket( args.frameTwoNumber, args.frameTwoStartSequence + 1, FrameMarker.Final, args.frameTwoPayloads[1], ), ); }