lgladsden/jellyfin-server
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

update live stream buffers

Browse Source
This commit is contained in:
Luke Pulverenti 2017-06-01 02:25:07 -04:00
parent 0f1d253278
commit f96e5c84a2
6 changed files with 200 additions and 14 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
Emby.Common.Implementations/Net/UdpSocket.cs
View File

@ -128,6 +128,11 @@ namespace Emby.Common.Implementations.Net
return _Socket.BeginReceiveFrom(buffer, offset, count, SocketFlags.None, ref receivedFromEndPoint, callback, buffer);
}
public int Receive(byte[] buffer, int offset, int count)
{
return _Socket.Receive(buffer, 0, buffer.Length, SocketFlags.None);
}
public SocketReceiveResult EndReceive(IAsyncResult result)
{
IPEndPoint sender = new IPEndPoint(IPAddress.Any, 0);

21
Emby.Server.Implementations/LiveTv/TunerHosts/HdHomerun/HdHomerunHttpStream.cs
View File

@ -26,7 +26,10 @@ namespace Emby.Server.Implementations.LiveTv.TunerHosts.HdHomerun
private readonly CancellationTokenSource _liveStreamCancellationTokenSource = new CancellationTokenSource();
private readonly TaskCompletionSource<bool> _liveStreamTaskCompletionSource = new TaskCompletionSource<bool>();
private readonly MulticastStream _multicastStream;
private readonly string _tempFilePath;
private bool _enableFileBuffer = false;
public HdHomerunHttpStream(MediaSourceInfo mediaSource, string originalStreamId, IFileSystem fileSystem, IHttpClient httpClient, ILogger logger, IServerApplicationPaths appPaths, IServerApplicationHost appHost, IEnvironmentInfo environment)
: base(mediaSource, environment, fileSystem)
@ -36,6 +39,7 @@ namespace Emby.Server.Implementations.LiveTv.TunerHosts.HdHomerun
_appHost = appHost;
OriginalStreamId = originalStreamId;
_multicastStream = new MulticastStream(_logger);
_tempFilePath = Path.Combine(appPaths.TranscodingTempPath, UniqueId + ".ts");
}
@ -103,13 +107,17 @@ namespace Emby.Server.Implementations.LiveTv.TunerHosts.HdHomerun
{
_logger.Info("Beginning multicastStream.CopyUntilCancelled");
if (_enableFileBuffer)
{
FileSystem.CreateDirectory(FileSystem.GetDirectoryName(_tempFilePath));
using (var fileStream = FileSystem.GetFileStream(_tempFilePath, FileOpenMode.Create, FileAccessMode.Write, FileShareMode.Read, FileOpenOptions.None))
{
//await response.Content.CopyToAsync(fileStream, 81920, cancellationToken).ConfigureAwait(false);
StreamHelper.CopyTo(response.Content, fileStream, 81920, () => Resolve(openTaskCompletionSource), cancellationToken);
//await AsyncStreamCopier.CopyStream(response.Content, fileStream, 81920, 4, cancellationToken).ConfigureAwait(false);
}
}
else
{
await _multicastStream.CopyUntilCancelled(response.Content, () => Resolve(openTaskCompletionSource), cancellationToken).ConfigureAwait(false);
}
}
}
@ -134,7 +142,7 @@ namespace Emby.Server.Implementations.LiveTv.TunerHosts.HdHomerun
}
_liveStreamTaskCompletionSource.TrySetResult(true);
await DeleteTempFile(_tempFilePath).ConfigureAwait(false);
//await DeleteTempFile(_tempFilePath).ConfigureAwait(false);
});
}
@ -147,9 +155,14 @@ namespace Emby.Server.Implementations.LiveTv.TunerHosts.HdHomerun
}
public Task CopyToAsync(Stream stream, CancellationToken cancellationToken)
{
if (_enableFileBuffer)
{
return CopyFileTo(_tempFilePath, stream, cancellationToken);
}
return _multicastStream.CopyToAsync(stream, cancellationToken);
//return CopyFileTo(_tempFilePath, stream, cancellationToken);
}
protected async Task CopyFileTo(string path, Stream outputStream, CancellationToken cancellationToken)
{

166
Emby.Server.Implementations/LiveTv/TunerHosts/HdHomerun/HdHomerunUdpStream.cs
View File

@ -34,6 +34,8 @@ namespace Emby.Server.Implementations.LiveTv.TunerHosts.HdHomerun
private readonly INetworkManager _networkManager;
private readonly string _tempFilePath;
private bool _enableFileBuffer = false;
private readonly MulticastStream _multicastStream;
public HdHomerunUdpStream(MediaSourceInfo mediaSource, string originalStreamId, IHdHomerunChannelCommands channelCommands, int numTuners, IFileSystem fileSystem, IHttpClient httpClient, ILogger logger, IServerApplicationPaths appPaths, IServerApplicationHost appHost, ISocketFactory socketFactory, INetworkManager networkManager, IEnvironmentInfo environment)
: base(mediaSource, environment, fileSystem)
@ -46,6 +48,7 @@ namespace Emby.Server.Implementations.LiveTv.TunerHosts.HdHomerun
_channelCommands = channelCommands;
_numTuners = numTuners;
_tempFilePath = Path.Combine(appPaths.TranscodingTempPath, UniqueId + ".ts");
_multicastStream = new MulticastStream(_logger);
}
protected override async Task OpenInternal(CancellationToken openCancellationToken)
@ -122,6 +125,8 @@ namespace Emby.Server.Implementations.LiveTv.TunerHosts.HdHomerun
_logger.Info("Opened HDHR UDP stream from {0}", remoteAddress);
if (!cancellationToken.IsCancellationRequested)
{
if (_enableFileBuffer)
{
FileSystem.CreateDirectory(FileSystem.GetDirectoryName(_tempFilePath));
using (var fileStream = FileSystem.GetFileStream(_tempFilePath, FileOpenMode.Create, FileAccessMode.Write, FileShareMode.Read, FileOpenOptions.None))
@ -129,6 +134,11 @@ namespace Emby.Server.Implementations.LiveTv.TunerHosts.HdHomerun
CopyTo(udpClient, fileStream, openTaskCompletionSource, cancellationToken);
}
}
else
{
await _multicastStream.CopyUntilCancelled(new UdpClientStream(udpClient), () => Resolve(openTaskCompletionSource), cancellationToken).ConfigureAwait(false);
}
}
}
catch (OperationCanceledException ex)
{
@ -170,6 +180,12 @@ namespace Emby.Server.Implementations.LiveTv.TunerHosts.HdHomerun
public async Task CopyToAsync(Stream outputStream, CancellationToken cancellationToken)
{
if (!_enableFileBuffer)
{
await _multicastStream.CopyToAsync(outputStream, cancellationToken).ConfigureAwait(false);
return;
}
var path = _tempFilePath;
long startPosition = -20000;
@ -285,5 +301,155 @@ namespace Emby.Server.Implementations.LiveTv.TunerHosts.HdHomerun
}
}
public class UdpClientStream : Stream
{
private static int RtpHeaderBytes = 12;
private static int PacketSize = 1316;
private readonly ISocket _udpClient;
bool disposed;
public UdpClientStream(ISocket udpClient) : base()
{
_udpClient = udpClient;
}
public override async Task<int> ReadAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
{
if (buffer == null)
throw new ArgumentNullException("buffer");
if (offset + count < 0)
throw new ArgumentOutOfRangeException("offset + count must not be negative", "offset+count");
if (offset + count > buffer.Length)
throw new ArgumentException("offset + count must not be greater than the length of buffer", "offset+count");
if (disposed)
throw new ObjectDisposedException(typeof(UdpClientStream).ToString());
// This will always receive a 1328 packet size (PacketSize + RtpHeaderSize)
// The RTP header will be stripped so see how many reads we need to make to fill the buffer.
int numReads = count / PacketSize;
int totalBytesRead = 0;
byte[] receiveBuffer = new byte[81920];
for (int i = 0; i < numReads; ++i)
{
var data = await _udpClient.ReceiveAsync(receiveBuffer, 0, receiveBuffer.Length, cancellationToken).ConfigureAwait(false);
var bytesRead = data.ReceivedBytes - RtpHeaderBytes;
// remove rtp header
Buffer.BlockCopy(data.Buffer, RtpHeaderBytes, buffer, offset, bytesRead);
offset += bytesRead;
totalBytesRead += bytesRead;
}
return totalBytesRead;
}
public override int Read(byte[] buffer, int offset, int count)
{
if (buffer == null)
throw new ArgumentNullException("buffer");
if (offset + count < 0)
throw new ArgumentOutOfRangeException("offset + count must not be negative", "offset+count");
if (offset + count > buffer.Length)
throw new ArgumentException("offset + count must not be greater than the length of buffer", "offset+count");
if (disposed)
throw new ObjectDisposedException(typeof(UdpClientStream).ToString());
// This will always receive a 1328 packet size (PacketSize + RtpHeaderSize)
// The RTP header will be stripped so see how many reads we need to make to fill the buffer.
int numReads = count / PacketSize;
int totalBytesRead = 0;
byte[] receiveBuffer = new byte[81920];
for (int i = 0; i < numReads; ++i)
{
var receivedBytes = _udpClient.Receive(receiveBuffer, 0, receiveBuffer.Length);
var bytesRead = receivedBytes - RtpHeaderBytes;
// remove rtp header
Buffer.BlockCopy(receiveBuffer, RtpHeaderBytes, buffer, offset, bytesRead);
offset += bytesRead;
totalBytesRead += bytesRead;
}
return totalBytesRead;
}
protected override void Dispose(bool disposing)
{
disposed = true;
}
public override bool CanRead
{
get
{
throw new NotImplementedException();
}
}
public override bool CanSeek
{
get
{
throw new NotImplementedException();
}
}
public override bool CanWrite
{
get
{
throw new NotImplementedException();
}
}
public override long Length
{
get
{
throw new NotImplementedException();
}
}
public override long Position
{
get
{
throw new NotImplementedException();
}
set
{
throw new NotImplementedException();
}
}
public override void Flush()
{
throw new NotImplementedException();
}
public override long Seek(long offset, SeekOrigin origin)
{
throw new NotImplementedException();
}
public override void SetLength(long value)
{
throw new NotImplementedException();
}
public override void Write(byte[] buffer, int offset, int count)
{
throw new NotImplementedException();
}
}
}
}

