Merge pull request #2025 from softworkz/AsyncStreamInterface

Async stream handling: Use interface instead of Func<Stream,Task>
This commit is contained in:
Luke 2016-08-15 17:59:28 -04:00 committed by GitHub
commit 02a4b90f65
8 changed files with 64 additions and 25 deletions

View File

@ -370,9 +370,9 @@ namespace MediaBrowser.Api.Playback.Progressive
outputHeaders[item.Key] = item.Value; outputHeaders[item.Key] = item.Value;
} }
Func<Stream, Task> streamWriter = stream => new ProgressiveFileCopier(FileSystem, job, Logger).StreamFile(outputPath, stream, CancellationToken.None); var streamSource = new ProgressiveFileCopier(FileSystem, outputPath, outputHeaders, job, Logger, CancellationToken.None);
return ResultFactory.GetAsyncStreamWriter(streamWriter, outputHeaders); return ResultFactory.GetAsyncStreamWriter(streamSource);
} }
finally finally
{ {

View File

@ -4,28 +4,45 @@ using System.IO;
using System.Threading; using System.Threading;
using System.Threading.Tasks; using System.Threading.Tasks;
using CommonIO; using CommonIO;
using MediaBrowser.Controller.Net;
using System.Collections.Generic;
using ServiceStack.Web;
namespace MediaBrowser.Api.Playback.Progressive namespace MediaBrowser.Api.Playback.Progressive
{ {
public class ProgressiveFileCopier public class ProgressiveFileCopier : IAsyncStreamSource, IHasOptions
{ {
private readonly IFileSystem _fileSystem; private readonly IFileSystem _fileSystem;
private readonly TranscodingJob _job; private readonly TranscodingJob _job;
private readonly ILogger _logger; private readonly ILogger _logger;
private readonly string _path;
private readonly CancellationToken _cancellationToken;
private readonly Dictionary<string, string> _outputHeaders;
// 256k // 256k
private const int BufferSize = 81920; private const int BufferSize = 81920;
private long _bytesWritten = 0; private long _bytesWritten = 0;
public ProgressiveFileCopier(IFileSystem fileSystem, TranscodingJob job, ILogger logger) public ProgressiveFileCopier(IFileSystem fileSystem, string path, Dictionary<string, string> outputHeaders, TranscodingJob job, ILogger logger, CancellationToken cancellationToken)
{ {
_fileSystem = fileSystem; _fileSystem = fileSystem;
_path = path;
_outputHeaders = outputHeaders;
_job = job; _job = job;
_logger = logger; _logger = logger;
_cancellationToken = cancellationToken;
} }
public async Task StreamFile(string path, Stream outputStream, CancellationToken cancellationToken) public IDictionary<string, string> Options
{
get
{
return _outputHeaders;
}
}
public async Task WriteToAsync(Stream outputStream)
{ {
try try
{ {

View File

@ -236,6 +236,7 @@
<Compile Include="Net\IAuthorizationContext.cs" /> <Compile Include="Net\IAuthorizationContext.cs" />
<Compile Include="Net\IAuthService.cs" /> <Compile Include="Net\IAuthService.cs" />
<Compile Include="Net\IHasAuthorization.cs" /> <Compile Include="Net\IHasAuthorization.cs" />
<Compile Include="Net\IAsyncStreamSource.cs" />
<Compile Include="Net\IHasResultFactory.cs" /> <Compile Include="Net\IHasResultFactory.cs" />
<Compile Include="Net\IHasSession.cs" /> <Compile Include="Net\IHasSession.cs" />
<Compile Include="Net\IHttpResultFactory.cs" /> <Compile Include="Net\IHttpResultFactory.cs" />

View File

@ -0,0 +1,18 @@
using ServiceStack.Web;
using System.IO;
using System.Threading.Tasks;
namespace MediaBrowser.Controller.Net
{
/// <summary>
/// Interface IAsyncStreamSource
/// Enables asynchronous writing to http resonse streams
/// </summary>
public interface IAsyncStreamSource
{
/// <summary>
/// Asynchronously write to the response stream.
/// </summary>
Task WriteToAsync(Stream responseStream);
}
}

View File

@ -28,7 +28,7 @@ namespace MediaBrowser.Controller.Net
/// <returns>System.Object.</returns> /// <returns>System.Object.</returns>
object GetResult(object content, string contentType, IDictionary<string,string> responseHeaders = null); object GetResult(object content, string contentType, IDictionary<string,string> responseHeaders = null);
object GetAsyncStreamWriter(Func<Stream,Task> streamWriter, IDictionary<string, string> responseHeaders = null); object GetAsyncStreamWriter(IAsyncStreamSource streamSource);
/// <summary> /// <summary>
/// Gets the optimized result. /// Gets the optimized result.

View File

@ -4,38 +4,41 @@ using System.IO;
using System.Threading.Tasks; using System.Threading.Tasks;
using ServiceStack; using ServiceStack;
using ServiceStack.Web; using ServiceStack.Web;
using MediaBrowser.Controller.Net;
namespace MediaBrowser.Server.Implementations.HttpServer namespace MediaBrowser.Server.Implementations.HttpServer
{ {
public class AsyncStreamWriterFunc : IStreamWriter, IAsyncStreamWriter, IHasOptions public class AsyncStreamWriter : IStreamWriter, IAsyncStreamWriter, IHasOptions
{ {
/// <summary> /// <summary>
/// Gets or sets the source stream. /// Gets or sets the source stream.
/// </summary> /// </summary>
/// <value>The source stream.</value> /// <value>The source stream.</value>
private Func<Stream, Task> Writer { get; set; } private IAsyncStreamSource _source;
/// <summary>
/// Gets the options.
/// </summary>
/// <value>The options.</value>
public IDictionary<string, string> Options { get; private set; }
public Action OnComplete { get; set; } public Action OnComplete { get; set; }
public Action OnError { get; set; } public Action OnError { get; set; }
/// <summary> /// <summary>
/// Initializes a new instance of the <see cref="StreamWriter" /> class. /// Initializes a new instance of the <see cref="AsyncStreamWriter" /> class.
/// </summary> /// </summary>
public AsyncStreamWriterFunc(Func<Stream, Task> writer, IDictionary<string, string> headers) public AsyncStreamWriter(IAsyncStreamSource source)
{ {
Writer = writer; _source = source;
}
if (headers == null) public IDictionary<string, string> Options
{
get
{ {
headers = new Dictionary<string, string>(StringComparer.OrdinalIgnoreCase); var hasOptions = _source as IHasOptions;
if (hasOptions != null)
{
return hasOptions.Options;
}
return new Dictionary<string, string>(StringComparer.OrdinalIgnoreCase);
} }
Options = headers;
} }
/// <summary> /// <summary>
@ -44,13 +47,13 @@ namespace MediaBrowser.Server.Implementations.HttpServer
/// <param name="responseStream">The response stream.</param> /// <param name="responseStream">The response stream.</param>
public void WriteTo(Stream responseStream) public void WriteTo(Stream responseStream)
{ {
var task = Writer(responseStream); var task = _source.WriteToAsync(responseStream);
Task.WaitAll(task); Task.WaitAll(task);
} }
public async Task WriteToAsync(Stream responseStream) public async Task WriteToAsync(Stream responseStream)
{ {
await Writer(responseStream).ConfigureAwait(false); await _source.WriteToAsync(responseStream).ConfigureAwait(false);
} }
} }
} }

View File

@ -704,9 +704,9 @@ namespace MediaBrowser.Server.Implementations.HttpServer
throw error; throw error;
} }
public object GetAsyncStreamWriter(Func<Stream, Task> streamWriter, IDictionary<string, string> responseHeaders = null) public object GetAsyncStreamWriter(IAsyncStreamSource streamSource)
{ {
return new AsyncStreamWriterFunc(streamWriter, responseHeaders); return new AsyncStreamWriter(streamSource);
} }
} }
} }

View File

@ -156,7 +156,7 @@
<Compile Include="EntryPoints\ServerEventNotifier.cs" /> <Compile Include="EntryPoints\ServerEventNotifier.cs" />
<Compile Include="EntryPoints\UserDataChangeNotifier.cs" /> <Compile Include="EntryPoints\UserDataChangeNotifier.cs" />
<Compile Include="FileOrganization\OrganizerScheduledTask.cs" /> <Compile Include="FileOrganization\OrganizerScheduledTask.cs" />
<Compile Include="HttpServer\AsyncStreamWriterFunc.cs" /> <Compile Include="HttpServer\AsyncStreamWriter.cs" />
<Compile Include="HttpServer\IHttpListener.cs" /> <Compile Include="HttpServer\IHttpListener.cs" />
<Compile Include="HttpServer\Security\AuthorizationContext.cs" /> <Compile Include="HttpServer\Security\AuthorizationContext.cs" />
<Compile Include="HttpServer\ContainerAdapter.cs" /> <Compile Include="HttpServer\ContainerAdapter.cs" />