rebroadcast: rtp mode that may handle audio better.

common/src/ffmpeg-rebroadcast.ts

@@ -2,11 +2,12 @@ import { createServer, Server } from 'net';
 import child_process, { StdioOptions } from 'child_process';
 import { ChildProcess } from 'child_process';
 import { FFMpegInput, MediaStreamOptions } from '@scrypted/sdk/types';
-import { listenZero } from './listen-cluster';
+import { bindZero, listenZero } from './listen-cluster';
 import { EventEmitter } from 'events';
 import sdk from "@scrypted/sdk";
 import { ffmpegLogInitialOutput, safePrintFFmpegArguments } from './media-helpers';
 import { StreamChunk, StreamParser } from './stream-parser';
+import dgram from 'dgram';
 
 const { mediaManager } = sdk;
 
@@ -18,6 +19,7 @@ export interface MP4Atom {
 }
 
 export interface ParserSession<T extends string> {
+    sdp: Buffer[];
     mediaStreamOptions: MediaStreamOptions;
     inputAudioCodec?: string;
     inputVideoCodec?: string;
@@ -142,15 +144,35 @@ export async function startParserSession<T extends string>(ffmpegInput: FFMpegIn
     const stdio: StdioOptions = ['pipe', 'pipe', 'pipe']
     let pipeCount = 3;
     for (const container of Object.keys(options.parsers)) {
-        const parser = options.parsers[container];
-        args.push(
-            ...parser.outputArguments,
-            `pipe:${pipeCount}`,
-        );
-        stdio.push('pipe');
-        pipeCount++;
+        const parser: StreamParser = options.parsers[container];
+        if (parser.parseDatagram) {
+            const socket = dgram.createSocket('udp4')
+            const udp = await bindZero(socket);
+            args.push(
+                ...parser.outputArguments,
+                udp.url,
+            );
+
+            (async () => {
+                for await (const chunk of parser.parseDatagram(socket, parseInt(inputVideoResolution?.[2]), parseInt(inputVideoResolution?.[3]))) {
+                    ffmpegStartedResolve?.(undefined);
+                    events.emit(container, chunk);
+                    resetActivityTimer();
+                }
+            })();
+        }
+        else {
+            args.push(
+                ...parser.outputArguments,
+                `pipe:${pipeCount++}`,
+            );
+            stdio.push('pipe');
+        }
     }
 
+    args.push('-sdp_file', `pipe:${pipeCount++}`);
+    stdio.push('pipe');
+
     // start ffmpeg process with child process pipes
     args.unshift('-hide_banner');
     safePrintFFmpegArguments(console, args);
@@ -160,13 +182,20 @@ export async function startParserSession<T extends string>(ffmpegInput: FFMpegIn
     ffmpegLogInitialOutput(console, cp);
     cp.on('exit', kill);
 
+    const sdp: Buffer[] = [];
+    (cp.stdio[pipeCount - 1]).on('data', buffer => sdp.push(buffer));
+
     // now parse the created pipes
-    Object.keys(options.parsers).forEach(async (container, index) => {
-        const pipe = cp.stdio[3 + index];
-        const parser = options.parsers[container];
+    let pipeIndex = 0;
+    Object.keys(options.parsers).forEach(async (container) => {
+        const parser: StreamParser = options.parsers[container];
+        if (!parser.parse)
+            return;
+        const pipe = cp.stdio[3 + pipeIndex];
+        pipeIndex++;
 
         try {
-            for await (const chunk of parser.parse(pipe, parseInt(inputVideoResolution?.[2]), parseInt(inputVideoResolution?.[3]))) {
+            for await (const chunk of parser.parse(pipe as any, parseInt(inputVideoResolution?.[2]), parseInt(inputVideoResolution?.[3]))) {
                 ffmpegStartedResolve?.(undefined);
                 events.emit(container, chunk);
                 resetActivityTimer();
@@ -189,6 +218,7 @@ export async function startParserSession<T extends string>(ffmpegInput: FFMpegIn
     clearTimeout(ffmpegIncomingConnectionTimeout);
 
     return {
+        sdp,
         inputAudioCodec,
         inputVideoCodec,
         inputVideoResolution,
@@ -225,7 +255,7 @@ export interface RebroadcastSessionCleanup {
 }
 
 export interface RebroadcasterOptions {
-    connect?: (writeData: (data: StreamChunk) => number, cleanup: () => void) => RebroadcastSessionCleanup|undefined;
+    connect?: (writeData: (data: StreamChunk) => number, cleanup: () => void) => RebroadcastSessionCleanup | undefined;
     console?: Console;
 }
 

common/src/listen-cluster.ts

@@ -11,7 +11,11 @@ export async function listenZero(server: net.Server) {
 export async function bindZero(server: dgram.Socket) {
     server.bind(0);
     await once(server, 'listening');
-    return (server.address() as net.AddressInfo).port;
+    const { port } = server.address() as net.AddressInfo;
+    return {
+        port,
+        url: `udp://127.0.0.1:${port}`,
+    }
 }
 
 export async function listenZeroSingleClient() {

common/src/stream-parser.ts

@@ -1,12 +1,14 @@
 import { once } from "events";
 import { Socket } from "net";
+import { Socket as DatagramSocket } from "dgram";
 import { Readable } from "stream";
 import { readLength } from "./read-length";
 
 export interface StreamParser {
     container: string;
     outputArguments: string[];
-    parse: (socket: Socket, width: number, height: number) => AsyncGenerator<StreamChunk>;
+    parse?: (socket: Socket, width: number, height: number) => AsyncGenerator<StreamChunk>;
+    parseDatagram?: (socket: DatagramSocket, width: number, height: number) => AsyncGenerator<StreamChunk>;
     findSyncFrame(streamChunks: StreamChunk[]): StreamChunk[];
 }
 
@@ -89,6 +91,30 @@ export function createPCMParser(): StreamParser {
     }
 }
 
+export function createDgramParser() {
+    async function* parse(socket: DatagramSocket) {
+        while (true) {
+            const [buffer] = await once(socket, 'message');
+            yield {
+                chunks: [buffer],
+            }
+        }
+    };
+    return parse;
+}
+
+export function createRtpParser(...codec: string[]): StreamParser {
+    return {
+        container: 'sdp',
+        outputArguments: [
+            ...codec,
+            '-f', 'rtp',
+        ],
+        parseDatagram: createDgramParser(),
+        findSyncFrame,
+    }
+}
+
 export function createMpegTsParser(options?: StreamParserOptions): StreamParser {
     return {
         container: 'mpegts',