diff --git a/meet/server/lib/Peer.js b/meet/server/lib/Peer.js index eb0984e6..05813a2a 100644 --- a/meet/server/lib/Peer.js +++ b/meet/server/lib/Peer.js @@ -1,281 +1,292 @@ const EventEmitter = require('events').EventEmitter; const Logger = require('./Logger'); +const Roles = require('./userRoles'); const logger = new Logger('Peer'); class Peer extends EventEmitter { constructor({ id, roomId }) { logger.info('constructor() [id:"%s"]', id); super(); this._id = id; this._roomId = roomId; this._socket = null; this._closed = false; this._role = 0; this._nickname = false; this._picture = null; - this._email = null; - this._routerId = null; this._rtpCapabilities = null; this._raisedHand = false; this._transports = new Map(); this._producers = new Map(); this._consumers = new Map(); } close() { logger.info('close()'); this._closed = true; // Iterate and close all mediasoup Transport associated to this Peer, so all // its Producers and Consumers will also be closed. for (const transport of this.transports.values()) { transport.close(); } if (this.socket) this.socket.disconnect(true); this.emit('close'); } get id() { return this._id; } set id(id) { this._id = id; } get roomId() { return this._roomId; } set roomId(roomId) { this._roomId = roomId; } get socket() { return this._socket; } set socket(socket) { this._socket = socket; if (this.socket) { this.socket.on('disconnect', () => { if (this.closed) return; logger.debug('"disconnect" event [id:%s]', this.id); this.close(); }); } } get closed() { return this._closed; } get role() { return this._role; } get nickname() { return this._nickname; } set nickname(nickname) { if (nickname !== this._nickname) { this._nickname = nickname; this.emit('nicknameChanged', {}); } } get picture() { return this._picture; } set picture(picture) { if (picture !== this._picture) { const oldPicture = this._picture; this._picture = picture; this.emit('pictureChanged', { oldPicture }); } } - get email() - { - return this._email; - } - - set email(email) - { - this._email = email; - } - get routerId() { return this._routerId; } set routerId(routerId) { this._routerId = routerId; } get rtpCapabilities() { return this._rtpCapabilities; } set rtpCapabilities(rtpCapabilities) { this._rtpCapabilities = rtpCapabilities; } get raisedHand() { return this._raisedHand; } set raisedHand(raisedHand) { this._raisedHand = raisedHand; } get transports() { return this._transports; } get producers() { return this._producers; } get consumers() { return this._consumers; } setRole(newRole) { if (this._role != newRole) { + // It is either screen sharing or publisher/subscriber + if (newRole & Roles.SCREEN) { + if (newRole & Roles.PUBLISHER) { + newRole ^= Roles.PUBLISHER; + } + if (newRole & Roles.SUBSCRIBER) { + newRole ^= Roles.SUBSCRIBER; + } + } + this._role = newRole; this.emit('gotRole', { newRole }); } } + isValidRole(newRole) + { + Object.keys(Roles).forEach(roleId => { + const role = Roles[roleId] + if (newRole & role) { + newRole = newRole ^ role; + } + }) + + return newRole == 0; + } + hasRole(role) { return !!(this._role & role); } addTransport(id, transport) { this.transports.set(id, transport); } getTransport(id) { return this.transports.get(id); } getConsumerTransport() { return Array.from(this.transports.values()) .find((t) => t.appData.consuming); } removeTransport(id) { this.transports.delete(id); } addProducer(id, producer) { this.producers.set(id, producer); } getProducer(id) { return this.producers.get(id); } removeProducer(id) { this.producers.delete(id); } addConsumer(id, consumer) { this.consumers.set(id, consumer); } getConsumer(id) { return this.consumers.get(id); } removeConsumer(id) { this.consumers.delete(id); } get peerInfo() { const peerInfo = { id: this.id, nickname: this.nickname, // picture: this.picture, role: this.role, raisedHand: this.raisedHand }; return peerInfo; } } module.exports = Peer; diff --git a/meet/server/lib/Room.js b/meet/server/lib/Room.js index dd76ca10..182397ee 100644 --- a/meet/server/lib/Room.js +++ b/meet/server/lib/Room.js @@ -1,1263 +1,1280 @@ const EventEmitter = require('events').EventEmitter; const AwaitQueue = require('awaitqueue'); const axios = require('axios'); const Logger = require('./Logger'); const { SocketTimeoutError } = require('./errors'); const Roles = require('./userRoles'); const config = require('../config/config'); const logger = new Logger('Room'); const ROUTER_SCALE_SIZE = config.routerScaleSize || 40; class Room extends EventEmitter { static calculateLoads(mediasoupWorkers, peers, mediasoupRouters) { const routerLoads = new Map(); const workerLoads = new Map(); const pipedRoutersIds = new Set(); - // Calculate router loads by adding up peers per router, - // and collecte piped routers + // Calculate router loads by adding up peers per router, and collected piped routers for (const peer of peers) { const routerId = peer.routerId; if (routerId) { if (mediasoupRouters.has(routerId)) { pipedRoutersIds.add(routerId); } if (routerLoads.has(routerId)) { routerLoads.set(routerId, routerLoads.get(routerId) + 1); } else { routerLoads.set(routerId, 1); } } } // Calculate worker loads by adding up router loads per worker for (const worker of mediasoupWorkers) { for (const router of worker._routers) { const routerId = router._internal.routerId; if (workerLoads.has(worker._pid)) { workerLoads.set(worker._pid, workerLoads.get(worker._pid) + (routerLoads.has(routerId)?routerLoads.get(routerId):0)); } else { workerLoads.set(worker._pid, (routerLoads.has(routerId)?routerLoads.get(routerId):0)); } } } return {routerLoads, workerLoads, pipedRoutersIds}; } /* * Find a router that is on a worker that is least loaded. * * A worker with a router that we are already piping to is preferred. */ static getLeastLoadedRouter(mediasoupWorkers, peers, mediasoupRouters) { const {routerLoads, workerLoads, pipedRoutersIds} = Room.calculateLoads(mediasoupWorkers, peers.values(), mediasoupRouters); const sortedWorkerLoads = new Map([ ...workerLoads.entries() ].sort( (a, b) => a[1] - b[1])); // we don't care about if router is piped, just choose the least loaded worker if (pipedRoutersIds.size === 0 || pipedRoutersIds.size === mediasoupRouters.size) { const workerId = sortedWorkerLoads.keys().next().value; for (const worker of mediasoupWorkers) { if (worker._pid === workerId) { for (const router of worker._routers) { const routerId = router._internal.routerId; if (mediasoupRouters.has(routerId)) { return routerId; } } } } } else { // find if there is a piped router that is on a worker that is below limit for (const [ workerId, workerLoad ] of sortedWorkerLoads.entries()) { for (const worker of mediasoupWorkers) { if (worker._pid === workerId) { for (const router of worker._routers) { const routerId = router._internal.routerId; // on purpose we check if the worker load is below the limit, // as in reality the worker load is imortant, // not the router load if (mediasoupRouters.has(routerId) && pipedRoutersIds.has(routerId) && workerLoad < ROUTER_SCALE_SIZE) { return routerId; } } } } } // no piped router found, we need to return router from least loaded worker const workerId = sortedWorkerLoads.keys().next().value; for (const worker of mediasoupWorkers) { if (worker._pid === workerId) { for (const router of worker._routers) { const routerId = router._internal.routerId; if (mediasoupRouters.has(routerId)) { return routerId; } } } } } } /** * Factory function that creates and returns Room instance. * * @async * * @param {mediasoup.Worker} mediasoupWorkers - The mediasoup Worker in which a new * mediasoup Router must be created. * @param {String} roomId - Id of the Room instance. */ static async create({ mediasoupWorkers, roomId, peers }) { logger.info('create() [roomId:"%s"]', roomId); // Router media codecs. const mediaCodecs = config.mediasoup.router.mediaCodecs; const mediasoupRouters = new Map(); for (const worker of mediasoupWorkers) { const router = await worker.createRouter({ mediaCodecs }); mediasoupRouters.set(router.id, router); } const firstRouter = mediasoupRouters.get(Room.getLeastLoadedRouter( mediasoupWorkers, peers, mediasoupRouters)); // Create a mediasoup AudioLevelObserver on first router const audioLevelObserver = await firstRouter.createAudioLevelObserver( { maxEntries : 1, threshold : -80, interval : 800 }); return new Room({ roomId, mediasoupRouters, audioLevelObserver, mediasoupWorkers, peers }); } constructor({ roomId, mediasoupRouters, audioLevelObserver, mediasoupWorkers, peers }) { logger.info('constructor() [roomId:"%s"]', roomId); super(); this.setMaxListeners(Infinity); // this._uuid = uuidv4(); this._mediasoupWorkers = mediasoupWorkers; this._allPeers = peers; // Room ID. this._roomId = roomId; // Closed flag. this._closed = false; // Joining queue this._queue = new AwaitQueue(); this._peers = {}; this._selfDestructTimeout = null; // Array of mediasoup Router instances. this._mediasoupRouters = mediasoupRouters; } dumpStats() { const peers = this.getPeers(); const {routerLoads, workerLoads, pipedRoutersIds} = Room.calculateLoads(this._mediasoupWorkers, peers, this._mediasoupRouters); let stats = { numberOfWorkers: this._mediasoupWorkers.length, numberOfRouters: this._mediasoupRouters.size, numberOfPeers: peers.length, routerLoads: routerLoads, workerLoads: workerLoads, pipedRoutersIds: pipedRoutersIds, }; console.log(stats); } close() { logger.debug('close()'); this._closed = true; this._queue.close(); this._queue = null; if (this._selfDestructTimeout) clearTimeout(this._selfDestructTimeout); this._selfDestructTimeout = null; // Close the peers. - for (const peer in this._peers) - { - if (!this._peers[peer].closed) - this._peers[peer].close(); + for (const peer of this._peers.values()) { + if (!peer.closed) + peer.close(); } this._peers = null; // Close the mediasoup Routers. - for (const router of this._mediasoupRouters.values()) - { + for (const router of this._mediasoupRouters.values()) { router.close(); } this._allPeers = null; this._mediasoupWorkers = null; this._mediasoupRouters.clear(); this._audioLevelObserver = null; // Emit 'close' event. this.emit('close'); } handlePeer({ peer }) { logger.info('handlePeer() [peer:"%s", role:%s]', peer.id, peer.role); // Should not happen if (this._peers[peer.id]) { logger.warn( 'handleConnection() | there is already a peer with same peerId [peer:"%s"]', peer.id); } this._peerJoining(peer); } logStatus() { logger.info( 'logStatus() [room id:"%s", peers:"%s"]', this._roomId, Object.keys(this._peers).length ); } dump() { return { roomId : this._roomId, peers : Object.keys(this._peers).length }; } get id() { return this._roomId; } selfDestructCountdown() { logger.debug('selfDestructCountdown() started'); if (this._selfDestructTimeout) clearTimeout(this._selfDestructTimeout); this._selfDestructTimeout = setTimeout(() => { if (this._closed) return; if (this.checkEmpty()) { logger.info( 'Room deserted for some time, closing the room [roomId:"%s"]', this._roomId); this.close(); } else logger.debug('selfDestructCountdown() aborted; room is not empty!'); }, 10000); } checkEmpty() { return Object.keys(this._peers).length === 0; } _peerJoining(peer) { this._queue.push(async () => { peer.socket.join(this._roomId); this._peers[peer.id] = peer; // Assign routerId peer.routerId = await this._getRouterId(); this._handlePeer(peer); let turnServers; if ('turnAPIURI' in config) { try { const { data } = await axios.get( config.turnAPIURI, { timeout : config.turnAPITimeout || 2000, params : { ...config.turnAPIparams, 'api_key' : config.turnAPIKey, 'ip' : peer.socket.request.connection.remoteAddress } }); turnServers = [ { urls : data.uris, username : data.username, credential : data.password } ]; } catch (error) { if ('backupTurnServers' in config && config.backupTurnServers.length) turnServers = config.backupTurnServers; logger.error('_peerJoining() | error on REST turn [error:"%o"]', error); } } else if ('backupTurnServers' in config && config.backupTurnServers.length) { turnServers = config.backupTurnServers; } this._notification(peer.socket, 'roomReady', { turnServers }); }) .catch((error) => { logger.error('_peerJoining() [error:"%o"]', error); }); } _handlePeer(peer) { logger.debug('_handlePeer() [peer:"%s"]', peer.id); peer.on('close', () => { this._handlePeerClose(peer); }); peer.on('nicknameChanged', () => { // Spread to others this._notification(peer.socket, 'changeNickname', { peerId: peer.id, nickname: peer.nickname }, true); }); peer.on('gotRole', ({ newRole }) => { // Spread to others - this._notification(peer.socket, 'gotRole', { + this._notification(peer.socket, 'changeRole', { peerId: peer.id, role: newRole }, true, true); }); peer.socket.on('request', (request, cb) => { logger.debug( 'Peer "request" event [method:"%s", peerId:"%s"]', request.method, peer.id); this._handleSocketRequest(peer, request, cb) .catch((error) => { logger.error('"request" failed [error:"%o"]', error); cb(error); }); }); // Peer left before we were done joining if (peer.closed) this._handlePeerClose(peer); } _handlePeerClose(peer) { logger.debug('_handlePeerClose() [peer:"%s"]', peer.id); if (this._closed) return; this._notification(peer.socket, 'peerClosed', { peerId: peer.id }, true); delete this._peers[peer.id]; // If this is the last Peer in the room close the room after a while. if (this.checkEmpty()) this.selfDestructCountdown(); } async _handleSocketRequest(peer, request, cb) { const router = this._mediasoupRouters.get(peer.routerId); console.log(request.method); switch (request.method) { case 'getRouterRtpCapabilities': { cb(null, router.rtpCapabilities); break; } case 'dumpStats': { this.dumpStats() cb(null); break; } case 'join': { const { nickname, picture, rtpCapabilities } = request.data; // Store client data into the Peer data object. peer.nickname = nickname; peer.picture = picture; peer.rtpCapabilities = rtpCapabilities; // Tell the new Peer about already joined Peers. // And also create Consumers for existing Producers. const otherPeers = this.getPeers(peer); const peerInfos = otherPeers .map((otherPeer) => (otherPeer.peerInfo)); cb(null, { id: peer.id, role: peer.role, peers: peerInfos, }); - for (const otherPeer of otherPeers) - { + for (const otherPeer of otherPeers) { // Create Consumers for existing Producers. - for (const producer of otherPeer.producers.values()) - { - this._createConsumer( - { - consumerPeer : peer, - producerPeer : otherPeer, + for (const producer of otherPeer.producers.values()) { + this._createConsumer({ + consumerPeer: peer, + producerPeer: otherPeer, producer - }); + }); } - } - // Notify the new Peer to all other Peers. - for (const otherPeer of this.getPeers(peer)) - { - this._notification( - otherPeer.socket, - 'newPeer', - peer.peerInfo - ); + // Notify the new Peer to all other Peers. + this._notification(otherPeer.socket, 'newPeer', peer.peerInfo); } logger.debug( 'peer joined [peer: "%s", nickname: "%s", picture: "%s"]', peer.id, nickname, picture); break; } case 'createPlainTransport': { const { producing, consuming } = request.data; const transport = await router.createPlainTransport( { //When consuming we manually connect using connectPlainTransport, //otherwise we let the port autodetection work. comedia: producing, // FFmpeg and GStreamer don't support RTP/RTCP multiplexing ("a=rtcp-mux" in SDP) rtcpMux: false, listenIp: { ip: "127.0.0.1", announcedIp: null }, appData : { producing, consuming } } ); // await transport.enableTraceEvent([ "probation", "bwe" ]); // transport.on("trace", (trace) => { // console.log(trace); // }); peer.addTransport(transport.id, transport); cb( null, { id : transport.id, ip : transport.tuple.localIp, port : transport.tuple.localPort, rtcpPort : transport.rtcpTuple ? transport.rtcpTuple.localPort : undefined }); break; } case 'connectPlainTransport': { const { transportId, ip, port, rtcpPort } = request.data; const transport = peer.getTransport(transportId); if (!transport) throw new Error(`transport with id "${transportId}" not found`); await transport.connect({ ip: ip, port: port, rtcpPort: rtcpPort, }); cb(); break; } case 'createWebRtcTransport': { // NOTE: Don't require that the Peer is joined here, so the client can // initiate mediasoup Transports and be ready when he later joins. const { forceTcp, producing, consuming } = request.data; const webRtcTransportOptions = { ...config.mediasoup.webRtcTransport, appData : { producing, consuming } }; webRtcTransportOptions.enableTcp = true; if (forceTcp) webRtcTransportOptions.enableUdp = false; else { webRtcTransportOptions.enableUdp = true; webRtcTransportOptions.preferUdp = true; } const transport = await router.createWebRtcTransport( webRtcTransportOptions ); transport.on('dtlsstatechange', (dtlsState) => { if (dtlsState === 'failed' || dtlsState === 'closed') { logger.warn('WebRtcTransport "dtlsstatechange" event [dtlsState:%s]', dtlsState); } }); // Store the WebRtcTransport into the Peer data Object. peer.addTransport(transport.id, transport); cb( null, { id : transport.id, iceParameters : transport.iceParameters, iceCandidates : transport.iceCandidates, dtlsParameters : transport.dtlsParameters }); const { maxIncomingBitrate } = config.mediasoup.webRtcTransport; // If set, apply max incoming bitrate limit. if (maxIncomingBitrate) { try { await transport.setMaxIncomingBitrate(maxIncomingBitrate); } catch (error) { logger.info("Setting the incoming bitrate failed") } } break; } case 'connectWebRtcTransport': { const { transportId, dtlsParameters } = request.data; const transport = peer.getTransport(transportId); if (!transport) throw new Error(`transport with id "${transportId}" not found`); await transport.connect({ dtlsParameters }); cb(); break; } /* case 'restartIce': { const { transportId } = request.data; const transport = peer.getTransport(transportId); if (!transport) throw new Error(`transport with id "${transportId}" not found`); const iceParameters = await transport.restartIce(); cb(null, iceParameters); break; } */ case 'produce': { let { appData } = request.data; if (!appData.source || ![ 'mic', 'webcam', 'screen' ].includes(appData.source)) throw new Error('invalid producer source'); if (appData.source === 'mic' && !peer.hasRole(Roles.PUBLISHER)) throw new Error('peer not authorized'); if (appData.source === 'webcam' && !peer.hasRole(Roles.PUBLISHER)) throw new Error('peer not authorized'); if (appData.source === 'screen' && !peer.hasRole(Roles.PUBLISHER)) throw new Error('peer not authorized'); const { transportId, kind, rtpParameters } = request.data; const transport = peer.getTransport(transportId); if (!transport) throw new Error(`transport with id "${transportId}" not found`); // Add peerId into appData to later get the associated Peer during // the 'loudest' event of the audioLevelObserver. appData = { ...appData, peerId: peer.id }; const producer = await transport.produce({ kind, rtpParameters, appData }); const pipeRouters = this._getRoutersToPipeTo(peer.routerId); for (const [ routerId, destinationRouter ] of this._mediasoupRouters) { if (pipeRouters.includes(routerId)) { await router.pipeToRouter({ producerId : producer.id, router : destinationRouter }); } } // Store the Producer into the Peer data Object. peer.addProducer(producer.id, producer); producer.on('videoorientationchange', (videoOrientation) => { logger.debug( 'producer "videoorientationchange" event [producerId:"%s", videoOrientation:"%o"]', producer.id, videoOrientation); }); // Trace individual packets for debugging // await producer.enableTraceEvent([ "rtp", "pli", "keyframe", "nack" ]); // producer.on("trace", (trace) => { // console.log(`Trace on ${producer.id}`, trace); // }); cb(null, { id: producer.id }); // Optimization: Create a server-side Consumer for each Peer. for (const otherPeer of this.getPeers(peer)) { this._createConsumer( { consumerPeer : otherPeer, producerPeer : peer, producer }); } // Add into the audioLevelObserver. if (kind === 'audio') { this._audioLevelObserver.addProducer({ producerId: producer.id }) .catch(() => {}); } break; } case 'closeProducer': { const { producerId } = request.data; const producer = peer.getProducer(producerId); if (!producer) throw new Error(`producer with id "${producerId}" not found`); producer.close(); // Remove from its map. peer.removeProducer(producer.id); cb(); break; } case 'pauseProducer': { const { producerId } = request.data; const producer = peer.getProducer(producerId); if (!producer) throw new Error(`producer with id "${producerId}" not found`); await producer.pause(); cb(); break; } case 'resumeProducer': { const { producerId } = request.data; const producer = peer.getProducer(producerId); if (!producer) throw new Error(`producer with id "${producerId}" not found`); await producer.resume(); cb(); break; } case 'pauseConsumer': { const { consumerId } = request.data; const consumer = peer.getConsumer(consumerId); if (!consumer) throw new Error(`consumer with id "${consumerId}" not found`); await consumer.pause(); cb(); break; } case 'resumeConsumer': { const { consumerId } = request.data; const consumer = peer.getConsumer(consumerId); if (!consumer) throw new Error(`consumer with id "${consumerId}" not found`); await consumer.resume(); cb(); break; } case 'changeNickname': { const { nickname } = request.data; peer.nickname = nickname; // This will be spread through events from the peer object // Return no error cb(); break; } case 'chatMessage': { const { message } = request.data; // Spread to others this._notification(peer.socket, 'chatMessage', { peerId: peer.id, nickname: peer.nickname, message: message }, true, true); // Return no error cb(); break; } - case 'moderator:setRole': + case 'moderator:addRole': { if (!peer.hasRole(Roles.MODERATOR)) throw new Error('peer not authorized'); const { peerId, role } = request.data; - const giveRolePeer = this._peers[peerId]; + const rolePeer = this._peers[peerId]; - if (!giveRolePeer) + if (!rolePeer) throw new Error(`peer with id "${peerId}" not found`); - // TODO: check if role is valid value + if (!rolePeer.isValidRole(role)) + throw new Error('invalid role'); - // This will propagate the event automatically - giveRolePeer.setRole(role); + if (!rolePeer.hasRole(role)) { + // This will propagate the event automatically + rolePeer.setRole(rolePeer.role | role); + } // Return no error cb(); break; } - case 'raisedHand': + case 'moderator:removeRole': { - const { raisedHand } = request.data; + if (!peer.hasRole(Roles.MODERATOR)) + throw new Error('peer not authorized'); - peer.raisedHand = raisedHand; + const { peerId, role } = request.data; - // Spread to others - this._notification(peer.socket, 'raisedHand', { - peerId: peer.id, - raisedHand: raisedHand, - }, true); + const rolePeer = this._peers[peerId]; + + if (!rolePeer) + throw new Error(`peer with id "${peerId}" not found`); + + if (!rolePeer.isValidRole(role)) + throw new Error('invalid role'); + + if (rolePeer.hasRole(role)) { + // This will propagate the event automatically + rolePeer.setRole(rolePeer.role ^ role); + } // Return no error cb(); break; } case 'moderator:closeRoom': { if (!peer.hasRole(Roles.OWNER)) throw new Error('peer not authorized'); this._notification(peer.socket, 'moderator:closeRoom', null, true); cb(); // Close the room this.close(); + // TODO: remove the room? + break; } case 'moderator:kickPeer': { if (!peer.hasRole(Roles.MODERATOR)) throw new Error('peer not authorized'); const { peerId } = request.data; const kickPeer = this._peers[peerId]; if (!kickPeer) throw new Error(`peer with id "${peerId}" not found`); this._notification(kickPeer.socket, 'moderator:kickPeer'); kickPeer.close(); cb(); break; } + case 'raisedHand': + { + const { raisedHand } = request.data; + + peer.raisedHand = raisedHand; + + // Spread to others + this._notification(peer.socket, 'raisedHand', { + peerId: peer.id, + raisedHand: raisedHand, + }, true); + + // Return no error + cb(); + + break; + } + default: { logger.error('unknown request.method "%s"', request.method); cb(500, `unknown request.method "${request.method}"`); } } } /** * Creates a mediasoup Consumer for the given mediasoup Producer. * * @async */ async _createConsumer({ consumerPeer, producerPeer, producer }) { logger.debug( '_createConsumer() [consumerPeer:"%s", producerPeer:"%s", producer:"%s"]', consumerPeer.id, producerPeer.id, producer.id ); const router = this._mediasoupRouters.get(producerPeer.routerId); // Optimization: // - Create the server-side Consumer. If video, do it paused. // - Tell its Peer about it and wait for its response. // - Upon receipt of the response, resume the server-side Consumer. // - If video, this will mean a single key frame requested by the // server-side Consumer (when resuming it). // NOTE: Don't create the Consumer if the remote Peer cannot consume it. if ( !consumerPeer.rtpCapabilities || !router.canConsume( { producerId : producer.id, rtpCapabilities : consumerPeer.rtpCapabilities }) ) { return; } // Must take the Transport the remote Peer is using for consuming. const transport = consumerPeer.getConsumerTransport(); // This should not happen. if (!transport) { logger.warn('_createConsumer() | Transport for consuming not found'); return; } // Create the Consumer in paused mode. let consumer; try { consumer = await transport.consume( { producerId : producer.id, rtpCapabilities : consumerPeer.rtpCapabilities, paused : producer.kind === 'video' }); if (producer.kind === 'audio') await consumer.setPriority(255); } catch (error) { logger.warn('_createConsumer() | [error:"%o"]', error); return; } // Trace individual packets for debugging // await consumer.enableTraceEvent([ "rtp", "pli", "fir" ]); // consumer.on("trace", (trace) => { // console.log(`Trace on ${consumer.id}`, trace); // }); // Store the Consumer into the consumerPeer data Object. consumerPeer.addConsumer(consumer.id, consumer); // Set Consumer events. consumer.on('transportclose', () => { // Remove from its map. consumerPeer.removeConsumer(consumer.id); }); consumer.on('producerclose', () => { // Remove from its map. consumerPeer.removeConsumer(consumer.id); this._notification(consumerPeer.socket, 'consumerClosed', { consumerId: consumer.id }); }); consumer.on('producerpause', () => { this._notification(consumerPeer.socket, 'consumerPaused', { consumerId: consumer.id }); }); consumer.on('producerresume', () => { this._notification(consumerPeer.socket, 'consumerResumed', { consumerId: consumer.id }); }); // Send a request to the remote Peer with Consumer parameters. try { await this._request( consumerPeer.socket, 'newConsumer', { peerId : producerPeer.id, kind : consumer.kind, producerId : producer.id, id : consumer.id, rtpParameters : consumer.rtpParameters, type : consumer.type, appData : producer.appData, producerPaused : consumer.producerPaused } ); // Now that we got the positive response from the remote Peer and, if // video, resume the Consumer to ask for an efficient key frame. await consumer.resume(); } catch (error) { logger.warn('_createConsumer() | [error:"%o"]', error); } } /** * Get the list of peers. */ getPeers(excludePeer = undefined) { return Object.values(this._peers) .filter((peer) => peer !== excludePeer); } _timeoutCallback(callback) { let called = false; const interval = setTimeout( () => { if (called) return; called = true; callback(new SocketTimeoutError('Request timed out')); }, config.requestTimeout || 20000 ); return (...args) => { if (called) return; called = true; clearTimeout(interval); callback(...args); }; } _sendRequest(socket, method, data = {}) { return new Promise((resolve, reject) => { socket.emit( 'request', { method, data }, this._timeoutCallback((err, response) => { if (err) { reject(err); } else { resolve(response); } }) ); }); } async _request(socket, method, data) { logger.debug('_request() [method:"%s", data:"%o"]', method, data); const { requestRetries = 3 } = config; for (let tries = 0; tries < requestRetries; tries++) { try { return await this._sendRequest(socket, method, data); } catch (error) { if ( error instanceof SocketTimeoutError && tries < requestRetries ) logger.warn('_request() | timeout, retrying [attempt:"%s"]', tries); else throw error; } } } _notification(socket, method, data = {}, broadcast = false, includeSender = false) { if (broadcast) { socket.broadcast.to(this._roomId).emit( 'notification', { method, data } ); if (includeSender) socket.emit('notification', { method, data }); } else { socket.emit('notification', { method, data }); } } /* - * Pipe producers of peers that are running under another routher to this router. + * Pipe producers of peers that are running under another router to this router. */ async _pipeProducersToRouter(routerId) { const router = this._mediasoupRouters.get(routerId); // All peers that have a different router const peersToPipe = Object.values(this._peers) .filter((peer) => peer.routerId !== routerId && peer.routerId !== null); for (const peer of peersToPipe) { const srcRouter = this._mediasoupRouters.get(peer.routerId); for (const producerId of peer.producers.keys()) { if (router._producers.has(producerId)) { continue; } await srcRouter.pipeToRouter({ producerId : producerId, router : router }); } } } async _getRouterId() { const routerId = Room.getLeastLoadedRouter( this._mediasoupWorkers, this._allPeers, this._mediasoupRouters); await this._pipeProducersToRouter(routerId); return routerId; } // Returns an array of router ids we need to pipe to: // The combined set of routers of all peers, exluding the router of the peer itself. _getRoutersToPipeTo(originRouterId) { return Object.values(this._peers) .map((peer) => peer.routerId) .filter((routerId, index, self) => routerId !== originRouterId && self.indexOf(routerId) === index ); } - } module.exports = Room; diff --git a/src/resources/js/meet/client.js b/src/resources/js/meet/client.js index 1d9add9a..cd38727c 100644 --- a/src/resources/js/meet/client.js +++ b/src/resources/js/meet/client.js @@ -1,650 +1,771 @@ 'use strict' import { Device, parseScalabilityMode } from 'mediasoup-client' import Config from './config.js' import { Media } from './media.js' import { Roles } from './constants.js' import { Socket } from './socket.js' function Client() { let eventHandlers = {} let camProducer let micProducer let screenProducer let consumers = {} let socket + let sendTransportInfo let sendTransport let recvTransport - let turnServers = [] + let iceServers = [] let nickname = '' let peers = {} + let joinProps = {} let videoSource let audioSource const VIDEO_CONSTRAINTS = { 'low': { width: { ideal: 320 } }, 'medium': { width: { ideal: 640 } }, 'high': { width: { ideal: 1280 } }, 'veryhigh': { width: { ideal: 1920 } }, 'ultra': { width: { ideal: 3840 } } } // Create a device (use browser auto-detection) const device = new Device() // A helper for basic browser media operations const media = new Media() this.media = media navigator.mediaDevices.addEventListener('devicechange', () => { trigger('deviceChange') }) /** * Start a session (join a room) */ this.joinSession = (token, props) => { + // Store the join properties for later + joinProps = props // Initialize the socket, 'roomReady' request handler will do the rest of the job socket = initSocket(token) - - nickname = props.nickname - videoSource = props.videoSource - audioSource = props.audioSource } /** * Close the session (disconnect) */ this.closeSession = async (reason) => { // If room owner, send the request to close the room - if (peers.self && peers.self.role & Roles.OWNER) { + if (reason === true && peers.self && peers.self.role & Roles.OWNER) { await socket.sendRequest('moderator:closeRoom') } trigger('closeSession', { reason: reason || 'disconnected' }) if (socket) { socket.close() } media.setupStop() // Close mediasoup Transports. if (sendTransport) { sendTransport.close() sendTransport = null } if (recvTransport) { recvTransport.close() recvTransport = null } // Remove peers' video elements Object.keys(peers).forEach(id => { let peer = peers[id] if (peer.videoElement) { $(peer.videoElement).remove() - peer.videoElement = null - peer.tracks = null } }) // Reset state eventHandlers = {} camProducer = null micProducer = null screenProducer = null consumers = {} peers = {} } + this.isJoined = () => { + return 'self' in peers + } + this.camMute = async () => { if (camProducer) { camProducer.pause() await socket.sendRequest('pauseProducer', { producerId: camProducer.id }) trigger('updatePeer', updatePeerState(peers.self)) } return this.camStatus() } this.camUnmute = async () => { if (camProducer) { camProducer.resume() await socket.sendRequest('resumeProducer', { producerId: camProducer.id }) trigger('updatePeer', updatePeerState(peers.self)) } return this.camStatus() } this.camStatus = () => { return camProducer && !camProducer.paused && !camProducer.closed } this.micMute = async () => { if (micProducer) { micProducer.pause() await socket.sendRequest('pauseProducer', { producerId: micProducer.id }) trigger('updatePeer', updatePeerState(peers.self)) } return this.micStatus() } this.micUnmute = async () => { if (micProducer) { micProducer.resume() await socket.sendRequest('resumeProducer', { producerId: micProducer.id }) trigger('updatePeer', updatePeerState(peers.self)) } return this.micStatus() } this.micStatus = () => { return micProducer && !micProducer.paused && !micProducer.closed } this.kickPeer = (peerId) => { socket.sendRequest('moderator:kickPeer', { peerId }) } this.chatMessage = (message) => { socket.sendRequest('chatMessage', { message }) } this.raiseHand = async (status) => { if (peers.self.raisedHand != status) { peers.self.raisedHand = status await socket.sendRequest('raisedHand', { raisedHand: status }) trigger('updatePeer', peers.self, ['raisedHand']) } return status } this.setNickname = (nickname) => { if (peers.self.nickname != nickname) { peers.self.nickname = nickname socket.sendRequest('changeNickname', { nickname }) trigger('updatePeer', peers.self, ['nickname']) } } + this.addRole = (peerId, role) => { + socket.sendRequest('moderator:addRole', { peerId, role }) + } + + this.removeRole = (peerId, role) => { + socket.sendRequest('moderator:removeRole', { peerId, role }) + } + /** * Register event handlers */ this.on = (eventName, callback) => { eventHandlers[eventName] = callback } /** * Execute an event handler */ const trigger = (...args) => { const eventName = args.shift() if (eventName in eventHandlers) { eventHandlers[eventName].apply(null, args) } } const initSocket = (token) => { // Connect to websocket socket = new Socket(token) socket.on('disconnect', reason => { // this.closeSession() }) socket.on('reconnectFailed', () => { // this.closeSession() }) socket.on('request', async (request, cb) => { switch (request.method) { case 'newConsumer': const { peerId, producerId, id, kind, rtpParameters, type, appData, producerPaused } = request.data const consumer = await recvTransport.consume({ id, producerId, kind, rtpParameters }) consumer.peerId = peerId consumer.on('transportclose', () => { // TODO: What actually else needs to be done here? delete consumers[consumer.id] }) consumers[consumer.id] = consumer // We are ready. Answer the request so the server will // resume this Consumer (which was paused for now). cb(null) let peer = peers[peerId] if (!peer) { return } - let tracks = (peer.tracks || []).filter(track => track.kind != kind) - - tracks.push(consumer.track) - - setPeerTracks(peer, tracks) + addPeerTrack(peer, consumer.track) trigger('updatePeer', peer) break default: console.error('Unknow request method: ' + request.method) } }) socket.on('notification', (notification) => { switch (notification.method) { case 'roomReady': - turnServers = notification.data.turnServers + iceServers = notification.data.turnServers joinRoom() return case 'newPeer': peers[notification.data.id] = notification.data trigger('addPeer', notification.data) return case 'peerClosed': const { peerId } = notification.data delete peers[peerId] trigger('removePeer', peerId) return case 'consumerClosed': { const { consumerId } = notification.data const consumer = consumers[consumerId] if (!consumer) { return } consumer.close() delete consumers[consumerId] let peer = peers[consumer.peerId] if (peer) { // TODO: Update peer state, remove track trigger('updatePeer', peer) } return } case 'consumerPaused': case 'consumerResumed': { const { consumerId } = notification.data const consumer = consumers[consumerId] if (!consumer) { return } consumer[notification.method == 'consumerPaused' ? 'pause' : 'resume']() let peer = peers[consumer.peerId] if (peer) { trigger('updatePeer', updatePeerState(peer)) } return } case 'changeNickname': { const { peerId, nickname } = notification.data const peer = peers[peerId] if (!peer) { return } peer.nickname = nickname trigger('updatePeer', peer, ['nickname']) return } + case 'changeRole': { + const { peerId, role } = notification.data + const peer = peers.self.id === peerId ? peers.self : peers[peerId] + + if (!peer) { + return + } + + let changes = ['role'] + + const rolePublisher = role & Roles.PUBLISHER + const roleModerator = role & Roles.MODERATOR + const isPublisher = peer.role & Roles.PUBLISHER + const isModerator = peer.role & Roles.MODERATOR + + if (isPublisher && !rolePublisher) { + // demoted to a subscriber + changes.push('publisherRole') + + if (peer.isSelf) { + // stop publishing any streams + this.setMic('', true) + this.setCamera('', true) + } else { + // remove the video element + peer.videoElement = null + // TODO: Do we need to remove/stop consumers? + } + } else if (!isPublisher && rolePublisher) { + // promoted to a publisher + changes.push('publisherRole') + + // create a video element with no tracks + setPeerTracks(peer, []) + } + + if ((!isModerator && roleModerator) || (isModerator && !roleModerator)) { + changes.push('moderatorRole') + } + + peer.role = role + + trigger('updatePeer', peer, changes) + return + } + case 'chatMessage': { trigger('chatMessage', notification.data) return } case 'moderator:closeRoom': { this.closeSession('session-closed') return } case 'moderator:kickPeer': { this.closeSession('session-closed') return } case 'raisedHand': { const { peerId, raisedHand } = notification.data const peer = peers[peerId] if (!peer) { return } peer.raisedHand = raisedHand trigger('updatePeer', peer, ['raisedHand']) return } default: console.error('Unknow notification method: ' + notification.method) - return } }) return socket } const joinRoom = async () => { const routerRtpCapabilities = await socket.getRtpCapabilities() routerRtpCapabilities.headerExtensions = routerRtpCapabilities.headerExtensions .filter(ext => ext.uri !== 'urn:3gpp:video-orientation') await device.load({ routerRtpCapabilities }) - const iceTransportPolicy = (device.handlerName.toLowerCase().includes('firefox') && turnServers) ? 'relay' : undefined; - - // Setup 'producer' transport - if (videoSource || audioSource) { - const transportInfo = await socket.sendRequest('createWebRtcTransport', { - forceTcp: false, - producing: true, - consuming: false - }) - - const { id, iceParameters, iceCandidates, dtlsParameters } = transportInfo - - sendTransport = device.createSendTransport({ - id, - iceParameters, - iceCandidates, - dtlsParameters, - iceServers: turnServers, - iceTransportPolicy: iceTransportPolicy, - proprietaryConstraints: { optional: [{ googDscp: true }] } - }) - - sendTransport.on('connect', ({ dtlsParameters }, callback, errback) => { - socket.sendRequest('connectWebRtcTransport', - { transportId: sendTransport.id, dtlsParameters }) - .then(callback) - .catch(errback) - }) - - sendTransport.on('produce', async ({ kind, rtpParameters, appData }, callback, errback) => { - try { - const { id } = await socket.sendRequest('produce', { - transportId: sendTransport.id, - kind, - rtpParameters, - appData - }) - callback({ id }) - } catch (error) { - errback(error) - } - }) - } - - // Setup 'consumer' transport - - const transportInfo = await socket.sendRequest('createWebRtcTransport', { - forceTcp: false, - producing: false, - consuming: true - }) - - const { id, iceParameters, iceCandidates, dtlsParameters } = transportInfo - - recvTransport = device.createRecvTransport({ - id, - iceParameters, - iceCandidates, - dtlsParameters, - iceServers: turnServers, - iceTransportPolicy: iceTransportPolicy - }) - - recvTransport.on('connect', ({ dtlsParameters }, callback, errback) => { - socket.sendRequest('connectWebRtcTransport', { transportId: recvTransport.id, dtlsParameters }) - .then(callback) - .catch(errback) - }) + // Setup the consuming transport (for handling streams of other participants) + await setRecvTransport() // Send the "join" request, get room data, participants, etc. const { peers: existing, role, id: peerId } = await socket.sendRequest('join', { - nickname: nickname, + nickname: joinProps.nickname, rtpCapabilities: device.rtpCapabilities }) trigger('joinSuccess') let peer = { id: peerId, role, - isSelf: true, - nickname, - audioActive: !!audioSource, - videoActive: !!videoSource - } - - // Start publishing webcam - if (videoSource) { - await setCamera(videoSource) - // Create the video element - peer.videoElement = media.createVideoElement([ camProducer.track ], { mirror: true }) - } - - // Start publishing microphone - if (audioSource) { - setMic(audioSource) - // Note: We're not adding this track to the video element + nickname: joinProps.nickname, + isSelf: true } - trigger('addPeer', peer) - // Add self to the list peers.self = peer - console.log(existing) + // Start publishing webcam and mic (and setup the producing transport) + await this.setCamera(joinProps.videoSource, true) + await this.setMic(joinProps.audioSource, true) + + updatePeerState(peer) + + trigger('addPeer', peer) // Trigger addPeer event for all peers already in the room, maintain peers list existing.forEach(peer => { let tracks = [] // We receive newConsumer requests before we add the peer to peers list, // therefore we look here for any consumers that belong to this peer and update // the peer. If we do not do this we have to wait about 20 seconds for repeated // newConsumer requests Object.keys(consumers).forEach(cid => { if (consumers[cid].peerId === peer.id) { tracks.push(consumers[cid].track) } }) if (tracks.length) { setPeerTracks(peer, tracks) } - trigger('addPeer', peer) peers[peer.id] = peer + + trigger('addPeer', peer) }) } - const setCamera = async (deviceId) => { + this.setCamera = async (deviceId, noUpdate) => { + // Actually selected device, do nothing + if (deviceId == videoSource) { + return + } + + // Remove current device, stop producer + if (camProducer && !camProducer.closed) { + camProducer.close() + await socket.sendRequest('closeProducer', { producerId: camProducer.id }) + setPeerTracks(peers.self, []) + } + + peers.self.videoSource = videoSource = deviceId + + if (!deviceId) { + if (!noUpdate) { + trigger('updatePeer', updatePeerState(peers.self), ['videoSource']) + } + return + } + if (!device.canProduce('video')) { throw new Error('cannot produce video') } const { aspectRatio, frameRate, resolution } = Config.videoOptions const track = await media.getTrack({ video: { deviceId: { ideal: deviceId }, ...VIDEO_CONSTRAINTS[resolution], frameRate } }) + await setSendTransport() + // TODO: Simulcast support? camProducer = await sendTransport.produce({ track, appData: { source : 'webcam' } }) /* camProducer.on('transportclose', () => { camProducer = null }) camProducer.on('trackended', () => { // disableWebcam() }) */ + // Create/Update the video element + addPeerTrack(peers.self, track) + if (!noUpdate) { + trigger('updatePeer', peers.self, ['videoSource']) + } } - const setMic = async (deviceId) => { + this.setMic = async (deviceId, noUpdate) => { + // Actually selected device, do nothing + if (deviceId == audioSource) { + return + } + + // Remove current device, stop producer + if (micProducer && !micProducer.closed) { + micProducer.close() + await socket.sendRequest('closeProducer', { producerId: micProducer.id }) + } + + peers.self.audioSource = audioSource = deviceId + + if (!deviceId) { + if (!noUpdate) { + trigger('updatePeer', updatePeerState(peers.self), ['audioSource']) + } + return + } + if (!device.canProduce('audio')) { throw new Error('cannot produce audio') } const { autoGainControl, echoCancellation, noiseSuppression, sampleRate, channelCount, volume, sampleSize, opusStereo, opusDtx, opusFec, opusPtime, opusMaxPlaybackRate } = Config.audioOptions const track = await media.getTrack({ audio: { sampleRate, channelCount, volume, autoGainControl, echoCancellation, noiseSuppression, sampleSize, deviceId: { ideal: deviceId } } }) + await setSendTransport() + micProducer = await sendTransport.produce({ track, codecOptions: { opusStereo, opusDtx, opusFec, opusPtime, opusMaxPlaybackRate }, appData: { source : 'mic' } }) /* micProducer.on('transportclose', () => { micProducer = null }) micProducer.on('trackended', () => { // disableMic() }) */ + // Note: We're not adding this track to the video element + if (!noUpdate) { + trigger('updatePeer', updatePeerState(peers.self), ['audioSource']) + } } const setPeerTracks = (peer, tracks) => { if (!peer.videoElement) { - peer.videoElement = media.createVideoElement(tracks, {}) + peer.videoElement = media.createVideoElement(tracks, { mirror: peer.isSelf }) } else { const stream = new MediaStream() tracks.forEach(track => stream.addTrack(track)) peer.videoElement.srcObject = stream } updatePeerState(peer) + } + + const addPeerTrack = (peer, track) => { + if (!peer.videoElement) { + setPeerTracks(peer, [ track ]) + return + } + + const stream = peer.videoElement.srcObject - peer.tracks = tracks + if (track.kind == 'video') { + media.removeTracksFromStream(stream, 'Video') + } else { + media.removeTracksFromStream(stream, 'Audio') + } + + stream.addTrack(track) + + updatePeerState(peer) } const updatePeerState = (peer) => { if (peer.isSelf) { peer.videoActive = this.camStatus() peer.audioActive = this.micStatus() } else { peer.videoActive = false peer.audioActive = false Object.keys(consumers).forEach(cid => { const consumer = consumers[cid] if (consumer.peerId == peer.id) { peer[consumer.kind + 'Active'] = !consumer.paused && !consumer.closed && !consumer.producerPaused } }) } return peer } + + const setSendTransport = async () => { + if (sendTransport && !sendTransport.closed) { + return + } + + if (!sendTransportInfo) { + sendTransportInfo = await socket.sendRequest('createWebRtcTransport', { + forceTcp: false, + producing: true, + consuming: false + }) + } + + const { id, iceParameters, iceCandidates, dtlsParameters } = sendTransportInfo + + const iceTransportPolicy = (device.handlerName.toLowerCase().includes('firefox') && iceServers) ? 'relay' : undefined + + sendTransport = device.createSendTransport({ + id, + iceParameters, + iceCandidates, + dtlsParameters, + iceServers, + iceTransportPolicy, + proprietaryConstraints: { optional: [{ googDscp: true }] } + }) + + sendTransport.on('connect', ({ dtlsParameters }, callback, errback) => { + socket.sendRequest('connectWebRtcTransport', { transportId: sendTransport.id, dtlsParameters }) + .then(callback) + .catch(errback) + }) + + sendTransport.on('produce', async ({ kind, rtpParameters, appData }, callback, errback) => { + try { + const { id } = await socket.sendRequest('produce', { + transportId: sendTransport.id, + kind, + rtpParameters, + appData + }) + callback({ id }) + } catch (error) { + errback(error) + } + }) + } + + const setRecvTransport = async () => { + const transportInfo = await socket.sendRequest('createWebRtcTransport', { + forceTcp: false, + producing: false, + consuming: true + }) + + const { id, iceParameters, iceCandidates, dtlsParameters } = transportInfo + + const iceTransportPolicy = (device.handlerName.toLowerCase().includes('firefox') && iceServers) ? 'relay' : undefined + + recvTransport = device.createRecvTransport({ + id, + iceParameters, + iceCandidates, + dtlsParameters, + iceServers, + iceTransportPolicy + }) + + recvTransport.on('connect', ({ dtlsParameters }, callback, errback) => { + socket.sendRequest('connectWebRtcTransport', { transportId: recvTransport.id, dtlsParameters }) + .then(callback) + .catch(errback) + }) + } } export { Client } diff --git a/src/resources/js/meet/config.js b/src/resources/js/meet/config.js index 8ecf4c5c..8a0c78a6 100644 --- a/src/resources/js/meet/config.js +++ b/src/resources/js/meet/config.js @@ -1,47 +1,47 @@ export default { // Default audio options audioOptions: { autoGainControl: false, echoCancellation: true, noiseSuppression: true, - voiceActivatedUnmute: false, // Automatically unmute speaking above noiseThereshold + voiceActivatedUnmute: false, // Automatically unmute speaking above noiseThreshold noiseThreshold: -60, // default -60 / This is only for voiceActivatedUnmute and audio-indicator sampleRate: 96000, // will not eat that much bandwith thanks to opus channelCount: 1, // usually mics are mono so this saves bandwidth volume: 1.0, sampleSize: 16, opusStereo: false, // usually mics are mono so this saves bandwidth opusDtx: true, // will save bandwidth opusFec: true, // forward error correction opusPtime: '20', // minimum packet time (3, 5, 10, 20, 40, 60, 120) opusMaxPlaybackRate: 96000 }, // Default video options videoOptions: { resolution: 'medium', aspectRatio: 1.777, // 16 : 9 - frameRate: 15, + frameRate: 15, // Note: OpenVidu default was 30 simulcast: true }, screenOptions: { resolution: 'veryhigh', frameRate: 5, simulcast: false }, // Simulcast encoding layers and levels simulcastEncodings: [ { scaleResolutionDownBy: 4 }, { scaleResolutionDownBy: 2 }, { scaleResolutionDownBy: 1 } ], // Socket.io request timeout requestTimeout: 20000, transportOptions: { tcp : true } } diff --git a/src/resources/js/meet/media.js b/src/resources/js/meet/media.js index 08f215cf..768e8c83 100644 --- a/src/resources/js/meet/media.js +++ b/src/resources/js/meet/media.js @@ -1,278 +1,281 @@ 'use strict' function Media() { let audioActive = false // True if the audio track is active let videoActive = false // True if the video track is active let audioSource = '' // Current audio device identifier let videoSource = '' // Current video device identifier let cameras = [] // List of user video devices let microphones = [] // List of user audio devices let setupVideoElement //