feat: support RTMP and SRT stream output (#5)

Dockerfile

@@ -1,6 +1,10 @@
 ARG NODE_IMAGE=node:18-alpine
 
 FROM ${NODE_IMAGE}
+USER root
+RUN apk update && \
+    apk upgrade && \
+    apk add ffmpeg
 ENV NODE_ENV=production
 EXPOSE 8000
 RUN mkdir /app

readme.md

@@ -32,6 +32,29 @@ docker run -d \
   eyevinntechnology/fast-engine-bridge
 ```
 
+To push an MPEG-TS stream using SRT protocol
+
+```
+docker run -d \
+  -e SOURCE=https://eyevinn.ce.prod.osaas.io/channels/demo/master.m3u8 \
+  -e DEST_TYPE=stream \
+  -e DEST_URL="srt://0.0.0.0:9998?mode=listener" \
+  -p 9998:9998/udp \
+  eyevinntechnology/fast-engine-bridge
+```
+
+To view the MPEG-TS stream you can open VLC on address `srt://127.0.0.1:9998`
+
+To bridge a channel to an RTMP destination
+
+```
+docker run -d \
+  -e SOURCE=https://eyevinn.ce.prod.osaas.io/channels/demo/master.m3u8 \
+  -e DEST_TYPE=stream \
+  -e DEST_URL="rtmp://172.232.130.157:10503/live/abc123" \
+  eyevinntechnology/fast-engine-bridge
+```
+
 <!--
 ## Development
 

src/server.ts

@@ -2,6 +2,7 @@ import { HLSPullPush, VoidOutput } from '@eyevinn/hls-pull-push';
 import api from './api';
 import { MyLogger } from './logger';
 import { getOutputDest } from './output';
+import { StreamOutput } from './stream_output';
 
 const server = api({ title: 'channel-engine-bridge' });
 
@@ -25,22 +26,35 @@ server.listen({ port: PORT, host: '0.0.0.0' }, (err, address) => {
       'https://demo.vc.eyevinn.technology/channels/demo/master.m3u8'
   );
 
-  const pullPushService = new HLSPullPush(new MyLogger(process.env.NODE_ENV));
-  const outputDest = getOutputDest({
-    pullPushService,
-    destinationUrl: process.env.DEST_URL
-      ? new URL(process.env.DEST_URL)
-      : undefined,
-    destinationType
-  });
-  if (outputDest) {
-    const sessionId = pullPushService.startFetcher({
+  if (destinationType == 'stream') {
+    const streamService = new StreamOutput(new MyLogger(process.env.NODE_ENV));
+    if (!process.env.DEST_URL) {
+      throw new Error(`Destination type 'stream' requires DEST_URL to be set`);
+    }
+    streamService.startFetcher({
       name: 'default',
       url: source.toString(),
-      destPlugin: outputDest,
-      destPluginName: destinationType
+      destinationUrl: new URL(process.env.DEST_URL),
+      h264encoder: process.env.H264ENCODER
     });
-    outputDest.attachSessionId(sessionId);
+  } else {
+    const pullPushService = new HLSPullPush(new MyLogger(process.env.NODE_ENV));
+    const outputDest = getOutputDest({
+      pullPushService,
+      destinationUrl: process.env.DEST_URL
+        ? new URL(process.env.DEST_URL)
+        : undefined,
+      destinationType
+    });
+    if (outputDest) {
+      const sessionId = pullPushService.startFetcher({
+        name: 'default',
+        url: source.toString(),
+        destPlugin: outputDest,
+        destPluginName: destinationType
+      });
+      outputDest.attachSessionId(sessionId);
+    }
   }
 });
 

src/stream_output.ts

@@ -0,0 +1,52 @@
+import { ILogger } from '@eyevinn/hls-pull-push';
+import { MyLogger } from './logger';
+import { spawn } from 'child_process';
+
+export class StreamOutput {
+  private logger: ILogger;
+
+  constructor(logger?: ILogger) {
+    this.logger = logger || new MyLogger(process.env.NODE_ENV);
+  }
+
+  async startFetcher({
+    name,
+    url,
+    destinationUrl,
+    h264encoder
+  }: {
+    name: string;
+    url: string;
+    destinationUrl: URL;
+    h264encoder?: string;
+  }) {
+    this.logger.info(
+      `Starting ffmpeg fetcher to fetch ${url} and push to ${destinationUrl}`
+    );
+    const container = destinationUrl.protocol === 'rtmp:' ? 'flv' : 'mpegts';
+    const process = spawn('ffmpeg', [
+      '-fflags',
+      '+genpts',
+      '-re',
+      '-i',
+      url,
+      '-c:v',
+      h264encoder || 'libx264',
+      '-b:v',
+      '8000k',
+      '-c:a',
+      'aac',
+      '-f',
+      container,
+      destinationUrl.toString()
+    ]);
+    this.logger.verbose(process.spawnargs.join(' '));
+    await new Promise((resolve, reject) => {
+      process.on('close', resolve);
+      process.stderr.on('data', (data) => {
+        this.logger.info(data.toString());
+      });
+      process.on('error', reject);
+    });
+  }
+}