/***

* ==++==

*

* Copyright (c) Microsoft Corporation. All rights reserved.

* Licensed under the Apache License, Version 2.0 (the "License");

* you may not use this file except in compliance with the License.

* You may obtain a copy of the License at

* http://www.apache.org/licenses/LICENSE-2.0

*

* Unless required by applicable law or agreed to in writing, software

* distributed under the License is distributed on an "AS IS" BASIS,

* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.

* See the License for the specific language governing permissions and

* limitations under the License.

*

* ==--==

* =+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+

*

* Asynchronous I/O: stream buffer implementation details

*

* We're going to some lengths to avoid exporting C++ class member functions and implementation details across

* module boundaries, and the factoring requires that we keep the implementation details away from the main header

* files. The supporting functions, which are in this file, use C-like signatures to avoid as many issues as

* possible.

*

* For the latest on this and related APIs, please see http://casablanca.codeplex.com.

*

* =-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-

****/

#include "stdafx.h"

#include "cpprest/details/fileio.h"

#include "robuffer.h"

using namespace ::Windows::Foundation;

using namespace ::Windows::Storage;

using namespace ::Windows::Storage::Streams;

using namespace ::Windows::Networking;

using namespace ::Windows::Networking::Sockets;

namespace Concurrency { namespace streams { namespace details {

/// <summary>

/// The public parts of the file information record contain only what is implementation-

/// independent. The actual allocated record is larger and has details that the implementation

/// require in order to function.

/// </summary>

struct _file_info_impl : _file_info

{

_file_info_impl(Streams::IRandomAccessStream^ stream, std::ios_base::openmode mode) :

m_stream(stream),

m_writer(nullptr),

_file_info(mode, 0)

{

m_pendingWrites = pplx::task_from_result();

}

_file_info_impl(Streams::IRandomAccessStream^ stream, std::ios_base::openmode mode, size_t buffer_size) :

m_stream(stream),

m_writer(nullptr),

_file_info(mode, buffer_size)

{

m_pendingWrites = pplx::task_from_result();

}

Streams::IRandomAccessStream^ m_stream;

Streams::IDataWriter^ m_writer;

pplx::task<void> m_pendingWrites;

};

}}}

using namespace Concurrency::streams::details;

#pragma warning(push)

#pragma warning(disable : 4100)

/// <summary>

/// Translate from C++ STL file open modes to Win32 flags.

/// </summary>

/// <param name="mode">The C++ file open mode</param>

/// <param name="prot">The C++ file open protection</param>

/// <param name="dwDesiredAccess">A pointer to a DWORD that will hold the desired access flags</param>

/// <param name="dwCreationDisposition">A pointer to a DWORD that will hold the creation disposition</param>

/// <param name="dwShareMode">A pointer to a DWORD that will hold the share mode</param>

void _get_create_flags(std::ios_base::openmode mode, int prot, FileAccessMode &acc_mode, CreationCollisionOption &options)

{

options = CreationCollisionOption::OpenIfExists;

acc_mode = FileAccessMode::ReadWrite;

if ( (mode & std::ios_base::in) && !(mode & std::ios_base::out) )

acc_mode = FileAccessMode::Read;

if ( mode & std::ios_base::trunc )

options = CreationCollisionOption::ReplaceExisting;

}

/// <summary>

/// Perform post-CreateFile processing.

/// </summary>

/// <param name="fh">The Win32 file handle</param>

/// <param name="callback">The callback interface pointer</param>

/// <param name="mode">The C++ file open mode</param>

/// <returns>The error code if there was an error in file creation.</returns>

void _finish_create(Streams::IRandomAccessStream^ stream, _In_ _filestream_callback *callback, std::ios_base::openmode mode, int prot)

{

_file_info_impl *info = nullptr;

info = new _file_info_impl(stream, mode, 512);

// Seek to end if it's in appending write mode

if ((mode & std::ios_base::out) && (mode & std::ios_base::app || mode & std::ios_base::ate ))

{

_seekwrpos_fsb(info, static_cast<size_t>(stream->Size), 1);

}

callback->on_opened(info);

}

/// <summary>

/// Create a streambuf instance to represent a WinRT file.

/// </summary>

/// <param name="callback">A pointer to the callback interface to invoke when the file has been opened.</param>

/// <param name="file">The file object</param>

