Add libmediasoup data channel demo (#101)

Author: tdrz

app/lib/RoomClient.js

@@ -2147,7 +2147,6 @@ export default class RoomClient
 
 				setTimeout(() => audioTrack.stop(), 120000);
 			}
-
 			// Create mediasoup Transport for sending (unless we don't want to produce).
 			if (this._produce)
 			{

server/lib/Room.js

@@ -643,6 +643,128 @@ class Room extends EventEmitter
 		};
 	}
 
+	/**
+	 * Create a mediasoup DataConsumer associated to a Broadcaster.
+	 *
+	 * @async
+	 *
+	 * @type {String} broadcasterId
+	 * @type {String} transportId
+	 * @type {String} dataProducerId
+	 */
+	async createBroadcasterDataConsumer(
+		{
+			broadcasterId,
+			transportId,
+			dataProducerId
+		}
+	)
+	{
+		const broadcaster = this._broadcasters.get(broadcasterId);
+
+		if (!broadcaster)
+			throw new Error(`broadcaster with id "${broadcasterId}" does not exist`);
+
+		if (!broadcaster.data.rtpCapabilities)
+			throw new Error('broadcaster does not have rtpCapabilities');
+
+		const transport = broadcaster.data.transports.get(transportId);
+
+		if (!transport)
+			throw new Error(`transport with id "${transportId}" does not exist`);
+
+		const dataConsumer = await transport.consumeData(
+			{
+				dataProducerId
+			});
+
+		// Store it.
+		broadcaster.data.dataConsumers.set(consumer.id, consumer);
+
+		// Set Consumer events.
+		dataConsumer.on('transportclose', () =>
+		{
+			// Remove from its map.
+			broadcaster.data.dataConsumers.delete(consumer.id);
+		});
+
+		dataConsumer.on('dataproducerclose', () =>
+		{
+			// Remove from its map.
+			broadcaster.data.dataConsumers.delete(consumer.id);
+		});
+
+		return {
+			id            : dataConsumer.id
+		};
+	}
+
+	/**
+	 * Create a mediasoup DataProducer associated to a Broadcaster.
+	 *
+	 * @async
+	 *
+	 * @type {String} broadcasterId
+	 * @type {String} transportId
+	 */
+	async createBroadcasterDataProducer(
+		{
+			broadcasterId,
+			transportId,
+			label,
+			protocol,
+			sctpStreamParameters,
+			appData
+		}
+	)
+	{
+		const broadcaster = this._broadcasters.get(broadcasterId);
+
+		if (!broadcaster)
+			throw new Error(`broadcaster with id "${broadcasterId}" does not exist`);
+
+		// if (!broadcaster.data.sctpCapabilities)
+		// 	throw new Error('broadcaster does not have sctpCapabilities');
+
+		const transport = broadcaster.data.transports.get(transportId);
+
+		if (!transport)
+			throw new Error(`transport with id "${transportId}" does not exist`);
+
+		const dataProducer = await transport.produceData(
+			{
+				sctpStreamParameters,
+				label,
+				protocol,
+				appData
+			});
+
+		// Store it.
+		broadcaster.data.dataProducers.set(consumer.id, consumer);
+
+		// Set Consumer events.
+		dataProducer.on('transportclose', () =>
+		{
+			// Remove from its map.
+			broadcaster.data.dataProducers.delete(consumer.id);
+		});
+
+		// // Optimization: Create a server-side Consumer for each Peer.
+		// for (const peer of this._getJoinedPeers())
+		// {
+		// 	this._createDataConsumer(
+		// 		{
+		// 			dataConsumerPeer : peer,
+		// 			dataProducerPeer : broadcaster,
+		// 			dataProducer: dataProducer
+		// 		});
+		// }
+
+		return {
+			id            : dataProducer.id
+		};
+	}
+
 	_handleAudioLevelObserver()
 	{
 		this._audioLevelObserver.on('volumes', (volumes) =>

server/server.js

@@ -224,7 +224,7 @@ async function createExpressApp()
 		async (req, res, next) =>
 		{
 			const { broadcasterId } = req.params;
-			const { type, rtcpMux, comedia } = req.body;
+			const { type, rtcpMux, comedia, sctpCapabilities } = req.body;
 
 			try
 			{
@@ -233,7 +233,8 @@ async function createExpressApp()
 						broadcasterId,
 						type,
 						rtcpMux,
-						comedia
+						comedia, 
+						sctpCapabilities
 					});
 
 				res.status(200).json(data);
@@ -333,6 +334,67 @@ async function createExpressApp()
 			}
 		});
 
+	/**
+	 * POST API to create a mediasoup DataConsumer associated to a Broadcaster.
+	 * The exact Transport in which the DataConsumer must be created is signaled in
+	 * the URL path. Query body must include the desired producerId to
+	 * consume.
+	 */
+	expressApp.post(
+		'/rooms/:roomId/broadcasters/:broadcasterId/transports/:transportId/consume/data',
+		async (req, res, next) =>
+		{
+			const { broadcasterId, transportId } = req.params;
+			const { dataProducerId } = req.body;
+
+			try
+			{
+				const data = await req.room.createBroadcasterDataConsumer(
+					{
+						broadcasterId,
+						transportId,
+						dataProducerId
+					});
+
+				res.status(200).json(data);
+			}
+			catch (error)
+			{
+				next(error);
+			}
+		});
+	
+	/**
+	 * POST API to create a mediasoup DataProducer associated to a Broadcaster.
+	 * The exact Transport in which the DataProducer must be created is signaled in
+	 */
+	expressApp.post(
+		'/rooms/:roomId/broadcasters/:broadcasterId/transports/:transportId/produce/data',
+		async (req, res, next) =>
+		{
+			const { broadcasterId, transportId } = req.params;
+			const { label, protocol, sctpStreamParameters, appData } = req.body;
+
+			try
+			{
+				const data = await req.room.createBroadcasterDataProducer(
+					{
+						broadcasterId,
+						transportId,
+						label,
+						protocol,
+						sctpStreamParameters,
+						appData
+					});
+
+				res.status(200).json(data);
+			}
+			catch (error)
+			{
+				next(error);
+			}
+		});
+
 	/**
 	 * Error handler.
 	 */