client volumes and dominant client

server/plugins/socket.ts

@@ -22,8 +22,8 @@ export default fp<Partial<ServerOptions>>(
       await fastify.io.close()
     })
 
-    fastify.ready(() => {
+    fastify.ready(async () => {
-      registerWebrtcSocket(fastify.io, fastify.mediasoupRouter)
+      await registerWebrtcSocket(fastify.io, fastify.mediasoupRouter)
     })
   },
   { name: 'socket-io', dependencies: ['mediasoup-worker', 'mediasoup-router'] },

server/socket/webrtc.ts

@@ -1,6 +1,7 @@
 import type { types } from 'mediasoup'
 import type { Server as SocketServer } from 'socket.io'
 import type {
+  ChadClient,
   Namespace,
   SomeSocket,
 } from '../types/webrtc.ts'
@@ -8,9 +9,39 @@ import { consola } from 'consola'
 import prisma from '../prisma/client.ts'
 import { socketToClient } from '../utils/socket-to-client.ts'
 
-export default function (io: SocketServer, router: types.Router) {
+export default async function (io: SocketServer, router: types.Router) {
   const namespace: Namespace = io.of('/webrtc')
 
+  const audioLevelObserver = await router.createAudioLevelObserver({
+    maxEntries: 10,
+    threshold: -80,
+    interval: 800,
+  })
+
+  const activeSpeakerObserver = await router.createActiveSpeakerObserver()
+
+  audioLevelObserver.on('volumes', async (volumes: types.AudioLevelObserverVolume[]) => {
+    namespace.emit('speakingPeers', volumes.map(({ producer, volume }) => {
+      const { clientId } = producer.appData as { clientId: ChadClient['socketId'] }
+
+      return {
+        clientId,
+        volume,
+      }
+    }))
+  })
+
+  audioLevelObserver.on('silence', () => {
+    namespace.emit('speakingPeers', [])
+    namespace.emit('activeSpeaker', undefined)
+  })
+
+  activeSpeakerObserver.on('dominantspeaker', ({ producer }) => {
+    const { clientId } = producer.appData as { clientId: ChadClient['socketId'] }
+
+    namespace.emit('activeSpeaker', clientId)
+  })
+
   namespace.on('connection', async (socket) => {
     consola.info('[WebRtc]', 'Client connected', socket.id)
 
@@ -182,8 +213,8 @@ export default function (io: SocketServer, router: types.Router) {
           )
         }
 
-        // TODO: Add into the AudioLevelObserver and ActiveSpeakerObserver.
+        await audioLevelObserver.addProducer({ producerId: producer.id })
-        // https://github.com/versatica/mediasoup-demo/blob/v3/server/lib/Room.js#L1276
+        await activeSpeakerObserver.addProducer({ producerId: producer.id })
       }
       catch (error) {
         if (error instanceof Error) {

server/types/webrtc.ts

@@ -118,6 +118,8 @@ export interface ServerToClientEvents {
   consumerResumed: (arg: { consumerId: string }) => void
   consumerScore: (arg: { consumerId: string, score: types.ConsumerScore }) => void
   clientChanged: (clientId: ChadClient['socketId'], client: ChadClient) => void
+  speakingPeers: (arg: { clientId: ChadClient['socketId'], volume: types.AudioLevelObserverVolume['volume'] }[]) => void
+  activeSpeaker: (clientId?: ChadClient['socketId']) => void
 }
 
 export interface InterServerEvent {}