/// <param name="mode">A creation mode for the stream buffer</param>

/// <returns>True if the opening operation could be initiated, false otherwise.</returns>

/// <remarks>

/// True does not signal that the file will eventually be successfully opened, just that the process was started.

/// This is only available for WinRT.

/// </remarks>

bool __cdecl _open_fsb_stf_str(_In_ Concurrency::streams::details::_filestream_callback *callback, ::Windows::Storage::StorageFile^ file, std::ios_base::openmode mode, int prot)

{

_ASSERTE(callback != nullptr);

_ASSERTE(file != nullptr);

CreationCollisionOption options;

FileAccessMode acc_mode;

_get_create_flags(mode, prot, acc_mode, options);

pplx::create_task(file->OpenAsync(acc_mode)).then(

[=](pplx::task<Streams::IRandomAccessStream^> sop)

{

try

{

_finish_create(sop.get(), callback, mode, prot);

}

catch(Platform::Exception^ exc)

{

callback->on_error(std::make_exception_ptr(utility::details::create_system_error(exc->HResult)));

}

});

return true;

}

bool __cdecl _sync_fsb_winrt(_In_ Concurrency::streams::details::_file_info *info, _In_opt_ Concurrency::streams::details::_filestream_callback *callback)

{

_ASSERTE(info != nullptr);

_file_info_impl *fInfo = static_cast<_file_info_impl *>(info);

pplx::extensibility::scoped_recursive_lock_t lck(fInfo->m_lock);

if ( fInfo->m_stream == nullptr || fInfo->m_writer == nullptr || !fInfo->m_stream->CanWrite)

return false;

// take a snapshot of current writer, since writer can be replaced during flush

auto writer = fInfo->m_writer;

// Flush operation will not begin until all previous writes (StoreAsync) finished, thus it could avoid race.

fInfo->m_pendingWrites = fInfo->m_pendingWrites.then([=] {

return writer->StoreAsync();

}).then([=](unsigned int) {

fInfo->m_buffill = 0;

return writer->FlushAsync();

}).then([=] (pplx::task<bool> result) {

// Rethrow exception if no callback attached.

if (callback == nullptr)

result.wait();

else

{

try

{

result.wait();

callback->on_completed(0);

}

catch (Platform::Exception^ exc)

{

callback->on_error(std::make_exception_ptr(utility::details::create_system_error(exc->HResult)));

}

}

});

return true;

}

/// <summary>

/// Close a file stream buffer.

/// </summary>

/// <param name="info">The file info record of the file</param>

/// <param name="callback">A pointer to the callback interface to invoke when the file has been opened.</param>

/// <returns>True if the closing operation could be initiated, false otherwise.</returns>

/// <remarks>

/// True does not signal that the file will eventually be successfully closed, just that the process was started.

/// </remarks>

bool __cdecl _close_fsb_nolock(_In_ _file_info **info, _In_ Concurrency::streams::details::_filestream_callback *callback)

{

_ASSERTE(callback != nullptr);

_ASSERTE(info != nullptr);

_ASSERTE(*info != nullptr);

_file_info_impl *fInfo = static_cast<_file_info_impl *>(*info);

*info = nullptr;

auto stream = fInfo->m_stream;

if (fInfo->m_stream->CanWrite)

{

_sync_fsb_winrt(fInfo, nullptr);

fInfo->m_pendingWrites.then([=] (pplx::task<void> t) {

try {

// The lock fInfo->m_lock must not be held at this point

delete fInfo;

t.wait();

callback->on_closed();

}

catch (Platform::Exception^ exc) {

callback->on_error(std::make_exception_ptr(utility::details::create_system_error(exc->HResult)));

}

});

}

else

{

// The lock fInfo->m_lock must not be held at this point

delete fInfo;

callback->on_closed();

}

return true;

}

bool __cdecl _close_fsb(_In_ _file_info **info, _In_ Concurrency::streams::details::_filestream_callback *callback)

{

return _close_fsb_nolock(info, callback);

}

/// <summary>

/// Initiate an asynchronous (overlapped) read from the file stream.

/// </summary>

/// <param name="info">The file info record of the file</param>

/// <param name="callback">A pointer to the callback interface to invoke when the write request is completed.</param>

/// <param name="ptr">A pointer to a buffer where the data should be placed</param>

