Fix memory leaks due to event listeners that keep objects alive (#335)

Author: ibc

src/Consumer.ts

@@ -184,7 +184,7 @@ export class Consumer<
 	/**
 	 * Closes the Consumer.
 	 */
-	close(): void {
+	override close(): void {
 		if (this._closed) {
 			return;
 		}
@@ -199,6 +199,10 @@ export class Consumer<
 
 		// Emit observer event.
 		this._observer.safeEmit('close');
+
+		// Invoke close() in EnhancedEventEmitter classes.
+		super.close();
+		this._observer.close();
 	}
 
 	/**

src/DataConsumer.ts

@@ -162,7 +162,7 @@ export class DataConsumer<
 	/**
 	 * Closes the DataConsumer.
 	 */
-	close(): void {
+	override close(): void {
 		if (this._closed) {
 			return;
 		}
@@ -177,6 +177,10 @@ export class DataConsumer<
 
 		// Emit observer event.
 		this._observer.safeEmit('close');
+
+		// Invoke close() in EnhancedEventEmitter classes.
+		super.close();
+		this._observer.close();
 	}
 
 	/**

src/DataProducer.ts

@@ -157,7 +157,7 @@ export class DataProducer<
 	/**
 	 * Closes the DataProducer.
 	 */
-	close(): void {
+	override close(): void {
 		if (this._closed) {
 			return;
 		}
@@ -172,6 +172,10 @@ export class DataProducer<
 
 		// Emit observer event.
 		this._observer.safeEmit('close');
+
+		// Invoke close() in EnhancedEventEmitter classes.
+		super.close();
+		this._observer.close();
 	}
 
 	/**

src/Producer.ts

@@ -237,7 +237,7 @@ export class Producer<
 	/**
 	 * Closes the Producer.
 	 */
-	close(): void {
+	override close(): void {
 		if (this._closed) {
 			return;
 		}
@@ -252,6 +252,10 @@ export class Producer<
 
 		// Emit observer event.
 		this._observer.safeEmit('close');
+
+		// Invoke close() in EnhancedEventEmitter classes.
+		super.close();
+		this._observer.close();
 	}
 
 	/**

src/Transport.ts

@@ -376,7 +376,7 @@ export class Transport<
 	/**
 	 * Close the Transport.
 	 */
-	close(): void {
+	override close(): void {
 		if (this._closed) {
 			return;
 		}
@@ -421,6 +421,10 @@ export class Transport<
 
 		// Emit observer event.
 		this._observer.safeEmit('close');
+
+		// Invoke close() in EnhancedEventEmitter classes.
+		super.close();
+		this._observer.close();
 	}
 
 	/**

src/enhancedEvents.ts

@@ -15,6 +15,13 @@ export class EnhancedEventEmitter<
 		this.setMaxListeners(Infinity);
 	}
 
+	/**
+	 * Empties all stored event listeners.
+	 */
+	close(): void {
+		super.removeAllListeners();
+	}
+
 	override emit<K extends keyof E & string>(
 		eventName: K,
 		...args: E[K]

src/handlers/Chrome111.ts

@@ -1,5 +1,6 @@
 import * as sdpTransform from 'sdp-transform';
 import type * as SdpTransform from 'sdp-transform';
+import { EnhancedEventEmitter } from '../enhancedEvents';
 import { Logger } from '../Logger';
 import * as utils from '../utils';
 import * as ortc from '../ortc';
@@ -15,18 +16,19 @@ import type { SctpCapabilities, SctpStreamParameters } from '../SctpParameters';
 import * as sdpCommonUtils from './sdp/commonUtils';
 import * as sdpUnifiedPlanUtils from './sdp/unifiedPlanUtils';
 import * as ortcUtils from './ortc/utils';
-import {
-	type HandlerFactory,
+import type {
+	HandlerFactory,
 	HandlerInterface,
-	type HandlerOptions,
-	type HandlerSendOptions,
-	type HandlerSendResult,
-	type HandlerReceiveOptions,
-	type HandlerReceiveResult,
-	type HandlerSendDataChannelOptions,
-	type HandlerSendDataChannelResult,
-	type HandlerReceiveDataChannelOptions,
-	type HandlerReceiveDataChannelResult,
+	HandlerEvents,
+	HandlerOptions,
+	HandlerSendOptions,
+	HandlerSendResult,
+	HandlerReceiveOptions,
+	HandlerReceiveResult,
+	HandlerSendDataChannelOptions,
+	HandlerSendDataChannelResult,
+	HandlerReceiveDataChannelOptions,
+	HandlerReceiveDataChannelResult,
 } from './HandlerInterface';
 import { RemoteSdp } from './sdp/RemoteSdp';
 
@@ -35,7 +37,10 @@ const logger = new Logger('Chrome111');
 const NAME = 'Chrome111';
 const SCTP_NUM_STREAMS = { OS: 1024, MIS: 1024 };
 
-export class Chrome111 extends HandlerInterface {
+export class Chrome111
+	extends EnhancedEventEmitter<HandlerEvents>
+	implements HandlerInterface
+{
 	// Closed flag.
 	private _closed = false;
 	// Handler direction.
@@ -76,7 +81,7 @@ export class Chrome111 extends HandlerInterface {
 			getNativeRtpCapabilities: async (): Promise<RtpCapabilities> => {
 				logger.debug('getNativeRtpCapabilities()');
 
-				const pc = new RTCPeerConnection({
+				let pc: RTCPeerConnection | undefined = new RTCPeerConnection({
 					iceServers: [],
 					iceTransportPolicy: 'all',
 					bundlePolicy: 'max-bundle',
@@ -97,6 +102,8 @@ export class Chrome111 extends HandlerInterface {
 						pc.close();
 					} catch (error) {}
 
+					pc = undefined;
+
 					const sdpObject = sdpTransform.parse(offer.sdp!);
 					const nativeRtpCapabilities = sdpCommonUtils.extractRtpCapabilities({
 						sdpObject,
@@ -108,9 +115,11 @@ export class Chrome111 extends HandlerInterface {
 					return nativeRtpCapabilities;
 				} catch (error) {
 					try {
-						pc.close();
+						pc?.close();
 					} catch (error2) {}
 
+					pc = undefined;
+
 					throw error;
 				}
 			},
@@ -177,68 +186,35 @@ export class Chrome111 extends HandlerInterface {
 			...additionalSettings,
 		});
 
-		this._pc.addEventListener('icegatheringstatechange', () => {
-			this.emit('@icegatheringstatechange', this._pc.iceGatheringState);
-		});
-
 		this._pc.addEventListener(
-			'icecandidateerror',
-			(event: RTCPeerConnectionIceErrorEvent) => {
-				this.emit('@icecandidateerror', event);
-			}
+			'icegatheringstatechange',
+			this.onIceGatheringStateChange
 		);
 
+		this._pc.addEventListener('icecandidateerror', this.onIceCandidateError);
+
 		if (this._pc.connectionState) {
-			this._pc.addEventListener('connectionstatechange', () => {
-				this.emit('@connectionstatechange', this._pc.connectionState);
-			});
+			this._pc.addEventListener(
+				'connectionstatechange',
+				this.onConnectionStateChange
+			);
 		} else {
 			logger.warn(
 				'run() | pc.connectionState not supported, using pc.iceConnectionState'
 			);
 
-			this._pc.addEventListener('iceconnectionstatechange', () => {
-				switch (this._pc.iceConnectionState) {
-					case 'checking': {
-						this.emit('@connectionstatechange', 'connecting');
-
-						break;
-					}
-
-					case 'connected':
-					case 'completed': {
-						this.emit('@connectionstatechange', 'connected');
-
-						break;
-					}
-
-					case 'failed': {
-						this.emit('@connectionstatechange', 'failed');
-
-						break;
-					}
-
-					case 'disconnected': {
-						this.emit('@connectionstatechange', 'disconnected');
-
-						break;
-					}
-
-					case 'closed': {
-						this.emit('@connectionstatechange', 'closed');
-
-						break;
-					}
-				}
-			});
+			this._pc.addEventListener(
+				'iceconnectionstatechange',
+				this.onIceConnectionStateChange
+			);
 		}
 	}
 
 	get name(): string {
 		return NAME;
 	}
 
-	close(): void {
+	override close(): void {
 		logger.debug('close()');
 
 		if (this._closed) {
@@ -248,13 +224,31 @@ export class Chrome111 extends HandlerInterface {
 		this._closed = true;
 
 		// Close RTCPeerConnection.
-		if (this._pc) {
-			try {
-				this._pc.close();
-			} catch (error) {}
-		}
+		try {
+			this._pc.close();
+		} catch (error) {}
+
+		this._pc.removeEventListener(
+			'icegatheringstatechange',
+			this.onIceGatheringStateChange
+		);
+
+		this._pc.removeEventListener('icecandidateerror', this.onIceCandidateError);
+
+		this._pc.removeEventListener(
+			'connectionstatechange',
+			this.onConnectionStateChange
+		);
+
+		this._pc.removeEventListener(
+			'iceconnectionstatechange',
+			this.onIceConnectionStateChange
+		);
 
 		this.emit('@close');
+
+		// Invoke close() in EnhancedEventEmitter classes.
+		super.close();
 	}
 
 	async updateIceServers(iceServers: RTCIceServer[]): Promise<void> {
@@ -1193,6 +1187,55 @@ export class Chrome111 extends HandlerInterface {
 		this._transportReady = true;
 	}
 
+	private onIceGatheringStateChange = (): void => {
+		this.emit('@icegatheringstatechange', this._pc.iceGatheringState);
+	};
+
+	private onIceCandidateError = (
+		event: RTCPeerConnectionIceErrorEvent
+	): void => {
+		this.emit('@icecandidateerror', event);
+	};
+
+	private onConnectionStateChange = (): void => {
+		this.emit('@connectionstatechange', this._pc.connectionState);
+	};
+
+	private onIceConnectionStateChange = (): void => {
+		switch (this._pc.iceConnectionState) {
+			case 'checking': {
+				this.emit('@connectionstatechange', 'connecting');
+
+				break;
+			}
+
+			case 'connected':
+			case 'completed': {
+				this.emit('@connectionstatechange', 'connected');
+
+				break;
+			}
+
+			case 'failed': {
+				this.emit('@connectionstatechange', 'failed');
+
+				break;
+			}
+
+			case 'disconnected': {
+				this.emit('@connectionstatechange', 'disconnected');
+
+				break;
+			}
+
+			case 'closed': {
+				this.emit('@connectionstatechange', 'closed');
+
+				break;
+			}
+		}
+	};
+
 	private assertNotClosed(): void {
 		if (this._closed) {
 			throw new InvalidStateError('method called in a closed handler');