TASK 4 ship step 5: live frame pipeline — FramePump paces the active scene's composite into the encoder, ScreenCaptureManager.GetLatestFrame, MainViewModel resolver/option-builder/pump wiring, FramePumpTests (7) + GetLatestFrame test — 147 tests passing, 0 warnings
This commit is contained in:
@@ -0,0 +1,256 @@
|
||||
using System;
|
||||
using System.Threading;
|
||||
using System.Threading.Tasks;
|
||||
using ytLive.Models;
|
||||
using ytLive.Services.Compositor;
|
||||
|
||||
namespace ytLive.Services.Encoder;
|
||||
|
||||
/// <summary>
|
||||
/// The live frame producer (TASK 4 ship step 5): the bridge between the capture
|
||||
/// managers + compositor and the encoder. While live it snapshots the active
|
||||
/// scene each tick, resolves every element to its latest frame, composites it
|
||||
/// into the tier's output frame, and paces frames into the encoder at the tier's
|
||||
/// FPS. All collaborators are constructor-injected seams (scene, resolver,
|
||||
/// options, encoder factory, pacing delay) so the pump stays free of WPF and of
|
||||
/// the capture managers and is fully hermetic in tests.
|
||||
///
|
||||
/// The RTMP URL comes from the options provider: until the live-stream create
|
||||
/// flow lands (TASK 5) it yields null, so go-live runs the existing visual flow
|
||||
/// without actually pushing.
|
||||
/// </summary>
|
||||
public sealed class FramePump : IDisposable
|
||||
{
|
||||
private readonly Func<Scene?> _sceneProvider;
|
||||
private readonly Func<SceneElement, VideoFrame?> _frameResolver;
|
||||
private readonly Func<CompositorOptions> _compositorOptions;
|
||||
private readonly Func<EncoderOptions?> _encoderOptions;
|
||||
private readonly Func<IFfmpegEncoder> _encoderFactory;
|
||||
private readonly Action<string>? _log;
|
||||
private readonly Func<TimeSpan, CancellationToken, Task> _pacingDelay;
|
||||
private readonly SceneCompositor _compositor = new();
|
||||
|
||||
private readonly object _gate = new();
|
||||
private IFfmpegEncoder? _encoder;
|
||||
private CancellationTokenSource? _cts;
|
||||
private Task? _pumpTask;
|
||||
private bool _started;
|
||||
|
||||
/// <summary>Forwards the encoder's parsed health — ship step 6 binds this to the bottom bar.</summary>
|
||||
public event EventHandler<StreamHealth>? HealthUpdated;
|
||||
|
||||
/// <summary>Raised when the encoder cannot start or dies mid-stream. The pump stops itself.</summary>
|
||||
public event EventHandler<string>? Failed;
|
||||
|
||||
public FramePump(
|
||||
Func<Scene?> sceneProvider,
|
||||
Func<SceneElement, VideoFrame?> frameResolver,
|
||||
Func<CompositorOptions> compositorOptions,
|
||||
Func<EncoderOptions?> encoderOptions,
|
||||
Func<IFfmpegEncoder> encoderFactory,
|
||||
Action<string>? log = null,
|
||||
Func<TimeSpan, CancellationToken, Task>? pacingDelay = null)
|
||||
{
|
||||
_sceneProvider = sceneProvider ?? throw new ArgumentNullException(nameof(sceneProvider));
|
||||
_frameResolver = frameResolver ?? throw new ArgumentNullException(nameof(frameResolver));
|
||||
_compositorOptions = compositorOptions ?? throw new ArgumentNullException(nameof(compositorOptions));
|
||||
_encoderOptions = encoderOptions ?? throw new ArgumentNullException(nameof(encoderOptions));
|
||||
_encoderFactory = encoderFactory ?? throw new ArgumentNullException(nameof(encoderFactory));
|
||||
_log = log;
|
||||
_pacingDelay = pacingDelay ?? ((delay, ct) => Task.Delay(delay, ct));
|
||||
}
|
||||
|
||||
public bool IsRunning { get; private set; }
|
||||
|
||||
/// <summary>Never throws: failures are logged and surfaced via <see cref="Failed"/>,
|
||||
/// so the VM can fire-and-forget it from a sync command handler.</summary>
|
||||
public async Task StartAsync(CancellationToken cancellationToken = default)
|
||||
{
|
||||
lock (_gate)
|
||||
{
|
||||
if (_started) return;
|
||||
_started = true;
|
||||
}
|
||||
|
||||
IFfmpegEncoder? encoder = null;
|
||||
try
|
||||
{
|
||||
var options = _encoderOptions();
|
||||
if (options == null)
|
||||
{
|
||||
_log?.Invoke("FramePump: no RTMP URL available (live-stream create lands in TASK 5) — encoder skipped");
|
||||
lock (_gate) _started = false;
|
||||
return;
|
||||
}
|
||||
|
||||
encoder = _encoderFactory();
|
||||
encoder.HealthUpdated += OnHealthUpdated;
|
||||
encoder.ProcessFailed += OnProcessFailed;
|
||||
await encoder.StartAsync(options, cancellationToken);
|
||||
|
||||
lock (_gate)
|
||||
{
|
||||
_encoder = encoder;
|
||||
}
|
||||
|
||||
// IsRunning must be true before the loop starts: the loop reads it on
|
||||
// its first iteration, and with a completed-task delay it can run
|
||||
// synchronously on this thread before PumpAsync even returns.
|
||||
IsRunning = true;
|
||||
_cts = new CancellationTokenSource();
|
||||
_pumpTask = PumpAsync(options, _cts.Token);
|
||||
_log?.Invoke($"FramePump started ({options.Width}×{options.Height} @ {options.Fps} fps)");
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_log?.Invoke($"FramePump: start failed: {ex.Message}");
|
||||
if (encoder != null)
|
||||
{
|
||||
encoder.HealthUpdated -= OnHealthUpdated;
|
||||
encoder.ProcessFailed -= OnProcessFailed;
|
||||
try
|
||||
{
|
||||
encoder.Dispose();
|
||||
}
|
||||
catch (Exception disposeEx)
|
||||
{
|
||||
_log?.Invoke($"FramePump: disposing failed encoder: {disposeEx.Message}");
|
||||
}
|
||||
}
|
||||
lock (_gate)
|
||||
{
|
||||
_started = false;
|
||||
IsRunning = false;
|
||||
}
|
||||
Failed?.Invoke(this, ex.Message);
|
||||
}
|
||||
}
|
||||
|
||||
public async Task StopAsync(CancellationToken cancellationToken = default)
|
||||
{
|
||||
IFfmpegEncoder? encoder;
|
||||
Task? pump;
|
||||
lock (_gate)
|
||||
{
|
||||
if (!_started && _encoder == null) return;
|
||||
_started = false;
|
||||
IsRunning = false;
|
||||
encoder = _encoder;
|
||||
pump = _pumpTask;
|
||||
_cts?.Cancel();
|
||||
}
|
||||
|
||||
// Stop the encoder BEFORE awaiting the pump: closing its stdin unblocks a
|
||||
// write stuck on pipe backpressure, otherwise the pump could await forever.
|
||||
if (encoder != null)
|
||||
{
|
||||
try
|
||||
{
|
||||
await encoder.StopAsync(cancellationToken);
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_log?.Invoke($"FramePump: encoder stop failed: {ex.Message}");
|
||||
}
|
||||
}
|
||||
|
||||
if (pump != null)
|
||||
{
|
||||
try
|
||||
{
|
||||
await pump;
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_log?.Invoke($"FramePump: pump loop faulted during stop: {ex.Message}");
|
||||
}
|
||||
}
|
||||
|
||||
if (encoder != null)
|
||||
{
|
||||
encoder.HealthUpdated -= OnHealthUpdated;
|
||||
encoder.ProcessFailed -= OnProcessFailed;
|
||||
try
|
||||
{
|
||||
encoder.Dispose();
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_log?.Invoke($"FramePump: encoder dispose failed: {ex.Message}");
|
||||
}
|
||||
}
|
||||
|
||||
lock (_gate)
|
||||
{
|
||||
_encoder = null;
|
||||
_cts = null;
|
||||
_pumpTask = null;
|
||||
}
|
||||
_log?.Invoke("FramePump stopped");
|
||||
}
|
||||
|
||||
public void Dispose()
|
||||
{
|
||||
try
|
||||
{
|
||||
StopAsync().GetAwaiter().GetResult();
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_log?.Invoke($"FramePump: dispose failed: {ex.Message}");
|
||||
}
|
||||
}
|
||||
|
||||
private async Task PumpAsync(EncoderOptions options, CancellationToken ct)
|
||||
{
|
||||
var interval = TimeSpan.FromSeconds(1d / Math.Max(1, options.Fps));
|
||||
|
||||
try
|
||||
{
|
||||
while (!ct.IsCancellationRequested)
|
||||
{
|
||||
var scene = _sceneProvider();
|
||||
if (scene != null)
|
||||
{
|
||||
var frame = _compositor.Render(scene, _frameResolver, null, _compositorOptions());
|
||||
IFfmpegEncoder? encoder;
|
||||
lock (_gate) encoder = _encoder;
|
||||
if (encoder == null) break; // _encoder is only cleared after the loop ends; defensive
|
||||
await encoder.SubmitFrameAsync(frame, ct);
|
||||
}
|
||||
await _pacingDelay(interval, ct);
|
||||
}
|
||||
}
|
||||
catch (OperationCanceledException)
|
||||
{
|
||||
// normal stop
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
// A failure while the pump is supposed to run (encoder died under us,
|
||||
// scene provider faulted, ...) stops the pump and surfaces once.
|
||||
if (ct.IsCancellationRequested)
|
||||
{
|
||||
_log?.Invoke($"FramePump: pump exited during stop: {ex.Message}");
|
||||
}
|
||||
else
|
||||
{
|
||||
_log?.Invoke($"FramePump: pump loop faulted: {ex.Message}");
|
||||
Failed?.Invoke(this, ex.Message);
|
||||
}
|
||||
}
|
||||
finally
|
||||
{
|
||||
lock (_gate) IsRunning = false;
|
||||
}
|
||||
}
|
||||
|
||||
private void OnHealthUpdated(object? sender, StreamHealth health) => HealthUpdated?.Invoke(this, health);
|
||||
|
||||
private void OnProcessFailed(object? sender, string message)
|
||||
{
|
||||
_log?.Invoke($"FramePump: encoder process failed: {message}");
|
||||
Failed?.Invoke(this, message);
|
||||
_ = StopAsync();
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user