using System.Threading.Channels; using Microsoft.Extensions.Logging; using DragonISO.Engine.Interop; namespace DragonISO.Engine.Pipeline; /// /// Pulls processed frames from a channel and forwards them to . /// public sealed class NdiSender : IDisposable { private readonly INdiInterop _interop; private readonly string _outputName; private readonly ChannelReader _input; private readonly ILogger _logger; private readonly NdiSenderHandle _handle; private long _framesSent; public NdiSender( INdiInterop interop, string outputName, ChannelReader input, ILogger logger, string? outputGroups = null, int frameRateN = 60000, int frameRateD = 1001) { _interop = interop; _outputName = outputName; _input = input; _logger = logger; _handle = interop.CreateSender(outputName, outputGroups, frameRateN, frameRateD); } public long FramesSent => Interlocked.Read(ref _framesSent); /// /// Awaits one frame and forwards it. Returns false if the channel is completed. /// Test seam. /// public async ValueTask SendNextAsync(CancellationToken cancellationToken) { if (!await _input.WaitToReadAsync(cancellationToken)) return false; if (!_input.TryRead(out var frame)) return false; _interop.SendFrame(_handle, frame); Interlocked.Increment(ref _framesSent); return true; } /// Long-running send loop. Run on a dedicated thread. public Task RunAsync(CancellationToken cancellationToken) => Task.Factory.StartNew(async () => { try { while (!cancellationToken.IsCancellationRequested) { var more = await SendNextAsync(cancellationToken); if (!more) break; } } catch (OperationCanceledException) { } catch (Exception ex) { _logger.LogError(ex, "NdiSender loop crashed for output {Output}.", _outputName); throw; } }, cancellationToken, TaskCreationOptions.LongRunning, TaskScheduler.Default).Unwrap(); public void Dispose() => _handle.Dispose(); }