add registerNodeGtkGst util

shinyoshiaki · Nov 8, 2024 · c9ac91f · c9ac91f
1 parent 4202691
commit c9ac91f
Show file tree

Hide file tree

Showing 3 changed files with 66 additions and 35 deletions.
diff --git a/packages/core/src/media/factory.ts b/packages/core/src/media/factory.ts
@@ -1,21 +1,22 @@
 import { Event, Logger } from '@skyway-sdk/common';
 
-import { errors } from '../errors';
+import {
+  MediaDevices,
+  MediaStreamTrackFactory,
+  Navigator,
+} from '../imports/mediasoup';
 import { createError, createWarnPayload } from '../util';
 import { LocalMediaStreamOptions } from './stream';
 import { LocalAudioStream } from './stream/local/audio';
-import {
-  LocalCustomVideoStream,
-  ProcessedStream,
-} from './stream/local/customVideo';
+import { ProcessedStream } from './stream/local/customVideo';
 import { DataStreamOptions, LocalDataStream } from './stream/local/data';
 import { LocalVideoStream } from './stream/local/video';
-import { MediaDevices, Navigator } from '../imports/mediasoup';
 
 const log = new Logger('packages/core/src/media/factory.ts');
 
 export class StreamFactory {
   readonly navigator: Navigator;
+  private gst: any;
 
   constructor(props: ConstructorParameters<typeof Navigator>[0] = {}) {
     this.navigator = new Navigator(props);
@@ -28,6 +29,49 @@ export class StreamFactory {
     });
   }
 
+  registerNodeGtkGst(gst: any) {
+    this.gst = gst;
+  }
+
+  async registerGstAudio({
+    wave,
+    rtpProcessor,
+  }: { wave?: string; rtpProcessor?: (b: Buffer) => Buffer } = {}) {
+    wave ??= 'ticks';
+
+    const [track, port, disposer] = await MediaStreamTrackFactory.rtpSource({
+      kind: 'audio',
+      cb: rtpProcessor,
+    });
+    const launch = this.gst.parseLaunch(
+      `audiotestsrc wave=${wave} ! audioconvert ! audioresample ! queue ! opusenc ! rtpopuspay ! udpsink host=127.0.0.1 port=${port}`
+    );
+    launch.setState(this.gst.State.PLAYING);
+    SkyWayStreamFactory.registerMediaDevices({ audio: track });
+
+    return () => {
+      disposer();
+      launch.setState(this.gst.State.NULL);
+    };
+  }
+
+  /**h264 only */
+  async registerGstVideo(_: any = {}) {
+    const [track, port, disposer] = await MediaStreamTrackFactory.rtpSource({
+      kind: 'video',
+    });
+    const launch = this.gst.parseLaunch(
+      `videotestsrc ! video/x-raw,width=640,height=480,format=I420 ! x264enc key-int-max=60 ! rtph264pay ! udpsink host=127.0.0.1 port=${port}`
+    );
+    launch.setState(this.gst.State.PLAYING);
+    SkyWayStreamFactory.registerMediaDevices({ video: track });
+
+    return () => {
+      disposer();
+      launch.setState(this.gst.State.NULL);
+    };
+  }
+
   /**
    * @description [japanese] CameraのVideoStreamを作成する
    */

diff --git a/submodules/mediasoup b/submodules/mediasoup
diff --git a/tests/large/loopback.test.ts b/tests/large/loopback.test.ts
@@ -1,21 +1,21 @@
+import Gst from '@girs/node-gst-1.0';
 import { describe, expect, it } from 'vitest';
 import {
+  dePacketizeRtpPackets,
+  deserializeAudioLevelIndication,
+  serializeAudioLevelIndication,
+} from 'werift';
+
+import {
+  MediaStreamTrackFactory,
   RemoteVideoStream,
+  RoomPublication,
   RtpPacket,
   SkyWayContext,
   SkyWayRoom,
-  MediaStreamTrackFactory,
   SkyWayStreamFactory,
-  RoomPublication,
 } from '../../packages/room/src';
-
-import Gst from '@girs/node-gst-1.0';
 import { testTokenString } from './fixture';
-import {
-  dePacketizeRtpPackets,
-  deserializeAudioLevelIndication,
-  serializeAudioLevelIndication,
-} from 'werift';
 
 const gst = require('node-gtk').require('Gst', '1.0') as typeof Gst;
 gst.init([]);
@@ -26,15 +26,15 @@ describe('loopback', () => {
       const context = await SkyWayContext.Create(testTokenString, {
         codecCapabilities: [{ mimeType: 'audio/opus' }],
       });
+      SkyWayStreamFactory.registerNodeGtkGst(gst);
       const room = await SkyWayRoom.Create(context, {
         type: 'sfu',
       });
       console.log('roomId', room.id);
       const sender = await room.join();
 
-      const [track, port, disposer] = await MediaStreamTrackFactory.rtpSource({
-        kind: 'audio',
-        cb: (buf) => {
+      const disposer = await SkyWayStreamFactory.registerGstAudio({
+        rtpProcessor: (buf) => {
           const rtp = RtpPacket.deSerialize(buf);
           rtp.header.extension = true;
           rtp.header.extensions.push({
@@ -44,11 +44,6 @@ describe('loopback', () => {
           return rtp.serialize();
         },
       });
-      const launch = gst.parseLaunch(
-        `audiotestsrc wave=ticks ! audioconvert ! audioresample ! queue ! opusenc ! rtpopuspay ! udpsink host=127.0.0.1 port=${port}`
-      );
-      launch.setState(gst.State.PLAYING);
-      SkyWayStreamFactory.registerMediaDevices({ audio: track });
 
       const publication = await sender.publish(
         await SkyWayStreamFactory.createMicrophoneAudioStream()
@@ -69,7 +64,6 @@ describe('loopback', () => {
           console.log('audioLevel', p);
           await room.close();
           context.dispose();
-          launch.setState(gst.State.NULL);
           disposer();
           done();
         }
@@ -157,21 +151,15 @@ describe('loopback', () => {
         ],
         rtcConfig: { turnPolicy: 'disable' },
       });
+      SkyWayStreamFactory.registerNodeGtkGst(gst);
+
       const room = await SkyWayRoom.Create(context, {
         type: 'sfu',
       });
       console.log('roomId', room.id);
       const sender = await room.join();
 
-      const [track, port, disposer] = await MediaStreamTrackFactory.rtpSource({
-        kind: 'video',
-      });
-      const launch = gst.parseLaunch(
-        `videotestsrc ! video/x-raw,width=640,height=480,format=I420 ! x264enc key-int-max=60 ! rtph264pay ! udpsink host=127.0.0.1 port=${port}`
-      );
-      launch.setState(gst.State.PLAYING);
-      SkyWayStreamFactory.registerMediaDevices({ video: track });
-
+      const disposer = await SkyWayStreamFactory.registerGstVideo();
       const publication = await sender.publish(
         await SkyWayStreamFactory.createCameraVideoStream()
       );
@@ -191,7 +179,6 @@ describe('loopback', () => {
 
           await room.close();
           context.dispose();
-          launch.setState(gst.State.NULL);
           disposer();
           done();
         }