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
|
using System;
using System.IO;
using System.Threading;
using System.Threading.Tasks;
using MediaBrowser.Model.IO;
using MediaBrowser.Common.IO;
using MediaBrowser.Common.Net;
using MediaBrowser.Controller.IO;
using MediaBrowser.Controller.Library;
using MediaBrowser.Model.Dto;
using MediaBrowser.Model.Logging;
namespace Emby.Server.Implementations.LiveTv.EmbyTV
{
public class DirectRecorder : IRecorder
{
private readonly ILogger _logger;
private readonly IHttpClient _httpClient;
private readonly IFileSystem _fileSystem;
public DirectRecorder(ILogger logger, IHttpClient httpClient, IFileSystem fileSystem)
{
_logger = logger;
_httpClient = httpClient;
_fileSystem = fileSystem;
}
public string GetOutputPath(MediaSourceInfo mediaSource, string targetFile)
{
return targetFile;
}
public Task Record(IDirectStreamProvider directStreamProvider, MediaSourceInfo mediaSource, string targetFile, TimeSpan duration, Action onStarted, CancellationToken cancellationToken)
{
if (directStreamProvider != null)
{
return RecordFromDirectStreamProvider(directStreamProvider, targetFile, duration, onStarted, cancellationToken);
}
return RecordFromMediaSource(mediaSource, targetFile, duration, onStarted, cancellationToken);
}
private async Task RecordFromDirectStreamProvider(IDirectStreamProvider directStreamProvider, string targetFile, TimeSpan duration, Action onStarted, CancellationToken cancellationToken)
{
using (var output = _fileSystem.GetFileStream(targetFile, FileOpenMode.Create, FileAccessMode.Write, FileShareMode.Read))
{
onStarted();
_logger.Info("Copying recording stream to file {0}", targetFile);
// The media source if infinite so we need to handle stopping ourselves
var durationToken = new CancellationTokenSource(duration);
cancellationToken = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken, durationToken.Token).Token;
await directStreamProvider.CopyToAsync(output, cancellationToken).ConfigureAwait(false);
}
_logger.Info("Recording completed to file {0}", targetFile);
}
private async Task RecordFromMediaSource(MediaSourceInfo mediaSource, string targetFile, TimeSpan duration, Action onStarted, CancellationToken cancellationToken)
{
var httpRequestOptions = new HttpRequestOptions
{
Url = mediaSource.Path,
BufferContent = false,
// Some remote urls will expect a user agent to be supplied
UserAgent = "Emby/3.0",
// Shouldn't matter but may cause issues
EnableHttpCompression = false
};
using (var response = await _httpClient.SendAsync(httpRequestOptions, "GET").ConfigureAwait(false))
{
_logger.Info("Opened recording stream from tuner provider");
using (var output = _fileSystem.GetFileStream(targetFile, FileOpenMode.Create, FileAccessMode.Write, FileShareMode.Read))
{
onStarted();
_logger.Info("Copying recording stream to file {0}", targetFile);
// The media source if infinite so we need to handle stopping ourselves
var durationToken = new CancellationTokenSource(duration);
cancellationToken = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken, durationToken.Token).Token;
await CopyUntilCancelled(response.Content, output, cancellationToken).ConfigureAwait(false);
}
}
_logger.Info("Recording completed to file {0}", targetFile);
}
private const int BufferSize = 81920;
public static Task CopyUntilCancelled(Stream source, Stream target, CancellationToken cancellationToken)
{
return CopyUntilCancelled(source, target, null, cancellationToken);
}
public static async Task CopyUntilCancelled(Stream source, Stream target, Action onStarted, CancellationToken cancellationToken)
{
while (!cancellationToken.IsCancellationRequested)
{
var bytesRead = await CopyToAsyncInternal(source, target, BufferSize, onStarted, cancellationToken).ConfigureAwait(false);
onStarted = null;
//var position = fs.Position;
//_logger.Debug("Streamed {0} bytes to position {1} from file {2}", bytesRead, position, path);
if (bytesRead == 0)
{
await Task.Delay(100).ConfigureAwait(false);
}
}
}
private static async Task<int> CopyToAsyncInternal(Stream source, Stream destination, Int32 bufferSize, Action onStarted, 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);
totalBytesRead += bytesRead;
if (onStarted != null)
{
onStarted();
}
onStarted = null;
}
return totalBytesRead;
}
}
}
|
