ProgressiveStreamWriter.cs 7.0 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193
  1. using MediaBrowser.Model.Logging;
  2. using System;
  3. using System.IO;
  4. using System.Threading;
  5. using System.Threading.Tasks;
  6. using MediaBrowser.Model.IO;
  7. using MediaBrowser.Controller.Net;
  8. using System.Collections.Generic;
  9. using MediaBrowser.Controller.IO;
  10. using MediaBrowser.Controller.Library;
  11. using MediaBrowser.Model.Services;
  12. using MediaBrowser.Model.System;
  13. namespace MediaBrowser.Api.Playback.Progressive
  14. {
  15. public class ProgressiveFileCopier : IAsyncStreamWriter, IHasHeaders
  16. {
  17. private readonly IFileSystem _fileSystem;
  18. private readonly TranscodingJob _job;
  19. private readonly ILogger _logger;
  20. private readonly string _path;
  21. private readonly CancellationToken _cancellationToken;
  22. private readonly Dictionary<string, string> _outputHeaders;
  23. const int StreamCopyToBufferSize = 81920;
  24. private long _bytesWritten = 0;
  25. public long StartPosition { get; set; }
  26. public bool AllowEndOfFile = true;
  27. private readonly IDirectStreamProvider _directStreamProvider;
  28. private readonly IEnvironmentInfo _environment;
  29. public ProgressiveFileCopier(IFileSystem fileSystem, string path, Dictionary<string, string> outputHeaders, TranscodingJob job, ILogger logger, IEnvironmentInfo environment, CancellationToken cancellationToken)
  30. {
  31. _fileSystem = fileSystem;
  32. _path = path;
  33. _outputHeaders = outputHeaders;
  34. _job = job;
  35. _logger = logger;
  36. _cancellationToken = cancellationToken;
  37. _environment = environment;
  38. }
  39. public ProgressiveFileCopier(IDirectStreamProvider directStreamProvider, Dictionary<string, string> outputHeaders, TranscodingJob job, ILogger logger, IEnvironmentInfo environment, CancellationToken cancellationToken)
  40. {
  41. _directStreamProvider = directStreamProvider;
  42. _outputHeaders = outputHeaders;
  43. _job = job;
  44. _logger = logger;
  45. _cancellationToken = cancellationToken;
  46. _environment = environment;
  47. }
  48. public IDictionary<string, string> Headers
  49. {
  50. get
  51. {
  52. return _outputHeaders;
  53. }
  54. }
  55. private Stream GetInputStream(bool allowAsyncFileRead)
  56. {
  57. var fileOpenOptions = FileOpenOptions.SequentialScan;
  58. if (allowAsyncFileRead)
  59. {
  60. fileOpenOptions |= FileOpenOptions.Asynchronous;
  61. }
  62. return _fileSystem.GetFileStream(_path, FileOpenMode.Open, FileAccessMode.Read, FileShareMode.ReadWrite, fileOpenOptions);
  63. }
  64. public async Task WriteToAsync(Stream outputStream, CancellationToken cancellationToken)
  65. {
  66. cancellationToken = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken, _cancellationToken).Token;
  67. try
  68. {
  69. if (_directStreamProvider != null)
  70. {
  71. await _directStreamProvider.CopyToAsync(outputStream, cancellationToken).ConfigureAwait(false);
  72. return;
  73. }
  74. var eofCount = 0;
  75. // use non-async filestream along with read due to https://github.com/dotnet/corefx/issues/6039
  76. var allowAsyncFileRead = _environment.OperatingSystem != OperatingSystem.Windows;
  77. using (var inputStream = GetInputStream(allowAsyncFileRead))
  78. {
  79. if (StartPosition > 0)
  80. {
  81. inputStream.Position = StartPosition;
  82. }
  83. while (eofCount < 20 || !AllowEndOfFile)
  84. {
  85. int bytesRead;
  86. if (allowAsyncFileRead)
  87. {
  88. bytesRead = await CopyToInternalAsync(inputStream, outputStream, cancellationToken).ConfigureAwait(false);
  89. }
  90. else
  91. {
  92. bytesRead = await CopyToInternalAsyncWithSyncRead(inputStream, outputStream, cancellationToken).ConfigureAwait(false);
  93. }
  94. //var position = fs.Position;
  95. //_logger.Debug("Streamed {0} bytes to position {1} from file {2}", bytesRead, position, path);
  96. if (bytesRead == 0)
  97. {
  98. if (_job == null || _job.HasExited)
  99. {
  100. eofCount++;
  101. }
  102. await Task.Delay(100, cancellationToken).ConfigureAwait(false);
  103. }
  104. else
  105. {
  106. eofCount = 0;
  107. }
  108. }
  109. }
  110. }
  111. finally
  112. {
  113. if (_job != null)
  114. {
  115. ApiEntryPoint.Instance.OnTranscodeEndRequest(_job);
  116. }
  117. }
  118. }
  119. private async Task<int> CopyToInternalAsyncWithSyncRead(Stream source, Stream destination, CancellationToken cancellationToken)
  120. {
  121. var array = new byte[StreamCopyToBufferSize];
  122. int bytesRead;
  123. int totalBytesRead = 0;
  124. while ((bytesRead = source.Read(array, 0, array.Length)) != 0)
  125. {
  126. var bytesToWrite = bytesRead;
  127. if (bytesToWrite > 0)
  128. {
  129. await destination.WriteAsync(array, 0, Convert.ToInt32(bytesToWrite), cancellationToken).ConfigureAwait(false);
  130. _bytesWritten += bytesRead;
  131. totalBytesRead += bytesRead;
  132. if (_job != null)
  133. {
  134. _job.BytesDownloaded = Math.Max(_job.BytesDownloaded ?? _bytesWritten, _bytesWritten);
  135. }
  136. }
  137. }
  138. return totalBytesRead;
  139. }
  140. private async Task<int> CopyToInternalAsync(Stream source, Stream destination, CancellationToken cancellationToken)
  141. {
  142. var array = new byte[StreamCopyToBufferSize];
  143. int bytesRead;
  144. int totalBytesRead = 0;
  145. while ((bytesRead = await source.ReadAsync(array, 0, array.Length, cancellationToken).ConfigureAwait(false)) != 0)
  146. {
  147. var bytesToWrite = bytesRead;
  148. if (bytesToWrite > 0)
  149. {
  150. await destination.WriteAsync(array, 0, Convert.ToInt32(bytesToWrite), cancellationToken).ConfigureAwait(false);
  151. _bytesWritten += bytesRead;
  152. totalBytesRead += bytesRead;
  153. if (_job != null)
  154. {
  155. _job.BytesDownloaded = Math.Max(_job.BytesDownloaded ?? _bytesWritten, _bytesWritten);
  156. }
  157. }
  158. }
  159. return totalBytesRead;
  160. }
  161. }
  162. }