/// <param name="count">The size (in bytes) of the buffer</param>

/// <param name="offset">The offset in the file to read from</param>

/// <returns>0 if the read request is still outstanding, -1 if the request failed, otherwise the size of the data read into the buffer</returns>

size_t __cdecl _read_file_async(_In_ Concurrency::streams::details::_file_info_impl *fInfo, _In_ Concurrency::streams::details::_filestream_callback *callback, _Out_writes_ (count) void *ptr, _In_ size_t count, size_t offset)

{

if ( fInfo->m_stream == nullptr )

{

if ( callback != nullptr )

{

// I don't know of a better error code, so this will have to do.

callback->on_error(std::make_exception_ptr(utility::details::create_system_error(ERROR_INVALID_ADDRESS)));

}

return 0;

}

auto reader = ref new Streams::DataReader(fInfo->m_stream->GetInputStreamAt(offset));

pplx::create_task(reader->LoadAsync(static_cast<unsigned int>(count))).then(

[=](pplx::task<unsigned int> result)

{

try

{

auto read = result.get();

if (read > 0)

{

reader->ReadBytes(Platform::ArrayReference<unsigned char>(static_cast<unsigned char *>(ptr), read));

}

callback->on_completed(read);

}

catch (Platform::Exception^ exc)

{

callback->on_error(std::make_exception_ptr(utility::details::create_system_error(exc->HResult)));

}

});

return 0;

}

template<typename Func>

class _filestream_callback_fill_buffer : public _filestream_callback

{

public:

_filestream_callback_fill_buffer(_In_ _file_info *info, const Func &func) : m_func(func), m_info(info) { }

virtual void on_completed(size_t result)

{

m_func(result);

delete this;

}

private:

_file_info *m_info;

Func m_func;

};

template<typename Func>

_filestream_callback_fill_buffer<Func> *create_callback(_In_ _file_info *info, const Func &func)

{

return new _filestream_callback_fill_buffer<Func>(info, func);

}

size_t _fill_buffer_fsb(_In_ _file_info_impl *fInfo, _In_ _filestream_callback *callback, size_t count, size_t char_size)

{

msl::safeint3::SafeInt<size_t> safeCount = count;

if ( fInfo->m_buffer == nullptr || safeCount > fInfo->m_bufsize )

{

if ( fInfo->m_buffer != nullptr )

delete fInfo->m_buffer;

fInfo->m_bufsize = safeCount.Max(fInfo->m_buffer_size);

fInfo->m_buffer = new char[fInfo->m_bufsize*char_size];

fInfo->m_bufoff = fInfo->m_rdpos;

auto cb = create_callback(fInfo,

[=] (size_t result)

{

{

pplx::extensibility::scoped_recursive_lock_t lck(fInfo->m_lock);

fInfo->m_buffill = result / char_size;

}

callback->on_completed(result);

});

auto read = _read_file_async(fInfo, cb, (uint8_t *)fInfo->m_buffer, fInfo->m_bufsize*char_size, fInfo->m_rdpos*char_size);

switch (read)

{

case 0:

// pending

return read;

case (-1):

// error

delete cb;

return read;

default:

// operation is complete. The pattern of returning synchronously

// has the expectation that we duplicate the callback code here...

// Do the expedient thing for now.

cb->on_completed(read);

// return pending

return 0;

};

}

// First, we need to understand how far into the buffer we have already read

// and how much remains.

size_t bufpos = fInfo->m_rdpos - fInfo->m_bufoff;

size_t bufrem = fInfo->m_buffill - bufpos;

// We have four different scenarios:

// 1. The read position is before the start of the buffer, in which case we will just reuse the buffer.

// 2. The read position is in the middle of the buffer, and we need to read some more.

// 3. The read position is beyond the end of the buffer. Do as in #1.

// 4. We have everything we need.

if ( (fInfo->m_rdpos < fInfo->m_bufoff) || (fInfo->m_rdpos >= (fInfo->m_bufoff+fInfo->m_buffill)) )

{

// Reuse the existing buffer.

fInfo->m_bufoff = fInfo->m_rdpos;

auto cb = create_callback(fInfo,

[=] (size_t result)

{

{

pplx::extensibility::scoped_recursive_lock_t lck(fInfo->m_lock);

fInfo->m_buffill = result / char_size;

}

callback->on_completed(bufrem*char_size+result);

});

auto read = _read_file_async(fInfo, cb, (uint8_t*)fInfo->m_buffer, fInfo->m_bufsize*char_size, fInfo->m_rdpos*char_size);

switch (read)

{

case 0:

// pending

return read;

case (-1):

// error

delete cb;

return read;

default:

// operation is complete. The pattern of returning synchronously

// has the expectation that we duplicate the callback code here...

// Do the expedient thing for now.

cb->on_completed(read);

// return pending

return 0;

};

}

else if ( bufrem < count )

{

fInfo->m_bufsize = safeCount.Max(fInfo->m_buffer_size);

// Then, we allocate a new buffer.

char *newbuf = new char[fInfo->m_bufsize*char_size];

// Then, we copy the unread part to the new buffer and delete the old buffer

if ( bufrem > 0 )

memcpy(newbuf, fInfo->m_buffer+bufpos*char_size, bufrem*char_size);

delete fInfo->m_buffer;

fInfo->m_buffer = newbuf;

// Then, we read the remainder of the count into the new buffer

fInfo->m_bufoff = fInfo->m_rdpos;

auto cb = create_callback(fInfo,

[=] (size_t result)

{

{

pplx::extensibility::scoped_recursive_lock_t lck(fInfo->m_lock);

fInfo->m_buffill = result / char_size;

}

callback->on_completed(bufrem*char_size+result);

});

auto read = _read_file_async(fInfo, cb, (uint8_t*)fInfo->m_buffer+bufrem*char_size, (fInfo->m_bufsize-bufrem)*char_size, (fInfo->m_rdpos+bufrem)*char_size);

switch (read)

{

case 0:

// pending

return read;

case (-1):

// error

delete cb;

return read;

default:

// operation is complete. The pattern of returning synchronously

// has the expectation that we duplicate the callback code here...

// Do the expedient thing for now.

cb->on_completed(read);

// return pending

return 0;

};

}

else

{

// If we are here, it means that we didn't need to read, we already have enough data in the buffer

return count*char_size;

}

}

