diff options
17 files changed, 308 insertions, 393 deletions
diff --git a/Emby.Server.Implementations/Library/ExclusiveLiveStream.cs b/Emby.Server.Implementations/Library/ExclusiveLiveStream.cs index 6c65b5899..868071a99 100644 --- a/Emby.Server.Implementations/Library/ExclusiveLiveStream.cs +++ b/Emby.Server.Implementations/Library/ExclusiveLiveStream.cs @@ -4,6 +4,7 @@ using System; using System.Globalization; +using System.IO; using System.Threading; using System.Threading.Tasks; using MediaBrowser.Controller.Library; @@ -41,6 +42,11 @@ namespace Emby.Server.Implementations.Library return _closeFn(); } + public Stream GetStream() + { + throw new NotSupportedException(); + } + public Task Open(CancellationToken openCancellationToken) { return Task.CompletedTask; diff --git a/Emby.Server.Implementations/Library/MediaSourceManager.cs b/Emby.Server.Implementations/Library/MediaSourceManager.cs index 6f83973ba..16231c73f 100644 --- a/Emby.Server.Implementations/Library/MediaSourceManager.cs +++ b/Emby.Server.Implementations/Library/MediaSourceManager.cs @@ -587,13 +587,6 @@ namespace Emby.Server.Implementations.Library mediaSource.InferTotalBitrate(); } - public Task<IDirectStreamProvider> GetDirectStreamProviderByUniqueId(string uniqueId, CancellationToken cancellationToken) - { - var info = _openStreams.FirstOrDefault(i => i.Value != null && string.Equals(i.Value.UniqueId, uniqueId, StringComparison.OrdinalIgnoreCase)); - - return Task.FromResult(info.Value as IDirectStreamProvider); - } - public async Task<LiveStreamResponse> OpenLiveStream(LiveStreamRequest request, CancellationToken cancellationToken) { var result = await OpenLiveStreamInternal(request, cancellationToken).ConfigureAwait(false); @@ -602,7 +595,8 @@ namespace Emby.Server.Implementations.Library public async Task<MediaSourceInfo> GetLiveStreamMediaInfo(string id, CancellationToken cancellationToken) { - var liveStreamInfo = await GetLiveStreamInfo(id, cancellationToken).ConfigureAwait(false); + // TODO probably shouldn't throw here but it is kept for "backwards compatibility" + var liveStreamInfo = GetLiveStreamInfo(id) ?? throw new ResourceNotFoundException(); var mediaSource = liveStreamInfo.MediaSource; @@ -771,18 +765,19 @@ namespace Emby.Server.Implementations.Library mediaSource.InferTotalBitrate(true); } - public async Task<Tuple<MediaSourceInfo, IDirectStreamProvider>> GetLiveStreamWithDirectStreamProvider(string id, CancellationToken cancellationToken) + public Task<Tuple<MediaSourceInfo, IDirectStreamProvider>> GetLiveStreamWithDirectStreamProvider(string id, CancellationToken cancellationToken) { if (string.IsNullOrEmpty(id)) { throw new ArgumentNullException(nameof(id)); } - var info = await GetLiveStreamInfo(id, cancellationToken).ConfigureAwait(false); - return new Tuple<MediaSourceInfo, IDirectStreamProvider>(info.MediaSource, info as IDirectStreamProvider); + // TODO probably shouldn't throw here but it is kept for "backwards compatibility" + var info = GetLiveStreamInfo(id) ?? throw new ResourceNotFoundException(); + return Task.FromResult(new Tuple<MediaSourceInfo, IDirectStreamProvider>(info.MediaSource, info as IDirectStreamProvider)); } - private Task<ILiveStream> GetLiveStreamInfo(string id, CancellationToken cancellationToken) + public ILiveStream GetLiveStreamInfo(string id) { if (string.IsNullOrEmpty(id)) { @@ -791,12 +786,16 @@ namespace Emby.Server.Implementations.Library if (_openStreams.TryGetValue(id, out ILiveStream info)) { - return Task.FromResult(info); - } - else - { - return Task.FromException<ILiveStream>(new ResourceNotFoundException()); + return info; } + + return null; + } + + /// <inheritdoc /> + public ILiveStream GetLiveStreamInfoByUniqueId(string uniqueId) + { + return _openStreams.Values.FirstOrDefault(stream => string.Equals(uniqueId, stream?.UniqueId, StringComparison.OrdinalIgnoreCase)); } public async Task<MediaSourceInfo> GetLiveStream(string id, CancellationToken cancellationToken) diff --git a/Emby.Server.Implementations/LiveTv/EmbyTV/DirectRecorder.cs b/Emby.Server.Implementations/LiveTv/EmbyTV/DirectRecorder.cs index c5a9a92ec..f6e0111b6 100644 --- a/Emby.Server.Implementations/LiveTv/EmbyTV/DirectRecorder.cs +++ b/Emby.Server.Implementations/LiveTv/EmbyTV/DirectRecorder.cs @@ -5,6 +5,7 @@ using System.IO; using System.Net.Http; using System.Threading; using System.Threading.Tasks; +using Jellyfin.Api.Helpers; using MediaBrowser.Common.Net; using MediaBrowser.Controller.Library; using MediaBrowser.Model.Dto; @@ -50,16 +51,23 @@ namespace Emby.Server.Implementations.LiveTv.EmbyTV { onStarted(); - _logger.LogInformation("Copying recording stream to file {0}", targetFile); + _logger.LogInformation("Copying recording to file {FilePath}", targetFile); // The media source is infinite so we need to handle stopping ourselves using var durationToken = new CancellationTokenSource(duration); using var cancellationTokenSource = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken, durationToken.Token); - - await directStreamProvider.CopyToAsync(output, cancellationTokenSource.Token).ConfigureAwait(false); + var linkedCancellationToken = cancellationTokenSource.Token; + + await using var fileStream = new ProgressiveFileStream(directStreamProvider.GetStream()); + await _streamHelper.CopyToAsync( + fileStream, + output, + IODefaults.CopyToBufferSize, + 1000, + linkedCancellationToken).ConfigureAwait(false); } - _logger.LogInformation("Recording completed to file {0}", targetFile); + _logger.LogInformation("Recording completed: {FilePath}", targetFile); } private async Task RecordFromMediaSource(MediaSourceInfo mediaSource, string targetFile, TimeSpan duration, Action onStarted, CancellationToken cancellationToken) diff --git a/Emby.Server.Implementations/LiveTv/TunerHosts/HdHomerun/HdHomerunUdpStream.cs b/Emby.Server.Implementations/LiveTv/TunerHosts/HdHomerun/HdHomerunUdpStream.cs index 58e0c7448..a8f761fde 100644 --- a/Emby.Server.Implementations/LiveTv/TunerHosts/HdHomerun/HdHomerunUdpStream.cs +++ b/Emby.Server.Implementations/LiveTv/TunerHosts/HdHomerun/HdHomerunUdpStream.cs @@ -156,11 +156,6 @@ namespace Emby.Server.Implementations.LiveTv.TunerHosts.HdHomerun await taskCompletionSource.Task.ConfigureAwait(false); } - public string GetFilePath() - { - return TempFilePath; - } - private async Task StartStreaming(UdpClient udpClient, HdHomerunManager hdHomerunManager, IPAddress remoteAddress, TaskCompletionSource<bool> openTaskCompletionSource, CancellationToken cancellationToken) { using (udpClient) @@ -184,7 +179,7 @@ namespace Emby.Server.Implementations.LiveTv.TunerHosts.HdHomerun EnableStreamSharing = false; } - await DeleteTempFiles(new List<string> { TempFilePath }).ConfigureAwait(false); + await DeleteTempFiles(TempFilePath).ConfigureAwait(false); } private async Task CopyTo(UdpClient udpClient, string file, TaskCompletionSource<bool> openTaskCompletionSource, CancellationToken cancellationToken) diff --git a/Emby.Server.Implementations/LiveTv/TunerHosts/LiveStream.cs b/Emby.Server.Implementations/LiveTv/TunerHosts/LiveStream.cs index 2c21a4a89..817b1f804 100644 --- a/Emby.Server.Implementations/LiveTv/TunerHosts/LiveStream.cs +++ b/Emby.Server.Implementations/LiveTv/TunerHosts/LiveStream.cs @@ -3,10 +3,8 @@ #pragma warning disable CS1591 using System; -using System.Collections.Generic; using System.Globalization; using System.IO; -using System.Linq; using System.Threading; using System.Threading.Tasks; using MediaBrowser.Common.Configuration; @@ -97,6 +95,18 @@ namespace Emby.Server.Implementations.LiveTv.TunerHosts return Task.CompletedTask; } + public Stream GetStream() + { + var stream = GetInputStream(TempFilePath, AsyncFile.UseAsyncIO); + bool seekFile = (DateTime.UtcNow - DateOpened).TotalSeconds > 10; + if (seekFile) + { + TrySeek(stream, -20000); + } + + return stream; + } + protected FileStream GetInputStream(string path, bool allowAsyncFileRead) => new FileStream( path, @@ -106,112 +116,29 @@ namespace Emby.Server.Implementations.LiveTv.TunerHosts IODefaults.FileStreamBufferSize, allowAsyncFileRead ? FileOptions.SequentialScan | FileOptions.Asynchronous : FileOptions.SequentialScan); - public Task DeleteTempFiles() - { - return DeleteTempFiles(GetStreamFilePaths()); - } - - protected async Task DeleteTempFiles(IEnumerable<string> paths, int retryCount = 0) + protected async Task DeleteTempFiles(string path, int retryCount = 0) { if (retryCount == 0) { - Logger.LogInformation("Deleting temp files {0}", paths); - } - - var failedFiles = new List<string>(); - - foreach (var path in paths) - { - if (!File.Exists(path)) - { - continue; - } - - try - { - FileSystem.DeleteFile(path); - } - catch (Exception ex) - { - Logger.LogError(ex, "Error deleting file {path}", path); - failedFiles.Add(path); - } - } - - if (failedFiles.Count > 0 && retryCount <= 40) - { - await Task.Delay(500).ConfigureAwait(false); - await DeleteTempFiles(failedFiles, retryCount + 1).ConfigureAwait(false); + Logger.LogInformation("Deleting temp file {FilePath}", path); } - } - protected virtual List<string> GetStreamFilePaths() - { - return new List<string> { TempFilePath }; - } - - public async Task CopyToAsync(Stream stream, CancellationToken cancellationToken) - { - using var linkedCancellationTokenSource = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken, LiveStreamCancellationTokenSource.Token); - cancellationToken = linkedCancellationTokenSource.Token; - - bool seekFile = (DateTime.UtcNow - DateOpened).TotalSeconds > 10; - - var nextFileInfo = GetNextFile(null); - var nextFile = nextFileInfo.file; - var isLastFile = nextFileInfo.isLastFile; - - var allowAsync = AsyncFile.UseAsyncIO; - while (!string.IsNullOrEmpty(nextFile)) - { - var emptyReadLimit = isLastFile ? EmptyReadLimit : 1; - - await CopyFile(nextFile, seekFile, emptyReadLimit, allowAsync, stream, cancellationToken).ConfigureAwait(false); - - seekFile = false; - nextFileInfo = GetNextFile(nextFile); - nextFile = nextFileInfo.file; - isLastFile = nextFileInfo.isLastFile; - } - - Logger.LogInformation("Live Stream ended."); - } - - private (string file, bool isLastFile) GetNextFile(string currentFile) - { - var files = GetStreamFilePaths(); - - if (string.IsNullOrEmpty(currentFile)) + try { - return (files[^1], true); + FileSystem.DeleteFile(path); } - - var nextIndex = files.FindIndex(i => string.Equals(i, currentFile, StringComparison.OrdinalIgnoreCase)) + 1; - - var isLastFile = nextIndex == files.Count - 1; - - return (files.ElementAtOrDefault(nextIndex), isLastFile); - } - - private async Task CopyFile(string path, bool seekFile, int emptyReadLimit, bool allowAsync, Stream stream, CancellationToken cancellationToken) - { - using (var inputStream = GetInputStream(path, allowAsync)) + catch (Exception ex) { - if (seekFile) + Logger.LogError(ex, "Error deleting file {FilePath}", path); + if (retryCount <= 40) { - TrySeek(inputStream, -20000); + await Task.Delay(500).ConfigureAwait(false); + await DeleteTempFiles(path, retryCount + 1).ConfigureAwait(false); } - - await StreamHelper.CopyToAsync( - inputStream, - stream, - IODefaults.CopyToBufferSize, - emptyReadLimit, - cancellationToken).ConfigureAwait(false); } } - private void TrySeek(FileStream stream, long offset) + private void TrySeek(Stream stream, long offset) { if (!stream.CanSeek) { diff --git a/Emby.Server.Implementations/LiveTv/TunerHosts/SharedHttpStream.cs b/Emby.Server.Implementations/LiveTv/TunerHosts/SharedHttpStream.cs index 862993877..84a878e51 100644 --- a/Emby.Server.Implementations/LiveTv/TunerHosts/SharedHttpStream.cs +++ b/Emby.Server.Implementations/LiveTv/TunerHosts/SharedHttpStream.cs @@ -3,7 +3,6 @@ #pragma warning disable CS1591 using System; -using System.Collections.Generic; using System.Globalization; using System.IO; using System.Net.Http; @@ -55,39 +54,26 @@ namespace Emby.Server.Implementations.LiveTv.TunerHosts Directory.CreateDirectory(Path.GetDirectoryName(TempFilePath)); var typeName = GetType().Name; - Logger.LogInformation("Opening " + typeName + " Live stream from {0}", url); + Logger.LogInformation("Opening {StreamType} Live stream from {Url}", typeName, url); // Response stream is disposed manually. var response = await _httpClientFactory.CreateClient(NamedClient.Default) .GetAsync(url, HttpCompletionOption.ResponseHeadersRead, CancellationToken.None) .ConfigureAwait(false); - var extension = "ts"; - var requiresRemux = false; - var contentType = response.Content.Headers.ContentType?.ToString() ?? string.Empty; - if (contentType.IndexOf("matroska", StringComparison.OrdinalIgnoreCase) != -1) - { - requiresRemux = true; - } - else if (contentType.IndexOf("mp4", StringComparison.OrdinalIgnoreCase) != -1 || - contentType.IndexOf("dash", StringComparison.OrdinalIgnoreCase) != -1 || - contentType.IndexOf("mpegURL", StringComparison.OrdinalIgnoreCase) != -1 || - contentType.IndexOf("text/", StringComparison.OrdinalIgnoreCase) != -1) - { - requiresRemux = true; - } - - // Close the stream without any sharing features - if (requiresRemux) + if (contentType.Contains("matroska", StringComparison.OrdinalIgnoreCase) + || contentType.Contains("mp4", StringComparison.OrdinalIgnoreCase) + || contentType.Contains("dash", StringComparison.OrdinalIgnoreCase) + || contentType.Contains("mpegURL", StringComparison.OrdinalIgnoreCase) + || contentType.Contains("text/", StringComparison.OrdinalIgnoreCase)) { - using (response) - { - return; - } + // Close the stream without any sharing features + response.Dispose(); + return; } - SetTempFilePath(extension); + SetTempFilePath("ts"); var taskCompletionSource = new TaskCompletionSource<bool>(); @@ -117,16 +103,11 @@ namespace Emby.Server.Implementations.LiveTv.TunerHosts if (!taskCompletionSource.Task.Result) { - Logger.LogWarning("Zero bytes copied from stream {0} to {1} but no exception raised", GetType().Name, TempFilePath); + Logger.LogWarning("Zero bytes copied from stream {StreamType} to {FilePath} but no exception raised", GetType().Name, TempFilePath); throw new EndOfStreamException(string.Format(CultureInfo.InvariantCulture, "Zero bytes copied from stream {0}", GetType().Name)); } } - public string GetFilePath() - { - return TempFilePath; - } - private Task StartStreaming(HttpResponseMessage response, TaskCompletionSource<bool> openTaskCompletionSource, CancellationToken cancellationToken) { return Task.Run( @@ -134,7 +115,7 @@ namespace Emby.Server.Implementations.LiveTv.TunerHosts { try { - Logger.LogInformation("Beginning {0} stream to {1}", GetType().Name, TempFilePath); + Logger.LogInformation("Beginning {StreamType} stream to {FilePath}", GetType().Name, TempFilePath); using var message = response; await using var stream = await response.Content.ReadAsStreamAsync(cancellationToken).ConfigureAwait(false); await using var fileStream = new FileStream(TempFilePath, FileMode.Create, FileAccess.Write, FileShare.Read, IODefaults.FileStreamBufferSize, AsyncFile.UseAsyncIO); @@ -147,19 +128,19 @@ namespace Emby.Server.Implementations.LiveTv.TunerHosts } catch (OperationCanceledException ex) { - Logger.LogInformation("Copying of {0} to {1} was canceled", GetType().Name, TempFilePath); + Logger.LogInformation("Copying of {StreamType} to {FilePath} was canceled", GetType().Name, TempFilePath); openTaskCompletionSource.TrySetException(ex); } catch (Exception ex) { - Logger.LogError(ex, "Error copying live stream {0} to {1}.", GetType().Name, TempFilePath); + Logger.LogError(ex, "Error copying live stream {StreamType} to {FilePath}", GetType().Name, TempFilePath); openTaskCompletionSource.TrySetException(ex); } openTaskCompletionSource.TrySetResult(false); EnableStreamSharing = false; - await DeleteTempFiles(new List<string> { TempFilePath }).ConfigureAwait(false); + await DeleteTempFiles(TempFilePath).ConfigureAwait(false); }, CancellationToken.None); } diff --git a/Jellyfin.Api/Controllers/LiveTvController.cs b/Jellyfin.Api/Controllers/LiveTvController.cs index 93dc76729..b131530c9 100644 --- a/Jellyfin.Api/Controllers/LiveTvController.cs +++ b/Jellyfin.Api/Controllers/LiveTvController.cs @@ -1199,15 +1199,15 @@ namespace Jellyfin.Api.Controllers [ProducesResponseType(StatusCodes.Status200OK)] [ProducesResponseType(StatusCodes.Status404NotFound)] [ProducesVideoFile] - public async Task<ActionResult> GetLiveStreamFile([FromRoute, Required] string streamId, [FromRoute, Required] string container) + public ActionResult GetLiveStreamFile([FromRoute, Required] string streamId, [FromRoute, Required] string container) { - var liveStreamInfo = await _mediaSourceManager.GetDirectStreamProviderByUniqueId(streamId, CancellationToken.None).ConfigureAwait(false); + var liveStreamInfo = _mediaSourceManager.GetLiveStreamInfoByUniqueId(streamId); if (liveStreamInfo == null) { return NotFound(); } - var liveStream = new ProgressiveFileStream(liveStreamInfo.GetFilePath(), null, _transcodingJobHelper); + var liveStream = new ProgressiveFileStream(liveStreamInfo.GetStream()); return new FileStreamResult(liveStream, MimeTypes.GetMimeType("file." + container)); } diff --git a/Jellyfin.Api/Controllers/VideosController.cs b/Jellyfin.Api/Controllers/VideosController.cs index bc6fc904a..150f22d1b 100644 --- a/Jellyfin.Api/Controllers/VideosController.cs +++ b/Jellyfin.Api/Controllers/VideosController.cs @@ -453,14 +453,15 @@ namespace Jellyfin.Api.Controllers { StreamingHelpers.AddDlnaHeaders(state, Response.Headers, true, startTimeTicks, Request, _dlnaManager); - await new ProgressiveFileCopier(state.DirectStreamProvider, null, _transcodingJobHelper, CancellationToken.None) + var liveStreamInfo = _mediaSourceManager.GetLiveStreamInfo(streamingRequest.LiveStreamId); + if (liveStreamInfo == null) { - AllowEndOfFile = false - }.WriteToAsync(Response.Body, CancellationToken.None) - .ConfigureAwait(false); + return NotFound(); + } + var liveStream = new ProgressiveFileStream(liveStreamInfo.GetStream()); // TODO (moved from MediaBrowser.Api): Don't hardcode contentType - return File(Response.Body, MimeTypes.GetMimeType("file.ts")!); + return File(liveStream, MimeTypes.GetMimeType("file.ts")!); } // Static remote stream @@ -492,13 +493,8 @@ namespace Jellyfin.Api.Controllers if (state.MediaSource.IsInfiniteStream) { - await new ProgressiveFileCopier(state.MediaPath, null, _transcodingJobHelper, CancellationToken.None) - { - AllowEndOfFile = false - }.WriteToAsync(Response.Body, CancellationToken.None) - .ConfigureAwait(false); - - return File(Response.Body, contentType); + var liveStream = new ProgressiveFileStream(state.MediaPath, null, _transcodingJobHelper); + return File(liveStream, contentType); } return FileStreamResponseHelpers.GetStaticFileResult( diff --git a/Jellyfin.Api/Helpers/AudioHelper.cs b/Jellyfin.Api/Helpers/AudioHelper.cs index ddcde1cf6..a5e47b8ec 100644 --- a/Jellyfin.Api/Helpers/AudioHelper.cs +++ b/Jellyfin.Api/Helpers/AudioHelper.cs @@ -1,4 +1,5 @@ -using System.Net.Http; +using System.IO; +using System.Net.Http; using System.Threading; using System.Threading.Tasks; using Jellyfin.Api.Models.StreamingDtos; @@ -120,14 +121,15 @@ namespace Jellyfin.Api.Helpers { StreamingHelpers.AddDlnaHeaders(state, _httpContextAccessor.HttpContext.Response.Headers, true, streamingRequest.StartTimeTicks, _httpContextAccessor.HttpContext.Request, _dlnaManager); - await new ProgressiveFileCopier(state.DirectStreamProvider, null, _transcodingJobHelper, CancellationToken.None) - { - AllowEndOfFile = false - }.WriteToAsync(_httpContextAccessor.HttpContext.Response.Body, CancellationToken.None) - .ConfigureAwait(false); + var liveStreamInfo = _mediaSourceManager.GetLiveStreamInfo(streamingRequest.LiveStreamId); + if (liveStreamInfo == null) + { + throw new FileNotFoundException(); + } + var liveStream = new ProgressiveFileStream(liveStreamInfo.GetStream()); // TODO (moved from MediaBrowser.Api): Don't hardcode contentType - return new FileStreamResult(_httpContextAccessor.HttpContext.Response.Body, MimeTypes.GetMimeType("file.ts")!); + return new FileStreamResult(liveStream, MimeTypes.GetMimeType("file.ts")); } // Static remote stream @@ -159,13 +161,8 @@ namespace Jellyfin.Api.Helpers if (state.MediaSource.IsInfiniteStream) { - await new ProgressiveFileCopier(state.MediaPath, null, _transcodingJobHelper, CancellationToken.None) - { - AllowEndOfFile = false - }.WriteToAsync(_httpContextAccessor.HttpContext.Response.Body, CancellationToken.None) - .ConfigureAwait(false); - - return new FileStreamResult(_httpContextAccessor.HttpContext.Response.Body, contentType); + var stream = new ProgressiveFileStream(state.MediaPath, null, _transcodingJobHelper); + return new FileStreamResult(stream, contentType); } return FileStreamResponseHelpers.GetStaticFileResult( diff --git a/Jellyfin.Api/Helpers/ProgressiveFileCopier.cs b/Jellyfin.Api/Helpers/ProgressiveFileCopier.cs deleted file mode 100644 index 81970b041..000000000 --- a/Jellyfin.Api/Helpers/ProgressiveFileCopier.cs +++ /dev/null @@ -1,187 +0,0 @@ -using System; -using System.Buffers; -using System.IO; -using System.Threading; -using System.Threading.Tasks; -using Jellyfin.Api.Models.PlaybackDtos; -using MediaBrowser.Common.Extensions; -using MediaBrowser.Controller.Library; -using MediaBrowser.Model.IO; - -namespace Jellyfin.Api.Helpers -{ - /// <summary> - /// Progressive file copier. - /// </summary> - public class ProgressiveFileCopier - { - private readonly TranscodingJobDto? _job; - private readonly string? _path; - private readonly CancellationToken _cancellationToken; - private readonly IDirectStreamProvider? _directStreamProvider; - private readonly TranscodingJobHelper _transcodingJobHelper; - private long _bytesWritten; - - /// <summary> - /// Initializes a new instance of the <see cref="ProgressiveFileCopier"/> class. - /// </summary> - /// <param name="path">The path to copy from.</param> - /// <param name="job">The transcoding job.</param> - /// <param name="transcodingJobHelper">Instance of the <see cref="TranscodingJobHelper"/>.</param> - /// <param name="cancellationToken">The cancellation token.</param> - public ProgressiveFileCopier(string path, TranscodingJobDto? job, TranscodingJobHelper transcodingJobHelper, CancellationToken cancellationToken) - { - _path = path; - _job = job; - _cancellationToken = cancellationToken; - _transcodingJobHelper = transcodingJobHelper; - } - - /// <summary> - /// Initializes a new instance of the <see cref="ProgressiveFileCopier"/> class. - /// </summary> - /// <param name="directStreamProvider">Instance of the <see cref="IDirectStreamProvider"/> interface.</param> - /// <param name="job">The transcoding job.</param> - /// <param name="transcodingJobHelper">Instance of the <see cref="TranscodingJobHelper"/>.</param> - /// <param name="cancellationToken">The cancellation token.</param> - public ProgressiveFileCopier(IDirectStreamProvider directStreamProvider, TranscodingJobDto? job, TranscodingJobHelper transcodingJobHelper, CancellationToken cancellationToken) - { - _directStreamProvider = directStreamProvider; - _job = job; - _cancellationToken = cancellationToken; - _transcodingJobHelper = transcodingJobHelper; - } - - /// <summary> - /// Gets or sets a value indicating whether allow read end of file. - /// </summary> - public bool AllowEndOfFile { get; set; } = true; - - /// <summary> - /// Gets or sets copy start position. - /// </summary> - public long StartPosition { get; set; } - - /// <summary> - /// Write source stream to output. - /// </summary> - /// <param name="outputStream">Output stream.</param> - /// <param name="cancellationToken">Cancellation token.</param> - /// <returns>A <see cref="Task"/>.</returns> - public async Task WriteToAsync(Stream outputStream, CancellationToken cancellationToken) - { - using var linkedCancellationTokenSource = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken, _cancellationToken); - cancellationToken = linkedCancellationTokenSource.Token; - - try - { - if (_directStreamProvider != null) - { - await _directStreamProvider.CopyToAsync(outputStream, cancellationToken).ConfigureAwait(false); - return; - } - - var fileOptions = FileOptions.SequentialScan; - var allowAsyncFileRead = false; - - if (AsyncFile.UseAsyncIO) - { - fileOptions |= FileOptions.Asynchronous; - allowAsyncFileRead = true; - } - - if (_path == null) - { - throw new ResourceNotFoundException(nameof(_path)); - } - - await using var inputStream = new FileStream(_path, FileMode.Open, FileAccess.Read, FileShare.ReadWrite, IODefaults.FileStreamBufferSize, fileOptions); - - var eofCount = 0; - const int EmptyReadLimit = 20; - if (StartPosition > 0) - { - inputStream.Position = StartPosition; - } - - while (eofCount < EmptyReadLimit || !AllowEndOfFile) - { - var bytesRead = await CopyToInternalAsync(inputStream, outputStream, allowAsyncFileRead, cancellationToken).ConfigureAwait(false); - - if (bytesRead == 0) - { - if (_job == null || _job.HasExited) - { - eofCount++; - } - - await Task.Delay(100, cancellationToken).ConfigureAwait(false); - } - else - { - eofCount = 0; - } - } - } - finally - { - if (_job != null) - { - _transcodingJobHelper.OnTranscodeEndRequest(_job); - } - } - } - - private async Task<int> CopyToInternalAsync(Stream source, Stream destination, bool readAsync, CancellationToken cancellationToken) - { - var array = ArrayPool<byte>.Shared.Rent(IODefaults.CopyToBufferSize); - try - { - int bytesRead; - int totalBytesRead = 0; - - if (readAsync) - { - bytesRead = await source.ReadAsync(array, 0, array.Length, cancellationToken).ConfigureAwait(false); - } - else - { - bytesRead = source.Read(array, 0, array.Length); - } - - while (bytesRead != 0) - { - var bytesToWrite = bytesRead; - - if (bytesToWrite > 0) - { - await destination.WriteAsync(array, 0, Convert.ToInt32(bytesToWrite), cancellationToken).ConfigureAwait(false); - - _bytesWritten += bytesRead; - totalBytesRead += bytesRead; - - if (_job != null) - { - _job.BytesDownloaded = Math.Max(_job.BytesDownloaded ?? _bytesWritten, _bytesWritten); - } - } - - if (readAsync) - { - bytesRead = await source.ReadAsync(array, 0, array.Length, cancellationToken).ConfigureAwait(false); - } - else - { - bytesRead = source.Read(array, 0, array.Length); - } - } - - return totalBytesRead; - } - finally - { - ArrayPool<byte>.Shared.Return(array); - } - } - } -} diff --git a/Jellyfin.Api/Helpers/ProgressiveFileStream.cs b/Jellyfin.Api/Helpers/ProgressiveFileStream.cs index d4cc0172d..c57018351 100644 --- a/Jellyfin.Api/Helpers/ProgressiveFileStream.cs +++ b/Jellyfin.Api/Helpers/ProgressiveFileStream.cs @@ -13,9 +13,9 @@ namespace Jellyfin.Api.Helpers /// </summary> public class ProgressiveFileStream : Stream { - private readonly FileStream _fileStream; + private readonly Stream _stream; private readonly TranscodingJobDto? _job; - private readonly TranscodingJobHelper _transcodingJobHelper; + private readonly TranscodingJobHelper? _transcodingJobHelper; private readonly int _timeoutMs; private readonly bool _allowAsyncFileRead; private int _bytesWritten; @@ -33,7 +33,6 @@ namespace Jellyfin.Api.Helpers _job = job; _transcodingJobHelper = transcodingJobHelper; _timeoutMs = timeoutMs; - _bytesWritten = 0; var fileOptions = FileOptions.SequentialScan; _allowAsyncFileRead = false; @@ -45,11 +44,25 @@ namespace Jellyfin.Api.Helpers _allowAsyncFileRead = true; } - _fileStream = new FileStream(filePath, FileMode.Open, FileAccess.Read, FileShare.ReadWrite, IODefaults.FileStreamBufferSize, fileOptions); + _stream = new FileStream(filePath, FileMode.Open, FileAccess.Read, FileShare.ReadWrite, IODefaults.FileStreamBufferSize, fileOptions); + } + + /// <summary> + /// Initializes a new instance of the <see cref="ProgressiveFileStream"/> class. + /// </summary> + /// <param name="stream">The stream to progressively copy.</param> + /// <param name="timeoutMs">The timeout duration in milliseconds.</param> + public ProgressiveFileStream(Stream stream, int timeoutMs = 30000) + { + _job = null; + _transcodingJobHelper = null; + _timeoutMs = timeoutMs; + _allowAsyncFileRead = AsyncFile.UseAsyncIO; + _stream = stream; } /// <inheritdoc /> - public override bool CanRead => _fileStream.CanRead; + public override bool CanRead => _stream.CanRead; /// <inheritdoc /> public override bool CanSeek => false; @@ -70,13 +83,13 @@ namespace Jellyfin.Api.Helpers /// <inheritdoc /> public override void Flush() { - _fileStream.Flush(); + _stream.Flush(); } /// <inheritdoc /> public override int Read(byte[] buffer, int offset, int count) { - return _fileStream.Read(buffer, offset, count); + return _stream.Read(buffer, offset, count); } /// <inheritdoc /> @@ -93,11 +106,11 @@ namespace Jellyfin.Api.Helpers int bytesRead; if (_allowAsyncFileRead) { - bytesRead = await _fileStream.ReadAsync(buffer, newOffset, remainingBytesToRead, cancellationToken).ConfigureAwait(false); + bytesRead = await _stream.ReadAsync(buffer, newOffset, remainingBytesToRead, cancellationToken).ConfigureAwait(false); } else { - bytesRead = _fileStream.Read(buffer, newOffset, remainingBytesToRead); + bytesRead = _stream.Read(buffer, newOffset, remainingBytesToRead); } remainingBytesToRead -= bytesRead; @@ -152,11 +165,11 @@ namespace Jellyfin.Api.Helpers { if (disposing) { - _fileStream.Dispose(); + _stream.Dispose(); if (_job != null) { - _transcodingJobHelper.OnTranscodeEndRequest(_job); + _transcodingJobHelper?.OnTranscodeEndRequest(_job); } } } diff --git a/Jellyfin.Api/Models/StreamingDtos/StreamState.cs b/Jellyfin.Api/Models/StreamingDtos/StreamState.cs index e95f2d1f4..0f84faeaf 100644 --- a/Jellyfin.Api/Models/StreamingDtos/StreamState.cs +++ b/Jellyfin.Api/Models/StreamingDtos/StreamState.cs @@ -60,6 +60,9 @@ namespace Jellyfin.Api.Models.StreamingDtos /// <summary> /// Gets or sets the direct stream provicer. /// </summary> + /// <remarks> + /// Deprecated. + /// </remarks> public IDirectStreamProvider? DirectStreamProvider { get; set; } /// <summary> diff --git a/Jellyfin.Server/Infrastructure/SymlinkFollowingPhysicalFileResultExecutor.cs b/Jellyfin.Server/Infrastructure/SymlinkFollowingPhysicalFileResultExecutor.cs new file mode 100644 index 000000000..e171fc145 --- /dev/null +++ b/Jellyfin.Server/Infrastructure/SymlinkFollowingPhysicalFileResultExecutor.cs @@ -0,0 +1,145 @@ +// The MIT License (MIT) +// +// Copyright (c) .NET Foundation and Contributors +// +// All rights reserved. +// +// Permission is hereby granted, free of charge, to any person obtaining a copy +// of this software and associated documentation files (the "Software"), to deal +// in the Software without restriction, including without limitation the rights +// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +// copies of the Software, and to permit persons to whom the Software is +// furnished to do so, subject to the following conditions: +// +// The above copyright notice and this permission notice shall be included in all +// copies or substantial portions of the Software. +// +// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +// SOFTWARE. + +using System; +using System.IO; +using System.Threading; +using System.Threading.Tasks; +using MediaBrowser.Model.IO; +using Microsoft.AspNetCore.Http; +using Microsoft.AspNetCore.Http.Extensions; +using Microsoft.AspNetCore.Mvc; +using Microsoft.AspNetCore.Mvc.Infrastructure; +using Microsoft.Extensions.Logging; +using Microsoft.Net.Http.Headers; + +namespace Jellyfin.Server.Infrastructure +{ + /// <inheritdoc /> + public class SymlinkFollowingPhysicalFileResultExecutor : PhysicalFileResultExecutor + { + /// <summary> + /// Initializes a new instance of the <see cref="SymlinkFollowingPhysicalFileResultExecutor"/> class. + /// </summary> + /// <param name="loggerFactory">An instance of the <see cref="ILoggerFactory"/> interface.</param> + public SymlinkFollowingPhysicalFileResultExecutor(ILoggerFactory loggerFactory) : base(loggerFactory) + { + } + + /// <inheritdoc /> + protected override FileMetadata GetFileInfo(string path) + { + var fileInfo = new FileInfo(path); + var length = fileInfo.Length; + // This may or may not be fixed in .NET 6, but looks like it will not https://github.com/dotnet/aspnetcore/issues/34371 + if ((fileInfo.Attributes & FileAttributes.ReparsePoint) == FileAttributes.ReparsePoint) + { + using Stream thisFileStream = File.OpenRead(path); + length = thisFileStream.Length; + } + + return new FileMetadata + { + Exists = fileInfo.Exists, + Length = length, + LastModified = fileInfo.LastWriteTimeUtc + }; + } + + /// <inheritdoc /> + protected override Task WriteFileAsync(ActionContext context, PhysicalFileResult result, RangeItemHeaderValue range, long rangeLength) + { + if (context == null) + { + throw new ArgumentNullException(nameof(context)); + } + + if (result == null) + { + throw new ArgumentNullException(nameof(result)); + } + + if (range != null && rangeLength == 0) + { + return Task.CompletedTask; + } + + // It's a bit of wasted IO to perform this check again, but non-symlinks shouldn't use this code + if (!IsSymLink(result.FileName)) + { + return base.WriteFileAsync(context, result, range, rangeLength); + } + + var response = context.HttpContext.Response; + + if (range != null) + { + return SendFileAsync( + result.FileName, + response, + offset: range.From ?? 0L, + count: rangeLength); + } + + return SendFileAsync( + result.FileName, + response, + offset: 0, + count: null); + } + + private async Task SendFileAsync(string filePath, HttpResponse response, long offset, long? count) + { + var fileInfo = GetFileInfo(filePath); + if (offset < 0 || offset > fileInfo.Length) + { + throw new ArgumentOutOfRangeException(nameof(offset), offset, string.Empty); + } + + if (count.HasValue + && (count.Value < 0 || count.Value > fileInfo.Length - offset)) + { + throw new ArgumentOutOfRangeException(nameof(count), count, string.Empty); + } + + // Copied from SendFileFallback.SendFileAsync + const int BufferSize = 1024 * 16; + + await using var fileStream = new FileStream( + filePath, + FileMode.Open, + FileAccess.Read, + FileShare.ReadWrite, + bufferSize: BufferSize, + options: (AsyncFile.UseAsyncIO ? FileOptions.Asynchronous : FileOptions.None) | FileOptions.SequentialScan); + + fileStream.Seek(offset, SeekOrigin.Begin); + await StreamCopyOperation + .CopyToAsync(fileStream, response.Body, count, BufferSize, CancellationToken.None) + .ConfigureAwait(true); + } + + private static bool IsSymLink(string path) => (File.GetAttributes(path) & FileAttributes.ReparsePoint) == FileAttributes.ReparsePoint; + } +} diff --git a/Jellyfin.Server/Startup.cs b/Jellyfin.Server/Startup.cs index 60cdc2f6f..8085c2630 100644 --- a/Jellyfin.Server/Startup.cs +++ b/Jellyfin.Server/Startup.cs @@ -7,6 +7,7 @@ using System.Text; using Jellyfin.Networking.Configuration; using Jellyfin.Server.Extensions; using Jellyfin.Server.Implementations; +using Jellyfin.Server.Infrastructure; using Jellyfin.Server.Middleware; using MediaBrowser.Common.Net; using MediaBrowser.Controller; @@ -14,6 +15,8 @@ using MediaBrowser.Controller.Configuration; using MediaBrowser.Controller.Extensions; using Microsoft.AspNetCore.Builder; using Microsoft.AspNetCore.Hosting; +using Microsoft.AspNetCore.Mvc; +using Microsoft.AspNetCore.Mvc.Infrastructure; using Microsoft.AspNetCore.StaticFiles; using Microsoft.Extensions.Configuration; using Microsoft.Extensions.DependencyInjection; @@ -56,6 +59,9 @@ namespace Jellyfin.Server { options.HttpsPort = _serverApplicationHost.HttpsPort; }); + + // TODO remove once this is fixed upstream https://github.com/dotnet/aspnetcore/issues/34371 + services.AddSingleton<IActionResultExecutor<PhysicalFileResult>, SymlinkFollowingPhysicalFileResultExecutor>(); services.AddJellyfinApi(_serverApplicationHost.GetApiPluginAssemblies(), _serverConfigurationManager.GetNetworkConfiguration()); services.AddJellyfinApiSwagger(); diff --git a/MediaBrowser.Controller/Library/IDirectStreamProvider.cs b/MediaBrowser.Controller/Library/IDirectStreamProvider.cs new file mode 100644 index 000000000..96f8b7eba --- /dev/null +++ b/MediaBrowser.Controller/Library/IDirectStreamProvider.cs @@ -0,0 +1,19 @@ +using System.IO; + +namespace MediaBrowser.Controller.Library +{ + /// <summary> + /// The direct live TV stream provider. + /// </summary> + /// <remarks> + /// Deprecated. + /// </remarks> + public interface IDirectStreamProvider + { + /// <summary> + /// Gets the live stream, shared streams seek to the end of the file first. + /// </summary> + /// <returns>The stream.</returns> + Stream GetStream(); + } +} diff --git a/MediaBrowser.Controller/Library/ILiveStream.cs b/MediaBrowser.Controller/Library/ILiveStream.cs index 323aa4876..4c44a17fd 100644 --- a/MediaBrowser.Controller/Library/ILiveStream.cs +++ b/MediaBrowser.Controller/Library/ILiveStream.cs @@ -2,6 +2,7 @@ #pragma warning disable CA1711, CS1591 +using System.IO; using System.Threading; using System.Threading.Tasks; using MediaBrowser.Model.Dto; @@ -25,5 +26,7 @@ namespace MediaBrowser.Controller.Library Task Open(CancellationToken openCancellationToken); Task Close(); + + Stream GetStream(); } } diff --git a/MediaBrowser.Controller/Library/IMediaSourceManager.cs b/MediaBrowser.Controller/Library/IMediaSourceManager.cs index fd3631da9..e802796d3 100644 --- a/MediaBrowser.Controller/Library/IMediaSourceManager.cs +++ b/MediaBrowser.Controller/Library/IMediaSourceManager.cs @@ -4,7 +4,6 @@ using System; using System.Collections.Generic; -using System.IO; using System.Threading; using System.Threading.Tasks; using Jellyfin.Data.Entities; @@ -111,6 +110,20 @@ namespace MediaBrowser.Controller.Library Task<Tuple<MediaSourceInfo, IDirectStreamProvider>> GetLiveStreamWithDirectStreamProvider(string id, CancellationToken cancellationToken); /// <summary> + /// Gets the live stream info. + /// </summary> + /// <param name="id">The identifier.</param> + /// <returns>An instance of <see cref="ILiveStream"/>.</returns> + public ILiveStream GetLiveStreamInfo(string id); + + /// <summary> + /// Gets the live stream info using the stream's unique id. + /// </summary> + /// <param name="uniqueId">The unique identifier.</param> + /// <returns>An instance of <see cref="ILiveStream"/>.</returns> + public ILiveStream GetLiveStreamInfoByUniqueId(string uniqueId); + + /// <summary> /// Closes the media source. /// </summary> /// <param name="id">The live stream identifier.</param> @@ -126,14 +139,5 @@ namespace MediaBrowser.Controller.Library void SetDefaultAudioAndSubtitleStreamIndexes(BaseItem item, MediaSourceInfo source, User user); Task AddMediaInfoWithProbe(MediaSourceInfo mediaSource, bool isAudio, string cacheKey, bool addProbeDelay, bool isLiveStream, CancellationToken cancellationToken); - - Task<IDirectStreamProvider> GetDirectStreamProviderByUniqueId(string uniqueId, CancellationToken cancellationToken); - } - - public interface IDirectStreamProvider - { - Task CopyToAsync(Stream stream, CancellationToken cancellationToken); - - string GetFilePath(); } } |