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)
{
_interop = interop;
_outputName = outputName;
_input = input;
_logger = logger;
_handle = interop.CreateSender(outputName, outputGroups);
}
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();
}