如何使用 BackgroundUploader 将未知数量的字节流式传输到服务器？

Question

我有一个繁重的 IO/CPU 转换“过程”，从文件类型 A 转换为 B。目前我将其写入 StorageFile，然后在转换完成后使用 BackgroundUploader 上传。

但是，我想尽快开始流式传输，同时仍在生成输出文件。此外，我什至不一定要创建输出 StorageFile，而是“随时随地”上传。

注意，我不知道转换过程中输出文件的最终大小，它可能比源文件小或大。

首先，我尝试在写入时简单地打开接收器输出 StorageFile，并将该流传递给 BackgroundUploader，但这会导致“竞争条件”，即上传在到达写入的字节末尾时终止StorageFile（当它赶上转换工作时）。

除了流式传输到 StorageFile，我还可以将输出字节写入缓冲区，例如 2KiB。我想在转换写入后上传这个缓冲区。

//simplified code...
uint bytesRead = 0;
byte[] buffer = new byte[buff_sz];
bytesRead = converter.Read(buffer);
while(bytesRead > 0)
{
    // here I would like to 'upload' the data (maybe using BackgroundUploader? or some other API?)
    bytesRead = converter.Read(buffer);
}

如果不遇到与以前相同的“竞争条件”，我不确定如何做到这一点。在将新字节放入缓冲区之前，如何保持 BackgroundUploader 运行？

注意 1：最后一次迭代将是 bytesRead

注意 2：转换代码位于跨平台 C++ 共享库中。

谢谢！

补充

根据 Nate Diamond 的建议，我研究了 IINputStream 接口。下面的代码可以工作，虽然可能是您编写的性​​能最差的版本，但足以证明这个概念。

以下基于“maxim pg”实现的包装代码从这里开始。 How can I implement IRandomAccessStream in C#?

 public IAsyncOperationWithProgress<IBuffer, UInt32> ReadAsync(
    IBuffer buffer,// The buffer into which the asynchronous read operation places the bytes that are read. 
    uint count,// The number of bytes to read that is less than or equal to the Capacity value.
    InputStreamOptions options) // Specifies the type of the asynchronous read operation.
{
    if (buffer == null) throw new ArgumentNullException("buffer");
    Func<CancellationToken, IProgress<uint>, Task<IBuffer>> taskProvider =
    (token, progress) => ReadBytesAsync(buffer, count, token, progress, options);
    return AsyncInfo.Run(taskProvider);
}

private Task<IBuffer> ReadBytesAsync(
    IBuffer buffer, 
    uint count, 
    CancellationToken token, 
    IProgress<uint> progress, 
    InputStreamOptions options)
{
    TaskCompletionSource<IBuffer> cts = new TaskCompletionSource<IBuffer>();
    try
    {
        var ignore = ThreadPool.RunAsync((handler) => {
            _buffer = new byte[count];
            uint bytesRead = _reader.Read(_buffer); // this triggers file conversion work
            buffer.Length = bytesRead; // this is important apparently, otherwise no data written!
            Stream stream = buffer.AsStream();
            stream.Write(_buffer, 0, (int)bytesRead);
            stream.Flush();
            cts.TrySetResult(buffer);
        });
    }
    catch(Exception e)
    {
        cts.SetException(e);
    }
    return cts.Task;
}

Answer 1

我相信最简单的方法是创建一个自定义IInputStream。这是一个只有一种方法的接口，ReadAsync。

然后您所要做的就是创建您的CustomInputStream，当它有足够的内容来填充请求的块时，它会不断地允许输入并从ReadAsync 返回。它还有一个Close 函数告诉它已经完成了。

Others have implemented something similar in IRandomAccessStream successfully.