/// <summary>

/// Read data from a file stream into a buffer

/// </summary>

/// <param name="info">The file info record of the file</param>

/// <param name="callback">A pointer to the callback interface to invoke when the write request is completed.</param>

/// <param name="ptr">A pointer to a buffer where the data should be placed</param>

/// <param name="count">The size (in characters) of the buffer</param>

/// <returns>0 if the read request is still outstanding, -1 if the request failed, otherwise the size of the data read into the buffer</returns>

size_t __cdecl _getn_fsb(_In_ Concurrency::streams::details::_file_info *info, _In_ Concurrency::streams::details::_filestream_callback *callback, _Out_writes_ (count) void *ptr, _In_ size_t count, size_t char_size)

{

_ASSERTE(callback != nullptr);

_ASSERTE(info != nullptr);

_ASSERTE(count > 0);

_file_info_impl *fInfo = static_cast<_file_info_impl *>(info);

pplx::extensibility::scoped_recursive_lock_t lck(info->m_lock);

if ( fInfo->m_buffer_size > 0 )

{

auto cb = create_callback(fInfo,

[=] (size_t read)

{

auto sz = count*char_size;

auto copy = (read < sz) ? read : sz;

auto bufoff = fInfo->m_rdpos - fInfo->m_bufoff;

memcpy(ptr, fInfo->m_buffer+bufoff*char_size, copy);

fInfo->m_atend = copy < sz;

callback->on_completed(copy);

});

size_t read = _fill_buffer_fsb(fInfo, cb, count, char_size);

if ( read > 0 )

{

auto sz = count*char_size;

auto copy = (read < sz) ? read : sz;

auto bufoff = fInfo->m_rdpos - fInfo->m_bufoff;

memcpy(ptr, fInfo->m_buffer+bufoff*char_size, copy);

fInfo->m_atend = copy < sz;

return copy;

}

return (size_t)read;

}

else

{

return _read_file_async(fInfo, callback, ptr, count*char_size, fInfo->m_rdpos*char_size);

}

}

/// <summary>

/// Write data from a buffer into the file stream.

/// </summary>

/// <param name="info">The file info record of the file</param>

/// <param name="callback">A pointer to the callback interface to invoke when the write request is completed.</param>

