TASK 4 ship step 3: encoder + RTMP push — FFmpeg subprocess with probed H.264 picker, BGRA stdin feed, stderr health parsing, graceful stop — 122 tests passing, 0 warnings
This commit is contained in:
@@ -0,0 +1,263 @@
|
||||
using System.Diagnostics;
|
||||
using ytLive.Helpers;
|
||||
using ytLive.Models;
|
||||
|
||||
namespace ytLive.Services.Encoder;
|
||||
|
||||
/// <summary>
|
||||
/// The default <see cref="IFfmpegEncoder"/>: spawns <c>ffmpeg.exe</c> (resolved via
|
||||
/// <see cref="IFfmpegLocator"/>), feeds raw BGRA frames into stdin, and parses the
|
||||
/// <c>-stats</c> progress lines into <see cref="StreamHealth"/>. Encoder choice is
|
||||
/// probed from the binary's <c>-encoders</c> listing (hardware NVENC/QSV/AMF first,
|
||||
/// OpenH264 software fallback — never libx264, see the license posture) unless
|
||||
/// <see cref="EncoderOptions.VideoEncoder"/> forces one.
|
||||
///
|
||||
/// Graceful stop = close stdin (EOF) → ffmpeg finalizes the FLV and exits by itself;
|
||||
/// a watchdogs kill fires only if it hasn't exited shortly after EOF.
|
||||
/// </summary>
|
||||
public sealed class FfmpegEncoder : IFfmpegEncoder
|
||||
{
|
||||
public event EventHandler<StreamHealth>? HealthUpdated;
|
||||
public event EventHandler<string>? ProcessFailed;
|
||||
|
||||
private readonly IFfmpegLocator _locator;
|
||||
private readonly Func<IEncoderProcess> _processFactory;
|
||||
|
||||
private readonly object _gate = new();
|
||||
private IEncoderProcess? _process;
|
||||
private EncoderOptions? _options;
|
||||
private StreamHealth _health = new() { Status = StreamStatus.Offline };
|
||||
private Task? _stderrLoop;
|
||||
private bool _stopRequested;
|
||||
|
||||
public FfmpegEncoder(
|
||||
IFfmpegLocator locator,
|
||||
Func<IEncoderProcess>? processFactory = null)
|
||||
{
|
||||
_locator = locator;
|
||||
_processFactory = processFactory ?? (() => new FfmpegEncoderProcess());
|
||||
}
|
||||
|
||||
public bool IsRunning { get; private set; }
|
||||
|
||||
public async Task StartAsync(EncoderOptions options, CancellationToken cancellationToken = default)
|
||||
{
|
||||
if (options == null) throw new ArgumentNullException(nameof(options));
|
||||
if (string.IsNullOrWhiteSpace(options.RtmpUrl))
|
||||
throw new ArgumentException("An RTMP ingestion URL is required.", nameof(options));
|
||||
|
||||
lock (_gate)
|
||||
{
|
||||
if (IsRunning) throw new InvalidOperationException("The encoder is already running.");
|
||||
_options = options;
|
||||
}
|
||||
|
||||
var ffmpegPath = await _locator.LocateAsync(cancellationToken).ConfigureAwait(false);
|
||||
var encoder = options.VideoEncoder ?? await ProbeEncoderAsync(ffmpegPath, cancellationToken).ConfigureAwait(false);
|
||||
var args = FfmpegArgs.Build(options, encoder);
|
||||
|
||||
var startInfo = new ProcessStartInfo
|
||||
{
|
||||
FileName = ffmpegPath,
|
||||
UseShellExecute = false,
|
||||
RedirectStandardInput = true,
|
||||
RedirectStandardOutput = true,
|
||||
RedirectStandardError = true,
|
||||
CreateNoWindow = true,
|
||||
};
|
||||
foreach (var arg in args) startInfo.ArgumentList.Add(arg);
|
||||
|
||||
IEncoderProcess process;
|
||||
try
|
||||
{
|
||||
process = _processFactory();
|
||||
process.Start(startInfo);
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
AppLog.Write(ex, "FFmpeg encoder: failed to start subprocess");
|
||||
lock (_gate) _options = null;
|
||||
throw;
|
||||
}
|
||||
|
||||
lock (_gate)
|
||||
{
|
||||
_process = process;
|
||||
IsRunning = true;
|
||||
_health = new StreamHealth { Status = StreamStatus.Streaming };
|
||||
}
|
||||
|
||||
_stopRequested = false;
|
||||
_stderrLoop = RunStderrLoopAsync(process);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Write one raw BGRA frame to ffmpeg's stdin. Serialized internally; callers
|
||||
/// (the compositor pump) may race freely. Frames are written as-is — the caller
|
||||
/// paces to capture rate (the compositor's job, ship step 5).
|
||||
/// </summary>
|
||||
public async Task SubmitFrameAsync(VideoFrame frame, CancellationToken cancellationToken = default)
|
||||
{
|
||||
if (frame == null) throw new ArgumentNullException(nameof(frame));
|
||||
IEncoderProcess? process;
|
||||
lock (_gate)
|
||||
{
|
||||
if (!IsRunning) throw new InvalidOperationException("The encoder is not running.");
|
||||
process = _process;
|
||||
}
|
||||
|
||||
var bytes = frame.BgraPixels;
|
||||
await process!.StandardInput.WriteAsync(bytes, cancellationToken).ConfigureAwait(false);
|
||||
await process.StandardInput.FlushAsync(cancellationToken).ConfigureAwait(false);
|
||||
}
|
||||
|
||||
public async Task StopAsync(CancellationToken cancellationToken = default)
|
||||
{
|
||||
IEncoderProcess? process;
|
||||
Task? loop;
|
||||
lock (_gate)
|
||||
{
|
||||
if (!IsRunning) return;
|
||||
_stopRequested = true;
|
||||
process = _process;
|
||||
loop = _stderrLoop;
|
||||
}
|
||||
|
||||
try
|
||||
{
|
||||
process!.StandardInput.Dispose(); // EOF → ffmpeg finalizes + exits
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
AppLog.Write(ex, "FFmpeg encoder: closing stdin failed");
|
||||
}
|
||||
|
||||
try
|
||||
{
|
||||
using var timeout = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
|
||||
timeout.CancelAfter(TimeSpan.FromSeconds(10));
|
||||
await process.WaitForExitAsync(timeout.Token).ConfigureAwait(false);
|
||||
}
|
||||
catch (OperationCanceledException)
|
||||
{
|
||||
AppLog.Write("FFmpeg encoder: did not exit after stdin EOF — killing");
|
||||
process!.Kill();
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
AppLog.Write(ex, "FFmpeg encoder: waiting for exit failed");
|
||||
process!.Kill();
|
||||
}
|
||||
|
||||
try
|
||||
{
|
||||
if (loop != null) await loop.ConfigureAwait(false);
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
AppLog.Write(ex, "FFmpeg encoder: stderr loop faulted during stop");
|
||||
}
|
||||
|
||||
process.Dispose();
|
||||
|
||||
lock (_gate)
|
||||
{
|
||||
IsRunning = false;
|
||||
_health.Status = StreamStatus.Offline;
|
||||
_health.LastError = null;
|
||||
_process = null;
|
||||
_options = null;
|
||||
}
|
||||
|
||||
AppLog.Write($"FFmpeg encoder stopped (exit {process.ExitCode})");
|
||||
}
|
||||
|
||||
public void Dispose()
|
||||
{
|
||||
lock (_gate)
|
||||
{
|
||||
if (!IsRunning) return;
|
||||
_process?.Kill();
|
||||
_process?.Dispose();
|
||||
_process = null;
|
||||
IsRunning = false;
|
||||
}
|
||||
}
|
||||
|
||||
private async Task<string> ProbeEncoderAsync(string ffmpegPath, CancellationToken cancellationToken)
|
||||
{
|
||||
try
|
||||
{
|
||||
var startInfo = new ProcessStartInfo
|
||||
{
|
||||
FileName = ffmpegPath,
|
||||
UseShellExecute = false,
|
||||
RedirectStandardOutput = true,
|
||||
RedirectStandardError = true,
|
||||
CreateNoWindow = true,
|
||||
};
|
||||
startInfo.ArgumentList.Add("-hide_banner");
|
||||
startInfo.ArgumentList.Add("-encoders");
|
||||
|
||||
using var probe = _processFactory();
|
||||
probe.Start(startInfo);
|
||||
using var cts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
|
||||
cts.CancelAfter(TimeSpan.FromSeconds(15));
|
||||
var output = await probe.StandardOutput.ReadToEndAsync(cts.Token).ConfigureAwait(false);
|
||||
return FfmpegEncoderPicker.Pick(output);
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
AppLog.Write(ex, "FFmpeg encoder: encoder probe failed — falling back to software");
|
||||
return FfmpegEncoderPicker.Preference[^1];
|
||||
}
|
||||
}
|
||||
|
||||
private Task RunStderrLoopAsync(IEncoderProcess process)
|
||||
{
|
||||
return Task.Run(async () =>
|
||||
{
|
||||
try
|
||||
{
|
||||
while (true)
|
||||
{
|
||||
var line = await process.StandardError.ReadLineAsync().ConfigureAwait(false);
|
||||
if (line == null) break;
|
||||
OnStderrLine(process, line);
|
||||
}
|
||||
|
||||
var code = process.ExitCode;
|
||||
var stillRunning = false;
|
||||
lock (_gate) stillRunning = IsRunning;
|
||||
if (stillRunning && !_stopRequested && code != 0)
|
||||
{
|
||||
AppLog.Write($"FFmpeg encoder: subprocess exited unexpectedly ({code})");
|
||||
_health.LastError = $"FFmpeg exited with code {code}";
|
||||
ProcessFailed?.Invoke(this, $"FFmpeg exited with code {code}");
|
||||
}
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
AppLog.Write(ex, "FFmpeg encoder: stderr loop faulted");
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
private void OnStderrLine(IEncoderProcess process, string line)
|
||||
{
|
||||
var progress = FfmpegProgressParser.TryParse(line);
|
||||
if (progress == null) return;
|
||||
|
||||
var dropped = Math.Max(0, (long)Math.Round(progress.Value.Fps * progress.Value.Duration.TotalSeconds) - progress.Value.Frame);
|
||||
|
||||
lock (_gate)
|
||||
{
|
||||
_health.CurrentBitrate = progress.Value.BitrateKbps;
|
||||
_health.FPS = progress.Value.Fps;
|
||||
_health.DroppedFrames = (int)dropped;
|
||||
_health.StreamDuration = progress.Value.Duration;
|
||||
}
|
||||
|
||||
HealthUpdated?.Invoke(this, _health);
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user