feat: add producer_peers api

.env.dev

@@ -67,6 +67,6 @@ MEDIASOUP_WEBRTC_TRANSPORT_INITIAL_AVAILABLE_OUTGOING_BITRATE = 1000000
 MEDIASOUP_WEBRTC_TRANSPORT_LISTEN_IPS = '[
    {
       "ip": "192.168.6.143",
-      "announcedIp": null,
+      "announcedIp": null
    }
 ]'

src/apis/master/room.ts

@@ -1,4 +1,4 @@
-import { RoomService } from '../../services/index.js';
+import { PeerService, RoomService } from '../../services/index.js';
 import { RouteConfig, getDataSource } from '../../utils/index.js';
 
 export const room: Array<RouteConfig> = [
@@ -9,4 +9,11 @@ export const room: Array<RouteConfig> = [
       return new RoomService(getDataSource()).create({});
     },
   },
+  {
+    method: 'POST',
+    url: '/rooms/:roomId/producer_peers',
+    handler: (data) => {
+      return new PeerService(getDataSource()).createProducer(data);
+    },
+  },
 ];

src/apis/slave/router.ts

@@ -14,9 +14,9 @@ export const router: Array<RouteConfig> = [
   },
   {
     method: 'POST',
-    url: '/routers/:routerId/transports',
+    url: '/routers/:routerId/producer_transports',
     handler: (data) => {
-      return mediasoupWebRTCTransportManager.create(data.routerId);
+      return mediasoupWebRTCTransportManager.createProducer(data.routerId);
     },
   },
 ];

src/entities/media.peer.ts

@@ -31,6 +31,9 @@ export class MediaPeer extends BaseEntity {
   @Column('text')
   routerId!: string;
 
+  @Column('text')
+  type!: string; // consumer | producer
+
   @Column({ type: 'jsonb', nullable: true })
   metadata?: any;
 

src/services/index.ts

@@ -1,3 +1,4 @@
 export * from './slave.js';
 export * from './room.js';
+export * from './peer.js';
 export * from './mediasoup/index.js';

src/services/mediasoup/webrtc.transport.ts

@@ -5,7 +5,7 @@ import { mediasoupRouterManager } from './router.js';
 class MediasoupWebRTCTransportManager {
   static transports: Array<types.Transport> = [];
 
-  async create(routerId: string) {
+  async createProducer(routerId: string) {
     const router = mediasoupRouterManager.get(routerId);
     if (router) {
       const maxIncomingBitrate = Number(

src/services/peer.ts

@@ -0,0 +1,30 @@
+import { constants } from '../constants.js';
+import { MediaPeer } from '../entities/index.js';
+import { fetchApi } from '../utils/api.js';
+import { BaseService, ServiceError } from './base.js';
+import { RoomService } from './room.js';
+
+export class PeerService extends BaseService {
+  async createProducer(data: { roomId: string; metadata?: any }) {
+    const room = await this.createService(RoomService).get({ id: data.roomId });
+    if (room) {
+      const result = await fetchApi({
+        host: room.slave.externalHost,
+        port: room.slave.apiPort,
+        path: '/routers/:routerId/producer_transports',
+        method: 'POST',
+        data: { routerId: room.routerId },
+      });
+      const mediaPeer = new MediaPeer();
+      mediaPeer.id = result.id;
+      mediaPeer.routerId = room.routerId;
+      mediaPeer.slaveId = room.slave.id;
+      mediaPeer.type = constants.PRODUCER;
+      mediaPeer.roomId = room.id;
+
+      await this.dataSource.getRepository(MediaPeer).save(mediaPeer);
+      return result;
+    }
+    throw new ServiceError(404, 'Room not found');
+  }
+}

src/services/room.ts

@@ -14,19 +14,20 @@ export class RoomService extends BaseService {
         path: '/routers',
         method: 'POST',
       });
-      const mediasoupRoom = new MediaRoom();
-      mediasoupRoom.routerId = result.id;
-      mediasoupRoom.slaveId = slave.id;
-      Object.assign(mediasoupRoom, data);
-      await this.dataSource.getRepository(MediaRoom).save(mediasoupRoom);
-      return mediasoupRoom.id;
+      const mediaRoom = new MediaRoom();
+      mediaRoom.routerId = result.id;
+      mediaRoom.slaveId = slave.id;
+      Object.assign(mediaRoom, data);
+      await this.dataSource.getRepository(MediaRoom).save(mediaRoom);
+      return { id: mediaRoom.id };
     }
     throw new ServiceError(404, 'Slave not found');
   }
 
   async get(data: { id: string }) {
-    return this.dataSource
-      .getRepository(MediaRoom)
-      .findOne({ where: { id: data.id } });
+    return this.dataSource.getRepository(MediaRoom).findOne({
+      relations: { slave: true },
+      where: { id: data.id },
+    });
   }
 }

src/utils/api.ts

@@ -1,16 +1,46 @@
+function generatePath(urlPattern: string, params: Record<string, any>) {
+  const retParams = { ...params };
+  const parts = urlPattern.split('/');
+  const result = [] as string[];
+  for (let i = 0; i < parts.length; i += 1) {
+    if (parts[i].startsWith(':')) {
+      const key = parts[i].slice(1);
+      result.push(encodeURIComponent(params[key]));
+      delete retParams[key];
+    } else {
+      result.push(parts[i]);
+    }
+  }
+  return {
+    path: result.join('/'),
+    params: retParams,
+  };
+}
+
 export function fetchApi({
   host,
   port,
   path,
   method,
+  data,
 }: {
   host: string;
   port?: string | number;
   path: string;
   method: 'POST' | 'GET' | 'PUT' | 'DELETE';
+  data?: Record<string, any>;
 }) {
-  return fetch('http://' + host + (port ? ':' + port : '') + path, {
+  const parsed = generatePath(path, data || {});
+  let body;
+  if (method === 'GET') {
+    parsed.path += '?' + new URLSearchParams(parsed.params);
+  } else {
+    body = JSON.stringify(parsed.params);
+  }
+  return fetch('http://' + host + (port ? ':' + port : '') + parsed.path, {
+    headers: { 'Content-Type': 'application/json' },
     method,
+    body,
   }).then(async (resp) => {
     if (resp.status > 400) {
       const message = await resp.text();