Compare commits

..

1 Commits

15 changed files with 560 additions and 79 deletions

39
bun.lock Normal file
View File

@@ -0,0 +1,39 @@
{
"lockfileVersion": 1,
"configVersion": 0,
"workspaces": {
"": {
"name": "@techniker-me/rtmp-push",
"devDependencies": {
"@types/bun": "latest",
"@types/node": "^25.0.3",
"prettier": "^3.7.4",
"typescript": "^5.9.3",
},
"peerDependencies": {
"typescript": "^5.9.3",
},
},
},
"packages": {
"@types/bun": ["@types/bun@1.2.20", "https://registry-node.techniker.me/@types/bun/-/bun-1.2.20.tgz", { "dependencies": { "bun-types": "1.2.20" } }, "sha512-dX3RGzQ8+KgmMw7CsW4xT5ITBSCrSbfHc36SNT31EOUg/LA9JWq0VDdEXDRSe1InVWpd2yLUM1FUF/kEOyTzYA=="],
"@types/node": ["@types/node@25.0.3", "", { "dependencies": { "undici-types": "~7.16.0" } }, "sha512-W609buLVRVmeW693xKfzHeIV6nJGGz98uCPfeXI1ELMLXVeKYZ9m15fAMSaUPBHYLGFsVRcMmSCksQOrZV9BYA=="],
"@types/react": ["@types/react@19.1.10", "", { "dependencies": { "csstype": "^3.0.2" } }, ""],
"bun-types": ["bun-types@1.2.20", "", { "dependencies": { "@types/node": "*" }, "peerDependencies": { "@types/react": "^19" } }, ""],
"csstype": ["csstype@3.1.3", "", {}, ""],
"prettier": ["prettier@3.7.4", "", { "bin": { "prettier": "bin/prettier.cjs" } }, "sha512-v6UNi1+3hSlVvv8fSaoUbggEM5VErKmmpGA7Pl3HF8V6uKY7rvClBOJlH6yNwQtfTueNkGVpOv/mtWL9L4bgRA=="],
"typescript": ["typescript@5.9.3", "", { "bin": { "tsc": "bin/tsc", "tsserver": "bin/tsserver" } }, "sha512-jl1vZzPDinLr9eUt3J/t7V6FgNEw9QjvBPdysz9KfQDD41fQrC2Y4vKQdiaUpFT4bXlb1RHhLpp8wtm6M5TgSw=="],
"undici-types": ["undici-types@7.16.0", "", {}, "sha512-Zz+aZWSj8LE6zoxD+xrjh4VfkIG8Ya6LvYkZqtUQGJPZjYl53ypCaUwWqo7eI0x66KBGeRo+mlBEkMSeSZ38Nw=="],
"bun-types/@types/node": ["@types/node@20.19.11", "https://registry-node.techniker.me/@types/node/-/node-20.19.11.tgz", { "dependencies": { "undici-types": "~6.21.0" } }, "sha512-uug3FEEGv0r+jrecvUUpbY8lLisvIjg6AAic6a2bSP5OEOLeJsDSnvhCDov7ipFFMXS3orMpzlmi0ZcuGkBbow=="],
"bun-types/@types/node/undici-types": ["undici-types@6.21.0", "https://registry-node.techniker.me/undici-types/-/undici-types-6.21.0.tgz", {}, "sha512-iwDZqg0QAGrg9Rav5H4n0M64c3mkR59cJ6wQp+7C4nI0gsmExaedaYLNO44eT4AtBBwjbTiGPMlt2Md0T9H9JQ=="],
}
}

View File