/// <param name="ptr">A pointer to a buffer where the data should be placed</param>

/// <param name="count">The size (in characters) of the buffer</param>

/// <returns>0 if the write request is still outstanding, -1 if the request failed, otherwise the size of the data read into the buffer</returns>

size_t __cdecl _putn_fsb(_In_ Concurrency::streams::details::_file_info *info, _In_ Concurrency::streams::details::_filestream_callback *callback, const void *ptr, size_t count, size_t char_size)

{

_ASSERTE(callback != nullptr);

_ASSERTE(info != nullptr);

_ASSERTE(count > 0);

_file_info_impl *fInfo = static_cast<_file_info_impl *>(info);

pplx::extensibility::scoped_recursive_lock_t lck(fInfo->m_lock);

if ( fInfo->m_stream == nullptr )

return static_cast<size_t>(-1);

// To preserve the async write order, we have to move the write head before read.

if (fInfo->m_wrpos != static_cast<size_t>(-1))

fInfo->m_wrpos += count;

msl::safeint3::SafeInt<unsigned int> safeWriteSize = count;

safeWriteSize *= char_size;

// In most of the time, we preserve the writer so that it would have better performance.

// However, after uer call seek, we will despose old writer. By doing so, users could

// write to new writer in new position while the old writer is still flushing data into stream.

if (fInfo->m_writer == nullptr)

{

fInfo->m_writer = ref new Streams::DataWriter(fInfo->m_stream);

fInfo->m_buffill = 0;

}

// It keeps tracking the number of bytes written into m_writer buffer.

fInfo->m_buffill += count;

// ArrayReference here is for avoiding data copy.

fInfo->m_writer->WriteBytes(Platform::ArrayReference<unsigned char>(const_cast<unsigned char *>(static_cast<const unsigned char*>(ptr)), safeWriteSize));

// Flush data from m_writer buffer into stream , if the buffer is full

if (fInfo->m_buffill >= fInfo->m_buffer_size)

{

fInfo->m_buffill = 0;

fInfo->m_pendingWrites = fInfo->m_pendingWrites.then([=] {

return fInfo->m_writer->StoreAsync();

}).then([=] (pplx::task<unsigned int> result) {

try

{

result.wait();

callback->on_completed(safeWriteSize);

}

catch (Platform::Exception^ exc)

{

callback->on_error(std::make_exception_ptr(utility::details::create_system_error(exc->HResult)));

}

});

return 0;

}

else

return safeWriteSize;

}

/// <summary>

/// Flush all buffered data to the underlying file.

/// </summary>

/// <param name="info">The file info record of the file</param>

/// <param name="callback">A pointer to the callback interface to invoke when the write request is completed.</param>

/// <returns>True if the request was initiated</returns>

bool __cdecl _sync_fsb(_In_ Concurrency::streams::details::_file_info *info, _In_ Concurrency::streams::details::_filestream_callback *callback)

{

return _sync_fsb_winrt(info, callback);

}

/// <summary>

/// Adjust the internal buffers and pointers when the application seeks to a new read location in the stream.

/// </summary>

/// <param name="info">The file info record of the file</param>

/// <param name="pos">The new position (offset from the start) in the file stream</param>

/// <returns>New file position or -1 if error</returns>

size_t __cdecl _seekrdpos_fsb(_In_ Concurrency::streams::details::_file_info *info, size_t pos, size_t char_size)

{

_ASSERTE(info != nullptr);

_file_info_impl *fInfo = static_cast<_file_info_impl *>(info);

pplx::extensibility::scoped_recursive_lock_t lck(info->m_lock);

if ( fInfo->m_stream == nullptr) return static_cast<size_t>(-1);;

if ( pos < fInfo->m_bufoff || pos > (fInfo->m_bufoff+fInfo->m_buffill) )

{

delete fInfo->m_buffer;

fInfo->m_buffer = nullptr;

fInfo->m_bufoff = fInfo->m_buffill = fInfo->m_bufsize = 0;

}

fInfo->m_rdpos = pos;

return fInfo->m_rdpos;

}

/// <summary>

/// Adjust the internal buffers and pointers when the application seeks to a new read location in the stream.

/// </summary>

/// <param name="info">The file info record of the file</param>

/// <param name="offset">The new position (offset from the end of the stream) in the file stream</param>