2
Emby.Server.Implementations/LiveTv/TunerHosts/MulticastStream.cs
View File

@ -43,7 +43,7 @@ namespace Emby.Server.Implementations.LiveTv.TunerHosts
//if (allStreams.Count == 1)
//{
// await allStreams[0].Value.WriteAsync(buffer, 0, bytesRead).ConfigureAwait(false);
// allStreams[0].Value.Write(buffer, 0, bytesRead);
//}
//else
{

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

@ -57,13 +57,13 @@ namespace Emby.Server.Implementations.LiveTv.TunerHosts
}
}
public async Task WriteAsync(byte[] bytes, int offset, int count, CancellationToken cancellationToken)
public void Write(byte[] bytes, int offset, int count)
{
//return _outputStream.WriteAsync(bytes, offset, count, cancellationToken);
try
{
await _outputStream.WriteAsync(bytes, offset, count, cancellationToken).ConfigureAwait(false);
_outputStream.Write(bytes, offset, count);
}
catch (OperationCanceledException)
{

2
MediaBrowser.Model/Net/ISocket.cs
View File

@ -16,6 +16,8 @@ namespace MediaBrowser.Model.Net
Task<SocketReceiveResult> ReceiveAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken);
int Receive(byte[] buffer, int offset, int count);
IAsyncResult BeginReceive(byte[] buffer, int offset, int count, AsyncCallback callback);
SocketReceiveResult EndReceive(IAsyncResult result);