jellyfin/Emby.Server.Implementations/LiveTv/TunerHosts/QueueStream.cs

121 lines
3.6 KiB
C#
Raw Normal View History

2016-10-07 15:08:13 +00:00
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.IO;
using System.Linq;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
using MediaBrowser.Model.Logging;
2016-11-03 23:35:19 +00:00
namespace Emby.Server.Implementations.LiveTv.TunerHosts
2016-10-07 15:08:13 +00:00
{
public class QueueStream
{
private readonly Stream _outputStream;
2017-03-27 19:31:24 +00:00
private readonly ConcurrentQueue<Tuple<byte[], int, int>> _queue = new ConcurrentQueue<Tuple<byte[], int, int>>();
2016-10-07 15:08:13 +00:00
private CancellationToken _cancellationToken;
public TaskCompletionSource<bool> TaskCompletion { get; private set; }
public Action<QueueStream> OnFinished { get; set; }
private readonly ILogger _logger;
2017-02-01 20:55:56 +00:00
public Guid Id = Guid.NewGuid();
2016-10-07 15:08:13 +00:00
public QueueStream(Stream outputStream, ILogger logger)
{
_outputStream = outputStream;
_logger = logger;
TaskCompletion = new TaskCompletionSource<bool>();
}
2017-03-26 16:26:52 +00:00
public void Queue(byte[] bytes, int offset, int count)
2016-10-07 15:08:13 +00:00
{
2017-03-26 16:26:52 +00:00
_queue.Enqueue(new Tuple<byte[], int, int>(bytes, offset, count));
2016-10-07 15:08:13 +00:00
}
public void Start(CancellationToken cancellationToken)
{
_cancellationToken = cancellationToken;
2016-10-07 15:13:57 +00:00
Task.Run(() => StartInternal());
2016-10-07 15:08:13 +00:00
}
2017-03-26 16:26:52 +00:00
private Tuple<byte[], int, int> Dequeue()
2016-10-07 15:08:13 +00:00
{
2017-03-26 16:26:52 +00:00
Tuple<byte[], int, int> result;
if (_queue.TryDequeue(out result))
2016-10-07 15:08:13 +00:00
{
2017-03-26 16:26:52 +00:00
return result;
2016-10-07 15:08:13 +00:00
}
return null;
}
2017-03-27 19:31:24 +00:00
private void OnClosed()
{
GC.Collect();
if (OnFinished != null)
{
OnFinished(this);
}
}
public async Task WriteAsync(byte[] bytes, int offset, int count)
{
//return _outputStream.WriteAsync(bytes, offset, count, cancellationToken);
var cancellationToken = _cancellationToken;
try
{
await _outputStream.WriteAsync(bytes, offset, count, cancellationToken).ConfigureAwait(false);
}
catch (OperationCanceledException)
{
_logger.Debug("QueueStream cancelled");
TaskCompletion.TrySetCanceled();
OnClosed();
}
catch (Exception ex)
{
_logger.ErrorException("Error in QueueStream", ex);
TaskCompletion.TrySetException(ex);
OnClosed();
}
}
2016-10-07 15:08:13 +00:00
private async Task StartInternal()
{
var cancellationToken = _cancellationToken;
try
{
2017-02-01 20:55:56 +00:00
while (true)
2016-10-07 15:08:13 +00:00
{
2017-03-26 16:26:52 +00:00
var result = Dequeue();
if (result != null)
2016-10-07 15:08:13 +00:00
{
2017-03-26 16:26:52 +00:00
await _outputStream.WriteAsync(result.Item1, result.Item2, result.Item3, cancellationToken).ConfigureAwait(false);
2016-10-07 15:08:13 +00:00
}
else
{
await Task.Delay(50, cancellationToken).ConfigureAwait(false);
}
}
}
catch (OperationCanceledException)
{
_logger.Debug("QueueStream cancelled");
TaskCompletion.TrySetCanceled();
}
catch (Exception ex)
{
_logger.ErrorException("Error in QueueStream", ex);
TaskCompletion.TrySetException(ex);
}
finally
{
2017-03-27 19:31:24 +00:00
OnClosed();
2016-10-07 15:08:13 +00:00
}
}
}
}