Cataloging

CHANGELOG.md

@@ -62,7 +62,6 @@
 
 * Downgrade OpenSSL to version 1.1.0h (27 Mar 2018) until issue [#257](https://github.com/versatica/mediasoup/issues/257) is fixed.
 
-
 ### 2.6.2
 
 * C++: Remove all `Destroy()` class methods and no longer do `delete this`.

lib/Consumer.js

@@ -29,6 +29,7 @@ class Consumer extends EnhancedEventEmitter
 		// - .kind
 		// - .peer (just if this Consumer belongs to a Peer)
 		// - .transport
+		// - .transportCallback
 		// - .rtpParameters
 		// - .source
 		this._data = data;
@@ -222,6 +223,14 @@ class Consumer extends EnhancedEventEmitter
 		// Remove notification subscriptions.
 		this._channel.removeAllListeners(this._internal.consumerId);
 
+		// Remove previously registered listener for transport's @close so that closed Consumer object can be picked up by GC
+		// while a Transport object it used to reffer to still stays alive
+		if (this._data.transport && this._data.transportCallback)
+			this._data.transport.removeListener('@close', this._data.transportCallback);
+
+		this._data.transportCallback = undefined; 
+		this._data.transport = undefined;
+
 		if (notifyChannel)
 		{
 			this._channel.request('consumer.close', this._internal)
@@ -338,13 +347,13 @@ class Consumer extends EnhancedEventEmitter
 
 				this.safeEmit('handled');
 
-				transport.on('@close', () =>
+				this._data.transportCallback = () =>
 				{
 					this._internal.transportId = undefined;
 					this._data.transport = undefined;
 
 					this.safeEmit('unhandled');
-				});
+				};
 			})
 			.catch((error) =>
 			{
@@ -709,7 +718,6 @@ class Consumer extends EnhancedEventEmitter
 
 		if (typeof interval !== 'number')
 		{
-			logger.error('enableStats() | invalid interval provided');
 
 			return;
 		}

lib/Peer.js

@@ -753,10 +753,18 @@ class Peer extends EnhancedEventEmitter
 			transportId : undefined
 		};
 
-		const consumerRtpParameters =
+		let consumerRtpParameters =
 			ortc.getConsumerRtpParameters(
 				producer.consumableRtpParameters, this.rtpCapabilities);
 
+		consumerRtpParameters.codecs.forEach((codec) => {
+			if (codec.rtcpFeedback !== undefined && Array.isArray(codec.rtcpFeedback)) 
+			{
+				let rtcpFeedback = codec.rtcpFeedback.filter((val) => val.type !== 'ffmpeg-proxy');
+				codec.rtcpFeedback = rtcpFeedback;
+			}
+		});
+
 		const data =
 		{
 			kind          : producer.kind,
@@ -790,7 +798,8 @@ class Peer extends EnhancedEventEmitter
 			{
 				logger.error('"router.createConsumer" request failed: %s', String(error));
 
-				throw error;
+				// remove throw as it's causing uncaught promise rejections
+				// throw error;
 			});
 
 		this.safeEmit('newconsumer', consumer);

lib/Producer.js

@@ -28,6 +28,7 @@ class Producer extends EnhancedEventEmitter
 		// - .kind
 		// - .peer
 		// - .transport
+		// -.transportCallback
 		// - .rtpParameters
 		// - .consumableRtpParameters
 		this._data = data;
@@ -184,6 +185,13 @@ class Producer extends EnhancedEventEmitter
 			this.emit('@notify', 'producerClosed', data);
 		}
 
+		// Remove previously registered listener for transport's @close so that closed Producer object can be picked up by GC
+		// while a Transport object it used to reffer to still stays alive
+		if (this.transport && !this.transport.closed)
+			this.transport.removeListener('@close', this._data.transportCallback);
+		this._data.transportCallback = undefined;
+		this._data.transport = undefined;
+
 		this.emit('@close');
 		this.safeEmit('close', 'local', appData);
 
@@ -214,6 +222,11 @@ class Producer extends EnhancedEventEmitter
 			clearInterval(this._statsInterval);
 		}
 
+		if (this.transport && !this.transport.closed)
+			this.transport.removeListener('@close', this._data.transportCallback);
+		this._data.transportCallback = undefined;
+		this._data.transport = undefined;
+
 		this.emit('@close');
 		this.safeEmit('close', 'remote', appData);
 
@@ -589,11 +602,12 @@ class Producer extends EnhancedEventEmitter
 	{
 		// On closure, the worker Transport closes all its Producers and the client
 		// side gets producer.on('unhandled').
-		this.transport.on('@close', () =>
+		this._data.transportCallback = () =>
 		{
 			if (!this._closed)
 				this.close(undefined, false);
-		});
+		};
+		this.transport.on('@close', this._data.transportCallback);
 	}
 
 	_handleWorkerNotifications()

lib/Room.js

