using System; using System.Threading; using System.Threading.Tasks; using ytLive.Models; using ytLive.Services.Compositor; namespace ytLive.Services.Encoder; /// /// 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. /// public sealed class FramePump : IDisposable { private readonly Func _sceneProvider; private readonly Func _frameResolver; private readonly Func _compositorOptions; private readonly Func _encoderOptions; private readonly Func _encoderFactory; private readonly Action? _log; private readonly Func _pacingDelay; private readonly Func<(VideoFrame? Frame, SocialBarPosition Position)>? _socialBar; private readonly SceneCompositor _compositor = new(); private readonly object _gate = new(); private IFfmpegEncoder? _encoder; private CancellationTokenSource? _cts; private Task? _pumpTask; private bool _started; /// Forwards the encoder's parsed health — ship step 6 binds this to the bottom bar. public event EventHandler? HealthUpdated; /// Raised when the encoder cannot start or dies mid-stream. The pump stops itself. public event EventHandler? Failed; public FramePump( Func sceneProvider, Func frameResolver, Func compositorOptions, Func encoderOptions, Func encoderFactory, Action? log = null, Func? pacingDelay = null, Func<(VideoFrame? Frame, SocialBarPosition Position)>? socialBar = 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)); _socialBar = socialBar; } public bool IsRunning { get; private set; } /// Never throws: failures are logged and surfaced via , /// so the VM can fire-and-forget it from a sync command handler. 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 compositorOptions = _compositorOptions(); VideoFrame? socialBarFrame = null; var socialBarTop = 0; if (_socialBar != null) { var (barFrame, position) = _socialBar(); socialBarFrame = barFrame; if (barFrame != null) socialBarTop = position == SocialBarPosition.Top ? 0 : compositorOptions.SourceRectHeight - barFrame.Height; } var frame = _compositor.Render( scene, _frameResolver, null, compositorOptions, socialBarFrame, socialBarTop); 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(); } }