import express from 'express' ; import { Consumer , MediaKind , Producer , RtpCapabilities , WebRtcTransport } from 'mediasoup/lib/types' ; import { io } from '../../../../shared/infra/http/app' ; import { meetingRepo } from '../../repos' ; import { config } from './config' ; import { createRoom , rooms , RoomState , runMediasoupWorkers } from './rooms' ; import { logger } from "../../../../logger" ; export const signalingRouter = express . Router ( ) ; async function main ( ) { logger . info ( 'starting mediasoup' ) ; try { await runMediasoupWorkers ( ) ; } catch ( error ) { logger . error ( { error : error } , 'unable to start mediasoup workers ...' ) ; logger . info ( 'unable to start mediasoup workers ... [exiting]' ) ; } } main ( ) ; signalingRouter . use ( async ( req , res , next ) => { if ( req . body . roomId ) { let room = rooms . get ( req . body . roomId ) ; if ( ! room ) { const meetingFound = await meetingRepo . getMeetingByMeetingId ( req . body . roomId ) ; if ( meetingFound == null ) { return res . status ( 404 ) . send ( 'Meeting not found' ) ; } room = await createRoom ( req . body . roomId ) ; } req . body . room = room ; next ( ) ; return ; } else { return res . status ( 400 ) . json ( { message : 'roomId is required' , } ) ; } } ) ; signalingRouter . post ( '/sync' , async ( req , res ) => { const { peerId , room } = req . body as { peerId : string ; room : RoomState ; } ; try { if ( ! room . peers [ peerId ] ) { throw new Error ( 'not connected' ) ; } room . peers [ peerId ] . lastSeenTs = Date . now ( ) ; res . send ( { peers : room . peers , activeSpeaker : room . activeSpeaker , } ) ; } catch ( e ) { logger . error ( e . message ) ; res . send ( { error : e . message } ) ; } } ) ; signalingRouter . post ( '/join-as-new-peer' , async ( req , res ) => { try { const { peerId , room , name } = req . body as { peerId : string ; name : string ; room : RoomState ; } ; const now = Date . now ( ) ; logger . info ( { peerId : peerId } , 'join-as-new-peer' , peerId ) ; const sendTransport = await createWebRtcTransport ( { peerId , direction : 'send' , room , } ) ; room . transports [ sendTransport . id ] = sendTransport ; const sendTransportOptions = { id : sendTransport . id , iceParameters : sendTransport . iceParameters , iceCandidates : sendTransport . iceCandidates , dtlsParameters : sendTransport . dtlsParameters , } ; const recvTransport = await createWebRtcTransport ( { peerId , direction : 'recv' , room , } ) ; room . transports [ recvTransport . id ] = recvTransport ; const recvTransportOptions = { id : recvTransport . id , iceParameters : recvTransport . iceParameters , iceCandidates : recvTransport . iceCandidates , dtlsParameters : recvTransport . dtlsParameters , } ; room . peers [ peerId ] = { id : peerId , name , joinTs : now , lastSeenTs : now , media : { } , consumerLayers : { } , stats : { } , } ; io . to ( room . id ) . emit ( 'newPeer' , { peerId , name , } ) ; res . send ( { routerRtpCapabilities : room . router ! . rtpCapabilities , sendTransportOptions , recvTransportOptions , } ) ; } catch ( e ) { logger . error ( { error : e } , 'error in /join-as-new-peer' ) ; res . send ( { error : e } ) ; } } ) ; signalingRouter . post ( '/init-consumers' , async ( req , res ) => { try { const { peerId , room } = req . body as { peerId : string ; room : RoomState ; } ; logger . info ( { peerId : peerId } , 'init-consumers' ) ; room . producers . forEach ( ( producer ) => { if ( producer . appData . peerId !== peerId ) { createConsumer ( { producer , mediaPeerId : producer . appData . peerId , mediaTag : producer . appData . mediaTag , peerId , room , rtpCapabilities : room . router ! . rtpCapabilities , } ) ; } } ) ; res . send ( { } ) ; } catch ( e ) { logger . error ( { error : e } , 'error in /join-as-new-peer' ) ; res . send ( { error : e } ) ; } } ) ; signalingRouter . post ( '/leave' , async ( req , res ) => { try { const { peerId , room } = req . body as { peerId : string ; room : RoomState ; } ; logger . info ( { peerId : peerId } , 'leave' ) ; closePeer ( peerId , room ) ; res . send ( { left : true } ) ; } catch ( e ) { logger . error ( { error : e } , 'error in /leave' ) ; res . send ( { error : e } ) ; } } ) ; export function closePeer ( peerId , room ) { logger . info ( { peerId : peerId } , 'closing peer %s' , peerId ) ; for ( const [ id , transport ] of Object . entries ( room . transports ) ) { if ( transport . appData . peerId === peerId ) { closeTransport ( transport , room ) ; } } delete room . peers [ peerId ] ; io . to ( room . id ) . emit ( 'peerClosed' , { peerId , } ) ; if ( Object . values ( room . peers ) . length === 0 ) { logger . info ( 'removing room as all peers have left' ) ; room . router . close ( ) ; room . audioLevelObserver . close ( ) ; rooms . delete ( room . id ) ; } } async function closeTransport ( transport , room ) { try { logger . info ( { transportId : transport . id , transportAppData : transport . appData } , 'closing transport' , transport . id , transport . appData ) ; transport . close ( ) ; delete room . transports [ transport . id ] ; } catch ( e ) { logger . error ( { error : e } , "error" ) ; } } async function closeProducer ( producer , room ) { logger . info ( { producerId : producer . id , producerAppData : producer . appData } , 'closing producer' , producer . id , producer . appData ) ; try { producer . close ( ) ; room . producers = room . producers . filter ( ( p ) => p . id !== producer . id ) ; if ( room . peers [ producer . appData . peerId ] ) { delete room . peers [ producer . appData . peerId ] . media [ producer . appData . mediaTag ] ; } } catch ( e ) { logger . error ( { error : e } , "error" ) ; } } async function closeConsumer ( consumer , room ) { logger . info ( { consumerId : consumer . id , consumerAppData : consumer . appData } , 'closing consumer' , consumer . id , consumer . appData ) ; consumer . close ( ) ; room . consumers = room . consumers . filter ( ( c ) => c . id !== consumer . id ) ; if ( room . peers [ consumer . appData . peerId ] ) { delete room . peers [ consumer . appData . peerId ] . consumerLayers [ consumer . id ] ; } } signalingRouter . post ( '/create-transport' , async ( req , res ) => { try { const { peerId , direction , room } = req . body as { peerId : string ; direction : string ; room : RoomState ; } ; logger . info ( { peerId : peerId , direction : direction } , 'create-transport' , peerId , direction ) ; const transport = await createWebRtcTransport ( { peerId , direction , room } ) ; room . transports [ transport . id ] = transport ; const { id , iceParameters , iceCandidates , dtlsParameters } = transport ; res . send ( { transportOptions : { id , iceParameters , iceCandidates , dtlsParameters } , } ) ; } catch ( e ) { logger . error ( { error : e } , 'error in /create-transport' , e ) ; res . send ( { error : e } ) ; } } ) ; async function createWebRtcTransport ( { peerId , direction , room , } ) { const { listenIps , initialAvailableOutgoingBitrate } = config . mediasoup . webRtcTransport ; const transport = await room . router . createWebRtcTransport ( { listenIps : listenIps , enableUdp : true , enableTcp : true , preferUdp : true , initialAvailableOutgoingBitrate : initialAvailableOutgoingBitrate , appData : { peerId , clientDirection : direction } , } ) ; return transport ; } signalingRouter . post ( '/connect-transport' , async ( req , res ) => { try { const { peerId , transportId , dtlsParameters , room } = req . body , transport = room . transports [ transportId ] ; if ( ! transport ) { logger . error ( `connect-transport: server-side transport ${ transportId } not found` ) ; res . send ( { error : `server-side transport ${ transportId } not found` } ) ; return ; } logger . info ( { peerId : peerId , transportAppData : transport . appData } , 'connect-transport' , peerId , transport . appData ) ; await transport . connect ( { dtlsParameters } ) ; res . send ( { connected : true } ) ; } catch ( e ) { logger . error ( { error : e } , 'error in /connect-transport' , e ) ; res . send ( { error : e } ) ; } } ) ; signalingRouter . post ( '/close-transport' , async ( req , res ) => { try { const { peerId , transportId , room } = req . body , transport = room . transports [ transportId ] ; if ( ! transport ) { logger . error ( `close-transport: server-side transport ${ transportId } not found` ) ; res . send ( { error : `server-side transport ${ transportId } not found` } ) ; return ; } logger . info ( { peerId : peerId , transportAppData : transport . appData } , 'close-transport' , peerId , transport . appData ) ; await closeTransport ( transport , room ) ; res . send ( { closed : true } ) ; } catch ( e ) { logger . error ( { error : e } , 'error in /close-transport' , e ) ; res . send ( { error : e . message } ) ; } } ) ; signalingRouter . post ( '/close-producer' , async ( req , res ) => { try { const { peerId , producerId , room } = req . body as { peerId : string ; producerId : string ; room : RoomState ; } ; const producer = room . producers . find ( ( p ) => p . id === producerId ) ; if ( ! producer ) { logger . error ( `close-producer: server-side producer ${ producerId } not found` ) ; res . send ( { error : `server-side producer ${ producerId } not found` } ) ; return ; } logger . info ( { peerId : peerId , producerAppData : producer . appData } , 'close-producer' , peerId , producer . appData ) ; await closeProducer ( producer , room ) ; res . send ( { closed : true } ) ; } catch ( e ) { logger . error ( { error : e } , "error" ) ; res . send ( { error : e . message } ) ; } } ) ; signalingRouter . post ( '/send-track' , async ( req , res ) => { try { const { peerId , transportId , kind , rtpParameters , paused = false , appData , room , } = req . body as { peerId : string ; transportId : string ; kind : MediaKind ; rtpParameters : any ; paused : boolean ; appData : any ; room : RoomState ; } ; const transport = room . transports [ transportId ] ; if ( ! transport ) { logger . error ( `send-track: server-side transport ${ transportId } not found` ) ; res . send ( { error : `server-side transport ${ transportId } not found` } ) ; return ; } const producer = await transport . produce ( { kind , rtpParameters , paused , appData : { ... appData , peerId , transportId } , } ) ; producer . on ( 'transportclose' , ( ) => { logger . log ( { producerId : producer . id } , "producer's transport closed" , producer . id ) ; const r = rooms . get ( room . id ) ; if ( r ) { closeProducer ( producer , r ) ; } } ) ; if ( producer . kind === 'audio' ) { room . audioLevelObserver ! . addProducer ( { producerId : producer . id } ) ; } room . producers . push ( producer ) ; room . peers [ peerId ] . media [ appData . mediaTag ] = { paused , encodings : rtpParameters . encodings , } ; res . send ( { id : producer . id } ) ; for ( const peer of Object . values ( room . peers ) ) { if ( peer . id === peerId ) continue ; createConsumer ( { producer , mediaPeerId : peerId , peerId : peer . id , mediaTag : producer . appData . mediaTag , room , rtpCapabilities : room . router . rtpCapabilities , } ) ; } } catch ( e ) { logger . error ( { error : e } , "error" ) ; } } ) ; interface ConsumerProps { producer : Producer ; rtpCapabilities : RtpCapabilities ; mediaPeerId : string ; mediaTag : string ; peerId : string ; room : RoomState ; } const createConsumer = async ( { producer , rtpCapabilities , room , mediaTag , mediaPeerId , peerId } ) => { if ( ! room . router . canConsume ( { producerId : producer . id , rtpCapabilities , } ) ) { const msg = `client cannot consume ${ mediaPeerId }:${ mediaTag }` ; logger . error ( { peerId : peerId } , `recv-track: ${ peerId } ${ msg }` ) ; return ; } const transport = Object . values ( room . transports ) . find ( ( t ) => t . appData . peerId === peerId && t . appData . clientDirection === 'recv' ) ; if ( ! transport ) { const msg = `server-side recv transport for ${ peerId } not found` ; logger . error ( 'recv-track: ' + msg ) ; return ; } const consumer = await transport . consume ( { producerId : producer . id , rtpCapabilities , paused : true , appData : { peerId , mediaPeerId , mediaTag } , } ) ; consumer . on ( 'transportclose' , ( ) => { logger . info ( { consumerId : consumer . id } , `consumer's transport closed` , consumer . id ) ; closeConsumer ( consumer , room ) ; } ) ; consumer . on ( 'producerclose' , ( ) => { logger . info ( { consumerId : consumer . id } , `consumer's producer closed` , consumer . id ) ; closeConsumer ( consumer , room ) ; io . to ( room . id ) . emit ( 'consumerClosed' , { consumerId : consumer . id } ) ; } ) ; consumer . on ( 'producerpause' , ( ) => { io . to ( room . id ) . emit ( 'consumerPaused' , { consumerId : consumer . id } ) ; } ) ; consumer . on ( 'producerresume' , ( ) => { io . to ( room . id ) . emit ( 'consumerResumed' , { consumerId : consumer . id } ) ; } ) ; consumer . on ( 'layerschange' , ( layers ) => { logger . info ( { mediaTag : mediaTag , layers : layers } , `consumer layerschange ${ mediaPeerId }->${ peerId }` , mediaTag , layers ) ; if ( room . peers [ peerId ] && room . peers [ peerId ] . consumerLayers [ consumer . id ] ) { room . peers [ peerId ] . consumerLayers [ consumer . id ] . currentLayer = layers && layers . spatialLayer ; } } ) ; room . consumers . push ( consumer ) ; room . peers [ peerId ] . consumerLayers [ consumer . id ] = { currentLayer : undefined , clientSelectedLayer : undefined , } ; io . to ( peerId ) . emit ( 'newConsumer' , { peerId : mediaPeerId , producerId : producer . id , id : consumer . id , kind : consumer . kind , rtpParameters : consumer . rtpParameters , type : consumer . type , appData : producer . appData , producerPaused : consumer . producerPaused , } ) ; return { producerId : producer . id , id : consumer . id , kind : consumer . kind , rtpParameters : consumer . rtpParameters , type : consumer . type , producerPaused : consumer . producerPaused , } ; } ; signalingRouter . post ( '/pause-consumer' , async ( req , res ) => { try { const { consumerId , room } = req . body as { consumerId : string ; room : RoomState ; } ; const consumer = room . consumers . find ( ( c ) => c . id === consumerId ) ; if ( ! consumer ) { logger . error ( { consumerId : consumerId } , `pause-consumer: server-side consumer ${ consumerId } not found` ) ; res . send ( { error : `server-side producer ${ consumerId } not found` } ) ; return ; } logger . info ( { consumerAppData : consumer . appData } , 'pause-consumer' , consumer . appData ) ; await consumer . pause ( ) ; res . send ( { paused : true } ) ; } catch ( e ) { logger . error ( { error : e } , 'error in /pause-consumer' , e ) ; res . send ( { error : e } ) ; } } ) ; signalingRouter . post ( '/resume-consumer' , async ( req , res ) => { try { const { consumerId , room } = req . body as { consumerId : string ; room : RoomState ; } ; const consumer = room . consumers . find ( ( c ) => c . id === consumerId ) ; if ( ! consumer ) { logger . error ( { consumerId : consumerId } , `pause-consumer: server-side consumer ${ consumerId } not found` ) ; res . send ( { error : `server-side consumer ${ consumerId } not found` } ) ; return ; } logger . info ( { consumerAppData : consumer . appData } , 'resume-consumer' , consumer . appData ) ; logger . info ( { consumerPaused : consumer . paused } , 'resume-consumer-paused' , consumer . paused ) ; await consumer . resume ( ) ; res . send ( { resumed : true } ) ; } catch ( e ) { logger . error ( { error : e } , 'error in /resume-consumer' , e ) ; res . send ( { error : e } ) ; } } ) ; signalingRouter . post ( '/close-consumer' , async ( req , res ) => { try { const { consumerId , room } = req . body as { consumerId : string ; room : RoomState ; } ; const consumer = room . consumers . find ( ( c ) => c . id === consumerId ) ; if ( ! consumer ) { logger . error ( { consumerId : consumerId } , `close-consumer: server-side consumer ${ consumerId } not found` ) ; res . send ( { error : `server-side consumer ${ consumerId } not found` } ) ; return ; } await closeConsumer ( consumer , room ) ; res . send ( { closed : true } ) ; } catch ( e ) { logger . error ( { error : e } , 'error in /close-consumer' , e ) ; res . send ( { error : e } ) ; } } ) ; signalingRouter . post ( '/consumer-set-layers' , async ( req , res ) => { try { const { consumerId , room , spatialLayer } = req . body as { consumerId : string ; room : RoomState ; spatialLayer : any ; } ; const consumer = room . consumers . find ( ( c ) => c . id === consumerId ) ; if ( ! consumer ) { logger . error ( { consumerId : consumerId } , `consumer-set-layers: server-side consumer ${ consumerId } not found` ) ; res . send ( { error : `server-side consumer ${ consumerId } not found` } ) ; return ; } logger . info ( { spatialLayer : spatialLayer , consumerAppData : consumer . appData } , 'consumer-set-layers' , spatialLayer , consumer . appData ) ; await consumer . setPreferredLayers ( { spatialLayer } ) ; res . send ( { layersSet : true } ) ; } catch ( e ) { logger . error ( { error : e } , 'error in /consumer-set-layers' , e ) ; res . send ( { error : e } ) ; } } ) ; signalingRouter . post ( '/pause-producer' , async ( req , res ) => { try { const { peerId , producerId , room } = req . body as { peerId : string ; producerId : string ; room : RoomState ; } ; const producer = room . producers . find ( ( p ) => p . id === producerId ) ; if ( ! producer ) { logger . error ( { producerId : producerId } , `pause-producer: server-side producer ${ producerId } not found` ) ; res . send ( { error : `server-side producer ${ producerId } not found` } ) ; return ; } logger . info ( { producerAppData : producer . appData } , 'pause-producer' , producer . appData ) ; await producer . pause ( ) ; room . peers [ peerId ] . media [ producer . appData . mediaTag ] . paused = true ; res . send ( { paused : true } ) ; } catch ( e ) { logger . error ( { error : e } , 'error in /pause-producer' , e ) ; res . send ( { error : e } ) ; } } ) ; signalingRouter . post ( '/resume-producer' , async ( req , res ) => { try { const { peerId , producerId , room } = req . body as { peerId : string ; producerId : string ; room : RoomState ; } ; const producer = room . producers . find ( ( p ) => p . id === producerId ) ; if ( ! producer ) { logger . error ( { producerId : producerId } , `resume-producer: server-side producer ${ producerId } not found` ) ; res . send ( { error : `server-side producer ${ producerId } not found` } ) ; return ; } logger . info ( { producerAppData : producer . appData } , 'resume-producer' , producer . appData ) ; await producer . resume ( ) ; room . peers [ peerId ] . media [ producer . appData . mediaTag ] . paused = false ; res . send ( { resumed : true } ) ; } catch ( e ) { logger . error ( { error : e } , 'error in /resume-producer' , e ) ; res . send ( { error : e } ) ; } } ) ;