Node: add enhancedOnce() typed function (#1375)

### Details - Same as [events.once(emmiter, eventName)](https://nodejs.org/api/events.html#eventsonceemitter-name-options) but with TypeScript typed events. - Rename `EnhancedEventEmitter.ts` file to `enhancedEvents.ts` for consistency with the Node `events` module.
versatica · Apr 10, 2024 · e516ebc · e516ebc
1 parent 8be8493
commit e516ebc
Show file tree

Hide file tree

Showing 29 changed files with 199 additions and 159 deletions.
diff --git a/node/src/ActiveSpeakerObserver.ts b/node/src/ActiveSpeakerObserver.ts
@@ -1,5 +1,5 @@
 import { Logger } from './Logger';
-import { EnhancedEventEmitter } from './EnhancedEventEmitter';
+import { EnhancedEventEmitter } from './enhancedEvents';
 import {
 	RtpObserver,
 	RtpObserverEvents,

diff --git a/node/src/AudioLevelObserver.ts b/node/src/AudioLevelObserver.ts
@@ -1,5 +1,5 @@
 import { Logger } from './Logger';
-import { EnhancedEventEmitter } from './EnhancedEventEmitter';
+import { EnhancedEventEmitter } from './enhancedEvents';
 import {
 	RtpObserver,
 	RtpObserverEvents,

diff --git a/node/src/Channel.ts b/node/src/Channel.ts
@@ -3,7 +3,7 @@ import { Duplex } from 'node:stream';
 import { info, warn } from 'node:console';
 import * as flatbuffers from 'flatbuffers';
 import { Logger } from './Logger';
-import { EnhancedEventEmitter } from './EnhancedEventEmitter';
+import { EnhancedEventEmitter } from './enhancedEvents';
 import { InvalidStateError } from './errors';
 import { Body as RequestBody, Method, Request } from './fbs/request';
 import { Response } from './fbs/response';

diff --git a/node/src/Consumer.ts b/node/src/Consumer.ts
@@ -1,5 +1,5 @@
 import { Logger } from './Logger';
-import { EnhancedEventEmitter } from './EnhancedEventEmitter';
+import { EnhancedEventEmitter } from './enhancedEvents';
 import { Channel } from './Channel';
 import { TransportInternal } from './Transport';
 import { ProducerStat } from './Producer';

diff --git a/node/src/DataConsumer.ts b/node/src/DataConsumer.ts
@@ -1,5 +1,5 @@
 import { Logger } from './Logger';
-import { EnhancedEventEmitter } from './EnhancedEventEmitter';
+import { EnhancedEventEmitter } from './enhancedEvents';
 import { Channel } from './Channel';
 import { TransportInternal } from './Transport';
 import {

diff --git a/node/src/DataProducer.ts b/node/src/DataProducer.ts
@@ -1,5 +1,5 @@
 import { Logger } from './Logger';
-import { EnhancedEventEmitter } from './EnhancedEventEmitter';
+import { EnhancedEventEmitter } from './enhancedEvents';
 import { Channel } from './Channel';
 import { TransportInternal } from './Transport';
 import {

diff --git a/node/src/Producer.ts b/node/src/Producer.ts
@@ -1,5 +1,5 @@
 import { Logger } from './Logger';
-import { EnhancedEventEmitter } from './EnhancedEventEmitter';
+import { EnhancedEventEmitter } from './enhancedEvents';
 import { Channel } from './Channel';
 import { TransportInternal } from './Transport';
 import { MediaKind, RtpParameters, parseRtpParameters } from './RtpParameters';

diff --git a/node/src/Router.ts b/node/src/Router.ts
@@ -1,5 +1,5 @@
 import { Logger } from './Logger';
-import { EnhancedEventEmitter } from './EnhancedEventEmitter';
+import { EnhancedEventEmitter } from './enhancedEvents';
 import * as ortc from './ortc';
 import { InvalidStateError } from './errors';
 import { Channel } from './Channel';

diff --git a/node/src/RtpObserver.ts b/node/src/RtpObserver.ts
@@ -1,5 +1,5 @@
 import { Logger } from './Logger';
-import { EnhancedEventEmitter } from './EnhancedEventEmitter';
+import { EnhancedEventEmitter } from './enhancedEvents';
 import { Channel } from './Channel';
 import { RouterInternal } from './Router';
 import { Producer } from './Producer';

diff --git a/node/src/Transport.ts b/node/src/Transport.ts
@@ -1,6 +1,6 @@
 import * as flatbuffers from 'flatbuffers';
 import { Logger } from './Logger';
-import { EnhancedEventEmitter } from './EnhancedEventEmitter';
+import { EnhancedEventEmitter } from './enhancedEvents';
 import * as ortc from './ortc';
 import { Channel } from './Channel';
 import { RouterInternal } from './Router';

diff --git a/node/src/WebRtcServer.ts b/node/src/WebRtcServer.ts
@@ -1,5 +1,5 @@
 import { Logger } from './Logger';
-import { EnhancedEventEmitter } from './EnhancedEventEmitter';
+import { EnhancedEventEmitter } from './enhancedEvents';
 import { Channel } from './Channel';
 import { TransportListenInfo } from './Transport';
 import { WebRtcTransport } from './WebRtcTransport';

diff --git a/node/src/Worker.ts b/node/src/Worker.ts
@@ -3,7 +3,7 @@ import * as path from 'node:path';
 import { spawn, ChildProcess } from 'node:child_process';
 import { version } from './';
 import { Logger } from './Logger';
-import { EnhancedEventEmitter } from './EnhancedEventEmitter';
+import { EnhancedEventEmitter } from './enhancedEvents';
 import * as ortc from './ortc';
 import { Channel } from './Channel';
 import { Router, RouterOptions } from './Router';

diff --git a/node/src/EnhancedEventEmitter.ts → node/src/enhancedEvents.ts b/node/src/EnhancedEventEmitter.ts → node/src/enhancedEvents.ts
@@ -1,7 +1,7 @@
-import { EventEmitter } from 'node:events';
+import { EventEmitter, once } from 'node:events';
 import { Logger } from './Logger';
 
-const logger = new Logger('EnhancedEventEmitter');
+const enhancedEventEmitterLogger = new Logger('EnhancedEventEmitter');
 
 type Events = Record<string, any[]>;
 
@@ -24,7 +24,7 @@ export class EnhancedEventEmitter<
 		try {
 			return super.emit(eventName, ...args);
 		} catch (error) {
-			logger.error(
+			enhancedEventEmitterLogger.error(
 				'safeEmit() | event listener threw an error [eventName:%s]:%o',
 				eventName,
 				error
@@ -121,3 +121,20 @@ export class EnhancedEventEmitter<
 		return super.rawListeners(eventName);
 	}
 }
+
+/**
+ * TypeScript version of events.once():
+ *   https://nodejs.org/api/events.html#eventsonceemitter-name-options
+ *
+ * Usage example:
+ * ```ts
+ * await enhancedOnce<ConsumerEvents>(videoConsumer, 'producerpause');
+ * ````
+ */
+export async function enhancedOnce<E extends Events = Events>(
+	emmiter: EnhancedEventEmitter<E>,
+	eventName: keyof E & string,
+	options?: any
+): Promise<any[]> {
+	return once(emmiter, eventName, options);
+}
diff --git a/node/src/index.ts b/node/src/index.ts
@@ -1,5 +1,5 @@
 import { Logger } from './Logger';
-import { EnhancedEventEmitter } from './EnhancedEventEmitter';
+import { EnhancedEventEmitter } from './enhancedEvents';
 import { workerBin, Worker, WorkerSettings } from './Worker';
 import * as utils from './utils';
 import { supportedRtpCapabilities } from './supportedRtpCapabilities';

diff --git a/node/src/test/test-ActiveSpeakerObserver.ts b/node/src/test/test-ActiveSpeakerObserver.ts
@@ -1,4 +1,6 @@
 import * as mediasoup from '../';
+import { enhancedOnce } from '../enhancedEvents';
+import { WorkerEvents, ActiveSpeakerObserverEvents } from '../types';
 import * as utils from '../utils';
 
 type TestContext = {
@@ -31,9 +33,7 @@ afterEach(async () => {
 	ctx.worker?.close();
 
 	if (ctx.worker?.subprocessClosed === false) {
-		await new Promise<void>(resolve =>
-			ctx.worker?.on('subprocessclose', resolve)
-		);
+		await enhancedOnce<WorkerEvents>(ctx.worker, 'subprocessclose');
 	}
 });
 
@@ -105,21 +105,27 @@ test('activeSpeakerObserver.close() succeeds', async () => {
 test('ActiveSpeakerObserver emits "routerclose" if Router is closed', async () => {
 	const activeSpeakerObserver = await ctx.router!.createAudioLevelObserver();
 
-	await new Promise<void>(resolve => {
-		activeSpeakerObserver.on('routerclose', resolve);
-		ctx.router!.close();
-	});
+	const promise = enhancedOnce<ActiveSpeakerObserverEvents>(
+		activeSpeakerObserver,
+		'routerclose'
+	);
+
+	ctx.router!.close();
+	await promise;
 
 	expect(activeSpeakerObserver.closed).toBe(true);
 }, 2000);
 
 test('ActiveSpeakerObserver emits "routerclose" if Worker is closed', async () => {
 	const activeSpeakerObserver = await ctx.router!.createAudioLevelObserver();
 
-	await new Promise<void>(resolve => {
-		activeSpeakerObserver.on('routerclose', resolve);
-		ctx.worker!.close();
-	});
+	const promise = enhancedOnce<ActiveSpeakerObserverEvents>(
+		activeSpeakerObserver,
+		'routerclose'
+	);
+
+	ctx.worker!.close();
+	await promise;
 
 	expect(activeSpeakerObserver.closed).toBe(true);
 }, 2000);
diff --git a/node/src/test/test-AudioLevelObserver.ts b/node/src/test/test-AudioLevelObserver.ts
@@ -1,4 +1,6 @@
 import * as mediasoup from '../';
+import { enhancedOnce } from '../enhancedEvents';
+import { WorkerEvents, AudioLevelObserverEvents } from '../types';
 import * as utils from '../utils';
 
 type TestContext = {
@@ -31,9 +33,7 @@ afterEach(async () => {
 	ctx.worker?.close();
 
 	if (ctx.worker?.subprocessClosed === false) {
-		await new Promise<void>(resolve =>
-			ctx.worker?.on('subprocessclose', resolve)
-		);
+		await enhancedOnce<WorkerEvents>(ctx.worker, 'subprocessclose');
 	}
 });
 
@@ -114,21 +114,27 @@ test('audioLevelObserver.close() succeeds', async () => {
 test('AudioLevelObserver emits "routerclose" if Router is closed', async () => {
 	const audioLevelObserver = await ctx.router!.createAudioLevelObserver();
 
-	await new Promise<void>(resolve => {
-		audioLevelObserver.on('routerclose', resolve);
-		ctx.router!.close();
-	});
+	const promise = enhancedOnce<AudioLevelObserverEvents>(
+		audioLevelObserver,
+		'routerclose'
+	);
+
+	ctx.router!.close();
+	await promise;
 
 	expect(audioLevelObserver.closed).toBe(true);
 }, 2000);
 
 test('AudioLevelObserver emits "routerclose" if Worker is closed', async () => {
 	const audioLevelObserver = await ctx.router!.createAudioLevelObserver();
 
-	await new Promise<void>(resolve => {
-		audioLevelObserver.on('routerclose', resolve);
-		ctx.worker!.close();
-	});
+	const promise = enhancedOnce<AudioLevelObserverEvents>(
+		audioLevelObserver,
+		'routerclose'
+	);
+
+	ctx.worker!.close();
+	await promise;
 
 	expect(audioLevelObserver.closed).toBe(true);
 }, 2000);
diff --git a/node/src/test/test-Consumer.ts b/node/src/test/test-Consumer.ts
@@ -1,6 +1,7 @@
-import { once } from 'node:events';
 import * as flatbuffers from 'flatbuffers';
 import * as mediasoup from '../';
+import { enhancedOnce } from '../enhancedEvents';
+import { WorkerEvents, ConsumerEvents } from '../types';
 import { UnsupportedError } from '../errors';
 import * as utils from '../utils';
 import {
@@ -248,9 +249,7 @@ afterEach(async () => {
 	ctx.worker?.close();
 
 	if (ctx.worker?.subprocessClosed === false) {
-		await new Promise<void>(resolve =>
-			ctx.worker?.on('subprocessclose', resolve)
-		);
+		await enhancedOnce<WorkerEvents>(ctx.worker, 'subprocessclose');
 	}
 });
 
@@ -1025,7 +1024,7 @@ test('Consumer emits "producerpause" and "producerresume"', async () => {
 	});
 
 	await Promise.all([
-		once(audioConsumer, 'producerpause'),
+		enhancedOnce<ConsumerEvents>(audioConsumer, 'producerpause'),
 
 		// Let's await for pause() to resolve to avoid aborted channel requests
 		// due to worker closure.
@@ -1036,7 +1035,7 @@ test('Consumer emits "producerpause" and "producerresume"', async () => {
 	expect(audioConsumer.producerPaused).toBe(true);
 
 	await Promise.all([
-		once(audioConsumer, 'producerresume'),
+		enhancedOnce<ConsumerEvents>(audioConsumer, 'producerresume'),
 
 		// Let's await for resume() to resolve to avoid aborted channel requests
 		// due to worker closure.
@@ -1162,10 +1161,10 @@ test('Consumer emits "producerclose" if Producer is closed', async () => {
 
 	audioConsumer.observer.once('close', onObserverClose);
 
-	await new Promise<void>(resolve => {
-		audioConsumer.on('producerclose', resolve);
-		ctx.audioProducer!.close();
-	});
+	const promise = enhancedOnce<ConsumerEvents>(audioConsumer, 'producerclose');
+
+	ctx.audioProducer!.close();
+	await promise;
 
 	expect(onObserverClose).toHaveBeenCalledTimes(1);
 	expect(audioConsumer.closed).toBe(true);
@@ -1181,10 +1180,10 @@ test('Consumer emits "transportclose" if Transport is closed', async () => {
 
 	videoConsumer.observer.once('close', onObserverClose);
 
-	await new Promise<void>(resolve => {
-		videoConsumer.on('transportclose', resolve);
-		ctx.webRtcTransport2!.close();
-	});
+	const promise = enhancedOnce<ConsumerEvents>(videoConsumer, 'transportclose');
+
+	ctx.webRtcTransport2!.close();
+	await promise;
 
 	expect(onObserverClose).toHaveBeenCalledTimes(1);
 	expect(videoConsumer.closed).toBe(true);

diff --git a/node/src/test/test-DataConsumer.ts b/node/src/test/test-DataConsumer.ts
@@ -1,4 +1,6 @@
 import * as mediasoup from '../';
+import { enhancedOnce } from '../enhancedEvents';
+import { WorkerEvents, DataConsumerEvents } from '../types';
 import * as utils from '../utils';
 
 type TestContext = {
@@ -44,9 +46,7 @@ afterEach(async () => {
 	ctx.worker?.close();
 
 	if (ctx.worker?.subprocessClosed === false) {
-		await new Promise<void>(resolve =>
-			ctx.worker?.on('subprocessclose', resolve)
-		);
+		await enhancedOnce<WorkerEvents>(ctx.worker, 'subprocessclose');
 	}
 });
 
@@ -374,11 +374,13 @@ test('DataConsumer emits "dataproducerclose" if DataProducer is closed', async (
 
 	dataConsumer.observer.once('close', onObserverClose);
 
-	await new Promise<void>(resolve => {
-		dataConsumer.on('dataproducerclose', resolve);
+	const promise = enhancedOnce<DataConsumerEvents>(
+		dataConsumer,
+		'dataproducerclose'
+	);
 
-		ctx.dataProducer!.close();
-	});
+	ctx.dataProducer!.close();
+	await promise;
 
 	expect(onObserverClose).toHaveBeenCalledTimes(1);
 	expect(dataConsumer.closed).toBe(true);
@@ -392,11 +394,13 @@ test('DataConsumer emits "transportclose" if Transport is closed', async () => {
 
 	dataConsumer.observer.once('close', onObserverClose);
 
-	await new Promise<void>(resolve => {
-		dataConsumer.on('transportclose', resolve);
+	const promise = enhancedOnce<DataConsumerEvents>(
+		dataConsumer,
+		'transportclose'
+	);
 
-		ctx.webRtcTransport2!.close();
-	});
+	ctx.webRtcTransport2!.close();
+	await promise;
 
 	expect(onObserverClose).toHaveBeenCalledTimes(1);
 	expect(dataConsumer.closed).toBe(true);