diff --git a/meet/server/lib/Room.js b/meet/server/lib/Room.js index c2825d34..7df9d0a1 100644 --- a/meet/server/lib/Room.js +++ b/meet/server/lib/Room.js @@ -1,1245 +1,1245 @@ 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 getLeastLoadedRouter(mediasoupWorkers, peers, mediasoupRouters) { const routerLoads = new Map(); const workerLoads = new Map(); const pipedRoutersIds = new Set(); for (const peer of peers.values()) { 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); } } } 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)); } } } 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._lastN = []; this._peers = {}; this._selfDestructTimeout = null; // Array of mediasoup Router instances. this._mediasoupRouters = mediasoupRouters; // mediasoup AudioLevelObserver. this._audioLevelObserver = audioLevelObserver; // Current active speaker. this._currentActiveSpeaker = null; this._handleAudioLevelObserver(); } 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(); } this._peers = null; // Close the mediasoup Routers. 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); } _handleAudioLevelObserver() { /* // Set audioLevelObserver events. this._audioLevelObserver.on('volumes', (volumes) => { const { producer, volume } = volumes[0]; // Notify all Peers. for (const peer of this.getPeers()) { this._notification( peer.socket, 'activeSpeaker', { peerId : producer.appData.peerId, volume : volume }); } }); this._audioLevelObserver.on('silence', () => { // Notify all Peers. for (const peer of this.getPeers()) { this._notification( peer.socket, 'activeSpeaker', { peerId: null } ); } }); */ } 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); // If we don't have this peer, add to end !this._lastN.includes(peer.id) && this._lastN.push(peer.id); 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, - nickame: peer.nickname + nickname: peer.nickname }, true); }); peer.on('gotRole', ({ newRole }) => { // Spread to others this._notification(peer.socket, 'gotRole', { 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); // Remove from lastN this._lastN = this._lastN.filter((id) => id !== peer.id); 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 '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 joinedPeers = this.getPeers(peer); const peerInfos = joinedPeers .map((joinedPeer) => (joinedPeer.peerInfo)); cb(null, { id: peer.id, role: peer.role, peers: peerInfos, }); for (const joinedPeer of joinedPeers) { // Create Consumers for existing Producers. for (const producer of joinedPeer.producers.values()) { this._createConsumer( { consumerPeer : peer, producerPeer : joinedPeer, producer }); } } // Notify the new Peer to all other Peers. for (const otherPeer of this.getPeers(peer)) { this._notification( otherPeer.socket, 'newPeer', peer.peerInfo ); } logger.debug( 'peer joined [peer: "%s", nickname: "%s", picture: "%s"]', peer.id, nickname, picture); 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', 'extravideo' ] .includes(appData.source) ) throw new Error('invalid producer source'); if ( appData.source === 'mic' && !this._hasPermission(peer, Roles.PUBLISHER) ) throw new Error('peer not authorized'); if ( appData.source === 'webcam' && !this._hasPermission(peer, Roles.PUBLISHER) ) throw new Error('peer not authorized'); if ( appData.source === 'screen' && !this._hasPermission(peer, 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); }); 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': { if (!this._hasPermission(peer, Roles.MODERATOR)) throw new Error('peer not authorized'); const { peerId, role } = request.data; const giveRolePeer = this._peers[peerId]; if (!giveRolePeer) throw new Error(`peer with id "${peerId}" not found`); // TODO: check if role is valid value // This will propagate the event automatically giveRolePeer.setRole(role); // Return no error cb(); break; } case 'raisedHand': { const { raisedHand } = request.data; peer.raisedHand = raisedHand; // Spread to others this._notification(peer.socket, 'raisedHand', { peerId : peer.id, raisedHand : raisedHand, raisedHandTimestamp : peer.raisedHandTimestamp }, true); // Return no error cb(); break; } case 'moderator:closeMeeting': { if (!this._hasPermission(peer, Roles.MODERATOR)) throw new Error('peer not authorized'); this._notification(peer.socket, 'moderator:kick', null, true); cb(); // Close the room this.close(); break; } /* case 'moderator:kickPeer': { if (!this._hasPermission(peer, 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:kick'); kickPeer.close(); cb(); break; } case 'moderator:lowerHand': { if (!this._hasPermission(peer, Roles.MODERATOR)) throw new Error('peer not authorized'); const { peerId } = request.data; const lowerPeer = this._peers[peerId]; if (!lowerPeer) throw new Error(`peer with id "${peerId}" not found`); this._notification(lowerPeer.socket, 'moderator:lowerHand'); 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; } // 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); } } _hasPermission(peer, role) { return !!(peer.role & role); } /** * 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 }); } } async _pipeProducersToRouter(routerId) { const router = this._mediasoupRouters.get(routerId); 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 _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 f6b5d2b5..7cbd1f6d 100644 --- a/src/resources/js/meet/client.js +++ b/src/resources/js/meet/client.js @@ -1,582 +1,588 @@ 'use strict' import { Device, parseScalabilityMode } from 'mediasoup-client' import Config from './config.js' import { Media } from './media.js' import { Socket } from './socket.js' function Client() { let eventHandlers = {} let camProducer let micProducer let screenProducer let consumers = {} let socket let sendTransport let recvTransport let turnServers = [] let nickname = '' let peers = {} 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 */ this.startSession = (token, 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 = () => { if (socket) { socket.close() } // Close mediasoup Transports. if (sendTransport) { sendTransport.close() sendTransport = null } if (recvTransport) { recvTransport.close() recvTransport = null } } 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.chatMessage = (message) => { socket.sendRequest('chatMessage', { message }) } + this.setNickname = (nickname) => { + peers.self.nickname = nickname + socket.sendRequest('changeNickname', { nickname }) + trigger('updatePeer', peers.self, ['nickname']) + } + /** * 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) peers[peerId] = peer trigger('updatePeer', peer) break default: console.error('Unknow request method: ' + request.method) } }) - socket.on('notification', async (notification) => { + socket.on('notification', (notification) => { switch (notification.method) { case 'roomReady': turnServers = 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) + trigger('updatePeer', peer, ['nickname']) return } case 'chatMessage': { trigger('chatMessage', notification.data) - return + 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) }) // Send the "join" request, get room data, participants, etc. const { peers: existing, role, id: peerId } = await socket.sendRequest('join', { nickname: 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 } trigger('addPeer', peer) // Add self to the list peers.self = peer console.log(existing) // 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 }) } const setCamera = async (deviceId) => { 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 } }) // TODO: Simulcast support? camProducer = await sendTransport.produce({ track, appData: { source : 'webcam' } }) camProducer.on('transportclose', () => { camProducer = null }) camProducer.on('trackended', () => { // disableWebcam() }) } const setMic = async (deviceId) => { 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 } } }) micProducer = await sendTransport.produce({ track, codecOptions: { opusStereo, opusDtx, opusFec, opusPtime, opusMaxPlaybackRate }, appData: { source : 'mic' } }) micProducer.on('transportclose', () => { micProducer = null }) micProducer.on('trackended', () => { // disableMic() }) } const setPeerTracks = (peer, tracks) => { if (!peer.videoElement) { peer.videoElement = media.createVideoElement(tracks, {}) } else { const stream = new MediaStream() tracks.forEach(track => stream.addTrack(track)) peer.videoElement.srcObject = stream } peer = updatePeerState(peer) peer.tracks = tracks peers[peer.id] = peer } const updatePeerState = (peer) => { if (peer.isSelf) { peer.videoActive = this.camStatus() peer.audioActive = this.micStatus() peers.self = peer } 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 } }) peers[peer.id] = peer } return peer } } export { Client } diff --git a/src/resources/js/meet/room.js b/src/resources/js/meet/room.js index 757df1ff..90eaaf80 100644 --- a/src/resources/js/meet/room.js +++ b/src/resources/js/meet/room.js @@ -1,1340 +1,1313 @@ 'use strict' import anchorme from 'anchorme' import { Client } from './client.js' import { Dropdown } from 'bootstrap' import { library } from '@fortawesome/fontawesome-svg-core' class Roles { static get SUBSCRIBER() { return 1 << 0; } static get PUBLISHER() { return 1 << 1; } static get MODERATOR() { return 1 << 2; } static get SCREEN() { return 1 << 3; } static get OWNER() { return 1 << 4; } } function Room(container) { let session // Session object where the user will connect let publisher // Publisher object which the user will publish let sessionData // Room session metadata let screenSession // Session object where the user will connect for screen sharing let screenPublisher // Publisher object which the user will publish the screen sharing /* let publisherDefaults = { publishAudio: true, // Whether to start publishing with your audio unmuted or not publishVideo: true, // Whether to start publishing with your video enabled or not resolution: '640x480', // The resolution of your video frameRate: 30, // The frame rate of your video mirror: true // Whether to mirror your local video or not } */ let connections = {} // Connected users in the session let peers = {} let chatCount = 0 let publishersContainer let subscribersContainer let scrollStop let $t const client = new Client() // Disconnect participant when browser's window close window.addEventListener('beforeunload', () => { leaveRoom() }) window.addEventListener('resize', resize) // Public methods this.isScreenSharingSupported = isScreenSharingSupported this.joinRoom = joinRoom this.leaveRoom = leaveRoom this.setupStart = setupStart this.setupStop = setupStop this.setupSetAudioDevice = setupSetAudioDevice this.setupSetVideoDevice = setupSetVideoDevice this.switchAudio = switchAudio this.switchChannel = switchChannel this.switchScreen = switchScreen this.switchVideo = switchVideo this.updateSession = updateSession /** * Join the room session * * @param data Session metadata and event handlers: * token - A token for the main connection, * shareToken - A token for screen-sharing connection, * nickname - Participant name, * role - connection (participant) role(s), * connections - Optional metadata for other users connections (current state), * channel - Selected interpreted language channel (two-letter language code) * languages - Supported languages (code-to-label map) * chatElement - DOM element for the chat widget, * counterElement - DOM element for the participants counter, * menuElement - DOM element of the room toolbar, * queueElement - DOM element for the Q&A queue (users with a raised hand) * onSuccess - Callback for session connection (join) success * onError - Callback for session connection (join) error * onDestroy - Callback for session disconnection event, * onDismiss - Callback for Dismiss action, * onJoinRequest - Callback for join request, * onConnectionChange - Callback for participant changes, e.g. role update, * onSessionDataUpdate - Callback for current user connection update, * onMediaSetup - Called when user clicks the Media setup button * translate - Translation function */ function joinRoom(data) { // Create a container for subscribers and publishers publishersContainer = $('
').appendTo(container).get(0) subscribersContainer = $('
').appendTo(container).get(0) resize(); $t = data.translate // Make sure all supported callbacks exist, so we don't have to check // their existence everywhere anymore let events = ['Success', 'Error', 'Destroy', 'Dismiss', 'JoinRequest', 'ConnectionChange', 'SessionDataUpdate', 'MediaSetup'] events.map(event => 'on' + event).forEach(event => { if (!data[event]) { data[event] = () => {} } }) sessionData = data // Participant added (including self) client.on('addPeer', (event) => { console.log('addPeer', event) - event.role = Roles.PUBLISHER // TODO - event.element = participantCreate(event) if (event.videoElement) { $(event.element).prepend(event.videoElement) } peers[event.id] = event }) // Participant removed client.on('removePeer', (peerId) => { console.log('removePeer', peerId) let peer = peers[peerId] if (peer) { // Remove elements related to the participant connectionHandDown(peerId) $(peer.element).remove() delete peers[peerId] } resize() }) // Participant properties changed e.g. audio/video muted/unmuted - client.on('updatePeer', (event) => { + client.on('updatePeer', (event, changed) => { console.log('updatePeer', event) let peer = peers[event.id] if (!peer) { return } - if (event.videoElement && event.videoElement.parentNode != peer.element) { - $(peer.element).prepend(event.videoElement) + event.element = peer.element + + if (event.videoElement && event.videoElement.parentNode != event.element) { + $(event.element).prepend(event.videoElement) } else if (!event.videoElement) { - $(peer.element).find('video').remove() + $(event.element).find('video').remove() } - // TODO: update peer properties + if (changed && changed.includes('nickname')) { + nicknameUpdate(event.nickname, event.id) + } - participantUpdate(peer.element, event) + participantUpdate(event.element, event) - peers[event.id] = peer + peers[event.id] = event }) client.on('joinSuccess', () => { data.onSuccess() }) /* // Handle session disconnection events client.on('sessionDisconnected', event => { data.onDestroy(event) client = null resize() }) */ const { audioSource, videoSource } = client.media.setupData() // Start the session client.startSession(data.token, { videoSource, audioSource, nickname: data.nickname }) // Prepare the chat setupChat() } /** * Leave the room (disconnect) */ function leaveRoom() { /* if (publisher) { // Release any media let mediaStream = publisher.stream.getMediaStream() if (mediaStream) { mediaStream.getTracks().forEach(track => track.stop()) } publisher = null } if (session) { session.disconnect(); session = null } if (screenSession) { screenSession.disconnect(); screenSession = null } */ } /** * Sets the audio and video devices for the session. * This will ask user for permission to access media devices. * * @param props Setup properties (videoElement, volumeElement, onSuccess, onError) */ function setupStart(props) { client.media.setupStart(props) } /** * Stop the setup "process", cleanup after it. */ function setupStop() { client.media.setupStop() } /** * Change the publisher audio device * * @param deviceId Device identifier string */ async function setupSetAudioDevice(deviceId) { return await client.media.setupSetAudio(deviceId) } /** * Change the publisher video device * * @param deviceId Device identifier string */ async function setupSetVideoDevice(deviceId) { return await client.media.setupSetVideo(deviceId) } /** * Setup the chat UI */ function setupChat() { // Handle arriving chat messages client.on('chatMessage', pushChatMessage) // The UI elements are created in the vue template // Here we add a logic for how they work const chat = $(sessionData.chatElement).find('.chat').get(0) const textarea = $(sessionData.chatElement).find('textarea') const button = $(sessionData.menuElement).find('.link-chat') textarea.on('keydown', e => { if (e.keyCode == 13 && !e.shiftKey) { if (textarea.val().length) { client.chatMessage(textarea.val()) textarea.val('') } return false } }) // Add an element for the count of unread messages on the chat button button.append('') .on('click', () => { button.find('.badge').text('') chatCount = 0 // When opening the chat scroll it to the bottom, or we shouldn't? scrollStop = false chat.scrollTop = chat.scrollHeight }) $(chat).on('scroll', event => { // Detect manual scrollbar moves, disable auto-scrolling until // the scrollbar is positioned on the element bottom again scrollStop = chat.scrollTop + chat.offsetHeight < chat.scrollHeight }) } - /** - * Signal events handler - */ +/* function signalEventHandler(signal) { let conn, data let connId = signal.from ? signal.from.connectionId : null switch (signal.type) { case 'signal:userChanged': // TODO: Use 'signal:connectionUpdate' for nickname updates? if (conn = connections[connId]) { data = JSON.parse(signal.data) conn.nickname = data.nickname participantUpdate(conn.element, conn) nicknameUpdate(data.nickname, connId) } break case 'signal:joinRequest': // accept requests from the server only if (!connId) { sessionData.onJoinRequest(JSON.parse(signal.data)) } break case 'signal:connectionUpdate': // accept requests from the server only if (!connId) { data = JSON.parse(signal.data) connectionUpdate(data) } break } } +*/ /** * Add a message to the chat * * @param data Object with a message, nickname, id (of the connection, empty for self) */ function pushChatMessage(data) { let message = $('').text(data.message).text() // make the message secure // Format the message, convert emails and urls to links message = anchorme({ input: message, options: { attributes: { target: "_blank" }, // any link above 20 characters will be truncated // to 20 characters and ellipses at the end truncate: 20, // characters will be taken out of the middle middleTruncation: true } // TODO: anchorme is extensible, we could support // github/phabricator's markup e.g. backticks for code samples }) message = message.replace(/\r?\n/, '
') // Display the message let isSelf = false // TODO let chat = $(sessionData.chatElement).find('.chat') let box = chat.find('.message').last() message = $('
').html(message) message.find('a').attr('rel', 'noreferrer') if (box.length && box.data('id') == data.peerId) { // A message from the same user as the last message, no new box needed message.appendTo(box) } else { box = $('
').data('id', data.peerId) .append($('
').text(data.nickname || '')) .append(message) .appendTo(chat) if (isSelf) { box.addClass('self') } } // Count unread messages if (!$(sessionData.chatElement).is('.open')) { if (!isSelf) { chatCount++ } } else { chatCount = 0 } $(sessionData.menuElement).find('.link-chat .badge').text(chatCount ? chatCount : '') // Scroll the chat element to the end if (!scrollStop) { chat.get(0).scrollTop = chat.get(0).scrollHeight } } - /** - * Send the user properties update signal to other participants - * - * @param connection Optional connection to which the signal will be sent - * If not specified the signal is sent to all participants - */ - function signalUserUpdate(connection) { - let data = { - nickname: sessionData.params.nickname - } - - session.signal({ - data: JSON.stringify(data), - type: 'userChanged', - to: connection ? [connection] : undefined - }) - - // The same nickname for screen sharing session - if (screenSession) { - screenSession.signal({ - data: JSON.stringify(data), - type: 'userChanged', - to: connection ? [connection] : undefined - }) - } - } - /** * Switch interpreted language channel * * @param channel Two-letter language code */ function switchChannel(channel) { sessionData.channel = channel // Mute/unmute all connections depending on the selected channel participantUpdateAll() } /** * Mute/Unmute audio for current session publisher */ async function switchAudio() { const isActive = client.micStatus() if (isActive) { return await client.micMute() } else { return await client.micUnmute() } } /** * Mute/Unmute video for current session publisher */ async function switchVideo() { const isActive = client.camStatus() if (isActive) { return await client.camMute() } else { return await client.camUnmute() } } /** * Switch on/off screen sharing */ function switchScreen(callback) { if (screenPublisher) { // Note: This is what the original openvidu-call app does. // It is probably better for performance reasons to close the connection, // than to use unpublish() and keep the connection open. screenSession.disconnect() screenSession = null screenPublisher = null if (callback) { // Note: Disconnecting invalidates the token, we have to inform the vue component // to update UI state (and be prepared to request a new token). callback(false) } return } screenConnect(callback) } /** * Detect if screen sharing is supported by the browser */ function isScreenSharingSupported() { - // TODO: Implement detection for screen sharing support in the browser - return true; + return false // TODO !!(navigator.mediaDevices && navigator.mediaDevices.getDisplayMedia) } /** * Update participant connection state */ function connectionUpdate(data) { let conn = connections[data.connectionId] let refresh = false let handUpdate = conn => { if ('hand' in data && data.hand != conn.hand) { if (data.hand) { connectionHandUp(conn) } else { connectionHandDown(data.connectionId) } } } // It's me if (session.connection.connectionId == data.connectionId) { const rolePublisher = data.role && data.role & Roles.PUBLISHER const roleModerator = data.role && data.role & Roles.MODERATOR const isPublisher = sessionData.role & Roles.PUBLISHER const isModerator = sessionData.role & Roles.MODERATOR // demoted to a subscriber if ('role' in data && isPublisher && !rolePublisher) { session.unpublish(publisher) // FIXME: There's a reference in OpenVidu to a video element that should not // exist anymore. It causes issues when we try to do publish/unpublish // sequence multiple times in a row. So, we're clearing the reference here. let videos = publisher.stream.streamManager.videos publisher.stream.streamManager.videos = videos.filter(video => video.video.parentNode != null) } handUpdate(sessionData) // merge the changed data into internal session metadata object sessionData = Object.assign({}, sessionData, data, { audioActive, videoActive }) // update the participant element sessionData.element = participantUpdate(sessionData.element, sessionData) // promoted/demoted to/from a moderator if ('role' in data) { // Update all participants, to enable/disable the popup menu refresh = (!isModerator && roleModerator) || (isModerator && !roleModerator) } // promoted to a publisher if ('role' in data && !isPublisher && rolePublisher) { publisher.createVideoElement(sessionData.element, 'PREPEND') session.publish(publisher).then(() => { sessionData.audioActive = publisher.stream.audioActive sessionData.videoActive = publisher.stream.videoActive sessionData.onSessionDataUpdate(sessionData) }) // Open the media setup dialog // Note: If user didn't give permission to media before joining the room // he will not be able to use them now. Changing permissions requires // a page refresh. // Note: In Firefox I'm always being asked again for media permissions. // It does not happen in Chrome. In Chrome the cam/mic will be just re-used. // I.e. streaming starts automatically. // It might make sense to not start streaming automatically in any cirmustances, // display the dialog and wait until user closes it, but this would be // a bigger refactoring. sessionData.onMediaSetup() } } else if (conn) { handUpdate(conn) // merge the changed data into internal session metadata object Object.keys(data).forEach(key => { conn[key] = data[key] }) conn.element = participantUpdate(conn.element, conn) } // Update channels list sessionData.channels = getChannels(connections) // The channel user was using has been removed (or rather the participant stopped being an interpreter) if (sessionData.channel && !sessionData.channels.includes(sessionData.channel)) { sessionData.channel = null refresh = true } if (refresh) { participantUpdateAll() } // Inform the vue component, so it can update some UI controls sessionData.onSessionDataUpdate(sessionData) } /** * Handler for Hand-Up "signal" */ function connectionHandUp(connection) { connection.isSelf = session.connection.connectionId == connection.connectionId let element = $(nicknameWidget(connection)) participantUpdate(element, connection) element.attr('id', 'qa' + connection.connectionId) .appendTo($(sessionData.queueElement).show()) setTimeout(() => element.addClass('widdle'), 50) } /** * Handler for Hand-Down "signal" */ function connectionHandDown(connectionId) { let list = $(sessionData.queueElement) list.find('#qa' + connectionId).remove(); if (!list.find('.meet-nickname').length) { list.hide(); } } /** * Update participant nickname in the UI * - * @param nickname Nickname - * @param connectionId Connection identifier of the user + * @param nickname Nickname + * @param peerId Connection identifier of the user */ - function nicknameUpdate(nickname, connectionId) { - if (connectionId) { + function nicknameUpdate(nickname, peerId) { + if (peerId) { $(sessionData.chatElement).find('.chat').find('.message').each(function() { let elem = $(this) - if (elem.data('id') == connectionId) { + if (elem.data('id') == peerId) { elem.find('.nickname').text(nickname || '') } }) - $(sessionData.queueElement).find('#qa' + connectionId + ' .content').text(nickname || '') + $(sessionData.queueElement).find('#qa' + peerId + ' .content').text(nickname || '') } } /** * Create a participant element in the matrix. Depending on the connection role * parameter it will be a video element wrapper inside the matrix or a simple * tag-like element on the subscribers list. * * @param params Connection metadata/params * @param content Optional content to prepend to the element * * @return The element */ function participantCreate(params, content) { let element if ((!params.language && params.role & Roles.PUBLISHER) || params.role & Roles.SCREEN) { // publishers and shared screens element = publisherCreate(params, content) } else { // subscribers and language interpreters element = subscriberCreate(params, content) } setTimeout(resize, 50); return element } /** * Create a