diff --git a/Services/Encoder/EncoderOptions.cs b/Services/Encoder/EncoderOptions.cs index 1b08334..dedf2de 100644 --- a/Services/Encoder/EncoderOptions.cs +++ b/Services/Encoder/EncoderOptions.cs @@ -50,6 +50,11 @@ public sealed class EncoderOptions /// Name of the mixer's audio pipe (default ). public string AudioPipeName { get; init; } = DefaultAudioPipeName; + /// TASK 65: when set, the encoder muxes a SECOND video output into this ring (fragmented + /// MP4 on stdout) for the replay buffer. Null = clips off, and the ffmpeg command line is exactly + /// what it was before the feature existed. + public Services.ClipRing? ClipBuffer { get; init; } + /// GOP in frames = Fps × 4s — the YouTube keyframe ≤ 4s compliance bound. public int GopSize => Fps * 4; } diff --git a/Services/Encoder/FfmpegArgs.cs b/Services/Encoder/FfmpegArgs.cs index 6078c26..f5d79c3 100644 --- a/Services/Encoder/FfmpegArgs.cs +++ b/Services/Encoder/FfmpegArgs.cs @@ -52,6 +52,19 @@ public static class FfmpegArgs args.AddRange(["-f", "mp4", options.RecordPath!]); } + // TASK 65: the replay buffer is a third mux of the same encoded frames. Fragmented MP4 on + // stdout (stats already own stderr) — one fragment per second, so a clip can start anywhere + // inside the chosen window instead of at the last 4-second keyframe. + if (options.ClipBuffer != null) + { + AddVideoTags(args, options, videoEncoder); + args.AddRange(["-c:a", "aac", + "-ar", options.AudioSampleRate.ToString(), "-ac", options.AudioChannels.ToString()]); + args.AddRange(["-movflags", "+frag_keyframe+empty_moov+default_base_moof"]); + args.AddRange(["-frag_duration", "1000000"]); + args.AddRange(["-f", "mp4", "pipe:1"]); + } + return args; } diff --git a/Services/Encoder/FfmpegEncoder.cs b/Services/Encoder/FfmpegEncoder.cs index a5447ed..2c8ba1e 100644 --- a/Services/Encoder/FfmpegEncoder.cs +++ b/Services/Encoder/FfmpegEncoder.cs @@ -118,6 +118,11 @@ public sealed class FfmpegEncoder : IFfmpegEncoder _stopRequested = false; _stderrLoop = RunStderrLoopAsync(process); StartDrainLoop(process); + + // TASK 65: the clip mux is a side channel. It reads its own thread, never blocks the frame + // path, and a fault disables clips instead of touching the stream. + if (options.ClipBuffer is { } clipRing) + _ = Task.Run(() => ReadClipOutputAsync(process, clipRing)); } /// @@ -239,6 +244,24 @@ public sealed class FfmpegEncoder : IFfmpegEncoder /// ArrayPool may return a larger buffer), returned to the pool after the write /// copies into the pipe, and stdin is closed (EOF) once the queue drains. /// + private static async Task ReadClipOutputAsync(IEncoderProcess process, Services.ClipRing ring) + { + var splitter = new Fmp4FragmentSplitter(ring); + var buffer = new byte[64 * 1024]; + try + { + var stdout = process.StandardOutputBytes; + int read; + while ((read = await stdout.ReadAsync(buffer).ConfigureAwait(false)) > 0) + splitter.Write(buffer.AsSpan(0, read)); + } + catch (Exception ex) + { + // Clips stop; the broadcast does not. + AppLog.Write(ex, "Clip buffer reader failed — replay clips disabled for this session"); + } + } + private void StartDrainLoop(IEncoderProcess process) { _drainTask = Task.Run(async () => diff --git a/Services/Encoder/FfmpegEncoderProcess.cs b/Services/Encoder/FfmpegEncoderProcess.cs index 9f7172e..7de388a 100644 --- a/Services/Encoder/FfmpegEncoderProcess.cs +++ b/Services/Encoder/FfmpegEncoderProcess.cs @@ -21,6 +21,8 @@ public sealed class FfmpegEncoderProcess : IEncoderProcess } public Stream StandardInput => _process.StandardInput.BaseStream; + + public Stream StandardOutputBytes => _process.StandardOutput.BaseStream; public TextReader StandardOutput => _process.StandardOutput; public TextReader StandardError => _process.StandardError; diff --git a/Services/Encoder/Fmp4FragmentSplitter.cs b/Services/Encoder/Fmp4FragmentSplitter.cs new file mode 100644 index 0000000..1ea4aa6 --- /dev/null +++ b/Services/Encoder/Fmp4FragmentSplitter.cs @@ -0,0 +1,102 @@ +using System; +using System.Collections.Generic; +using System.IO; +using System.Text; + +namespace ytLive.Services.Encoder; + +/// +/// TASK 65: turns the fragmented-MP4 byte stream ffmpeg writes to stdout into a playable ring. +/// +/// MP4 boxes are length-prefixed, so this needs no muxer cooperation: ftyp+moov are +/// the initialisation segment, and each moof+mdat pair is one fragment. The ring only ever +/// sees whole fragments, which is what makes a snapshot playable and a trim safe. +/// +/// Deliberately forgiving: an unparseable header clears the buffer and waits for the next one. A +/// clip that fails must never take the encoder down with it — the stream is running through the same +/// process. +/// +public sealed class Fmp4FragmentSplitter +{ + private readonly ClipRing _ring; + private readonly List _pending = new(capacity: 64 * 1024); + private readonly MemoryStream _init = new(); + private readonly MemoryStream _fragment = new(); + private bool _inFragment; + + public Fmp4FragmentSplitter(ClipRing ring) => _ring = ring; + + public void Write(ReadOnlySpan data) + { + foreach (var b in data) _pending.Add(b); + Drain(); + } + + private void Drain() + { + while (true) + { + if (_pending.Count < 8) return; + + long size = ((long)_pending[0] << 24) | ((long)_pending[1] << 16) | ((long)_pending[2] << 8) | _pending[3]; + var type = Encoding.ASCII.GetString(_pending.GetRange(4, 4).ToArray()); + var header = 8; + + if (size == 1) + { + if (_pending.Count < 16) return; + size = 0; + for (var i = 8; i < 16; i++) size = (size << 8) | _pending[i]; + header = 16; + } + + if (size < header || size > int.MaxValue) + { + // Not a box boundary any more; drop what we have and resync on the next write. + _pending.Clear(); + return; + } + + if (_pending.Count < size) return; + + var box = _pending.GetRange(0, (int)size).ToArray(); + _pending.RemoveRange(0, (int)size); + Route(type, box); + } + } + + private void Route(string type, byte[] box) + { + switch (type) + { + case "ftyp": + _init.Write(box); + break; + case "moov": + _init.Write(box); + _ring.SetInit(_init.ToArray()); // init is complete once moov lands + break; + case "moof": + FlushFragment(); + _fragment.Write(box); + _inFragment = true; + break; + case "mdat": + if (!_inFragment) return; // mdat without a moof is not replayable + _fragment.Write(box); + FlushFragment(); + break; + default: + if (_inFragment) _fragment.Write(box); + break; + } + } + + private void FlushFragment() + { + if (_inFragment && _fragment.Length > 0) + _ring.Append(_fragment.ToArray()); + _fragment.SetLength(0); + _inFragment = false; + } +} diff --git a/Services/Encoder/IEncoderProcess.cs b/Services/Encoder/IEncoderProcess.cs index cd6e3e8..58a70ac 100644 --- a/Services/Encoder/IEncoderProcess.cs +++ b/Services/Encoder/IEncoderProcess.cs @@ -17,6 +17,10 @@ public interface IEncoderProcess : IDisposable Stream StandardInput { get; } TextReader StandardOutput { get; } + + /// TASK 65: the same stdout as raw bytes — the replay-buffer mux writes fragmented MP4 + /// here while -stats keeps stderr for itself. + Stream StandardOutputBytes { get; } TextReader StandardError { get; } bool HasExited { get; } diff --git a/ytLive.Tests/ClipStreamTests.cs b/ytLive.Tests/ClipStreamTests.cs new file mode 100644 index 0000000..bc5b726 --- /dev/null +++ b/ytLive.Tests/ClipStreamTests.cs @@ -0,0 +1,101 @@ +using System; +using System.Collections.Generic; +using System.Linq; +using System.Text; +using Xunit; +using ytLive.Services; +using ytLive.Services.Encoder; + +namespace ytLive.Tests; + +/// +/// TASK 65 slice 2 — the clip mux as a side channel. +/// +/// Two things are pinned here and both protect the broadcast: the command line gains the clip +/// output ONLY when clips are enabled (otherwise ffmpeg's args are byte-for-byte what they were), and the +/// byte splitter turns ffmpeg's fragmented MP4 into whole fragments without ever seeing the stream path. +/// +public sealed class ClipStreamTests +{ + private static byte[] Box(string type, int payloadBytes) + { + var payload = Enumerable.Repeat((byte)0x41, payloadBytes).ToArray(); + var box = new byte[8 + payloadBytes]; + var size = box.Length; + box[0] = (byte)(size >> 24); + box[1] = (byte)(size >> 16); + box[2] = (byte)(size >> 8); + box[3] = (byte)size; + Encoding.ASCII.GetBytes(type).CopyTo(box, 4); + payload.CopyTo(box, 8); + return box; + } + + [Fact] + public void TheClipOutputIsAppendedOnlyWhenClipsAreEnabled() + { + var plain = FfmpegArgs.Build(Options(clip: null), "libopenh264"); + var withClips = FfmpegArgs.Build(Options(clip: new ClipRing(1 << 20)), "libopenh264"); + + Assert.DoesNotContain("pipe:1", plain); + Assert.Contains("pipe:1", withClips); + + // The stream block must not move: everything up to the clip output is identical. + // The clip block is appended, so the original command line is a strict prefix of the new one. + Assert.Equal(plain, withClips.Take(plain.Count).ToList()); + } + + [Fact] + public void SplitterFeedsInitAndWholeFragments_EvenWhenBytesArriveInPieces() + { + var ring = new ClipRing(1 << 20); + var splitter = new Fmp4FragmentSplitter(ring); + + var stream = new List(); + stream.AddRange(Box("ftyp", 16)); + stream.AddRange(Box("moov", 32)); + stream.AddRange(Box("moof", 8)); + stream.AddRange(Box("mdat", 40)); + stream.AddRange(Box("moof", 8)); + stream.AddRange(Box("mdat", 40)); + + // Deliberately ragged writes: a box must survive being cut anywhere. + var bytes = stream.ToArray(); + var offset = 0; + foreach (var chunk in new[] { 3, 7, 11, 29, 1000 }) + { + var take = Math.Min(chunk, bytes.Length - offset); + if (take <= 0) break; + splitter.Write(bytes.AsSpan(offset, take)); + offset += take; + } + if (offset < bytes.Length) splitter.Write(bytes.AsSpan(offset)); + + Assert.Equal(2, ring.FragmentCount); + + var snapshot = ring.Snapshot(); + Assert.True(snapshot.Length > 0); + Assert.Equal("ftyp", Encoding.ASCII.GetString(snapshot, 4, 4)); // init comes first + } + + [Fact] + public void MdatWithoutAMoofIsNotReplayable_AndIsIgnored() + { + var ring = new ClipRing(1 << 20); + var splitter = new Fmp4FragmentSplitter(ring); + splitter.Write(Box("ftyp", 8)); + splitter.Write(Box("moov", 8)); + splitter.Write(Box("mdat", 32)); // no moof → not a fragment + + Assert.Empty(ring.Snapshot()); // init alone is not a clip + Assert.Equal(0, ring.FragmentCount); + } + + private static EncoderOptions Options(ClipRing? clip) => new() + { + RtmpUrl = "rtmp://a.rtmp.youtube.com/live2/key", + Width = 1920, Height = 1080, Fps = 30, BitrateKbps = 6000, + StreamEnabled = true, + ClipBuffer = clip, + }; +} diff --git a/ytLive.Tests/FfmpegEncoderTests.cs b/ytLive.Tests/FfmpegEncoderTests.cs index 388316d..a930131 100644 --- a/ytLive.Tests/FfmpegEncoderTests.cs +++ b/ytLive.Tests/FfmpegEncoderTests.cs @@ -42,6 +42,7 @@ public class FfmpegEncoderTests public Stream Stdin { get; } public Stream StandardInput => Stdin; public TextReader StandardOutput { get; } + public Stream StandardOutputBytes => Stream.Null; public bool HasExited { get; private set; } public int ExitCode { get; private set; } public bool Killed { get; private set; }