@@ -12,7 +12,7 @@ const logger = new Logger('Room');
 
 class Room extends EnhancedEventEmitter
 {
-	constructor(internal, data, channel)
+	constructor(internal, data, channel, getCpuPercentage)
 	{
 		super(logger);
 
@@ -22,9 +22,12 @@ class Room extends EnhancedEventEmitter
 		this._closed = false;
 
 		// Internal data.
+		// - .workerId
 		// - .routerId
 		this._internal = internal;
 
+		this._getCpuPercentage = getCpuPercentage;
+
 		// Room data.
 		// - .rtpCapabilities
 		this._data = data;
@@ -48,6 +51,16 @@ class Room extends EnhancedEventEmitter
 		return this._internal.routerId;
 	}
 
+	get workerId()
+	{
+		return this._internal.workerId;
+	}
+
+	get cpuPercentage()
+	{
+		return this._getCpuPercentage();
+	}
+
 	get closed()
 	{
 		return this._closed;
@@ -227,7 +240,7 @@ class Room extends EnhancedEventEmitter
 
 	createRtpStreamer(producer, options)
 	{
-		logger.debug('createRtpStreamer()');
+		logger.debug('"createRtpStreamer()" sendOldNack is %s producer.kind is %s', options.sendOldNack, producer.kind);
 
 		if (!this._producers.has(producer.id))
 			return Promise.reject(new Error('Producer not found'));
@@ -257,10 +270,22 @@ class Room extends EnhancedEventEmitter
 				};
 
 				// TODO: The app should provide his own RTP capabilities.
-				const consumerRtpParameters =
+				let consumerRtpParameters =
 					ortc.getConsumerRtpParameters(
 						producer.consumableRtpParameters, this.rtpCapabilities);
 
+				// Remove ffmpeg-proxy from capabilities if sendOldNack is not present
+				if (options.sendOldNack !== true)
+				{
+					consumerRtpParameters.codecs.forEach((codec) => {
+						if (codec.rtcpFeedback !== undefined && Array.isArray(codec.rtcpFeedback)) 
+						{
+							let rtcpFeedback = codec.rtcpFeedback.filter((val) => val.type !== 'ffmpeg-proxy');
+							codec.rtcpFeedback = rtcpFeedback;
+						}
+					});
+				}
+
 				const data =
 				{
 					kind          : producer.kind,

lib/Server.js

@@ -56,6 +56,9 @@ class Server extends EnhancedEventEmitter
 		// Remove numWorkers.
 		delete options.numWorkers;
 
+		const workerOptions = options.worker;
+		delete options.worker;
+
 		// Normalize some options.
 
 		if (!check.array(options.logTags))
@@ -142,7 +145,7 @@ class Server extends EnhancedEventEmitter
 
 			// Create a Worker instance (do it in a separate method to avoid creating
 			// a callback function within a loop).
-			this._addWorker(new Worker(workerId, workerParameters));
+			this._addWorker(new Worker(workerId, workerParameters, workerOptions));
 		}
 	}
 
@@ -161,6 +164,11 @@ class Server extends EnhancedEventEmitter
 		return this._workers.size;
 	}
 
+	get workers()
+	{
+		return Array.from(this._workers);
+	}
+
 	/**
 	 * Close the Server.
 	 */
@@ -257,22 +265,49 @@ class Server extends EnhancedEventEmitter
 		// This may throw.
 		const rtpCapabilities = ortc.generateRoomRtpCapabilities(mediaCodecs);
 
-		if (typeof workerIdx === 'number')
-		{
-			if (workerIdx < 0 || workerIdx >= this._workers.size)
-				throw new Error('requested worker idx out of range');
 
-			// Update latest worker idx.
-			this._latestWorkerIdx = workerIdx;
+		const array = Array.from(this._workers);
+
+		let lowestCpu;
+		let lowestCpuIndex;
+		const zeroCpus = [];
+		for (let i = array.length - 1; i >= 0; i--) {
+			const worker = array[i];
+			const wcpu = worker.cpuPercentage;
+			if (wcpu) {
+				if (!lowestCpu || wcpu < lowestCpu) {
+					lowestCpu = wcpu;
+					lowestCpuIndex = i;
+				}
+			} else {
+				zeroCpus.push(worker);
+			}
 		}
-		else
-		{
-			// Increase latest worker idx.
-			this._latestWorkerIdx++;
 
-			// Move to 0 if out of range.
-			if (this._latestWorkerIdx === this._workers.size)
-				this._latestWorkerIdx = 0;
+		if (zeroCpus.length) {
+			if (zeroCpus.length === array.length) {
+				if (typeof workerIdx === 'number')
+				{
+					if (workerIdx < 0 || workerIdx >= this._workers.size)
+						throw new Error('requested worker idx out of range');
+
+					// Update latest worker idx.
+					this._latestWorkerIdx = workerIdx;
+				}
+				else
+				{
+					// Increase latest worker idx.
+					this._latestWorkerIdx++;
+
+					// Move to 0 if out of range.
+					if (this._latestWorkerIdx === this._workers.size)
+						this._latestWorkerIdx = 0;
+				}
+			} else {
+				this._latestWorkerIdx = array.indexOf(zeroCpus[Math.floor(Math.random() * Math.floor(zeroCpus.length))]);
+			}
+		} else {
+			this._latestWorkerIdx = lowestCpuIndex
 		}
 
 		const worker = Array.from(this._workers)[this._latestWorkerIdx];