From b38b7a706268fe5d92d8cbe703a188b58ed7ec4d Mon Sep 17 00:00:00 2001 From: Luke Pulverenti Date: Mon, 13 Mar 2017 00:08:23 -0400 Subject: rework filestream --- .../Library/LibraryManager.cs | 27 ++++++++++++++-------- 1 file changed, 17 insertions(+), 10 deletions(-) (limited to 'Emby.Server.Implementations/Library') diff --git a/Emby.Server.Implementations/Library/LibraryManager.cs b/Emby.Server.Implementations/Library/LibraryManager.cs index f7706db47..026486efc 100644 --- a/Emby.Server.Implementations/Library/LibraryManager.cs +++ b/Emby.Server.Implementations/Library/LibraryManager.cs @@ -513,6 +513,11 @@ namespace Emby.Server.Implementations.Library } public Guid GetNewItemId(string key, Type type) + { + return GetNewItemIdInternal(key, type, false); + } + + private Guid GetNewItemIdInternal(string key, Type type, bool forceCaseInsensitive) { if (string.IsNullOrWhiteSpace(key)) { @@ -531,7 +536,7 @@ namespace Emby.Server.Implementations.Library .Replace("/", "\\"); } - if (!ConfigurationManager.Configuration.EnableCaseSensitiveItemIds) + if (forceCaseInsensitive || !ConfigurationManager.Configuration.EnableCaseSensitiveItemIds) { key = key.ToLower(); } @@ -865,7 +870,7 @@ namespace Emby.Server.Implementations.Library /// Task{Person}. public Person GetPerson(string name) { - return CreateItemByName(Person.GetPath(name), name); + return CreateItemByName(Person.GetPath, name); } /// @@ -875,7 +880,7 @@ namespace Emby.Server.Implementations.Library /// Task{Studio}. public Studio GetStudio(string name) { - return CreateItemByName(Studio.GetPath(name), name); + return CreateItemByName(Studio.GetPath, name); } /// @@ -885,7 +890,7 @@ namespace Emby.Server.Implementations.Library /// Task{Genre}. public Genre GetGenre(string name) { - return CreateItemByName(Genre.GetPath(name), name); + return CreateItemByName(Genre.GetPath, name); } /// @@ -895,7 +900,7 @@ namespace Emby.Server.Implementations.Library /// Task{MusicGenre}. public MusicGenre GetMusicGenre(string name) { - return CreateItemByName(MusicGenre.GetPath(name), name); + return CreateItemByName(MusicGenre.GetPath, name); } /// @@ -905,7 +910,7 @@ namespace Emby.Server.Implementations.Library /// Task{GameGenre}. public GameGenre GetGameGenre(string name) { - return CreateItemByName(GameGenre.GetPath(name), name); + return CreateItemByName(GameGenre.GetPath, name); } /// @@ -923,7 +928,7 @@ namespace Emby.Server.Implementations.Library var name = value.ToString(CultureInfo.InvariantCulture); - return CreateItemByName(Year.GetPath(name), name); + return CreateItemByName(Year.GetPath, name); } /// @@ -933,10 +938,10 @@ namespace Emby.Server.Implementations.Library /// Task{Genre}. public MusicArtist GetArtist(string name) { - return CreateItemByName(MusicArtist.GetPath(name), name); + return CreateItemByName(MusicArtist.GetPath, name); } - private T CreateItemByName(string path, string name) + private T CreateItemByName(Func getPathFn, string name) where T : BaseItem, new() { if (typeof(T) == typeof(MusicArtist)) @@ -957,7 +962,9 @@ namespace Emby.Server.Implementations.Library } } - var id = GetNewItemId(path, typeof(T)); + var path = getPathFn(name); + var forceCaseInsensitiveId = ConfigurationManager.Configuration.EnableNormalizedItemByNameIds; + var id = GetNewItemIdInternal(path, typeof(T), forceCaseInsensitiveId); var item = GetItemById(id) as T; -- cgit v1.2.3 From 7d3aa60db0800a6dd8a59dbdb9ce28e3ca06ba26 Mon Sep 17 00:00:00 2001 From: Luke Pulverenti Date: Mon, 27 Mar 2017 15:31:24 -0400 Subject: update hdhr udp stream --- .../Library/MediaSourceManager.cs | 2 + .../LiveTv/TunerHosts/HdHomerun/HdHomerunHost.cs | 2 +- .../TunerHosts/HdHomerun/HdHomerunUdpStream.cs | 125 ++++++++++++++++++++- .../LiveTv/TunerHosts/MulticastStream.cs | 37 ++++-- .../LiveTv/TunerHosts/QueueStream.cs | 39 ++++++- 5 files changed, 187 insertions(+), 18 deletions(-) (limited to 'Emby.Server.Implementations/Library') diff --git a/Emby.Server.Implementations/Library/MediaSourceManager.cs b/Emby.Server.Implementations/Library/MediaSourceManager.cs index c1bd8fe91..ccd4c3631 100644 --- a/Emby.Server.Implementations/Library/MediaSourceManager.cs +++ b/Emby.Server.Implementations/Library/MediaSourceManager.cs @@ -369,6 +369,8 @@ namespace Emby.Server.Implementations.Library { await _liveStreamSemaphore.WaitAsync(cancellationToken).ConfigureAwait(false); + enableAutoClose = false; + try { var tuple = GetProvider(request.OpenToken); diff --git a/Emby.Server.Implementations/LiveTv/TunerHosts/HdHomerun/HdHomerunHost.cs b/Emby.Server.Implementations/LiveTv/TunerHosts/HdHomerun/HdHomerunHost.cs index 2fac96169..8fa1bbe23 100644 --- a/Emby.Server.Implementations/LiveTv/TunerHosts/HdHomerun/HdHomerunHost.cs +++ b/Emby.Server.Implementations/LiveTv/TunerHosts/HdHomerun/HdHomerunHost.cs @@ -510,7 +510,7 @@ namespace Emby.Server.Implementations.LiveTv.TunerHosts.HdHomerun // The UDP method is not working reliably on OSX, and on BSD it hasn't been tested yet var enableHttpStream = _environment.OperatingSystem == OperatingSystem.OSX || _environment.OperatingSystem == OperatingSystem.BSD; - + enableHttpStream = true; if (enableHttpStream) { mediaSource.Protocol = MediaProtocol.Http; diff --git a/Emby.Server.Implementations/LiveTv/TunerHosts/HdHomerun/HdHomerunUdpStream.cs b/Emby.Server.Implementations/LiveTv/TunerHosts/HdHomerun/HdHomerunUdpStream.cs index 92000a1b3..e1572ea3f 100644 --- a/Emby.Server.Implementations/LiveTv/TunerHosts/HdHomerun/HdHomerunUdpStream.cs +++ b/Emby.Server.Implementations/LiveTv/TunerHosts/HdHomerun/HdHomerunUdpStream.cs @@ -130,7 +130,7 @@ namespace Emby.Server.Implementations.LiveTv.TunerHosts.HdHomerun onStarted = () => openTaskCompletionSource.TrySetResult(true); } - await _multicastStream.CopyUntilCancelled(udpClient, onStarted, cancellationToken).ConfigureAwait(false); + await _multicastStream.CopyUntilCancelled(new UdpClientStream(udpClient), onStarted, cancellationToken).ConfigureAwait(false); } } catch (OperationCanceledException ex) @@ -167,4 +167,127 @@ namespace Emby.Server.Implementations.LiveTv.TunerHosts.HdHomerun return _multicastStream.CopyToAsync(stream); } } + + // This handles the ReadAsync function only of a Stream object + // This is used to wrap a UDP socket into a stream for MulticastStream which only uses ReadAsync + 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 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; + + for (int i = 0; i < numReads; ++i) + { + var data = await _udpClient.ReceiveAsync(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; + } + + 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 int Read(byte[] buffer, int offset, int count) + { + 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(); + } + } } \ No newline at end of file diff --git a/Emby.Server.Implementations/LiveTv/TunerHosts/MulticastStream.cs b/Emby.Server.Implementations/LiveTv/TunerHosts/MulticastStream.cs index e3d0d1eba..281632590 100644 --- a/Emby.Server.Implementations/LiveTv/TunerHosts/MulticastStream.cs +++ b/Emby.Server.Implementations/LiveTv/TunerHosts/MulticastStream.cs @@ -35,13 +35,21 @@ namespace Emby.Server.Implementations.LiveTv.TunerHosts if (bytesRead > 0) { - byte[] copy = new byte[bytesRead]; - Buffer.BlockCopy(buffer, 0, copy, 0, bytesRead); - var allStreams = _outputStreams.ToList(); - foreach (var stream in allStreams) + + if (allStreams.Count == 1) { - stream.Value.Queue(copy, 0, copy.Length); + await allStreams[0].Value.WriteAsync(buffer, 0, bytesRead).ConfigureAwait(false); + } + else + { + byte[] copy = new byte[bytesRead]; + Buffer.BlockCopy(buffer, 0, copy, 0, bytesRead); + + foreach (var stream in allStreams) + { + stream.Value.Queue(copy, 0, copy.Length); + } } if (onStarted != null) @@ -79,14 +87,21 @@ namespace Emby.Server.Implementations.LiveTv.TunerHosts if (bytesRead > 0) { - byte[] copy = new byte[bytesRead]; - Buffer.BlockCopy(data.Buffer, RtpHeaderBytes, copy, 0, bytesRead); - var allStreams = _outputStreams.ToList(); - foreach (var stream in allStreams) + + if (allStreams.Count == 1) { - //stream.Value.Queue(data.Buffer, RtpHeaderBytes, bytesRead); - stream.Value.Queue(copy, 0, copy.Length); + await allStreams[0].Value.WriteAsync(data.Buffer, 0, bytesRead).ConfigureAwait(false); + } + else + { + byte[] copy = new byte[bytesRead]; + Buffer.BlockCopy(data.Buffer, RtpHeaderBytes, copy, 0, bytesRead); + + foreach (var stream in allStreams) + { + stream.Value.Queue(copy, 0, copy.Length); + } } if (onStarted != null) diff --git a/Emby.Server.Implementations/LiveTv/TunerHosts/QueueStream.cs b/Emby.Server.Implementations/LiveTv/TunerHosts/QueueStream.cs index 27dd288a7..543d2e373 100644 --- a/Emby.Server.Implementations/LiveTv/TunerHosts/QueueStream.cs +++ b/Emby.Server.Implementations/LiveTv/TunerHosts/QueueStream.cs @@ -13,7 +13,7 @@ namespace Emby.Server.Implementations.LiveTv.TunerHosts public class QueueStream { private readonly Stream _outputStream; - private readonly ConcurrentQueue> _queue = new ConcurrentQueue>(); + private readonly ConcurrentQueue> _queue = new ConcurrentQueue>(); private CancellationToken _cancellationToken; public TaskCompletionSource TaskCompletion { get; private set; } @@ -50,6 +50,38 @@ namespace Emby.Server.Implementations.LiveTv.TunerHosts return null; } + 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(); + } + } + private async Task StartInternal() { var cancellationToken = _cancellationToken; @@ -81,10 +113,7 @@ namespace Emby.Server.Implementations.LiveTv.TunerHosts } finally { - if (OnFinished != null) - { - OnFinished(this); - } + OnClosed(); } } } -- cgit v1.2.3