michael.heier/jellyfin
1
0
Fork 0
Code Issues Pull Requests Actions 3 Packages Projects Releases Wiki Activity

Merge pull request #6538 from cvium/livetv_oh_no

Browse Source
This commit is contained in:
Bond-009 2021-09-20 22:38:44 +02:00 committed by GitHub
commit 32ea4806f8
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
15 changed files with 158 additions and 394 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
Emby.Server.Implementations/Library/ExclusiveLiveStream.cs
View File

@ -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;

33
Emby.Server.Implementations/Library/MediaSourceManager.cs
View File

@ -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)

14
Emby.Server.Implementations/LiveTv/EmbyTV/DirectRecorder.cs
View File

@ -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);
var linkedCancellationToken = cancellationTokenSource.Token;
await directStreamProvider.CopyToAsync(output, cancellationTokenSource.Token).ConfigureAwait(false);
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)

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

@ -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)

121
Emby.Server.Implementations/LiveTv/TunerHosts/LiveStream.cs
View File

@ -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);
Logger.LogInformation("Deleting temp file {FilePath}", path);
}
var failedFiles = new List<string>();
foreach (var path in paths)
try
{
if (!File.Exists(path))
FileSystem.DeleteFile(path);
}
catch (Exception ex)
{
Logger.LogError(ex, "Error deleting file {FilePath}", path);
if (retryCount <= 40)
{
continue;
}
try
{
FileSystem.DeleteFile(path);
}
catch (Exception ex)
{
Logger.LogError(ex, "Error deleting file {path}", path);
failedFiles.Add(path);
await Task.Delay(500).ConfigureAwait(false);
await DeleteTempFiles(path, retryCount + 1).ConfigureAwait(false);
}
}
if (failedFiles.Count > 0 && retryCount <= 40)
{
await Task.Delay(500).ConfigureAwait(false);
await DeleteTempFiles(failedFiles, retryCount + 1).ConfigureAwait(false);
}
}
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))
{
return (files[^1], true);
}
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))
{
if (seekFile)
{
TrySeek(inputStream, -20000);
}
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)
{

49
Emby.Server.Implementations/LiveTv/TunerHosts/SharedHttpStream.cs
View File

@ -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)
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))
{
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
response.Dispose();
return;
}
// Close the stream without any sharing features
if (requiresRemux)
{
using (response)
{
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);
}

6
Jellyfin.Api/Controllers/LiveTvController.cs
View File

@ -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));
}

20
Jellyfin.Api/Controllers/VideosController.cs
View File

@ -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(

25
Jellyfin.Api/Helpers/AudioHelper.cs
View File

@ -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(

187
Jellyfin.Api/Helpers/ProgressiveFileCopier.cs
View File

@ -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);
}
}
}
}

35
Jellyfin.Api/Helpers/ProgressiveFileStream.cs
View File

@ -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);
}
}
}

3
Jellyfin.Api/Models/StreamingDtos/StreamState.cs
View File

@ -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>

19
MediaBrowser.Controller/Library/IDirectStreamProvider.cs Normal file
View File

@ -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();
}
}

3
MediaBrowser.Controller/Library/ILiveStream.cs
View File

@ -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();
}
}

24
MediaBrowser.Controller/Library/IMediaSourceManager.cs
View File

@ -4,7 +4,6 @@
using System;
using System.Collections.Generic;
using System.IO;
using System.Threading;
using System.Threading.Tasks;
using Jellyfin.Data.Entities;
@ -110,6 +109,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>
@ -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();
}
}