/// <param name="char_size">The size of the character type used for this stream</param>

/// <returns>New file position or -1 if error</returns>

_ASYNCRTIMP size_t __cdecl _seekrdtoend_fsb(_In_ Concurrency::streams::details::_file_info *info, int64_t offset, size_t char_size)

{

_ASSERTE(info != nullptr);

_file_info_impl *fInfo = static_cast<_file_info_impl *>(info);

return _seekrdpos_fsb(info, static_cast<size_t>(fInfo->m_stream->Size / char_size + offset), char_size);

}

utility::size64_t __cdecl _get_size(_In_ concurrency::streams::details::_file_info *info, size_t char_size)

{

_ASSERTE(info != nullptr);

_file_info_impl *fInfo = static_cast<_file_info_impl *>(info);

pplx::extensibility::scoped_recursive_lock_t lck(info->m_lock);

if ( fInfo->m_stream == nullptr) return 0;

return utility::size64_t(fInfo->m_stream->Size/char_size);

}

/// <summary>

/// Adjust the internal buffers and pointers when the application seeks to a new write location in the stream.

/// </summary>

/// <param name="info">The file info record of the file</param>

/// <param name="pos">The new position (offset from the start) in the file stream</param>

/// <returns>New file position or -1 if error</returns>

size_t __cdecl _seekwrpos_fsb(_In_ Concurrency::streams::details::_file_info *info, size_t pos, size_t char_size)

{

_ASSERTE(info != nullptr);

_file_info_impl *fInfo = static_cast<_file_info_impl *>(info);

pplx::extensibility::scoped_recursive_lock_t lck(info->m_lock);

if ( fInfo->m_stream == nullptr) return static_cast<size_t>(-1);

fInfo->m_wrpos = pos;

// m_buffill keeps number of chars written into the m_writer buffer.

// We need to flush it into stream before seek the write head of the stream

if (fInfo->m_buffill > 0)

_sync_fsb_winrt(fInfo, nullptr);

// Moving write head should follow the flush operation. is_done test is for perf optimization.

if (fInfo->m_pendingWrites.is_done())

fInfo->m_stream->Seek(static_cast<long long>(pos) * char_size);

else

{

auto lastWriter = fInfo->m_writer;

fInfo->m_writer = nullptr;

fInfo->m_pendingWrites = fInfo->m_pendingWrites.then([=] {

// Detach stream could avoid stream destruction after writer get destructed.

lastWriter->DetachStream();

fInfo->m_stream->Seek(static_cast<long long>(pos) * char_size);

});

}

return fInfo->m_wrpos;

}

