feat: add PeerService.produce

src/apis/master/room.ts

@@ -16,4 +16,11 @@ export const room: Array<RouteConfig> = [
       return new PeerService(getDataSource()).createProducer(data);
     },
   },
+  {
+    method: 'POST',
+    url: '/peers/:peerId/produce',
+    handler: (data) => {
+      return new PeerService(getDataSource()).produce(data);
+    },
+  },
 ];

src/apis/slave/router.ts

@@ -1,6 +1,7 @@
 import {
   mediasoupRouterManager,
-  mediasoupWebRTCTransportManager,
+  mediasoupProducerWebRTCTransportManager,
+  mediasoupProducerManager,
 } from '../../services/index.js';
 import { RouteConfig } from '../../utils/index.js';
 
@@ -16,7 +17,14 @@ export const router: Array<RouteConfig> = [
     method: 'POST',
     url: '/routers/:routerId/producer_transports',
     handler: (data) => {
-      return mediasoupWebRTCTransportManager.createProducer(data.routerId);
+      return mediasoupProducerWebRTCTransportManager.create(data);
+    },
+  },
+  {
+    method: 'POST',
+    url: '/transports/:transportId/producer',
+    handler: (data) => {
+      return mediasoupProducerManager.create(data);
     },
   },
 ];

src/entities/media.peer.ts

@@ -31,6 +31,12 @@ export class MediaPeer extends BaseEntity {
   @Column('text')
   routerId!: string;
 
+  @Column('text')
+  producerId?: string;
+
+  @Column('text')
+  consumerId?: string;
+
   @Column('text')
   type!: string; // consumer | producer
 

src/services/mediasoup/index.ts

@@ -1,3 +1,4 @@
 export * from './worker.js';
 export * from './router.js';
 export * from './webrtc.transport.js';
+export * from './producer.js';

src/services/mediasoup/producer.ts

@@ -0,0 +1,27 @@
+import { types } from 'mediasoup';
+import { ServiceError } from '../base.js';
+import { mediasoupProducerWebRTCTransportManager } from './webrtc.transport.js';
+
+class MediasoupProducerManager {
+  static producers: Array<types.Producer> = [];
+
+  async create(data: {
+    transportId: string;
+    kind: types.MediaKind;
+    rtpParameters: types.RtpParameters;
+  }) {
+    const transport = mediasoupProducerWebRTCTransportManager.get(
+      data.transportId
+    );
+    if (transport) {
+      const { kind, rtpParameters } = data;
+      const producer = await transport.produce({ kind, rtpParameters });
+      MediasoupProducerManager.producers.push(producer);
+
+      return { id: producer.id };
+    }
+    throw new ServiceError(404, 'Transport not found');
+  }
+}
+
+export const mediasoupProducerManager = new MediasoupProducerManager();

src/services/mediasoup/webrtc.transport.ts

@@ -5,8 +5,18 @@ import { mediasoupRouterManager } from './router.js';
 class MediasoupWebRTCTransportManager {
   static transports: Array<types.Transport> = [];
 
-  async createProducer(routerId: string) {
-    const router = mediasoupRouterManager.get(routerId);
+  get(transportId: string) {
+    return (
+      this.constructor as typeof MediasoupWebRTCTransportManager
+    ).transports.find((item) => item.id === transportId);
+  }
+}
+
+class MediasoupProducerWebRTCTransportManager extends MediasoupWebRTCTransportManager {
+  static transports: Array<types.Transport> = [];
+
+  async create(data: { routerId: string }) {
+    const router = mediasoupRouterManager.get(data.routerId);
     if (router) {
       const maxIncomingBitrate = Number(
         process.env.MEDIASOUP_WEBRTC_TRANSPORT_MAX_INCOMING_BITRATE
@@ -31,7 +41,10 @@ class MediasoupWebRTCTransportManager {
           await transport.setMaxIncomingBitrate(maxIncomingBitrate);
         } catch (error) {}
       }
-      MediasoupWebRTCTransportManager.transports.push(transport);
+      (
+        this.constructor as typeof MediasoupProducerWebRTCTransportManager
+      ).transports.push(transport);
+
       return {
         id: transport.id,
         iceParameters: transport.iceParameters,
@@ -43,5 +56,5 @@ class MediasoupWebRTCTransportManager {
   }
 }
 
-export const mediasoupWebRTCTransportManager =
-  new MediasoupWebRTCTransportManager();
+export const mediasoupProducerWebRTCTransportManager =
+  new MediasoupProducerWebRTCTransportManager();

src/services/peer.ts

@@ -6,7 +6,9 @@ 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 });
+    const room = await this.createService(RoomService).get({
+      roomId: data.roomId,
+    });
     if (room) {
       const result = await fetchApi({
         host: room.slave.externalHost,
@@ -27,4 +29,32 @@ export class PeerService extends BaseService {
     }
     throw new ServiceError(404, 'Room not found');
   }
+
+  async produce(data: { peerId: string; kind: any; rtpParameters: any }) {
+    const peer = await this.get({ peerId: data.peerId });
+    if (peer && peer.type === constants.PRODUCER) {
+      const result = await fetchApi({
+        host: peer.slave.externalHost,
+        port: peer.slave.apiPort,
+        path: '/transports/:transportId/producer',
+        method: 'POST',
+        data: {
+          transportId: peer.id,
+          kind: data.kind,
+          rtpParameters: data.rtpParameters,
+        },
+      });
+
+      peer.producerId = result.id;
+      await this.dataSource.getRepository(MediaPeer).save(peer);
+    }
+    throw new ServiceError(404, 'Peer not found');
+  }
+
+  async get(data: { peerId: string }) {
+    return this.dataSource.getRepository(MediaPeer).findOne({
+      relations: { slave: true },
+      where: { id: data.peerId },
+    });
+  }
 }

src/services/room.ts

@@ -24,10 +24,10 @@ export class RoomService extends BaseService {
     throw new ServiceError(404, 'Slave not found');
   }
 
-  async get(data: { id: string }) {
+  async get(data: { roomId: string }) {
     return this.dataSource.getRepository(MediaRoom).findOne({
       relations: { slave: true },
-      where: { id: data.id },
+      where: { id: data.roomId },
     });
   }
 }