alteropen/jellyfin
1
0
Fork 0
mirror of https://github.com/jellyfin/jellyfin.git synced 2026-05-26 18:48:21 +01:00
Code Issues Actions 8 Packages Projects Releases Wiki Activity

update live stream management

Browse Source
This commit is contained in:
Luke Pulverenti
2017-08-19 18:37:15 -04:00
parent 1ad990ad72
commit b783f317fe
6 changed files with 25 additions and 99 deletions
Download Patch File Download Diff File Expand all files Collapse all files

62
Emby.Server.Implementations/LiveTv/TunerHosts/QueueStream.cs
View File

@@ -14,9 +14,7 @@ namespace Emby.Server.Implementations.LiveTv.TunerHosts
{
private readonly Stream _outputStream;
private readonly BlockingCollection<Tuple<byte[], int, int>> _queue = new BlockingCollection<Tuple<byte[], int, int>>();
public TaskCompletionSource<bool> TaskCompletion { get; private set; }
public Action<QueueStream> OnFinished { get; set; }
private readonly ILogger _logger;
public Guid Id = Guid.NewGuid();
@@ -24,7 +22,6 @@ namespace Emby.Server.Implementations.LiveTv.TunerHosts
{
_outputStream = outputStream;
_logger = logger;
TaskCompletion = new TaskCompletionSource<bool>();
}
public void Queue(byte[] bytes, int offset, int count)
@@ -34,68 +31,15 @@ namespace Emby.Server.Implementations.LiveTv.TunerHosts
public void Start(CancellationToken cancellationToken)
{
Task.Run(() => StartInternal(cancellationToken));
}
private void OnClosed()
{
GC.Collect();
if (OnFinished != null)
while (true)
{
OnFinished(this);
}
}
public void Write(byte[] bytes, int offset, int count)
{
//return _outputStream.WriteAsync(bytes, offset, count, cancellationToken);
try
{
_outputStream.Write(bytes, offset, count);
}
catch (OperationCanceledException)
{
_logger.Debug("QueueStream cancelled");
TaskCompletion.TrySetCanceled();
OnClosed();
}
catch (Exception ex)
{
_logger.ErrorException("Error in QueueStream", ex);
TaskCompletion.TrySetException(ex);
OnClosed();
}
}
private void StartInternal(CancellationToken cancellationToken)
{
try
{
while (true)
foreach (var result in _queue.GetConsumingEnumerable())
{
cancellationToken.ThrowIfCancellationRequested();
foreach (var result in _queue.GetConsumingEnumerable())
{
_outputStream.Write(result.Item1, result.Item2, result.Item3);
}
_outputStream.Write(result.Item1, result.Item2, result.Item3);
}
}
catch (OperationCanceledException)
{
_logger.Debug("QueueStream cancelled");
TaskCompletion.TrySetCanceled();
}
catch (Exception ex)
{
_logger.ErrorException("Error in QueueStream", ex);
TaskCompletion.TrySetException(ex);
}
finally
{
OnClosed();
}
}
}
}