@@ -1,6 +1,6 @@
{
"name": "@techniker-me/rtmp-push",
"version": "2025.0.3",
"version": "2025.1.0",
"description": "A TypeScript library for pushing media streams to RTMP servers using FFmpeg",
"main": "dist/node/index.js",
"module": "src/index.ts",
@@ -14,18 +14,18 @@
"ci-build": "bun run build:node && bun run build:browser && bun run build:types",
"build:node": "bun build src/index.ts --outdir dist/node --target node --format esm --minify --production",
"build:browser": "bun build src/index.ts --outdir dist/browser --target browser --format esm --minify --production",
"build:types": "tsc --outDir dist/types",
"build:types": "tsc --project tsconfig.build.json",
"prepublishOnly": "bun run build",
"clean": "rm -rf dist"
},
"devDependencies": {
"@types/bun": "latest",
"@types/node": "20.0.0",
"prettier": "3.6.2",
"typescript": "5.9.2"
"@types/node": "^25.0.3",
"prettier": "^3.7.4",
"typescript": "^5.9.3"
},
"peerDependencies": {
"typescript": "5.9.2"
"typescript": "^5.9.3"
},
"exports": {
".": {

View File

@@ -1,13 +1,49 @@
import {spawn, ChildProcess} from 'node:child_process';
import type {ICommandBuilder} from './interfaces/ICommandBuilder.js';
import type {IProcessManager} from './interfaces/IProcessManager.js';
import {CommandBuilder} from './implementations/CommandBuilder.js';
import {ProcessManager} from './implementations/ProcessManager.js';
import {ProcessSpawner} from './implementations/ProcessSpawner.js';
import {ConsoleLogger} from './implementations/ConsoleLogger.js';
/**
* Configuration options for RtmpPush
*/
export interface RtmpPushOptions {
commandBuilder?: ICommandBuilder;
processManager?: IProcessManager;
ffmpegCommand?: string;
}
/**
* RTMP Push class for streaming media to RTMP servers
*
* Follows SOLID principles:
* - Single Responsibility: Orchestrates RTMP streaming workflow
* - Open/Closed: Extensible via dependency injection
* - Liskov Substitution: Works with any ICommandBuilder/IProcessManager implementation
* - Interface Segregation: Uses focused interfaces
* - Dependency Inversion: Depends on abstractions, not concretions
*/
export class RtmpPush {
private _mediaSourceUri: string;
private _rtmpIngestUri: string;
private _activeProcess: ChildProcess | null = null;
private readonly commandBuilder: ICommandBuilder;
private readonly processManager: IProcessManager;
private readonly _mediaSourceUri: string;
private readonly _rtmpIngestUri: string;
constructor(mediaSourceUri: string, rtmpIngestUri: string) {
constructor(
mediaSourceUri: string,
rtmpIngestUri: string,
options: RtmpPushOptions = {}
) {
this._mediaSourceUri = mediaSourceUri;
this._rtmpIngestUri = rtmpIngestUri;
// Dependency Injection with defaults (Dependency Inversion Principle)
this.commandBuilder = options.commandBuilder ?? new CommandBuilder(options.ffmpegCommand);
this.processManager = options.processManager ?? new ProcessManager(
new ProcessSpawner(),
new ConsoleLogger()
);
}
get rtmpIngestUri(): string {
@@ -18,80 +54,57 @@ export class RtmpPush {
return this._mediaSourceUri;
}
/**
* Start streaming to RTMP server
* @param streamKey - The stream key for the RTMP stream
* @param capabilities - Array of capability strings
* @returns This instance for method chaining
*/
public start(streamKey: string, capabilities: string[]): this {
// Use the constructor parameters instead of hardcoded values
const mediaSourceUri = this._mediaSourceUri;
const ingestUri = `${this._rtmpIngestUri}/${streamKey};capabilities=${capabilities.join(',')};tags=`;
const command = [
'ffmpeg',
'-re',
'-hide_banner',
'-stream_loop',
'-1',
'-y',
'-flags',
'low_delay',
'-fflags',
'+nobuffer+flush_packets',
'-i',
mediaSourceUri,
'-c:a',
'copy',
'-c:v',
'copy',
'-flush_packets',
'1',
'-copyts',
'-f',
'flv',
ingestUri
];
try {
if (!command[0]) {
throw new Error('Invalid command: ffmpeg not found');
}
this._activeProcess = spawn(command[0], command.slice(1));
if (!this._activeProcess) {
throw new Error('Failed to spawn ffmpeg process');
}
this._activeProcess.stdout?.on('data', data => {
console.log(`stdout: ${data}`);
});
this._activeProcess.stderr?.on('data', data => {
console.log(`stderr: ${data}`);
});
this._activeProcess.on('close', code => {
console.log(`child process exited with code ${code}`);
this._activeProcess = null;
});
this._activeProcess.on('error', error => {
console.error('ffmpeg process error:', error);
this._activeProcess = null;
});
} catch (error) {
console.error('Failed to start RTMP push:', error);
throw error;
if (this.processManager.isRunning()) {
throw new Error('Stream is already running');
}
const command = this.commandBuilder.buildCommand(
this._mediaSourceUri,
this._rtmpIngestUri,
streamKey,
capabilities
);
const [ffmpegCommand, ...args] = command;
if (!ffmpegCommand) {
throw new Error('Invalid command: ffmpeg command not found');
}
this.processManager.start(ffmpegCommand, args);
return this;
}
/**
* Stop the current stream
* @returns This instance for method chaining
*/
public stop(): this {
if (this._activeProcess) {
this._activeProcess.kill();
this._activeProcess = null;
}
this.processManager.stop();
return this;
}
/**
* Check if stream is currently running
* @returns True if stream is active
*/
public isRunning(): boolean {
return this._activeProcess !== null && !this._activeProcess.killed;
return this.processManager.isRunning();
}
/**
* Get the process manager for event handling
* Allows access to process events (data, error, close)
*/
public getProcessManager(): IProcessManager {
return this.processManager;
}
}

View File

@@ -0,0 +1,48 @@
import type {ICommandBuilder} from '../interfaces/ICommandBuilder.js';
/**
* Builds FFmpeg commands for RTMP streaming
* Single Responsibility: Command construction only
*/
export class CommandBuilder implements ICommandBuilder {
private readonly ffmpegCommand: string;
constructor(ffmpegCommand: string = 'ffmpeg') {
this.ffmpegCommand = ffmpegCommand;
}
buildCommand(
mediaSourceUri: string,
rtmpIngestUri: string,
streamKey: string,
capabilities: string[]
): string[] {
const ingestUri = `${rtmpIngestUri}/${streamKey};capabilities=${capabilities.join(',')};tags=`;
return [
this.ffmpegCommand,
'-re',
'-hide_banner',
'-stream_loop',
'-1',
'-y',
'-flags',
'low_delay',
'-fflags',
'+nobuffer+flush_packets',
'-i',
mediaSourceUri,
'-c:a',
'copy',
'-c:v',
'copy',
'-flush_packets',
'1',
'-copyts',
'-f',
'flv',
ingestUri
];
}
}

View File

@@ -0,0 +1,16 @@
import type {ILogger} from '../interfaces/ILogger.js';
/**
* Console-based logger implementation
* Single Responsibility: Logging only
*/
export class ConsoleLogger implements ILogger {
log(message: string): void {
console.log(message);
}
error(message: string, error?: unknown): void {
console.error(message, error);
}
}

View File

@@ -0,0 +1,85 @@
import {ChildProcess} from 'node:child_process';
import type {IProcessManager, ProcessManagerEvents} from '../interfaces/IProcessManager.js';
import type {IProcessSpawner} from '../interfaces/IProcessSpawner.js';
import type {ILogger} from '../interfaces/ILogger.js';
import {EventEmitter} from 'node:events';
/**
* Manages process lifecycle
* Single Responsibility: Process management only
* Open/Closed: Can be extended without modification
*/
export class ProcessManager extends EventEmitter implements IProcessManager {
private activeProcess: ChildProcess | null = null;
constructor(
private readonly processSpawner: IProcessSpawner,
private readonly logger: ILogger
) {
super();
}
start(command: string, args: string[]): void {
if (this.isRunning()) {
throw new Error('Process is already running');
}
if (!command) {
throw new Error('Invalid command: command cannot be empty');
}
try {
this.activeProcess = this.processSpawner.spawn(command, args);
if (!this.activeProcess) {
throw new Error('Failed to spawn process');
}
this.setupEventHandlers();
} catch (error) {
this.logger.error('Failed to start process:', error);
this.activeProcess = null;
throw error;
}
}
stop(): void {
if (this.activeProcess) {
this.activeProcess.kill();
this.activeProcess = null;
}
}
isRunning(): boolean {
return this.activeProcess !== null && !this.activeProcess.killed;
}
private setupEventHandlers(): void {
if (!this.activeProcess) {
return;
}
this.activeProcess.stdout?.on('data', (data: Buffer) => {
this.logger.log(`stdout: ${data}`);
this.emit('data', data);
});
this.activeProcess.stderr?.on('data', (data: Buffer) => {
this.logger.log(`stderr: ${data}`);
this.emit('data', data);
});
this.activeProcess.on('close', (code: number | null) => {
this.logger.log(`child process exited with code ${code}`);
this.activeProcess = null;
this.emit('close', code);
});
this.activeProcess.on('error', (error: Error) => {
this.logger.error('Process error:', error);
this.activeProcess = null;
this.emit('error', error);
});
}
}

View File

@@ -0,0 +1,13 @@
import {spawn, ChildProcess} from 'node:child_process';
import type {IProcessSpawner} from '../interfaces/IProcessSpawner.js';
/**
* Default implementation of process spawner
* Single Responsibility: Process spawning only
*/
export class ProcessSpawner implements IProcessSpawner {
spawn(command: string, args: string[]): ChildProcess {
return spawn(command, args);
}
}

View File

@@ -1,4 +1,19 @@
import {RtmpPush} from './RtmpPush.js';
// Main export
export {RtmpPush} from './RtmpPush.js';
export type {RtmpPushOptions} from './RtmpPush.js';
export {RtmpPush};
// Interfaces (for extensibility)
export type {ICommandBuilder} from './interfaces/ICommandBuilder.js';
export type {IProcessManager, ProcessManagerEvents} from './interfaces/IProcessManager.js';
export type {IProcessSpawner} from './interfaces/IProcessSpawner.js';
export type {ILogger} from './interfaces/ILogger.js';
// Default implementations (for convenience)
export {CommandBuilder} from './implementations/CommandBuilder.js';
export {ProcessManager} from './implementations/ProcessManager.js';
export {ProcessSpawner} from './implementations/ProcessSpawner.js';
export {ConsoleLogger} from './implementations/ConsoleLogger.js';
// Default export
import {RtmpPush} from './RtmpPush.js';
export default {RtmpPush};

View File

@@ -0,0 +1,8 @@
/**
* Interface for building FFmpeg commands (Interface Segregation Principle)
* Separates command building logic from process management
*/
export interface ICommandBuilder {
buildCommand(mediaSourceUri: string, rtmpIngestUri: string, streamKey: string, capabilities: string[]): string[];
}

View File

@@ -0,0 +1,9 @@
/**
* Interface for logging (Dependency Inversion Principle)
* Allows for different logging implementations
*/
export interface ILogger {
log(message: string): void;
error(message: string, error?: unknown): void;
}

View File

@@ -0,0 +1,32 @@
import {EventEmitter} from 'node:events';
/**
* Events emitted by ProcessManager
*/
export interface ProcessManagerEvents {
data: [Buffer];
error: [Error];
close: [number | null];
}
/**
* Interface for process management (Interface Segregation Principle)
* Separates process lifecycle management from business logic
*/
export interface IProcessManager extends EventEmitter {
start(command: string, args: string[]): void;
stop(): void;
isRunning(): boolean;
// Event emitter methods with proper typing
on(event: 'data', listener: (data: Buffer) => void): this;
on(event: 'error', listener: (error: Error) => void): this;
on(event: 'close', listener: (code: number | null) => void): this;
on(event: string | symbol, listener: (...args: unknown[]) => void): this;
emit(event: 'data', data: Buffer): boolean;
emit(event: 'error', error: Error): boolean;
emit(event: 'close', code: number | null): boolean;
emit(event: string | symbol, ...args: unknown[]): boolean;
}

View File

@@ -0,0 +1,10 @@
import {ChildProcess} from 'node:child_process';
/**
* Interface for spawning child processes (Dependency Inversion Principle)
* Allows for dependency injection and easier testing
*/
export interface IProcessSpawner {
spawn(command: string, args: string[]): ChildProcess;
}

View File

@@ -1,5 +1,5 @@
import {describe, it, expect, beforeEach, afterEach} from 'bun:test';
import {RtmpPush} from '../src/RtmpPush';
import {RtmpPush} from '../src/index.js';
describe('RtmpPush Integration Tests', () => {
let rtmpPush: RtmpPush;

View File

@@ -1,5 +1,7 @@
import {describe, it, expect, beforeEach, afterEach} from 'bun:test';
import {RtmpPush} from '../src/RtmpPush';
import {RtmpPush} from '../src/index.js';
import type {ICommandBuilder, IProcessManager, ILogger, IProcessSpawner} from '../src/index.js';
import {CommandBuilder, ProcessManager, ProcessSpawner, ConsoleLogger} from '../src/index.js';
describe('RtmpPush', () => {
let rtmpPush: RtmpPush;
@@ -252,5 +254,179 @@ describe('RtmpPush', () => {
expect(error).toBeInstanceOf(Error);
}
});
it('should throw error when starting an already running stream', () => {
const streamKey = 'test-stream';
const capabilities = ['h264'];
// Mock process manager that reports as running
const mockProcessManager: IProcessManager = {
isRunning: () => true,
start: () => {},
stop: () => {},
on: () => mockProcessManager,
emit: () => false,
once: () => mockProcessManager,
off: () => mockProcessManager,
removeListener: () => mockProcessManager,
removeAllListeners: () => mockProcessManager,
addListener: () => mockProcessManager,
setMaxListeners: () => mockProcessManager,
getMaxListeners: () => 10,
listeners: () => [],
rawListeners: () => [],
listenerCount: () => 0,
prependListener: () => mockProcessManager,
prependOnceListener: () => mockProcessManager,
eventNames: () => []
} as IProcessManager;
const customRtmpPush = new RtmpPush(mockMediaSourceUri, mockRtmpIngestUri, {
processManager: mockProcessManager
});
expect(() => {
customRtmpPush.start(streamKey, capabilities);
}).toThrow('Stream is already running');
});
});
describe('Dependency Injection', () => {
it('should accept custom command builder', () => {
const mockCommandBuilder: ICommandBuilder = {
buildCommand: (mediaUri, rtmpUri, streamKey, capabilities) => {
expect(mediaUri).toBe(mockMediaSourceUri);
expect(rtmpUri).toBe(mockRtmpIngestUri);
expect(streamKey).toBe('test-key');
expect(capabilities).toEqual(['h264']);
return ['custom-ffmpeg', '-custom', 'args'];
}
};
const customRtmpPush = new RtmpPush(mockMediaSourceUri, mockRtmpIngestUri, {
commandBuilder: mockCommandBuilder
});
expect(customRtmpPush.mediaSourceUri).toBe(mockMediaSourceUri);
expect(customRtmpPush.rtmpIngestUri).toBe(mockRtmpIngestUri);
});
it('should accept custom process manager', () => {
let startCalled = false;
const mockProcessManager: IProcessManager = {
isRunning: () => false,
start: (command, args) => {
startCalled = true;
expect(command).toBe('ffmpeg');
expect(args).toBeInstanceOf(Array);
},
stop: () => {},
on: () => mockProcessManager,
emit: () => false,
once: () => mockProcessManager,
off: () => mockProcessManager,
removeListener: () => mockProcessManager,
removeAllListeners: () => mockProcessManager,
addListener: () => mockProcessManager,
setMaxListeners: () => mockProcessManager,
getMaxListeners: () => 10,
listeners: () => [],
rawListeners: () => [],
listenerCount: () => 0,
prependListener: () => mockProcessManager,
prependOnceListener: () => mockProcessManager,
eventNames: () => []
} as IProcessManager;
const customRtmpPush = new RtmpPush(mockMediaSourceUri, mockRtmpIngestUri, {
processManager: mockProcessManager
});
try {
customRtmpPush.start('test-key', ['h264']);
expect(startCalled).toBe(true);
} catch (error) {
// If ffmpeg is not available, that's expected
expect(error).toBeInstanceOf(Error);
}
});
it('should accept custom ffmpeg command', () => {
const customRtmpPush = new RtmpPush(mockMediaSourceUri, mockRtmpIngestUri, {
ffmpegCommand: 'custom-ffmpeg-path'
});
expect(customRtmpPush.mediaSourceUri).toBe(mockMediaSourceUri);
expect(customRtmpPush.rtmpIngestUri).toBe(mockRtmpIngestUri);
});
it('should use default implementations when none provided', () => {
const defaultRtmpPush = new RtmpPush(mockMediaSourceUri, mockRtmpIngestUri);
expect(defaultRtmpPush).toBeInstanceOf(RtmpPush);
expect(defaultRtmpPush.mediaSourceUri).toBe(mockMediaSourceUri);
expect(defaultRtmpPush.rtmpIngestUri).toBe(mockRtmpIngestUri);
});
});
describe('Process Manager Events', () => {
it('should provide access to process manager for event handling', () => {
const processManager = rtmpPush.getProcessManager();
expect(processManager).toBeDefined();
expect(typeof processManager.on).toBe('function');
expect(typeof processManager.emit).toBe('function');
});
it('should allow listening to process events', () => {
const processManager = rtmpPush.getProcessManager();
let dataReceived = false;
let errorReceived = false;
let closeReceived = false;
processManager.on('data', () => {
dataReceived = true;
});
processManager.on('error', () => {
errorReceived = true;
});
processManager.on('close', () => {
closeReceived = true;
});
// Events would be emitted by the actual process, but we can verify the listeners are set up
expect(typeof processManager.on).toBe('function');
});
});
describe('CommandBuilder', () => {
it('should build correct FFmpeg commands', () => {
const builder = new CommandBuilder();
const command = builder.buildCommand(
'https://example.com/video.ts',
'rtmp://ingest.example.com/ingest',
'test-key',
['h264', 'aac']
);
expect(command[0]).toBe('ffmpeg');
expect(command).toContain('-re');
expect(command).toContain('-i');
expect(command).toContain('https://example.com/video.ts');
expect(command[command.length - 1]).toContain('rtmp://ingest.example.com/ingest/test-key');
expect(command[command.length - 1]).toContain('capabilities=h264,aac');
});
it('should use custom ffmpeg command when provided', () => {
const builder = new CommandBuilder('custom-ffmpeg');
const command = builder.buildCommand(
'https://example.com/video.ts',
'rtmp://ingest.example.com/ingest',
'test-key',
['h264']
);
expect(command[0]).toBe('custom-ffmpeg');
});
});
});

17
tsconfig.build.json Normal file
View File

@@ -0,0 +1,17 @@
{
"extends": "./tsconfig.json",
"compilerOptions": {
"noEmit": false,
"declaration": true,
"emitDeclarationOnly": true,
"outDir": "./dist/types",
"rootDir": "./src",
"moduleResolution": "node",
"module": "ESNext",
"allowImportingTsExtensions": false,
"declarationMap": true
},
"include": ["src/**/*"],
"exclude": ["node_modules", "dist", "tests"]
}