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