namespace Concurrency { namespace streams { namespace details

{

/// <summary>

/// This class acts as a bridge between WinRT input streams and Casablanca asynchronous streams.

/// </summary>

ref class IRandomAccessStream_bridge sealed : public Windows::Storage::Streams::IRandomAccessStream

{

public:

virtual property bool CanRead { bool get() { return m_buffer.can_read(); } }

virtual property bool CanWrite { bool get() { return m_buffer.can_write(); } }

virtual property uint64_t Position { uint64_t get() { return m_position; } }

virtual property uint64_t Size

{

uint64_t get()

{

if (!m_buffer.has_size())

return m_remembered_size;

return m_buffer.size();

}

void set(uint64_t sz)

{

if (!m_buffer.has_size() || !m_buffer.can_write())

m_remembered_size = sz;

else

m_buffer.seekoff(basic_streambuf<uint8_t>::pos_type(sz), std::ios_base::beg, std::ios_base::out);

}

}

virtual Windows::Storage::Streams::IRandomAccessStream^ CloneStream()

{

return ref new IRandomAccessStream_bridge(m_buffer);

}

virtual Windows::Storage::Streams::IInputStream^ GetInputStreamAt(uint64_t position)

{

if ( !m_buffer.can_read() ) return nullptr;

concurrency::streams::streambuf<uint8_t>::pos_type pos = position;

if ( m_buffer.can_seek() || pos == m_buffer.getpos(std::ios_base::in) )

{

return ref new IRandomAccessStream_bridge(m_buffer,position);

}

return nullptr;

}

virtual Windows::Storage::Streams::IOutputStream^ GetOutputStreamAt(uint64_t position)

{

if ( !m_buffer.can_write() ) return nullptr;

concurrency::streams::streambuf<uint8_t>::pos_type pos = position;

if ( m_buffer.can_seek() || pos == m_buffer.getpos(std::ios_base::out) )

{

return ref new IRandomAccessStream_bridge(m_buffer,position);

}

return nullptr;

};

virtual void Seek(uint64_t position)

{

if (!m_buffer.can_seek())

throw ref new Platform::InvalidArgumentException(L"underlying buffer cannot seek");

m_position = position;

m_buffer.seekpos(concurrency::streams::streambuf<uint8_t>::pos_type(m_position),std::ios_base::in);

m_buffer.seekpos(concurrency::streams::streambuf<uint8_t>::pos_type(m_position),std::ios_base::out);

}

virtual Windows::Foundation::IAsyncOperationWithProgress<unsigned int, unsigned int>^ WriteAsync(Windows::Storage::Streams::IBuffer^ buffer);

virtual Windows::Foundation::IAsyncOperationWithProgress<::Windows::Storage::Streams::IBuffer^, unsigned int>^ ReadAsync(::Windows::Storage::Streams::IBuffer^ buffer, unsigned int count, Windows::Storage::Streams::InputStreamOptions options);

virtual Windows::Foundation::IAsyncOperation<bool>^ FlushAsync();

virtual ~IRandomAccessStream_bridge()

{

}

internal:

IRandomAccessStream_bridge(const concurrency::streams::streambuf<uint8_t> &buffer) :

m_buffer(buffer),

m_remembered_size(0),

m_position(0)

{

}

IRandomAccessStream_bridge(const concurrency::streams::streambuf<uint8_t> &buffer,

concurrency::streams::streambuf<uint8_t>::pos_type position) :

m_buffer(buffer),

m_remembered_size(0),

m_position(position)

{

}

private:

uint64_t m_remembered_size;

concurrency::streams::streambuf<uint8_t>::pos_type m_position;

concurrency::streams::streambuf<uint8_t> m_buffer;

};

struct _alloc_protector

{

_alloc_protector(concurrency::streams::streambuf<uint8_t>& buffer) :

m_buffer(buffer), m_size(0)

{

}

~_alloc_protector()

{

m_buffer.commit(m_size);

}

size_t m_size;

private:

_alloc_protector& operator=(const _alloc_protector&);

concurrency::streams::streambuf<uint8_t>& m_buffer;

};

struct _acquire_protector

{

_acquire_protector(concurrency::streams::streambuf<uint8_t>& buffer, uint8_t* ptr) :

m_buffer(buffer), m_ptr(ptr), m_size(0)

{

}

~_acquire_protector()

{

m_buffer.release(m_ptr, m_size);

}

size_t m_size;

private:

_acquire_protector& operator=(const _acquire_protector&);

uint8_t* m_ptr;

concurrency::streams::streambuf<uint8_t>& m_buffer;

};

// Rather than using ComPtr, which is somewhat complex, a simple RAII class

// to make sure that Release() is called is useful here.

struct _IUnknown_protector

{

_IUnknown_protector(IUnknown* unk_ptr) : m_unknown(unk_ptr) {}

~_IUnknown_protector() { if (m_unknown != nullptr) m_unknown->Release(); }

private:

IUnknown* m_unknown;

};

Windows::Foundation::IAsyncOperationWithProgress<::Windows::Storage::Streams::IBuffer^, unsigned int>^

IRandomAccessStream_bridge::ReadAsync(::Windows::Storage::Streams::IBuffer^ buffer, unsigned int count, Windows::Storage::Streams::InputStreamOptions options)

{

if (!m_buffer.can_read())

{

return pplx::create_async([buffer](pplx::progress_reporter<uint32> reporter) { return buffer; });

}

if (buffer->Capacity < count)

return pplx::create_async([buffer](pplx::progress_reporter<uint32> reporter) { return buffer; });

m_buffer.seekpos(concurrency::streams::streambuf<uint8_t>::pos_type(m_position),std::ios_base::in);

concurrency::streams::streambuf<uint8_t> streambuf = m_buffer;

return pplx::create_async(

[streambuf,buffer,options,count](pplx::progress_reporter<uint32> reporter)

{

auto sbuf = streambuf;

auto local_buf = ref new ::Platform::Array<unsigned char, 1>(count);

uint8_t* ptr = nullptr;

size_t acquired_size = 0;

if ( sbuf.acquire(ptr, acquired_size) && acquired_size >= count )

{

_acquire_protector prot(sbuf, ptr);

IUnknown* pUnk = reinterpret_cast<IUnknown*>(buffer);

::Windows::Storage::Streams::IBufferByteAccess* pBufferByteAccess = nullptr;

HRESULT hr = pUnk->QueryInterface(IID_PPV_ARGS(&pBufferByteAccess));

__abi_ThrowIfFailed(hr);

_IUnknown_protector unkprot(pBufferByteAccess);

byte* buffer_data = nullptr;

hr = pBufferByteAccess->Buffer(&buffer_data);

__abi_ThrowIfFailed(hr);

memcpy(buffer_data,ptr,count);

prot.m_size = count;

buffer->Length = count;

return pplx::task_from_result(buffer);

}

else

{

if ( acquired_size > 0 )

{

sbuf.release(ptr, 0);

}

IUnknown* pUnk = reinterpret_cast<IUnknown*>(buffer);

::Windows::Storage::Streams::IBufferByteAccess* pBufferByteAccess = nullptr;

HRESULT hr = pUnk->QueryInterface(IID_PPV_ARGS(&pBufferByteAccess));

__abi_ThrowIfFailed(hr);

_IUnknown_protector unkprot(pBufferByteAccess);

byte* buffer_data = nullptr;

hr = pBufferByteAccess->Buffer(&buffer_data);

__abi_ThrowIfFailed(hr);

pBufferByteAccess->AddRef();

return sbuf.getn(buffer_data,count).then(

[buffer,pBufferByteAccess,count](pplx::task<size_t> written)

{

_IUnknown_protector unkprot(pBufferByteAccess);

buffer->Length = (unsigned int)written.get();

return pplx::task_from_result(buffer);

});

}

});

}

Windows::Foundation::IAsyncOperationWithProgress<unsigned int, unsigned int>^

IRandomAccessStream_bridge::WriteAsync(Windows::Storage::Streams::IBuffer^ buffer)

{

if (!m_buffer.can_write())

{

return pplx::create_async([](pplx::progress_reporter<uint32> reporter) { return 0U; });

}

m_buffer.seekpos(concurrency::streams::streambuf<uint8_t>::pos_type(m_position),std::ios_base::out);

concurrency::streams::streambuf<uint8_t> streambuf = m_buffer;

return pplx::create_async(

[buffer,streambuf](pplx::progress_reporter<uint32> reporter)

{

auto size = buffer->Length;

auto sbuf = streambuf;

uint8_t* ptr = sbuf.alloc(size);

if ( ptr != nullptr)

{

{

_alloc_protector prot(sbuf);

IUnknown* pUnk = reinterpret_cast<IUnknown*>(buffer);

::Windows::Storage::Streams::IBufferByteAccess* pBufferByteAccess = nullptr;

HRESULT hr = pUnk->QueryInterface(IID_PPV_ARGS(&pBufferByteAccess));

__abi_ThrowIfFailed(hr);

_IUnknown_protector unkprot(pBufferByteAccess);

byte* buffer_data = nullptr;

hr = pBufferByteAccess->Buffer(&buffer_data);

__abi_ThrowIfFailed(hr);

memcpy(ptr,buffer_data,size);

prot.m_size = size;

}

return pplx::task_from_result((unsigned int)size);

}

else

{

IUnknown* pUnk = reinterpret_cast<IUnknown*>(buffer);

::Windows::Storage::Streams::IBufferByteAccess* pBufferByteAccess = nullptr;

HRESULT hr = pUnk->QueryInterface(IID_PPV_ARGS(&pBufferByteAccess));

__abi_ThrowIfFailed(hr);

_IUnknown_protector unkprot(pBufferByteAccess);

byte* buffer_data = nullptr;

hr = pBufferByteAccess->Buffer(&buffer_data);

__abi_ThrowIfFailed(hr);

pBufferByteAccess->AddRef();

return sbuf.putn_nocopy(buffer_data, size).then(

[pBufferByteAccess](pplx::task<size_t> size)

{

pBufferByteAccess->Release();

return (unsigned int)size.get();

});

}

});

}

Windows::Foundation::IAsyncOperation<bool>^

IRandomAccessStream_bridge::FlushAsync()