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
|
using MediaBrowser.Model.Logging;
using System;
using System.IO;
using System.Threading;
using System.Threading.Tasks;
using CommonIO;
using MediaBrowser.Controller.Net;
using System.Collections.Generic;
using ServiceStack.Web;
namespace MediaBrowser.Api.Playback.Progressive
{
public class ProgressiveFileCopier : IAsyncStreamSource, IHasOptions
{
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;
// 256k
private const int BufferSize = 81920;
private long _bytesWritten = 0;
public long StartPosition { get; set; }
public bool AllowEndOfFile = true;
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 IDictionary<string, string> Options
{
get
{
return _outputHeaders;
}
}
public async Task WriteToAsync(Stream outputStream)
{
try
{
var eofCount = 0;
using (var fs = _fileSystem.GetFileStream(_path, FileMode.Open, FileAccess.Read, FileShare.ReadWrite, true))
{
if (StartPosition > 0)
{
fs.Position = StartPosition;
}
while (eofCount < 15 || !AllowEndOfFile)
{
var bytesRead = await CopyToAsyncInternal(fs, outputStream, BufferSize, _cancellationToken).ConfigureAwait(false);
//var position = fs.Position;
//_logger.Debug("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> CopyToAsyncInternal(Stream source, Stream destination, Int32 bufferSize, CancellationToken cancellationToken)
{
byte[] buffer = new byte[bufferSize];
int bytesRead;
int totalBytesRead = 0;
while ((bytesRead = await source.ReadAsync(buffer, 0, buffer.Length, cancellationToken).ConfigureAwait(false)) != 0)
{
await destination.WriteAsync(buffer, 0, bytesRead, cancellationToken).ConfigureAwait(false);
_bytesWritten += bytesRead;
totalBytesRead += bytesRead;
if (_job != null)
{
_job.BytesDownloaded = Math.Max(_job.BytesDownloaded ?? _bytesWritten, _bytesWritten);
}
}
return totalBytesRead;
}
}
}
|
