filestream.h

/***
 * Copyright (C) Microsoft. All rights reserved.
 * Licensed under the MIT license. See LICENSE.txt file in the project root for full license information.
 *
 * =+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+
 *
 * Asynchronous File streams
 *
 * For the latest on this and related APIs, please see: https://github.com/Microsoft/cpprestsdk
 *
 * =-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-
 ****/
#pragma once

#ifndef CASA_FILE_STREAMS_H
#define CASA_FILE_STREAMS_H

#include "cpprest/astreambuf.h"
#include "cpprest/details/fileio.h"
#include "cpprest/streams.h"
#include <assert.h>

#ifndef _CONCRT_H
#ifndef _LWRCASE_CNCRRNCY
#define _LWRCASE_CNCRRNCY
// Note to reader: we're using lower-case namespace names everywhere, but the 'Concurrency' namespace
// is capitalized for historical reasons. The alias let's us pretend that style issue doesn't exist.
namespace Concurrency
{
}
namespace concurrency = Concurrency;
#endif
#endif

namespace Concurrency
{
namespace streams
{
// Forward declarations
template<typename _CharType>
class file_buffer;

namespace details
{
// This operation queue is NOT thread safe
class async_operation_queue
{
    pplx::task<void> m_lastOperation;

public:
    async_operation_queue() { m_lastOperation = pplx::task_from_result(); }

    // It only accepts functors that take no argument and return pplx::task<T>
    // This function may execute op inline, thus it could throw immediately
    template<typename Func>
    auto enqueue_operation(Func&& op) -> decltype(op())
    {
        decltype(op()) res; // res is task<T> , which always has default constructor
        if (m_lastOperation.is_done())
        {
            res = op(); // Exceptions are expected to be thrown directly without catching
            if (res.is_done()) return res;
        }
        else
        {
            res = m_lastOperation.then([=] {
                return op(); // It will cause task unwrapping
            });
        }
        m_lastOperation = res.then([&](decltype(op())) {
            // This empty task is necessary for keeping the rest of the operations on the list running
            // even when the previous operation gets error.
            // Don't observe exception here.
        });
        return res;
    }

    void wait() const { m_lastOperation.wait(); }
};

/// <summary>
/// Private stream buffer implementation for file streams.
/// The class itself should not be used in application code, it is used by the stream definitions farther down in the
/// header file.
/// </summary>
template<typename _CharType>
class basic_file_buffer : public details::streambuf_state_manager<_CharType>
{
public:
    typedef typename basic_streambuf<_CharType>::traits traits;
    typedef typename basic_streambuf<_CharType>::int_type int_type;
    typedef typename basic_streambuf<_CharType>::pos_type pos_type;
    typedef typename basic_streambuf<_CharType>::off_type off_type;

    virtual ~basic_file_buffer()
    {
        if (this->can_read())
        {
            this->_close_read().wait();
        }

        if (this->can_write())
        {
            this->_close_write().wait();
        }
    }

protected:
    /// <summary>
    /// <c>can_seek</c> is used to determine whether a stream buffer supports seeking.
    /// </summary>
    virtual bool can_seek() const { return this->is_open(); }

    /// <summary>
    /// <c>has_size<c/> is used to determine whether a stream buffer supports size().
    /// </summary>
    virtual bool has_size() const { return this->is_open(); }

    virtual utility::size64_t size() const
    {
        if (!this->is_open()) return 0;
        return _get_size(m_info, sizeof(_CharType));
    }

    /// <summary>
    /// Gets the stream buffer size, if one has been set.
    /// </summary>
    /// <param name="direction">The direction of buffering (in or out)</param>
    /// <remarks>An implementation that does not support buffering will always return '0'.</remarks>
    virtual size_t buffer_size(std::ios_base::openmode direction = std::ios_base::in) const
    {
        if (direction == std::ios_base::in)
            return m_info->m_buffer_size;
        else
            return 0;
    }

    /// <summary>
    /// Sets the stream buffer implementation to buffer or not buffer.
    /// </summary>
    /// <param name="size">The size to use for internal buffering, 0 if no buffering should be done.</param>
    /// <param name="direction">The direction of buffering (in or out)</param>
    /// <remarks>An implementation that does not support buffering will silently ignore calls to this function and it
    /// will not have
    ///          any effect on what is returned by subsequent calls to buffer_size().</remarks>
    virtual void set_buffer_size(size_t size, std::ios_base::openmode direction = std::ios_base::in)
    {
        if (direction == std::ios_base::out) return;

        m_info->m_buffer_size = size;

        if (size == 0 && m_info->m_buffer != nullptr)
        {
            delete m_info->m_buffer;
            m_info->m_buffer = nullptr;
        }
    }

    /// <summary>
    /// For any input stream, <c>in_avail</c> returns the number of characters that are immediately available
    /// to be consumed without blocking. May be used in conjunction with <cref="::sbumpc method"/> to read data without
    /// incurring the overhead of using tasks.
    /// </summary>
    virtual size_t in_avail() const
    {
        pplx::extensibility::scoped_recursive_lock_t lck(m_info->m_lock);

        return _in_avail_unprot();
    }

    size_t _in_avail_unprot() const
    {
        if (!this->is_open()) return 0;

        if (m_info->m_buffer == nullptr || m_info->m_buffill == 0) return 0;
        if (m_info->m_bufoff > m_info->m_rdpos || (m_info->m_bufoff + m_info->m_buffill) < m_info->m_rdpos) return 0;

        msl::safeint3::SafeInt<size_t> rdpos(m_info->m_rdpos);
        msl::safeint3::SafeInt<size_t> buffill(m_info->m_buffill);
        msl::safeint3::SafeInt<size_t> bufpos = rdpos - m_info->m_bufoff;

        return buffill - bufpos;
    }

    _file_info* _close_stream()
    {
        // indicate that we are no longer open
        auto fileInfo = m_info;
        m_info = nullptr;
        return fileInfo;
    }

    static pplx::task<void> _close_file(_In_ _file_info* fileInfo)
    {
        pplx::task_completion_event<void> result_tce;
        auto callback = new _filestream_callback_close(result_tce);

        if (!_close_fsb_nolock(&fileInfo, callback))
        {
            delete callback;
            return pplx::task_from_result();
        }
        return pplx::create_task(result_tce);
    }

    // Workaround GCC compiler bug https://gcc.gnu.org/bugzilla/show_bug.cgi?id=58972
    void _invoke_parent_close_read() { streambuf_state_manager<_CharType>::_close_read(); }

    pplx::task<void> _close_read()
    {
        return m_readOps.enqueue_operation([this] {
            _invoke_parent_close_read();

            if (this->can_write())
            {
                return pplx::task_from_result();
            }
            else
            {
                // Neither heads are open. Close the underlying device
                // to indicate that we are no longer open
                auto fileInfo = _close_stream();

                return _close_file(fileInfo);
            }
        });
    }

    pplx::task<void> _close_write()
    {
        streambuf_state_manager<_CharType>::_close_write();
        if (this->can_read())
        {
            // Read head is still open. Just flush the write data
            return flush_internal();
        }
        else
        {
            // Neither heads are open. Close the underlying device

            // We need to flush all writes if the file was opened for writing.
            return flush_internal().then([=](pplx::task<void> flushTask) -> pplx::task<void> {
                // swallow exception from flush
                try
                {
                    flushTask.wait();
                }
                catch (...)
                {
                }

                // indicate that we are no longer open
                auto fileInfo = this->_close_stream();

                return this->_close_file(fileInfo);
            });
        }
    }

    /// <summary>
    /// Writes a single byte to an output stream.
    /// </summary>
    /// <param name="ch">The byte to write</param>
    /// <returns>A <c>task</c> that holds the value of the byte written. This is EOF if the write operation
    /// fails.</returns>
    virtual pplx::task<int_type> _putc(_CharType ch)
    {
        auto result_tce = pplx::task_completion_event<size_t>();
        auto callback = new _filestream_callback_write<size_t>(m_info, result_tce);

        // Potentially we should consider deprecating this API, it is TERRIBLY inefficient.
        std::shared_ptr<_CharType> sharedCh;
        try
        {
            sharedCh = std::make_shared<_CharType>(ch);
        }
        catch (const std::bad_alloc&)
        {
            delete callback;
            throw;
        }

        size_t written = _putn_fsb(m_info, callback, sharedCh.get(), 1, sizeof(_CharType));
        if (written == sizeof(_CharType))
        {
            delete callback;
            return pplx::task_from_result<int_type>(ch);
        }

        return pplx::create_task(result_tce).then([sharedCh](size_t) { return static_cast<int_type>(*sharedCh); });
    }

    /// <summary>
    /// Allocates a contiguous memory block and returns it.
    /// </summary>
    /// <param name="count">The number of characters to allocate.</param>
    /// <returns>A pointer to a block to write to, null if the stream buffer implementation does not support
    /// alloc/commit.</returns>
    _CharType* _alloc(size_t) { return nullptr; }

    /// <summary>
    /// Submits a block already allocated by the stream buffer.
    /// </summary>
    /// <param name="ptr">Count of characters to be committed.</param>
    void _commit(size_t) {}

    /// <summary>
    /// Gets a pointer to the next already allocated contiguous block of data.
    /// </summary>
    /// <param name="ptr">A reference to a pointer variable that will hold the address of the block on success.</param>
    /// <param name="count">The number of contiguous characters available at the address in 'ptr'.</param>
    /// <returns><c>true</c> if the operation succeeded, <c>false</c> otherwise.</returns>
    /// <remarks>
    /// A return of false does not necessarily indicate that a subsequent read operation would fail, only that
    /// there is no block to return immediately or that the stream buffer does not support the operation.
    /// The stream buffer may not de-allocate the block until <see cref="::release method" /> is called.
    /// If the end of the stream is reached, the function will return <c>true</c>, a null pointer, and a count of zero;
    /// a subsequent read will not succeed.
    /// </remarks>
    virtual bool acquire(_Out_ _CharType*& ptr, _Out_ size_t& count)
    {
        ptr = nullptr;
        count = 0;
        return false;
    }

    /// <summary>
    /// Releases a block of data acquired using <see cref="::acquire method"/>. This frees the stream buffer to
    /// de-allocate the memory, if it so desires. Move the read position ahead by the count.
    /// </summary>
    /// <param name="ptr">A pointer to the block of data to be released.</param>
    /// <param name="count">The number of characters that were read.</param>
    virtual void release(_Out_writes_(count) _CharType*, _In_ size_t count) { (void)(count); }

    /// <summary>
    /// Writes a number of characters to the stream.
    /// </summary>
    /// <param name="ptr">A pointer to the block of data to be written.</param>
    /// <param name="count">The number of characters to write.</param>
    /// <returns>A <c>task</c> that holds the number of characters actually written, either 'count' or 0.</returns>
    virtual pplx::task<size_t> _putn(const _CharType* ptr, size_t count)
    {
        auto result_tce = pplx::task_completion_event<size_t>();
        auto callback = new _filestream_callback_write<size_t>(m_info, result_tce);

        size_t written = _putn_fsb(m_info, callback, ptr, count, sizeof(_CharType));

        if (written != 0 && written != size_t(-1))
        {
            delete callback;
            written = written / sizeof(_CharType);
            return pplx::task_from_result<size_t>(written);
        }
        return pplx::create_task(result_tce);
    }

    // Temporarily needed until the deprecated putn is removed.
    virtual pplx::task<size_t> _putn(const _CharType* ptr, size_t count, bool copy)
    {
        if (copy)
        {
            auto sharedData = std::make_shared<std::vector<_CharType>>(ptr, ptr + count);
            return _putn(ptr, count).then([sharedData](size_t size) { return size; });
        }
        else
        {
            return _putn(ptr, count);
        }
    }

    /// <summary>
    /// Reads a single byte from the stream and advance the read position.
    /// </summary>
    /// <returns>A <c>task</c> that holds the value of the byte read. This is EOF if the read fails.</returns>
    virtual pplx::task<int_type> _bumpc()
    {
        return m_readOps.enqueue_operation([this]() -> pplx::task<int_type> {
            if (_in_avail_unprot() > 0)
            {
                pplx::extensibility::scoped_recursive_lock_t lck(m_info->m_lock);

                // Check again once the lock is held.

                if (_in_avail_unprot() > 0)
                {
                    auto bufoff = m_info->m_rdpos - m_info->m_bufoff;
                    _CharType ch = m_info->m_buffer[bufoff * sizeof(_CharType)];
                    m_info->m_rdpos += 1;
                    return pplx::task_from_result<int_type>(ch);
                }
            }

            auto result_tce = pplx::task_completion_event<int_type>();
            auto callback = new _filestream_callback_bumpc(m_info, result_tce);

            size_t ch = _getn_fsb(m_info, callback, &callback->m_ch, 1, sizeof(_CharType));

            if (ch == sizeof(_CharType))
            {
                pplx::extensibility::scoped_recursive_lock_t lck(m_info->m_lock);
                m_info->m_rdpos += 1;
                _CharType ch1 = (_CharType)callback->m_ch;
                delete callback;
                return pplx::task_from_result<int_type>(ch1);
            }
            return pplx::create_task(result_tce);
        });
    }

    /// <summary>
    /// Reads a single byte from the stream and advance the read position.
    /// </summary>
    /// <returns>The value of the byte. EOF if the read fails. <see cref="::requires_async method" /> if an asynchronous
    /// read is required</returns> <remarks>This is a synchronous operation, but is guaranteed to never block.</remarks>
    virtual int_type _sbumpc()
    {
        m_readOps.wait();
        if (m_info->m_atend) return traits::eof();

        if (_in_avail_unprot() == 0) return traits::requires_async();

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

        if (_in_avail_unprot() == 0) return traits::requires_async();

        auto bufoff = m_info->m_rdpos - m_info->m_bufoff;
        _CharType ch = m_info->m_buffer[bufoff * sizeof(_CharType)];
        m_info->m_rdpos += 1;
        return (int_type)ch;
    }

    pplx::task<int_type> _getcImpl()
    {
        if (_in_avail_unprot() > 0)
        {
            pplx::extensibility::scoped_recursive_lock_t lck(m_info->m_lock);

            // Check again once the lock is held.

            if (_in_avail_unprot() > 0)
            {
                auto bufoff = m_info->m_rdpos - m_info->m_bufoff;
                _CharType ch = m_info->m_buffer[bufoff * sizeof(_CharType)];
                return pplx::task_from_result<int_type>(ch);
            }
        }

        auto result_tce = pplx::task_completion_event<int_type>();
        auto callback = new _filestream_callback_getc(m_info, result_tce);

        size_t ch = _getn_fsb(m_info, callback, &callback->m_ch, 1, sizeof(_CharType));

        if (ch == sizeof(_CharType))
        {
            pplx::extensibility::scoped_recursive_lock_t lck(m_info->m_lock);
            _CharType ch1 = (_CharType)callback->m_ch;
            delete callback;
            return pplx::task_from_result<int_type>(ch1);
        }
        return pplx::create_task(result_tce);
    }

    /// <summary>
    /// Reads a single byte from the stream without advancing the read position.
    /// </summary>
    /// <returns>The value of the byte. EOF if the read fails.</returns>
    pplx::task<int_type> _getc()
    {
        return m_readOps.enqueue_operation([this]() -> pplx::task<int_type> { return _getcImpl(); });
    }

    /// <summary>
    /// Reads a single byte from the stream without advancing the read position.
    /// </summary>
    /// <returns>The value of the byte. EOF if the read fails. <see cref="::requires_async method" /> if an asynchronous
    /// read is required</returns> <remarks>This is a synchronous operation, but is guaranteed to never block.</remarks>
    int_type _sgetc()
    {
        m_readOps.wait();
        if (m_info->m_atend) return traits::eof();

        if (_in_avail_unprot() == 0) return traits::requires_async();

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

        if (_in_avail_unprot() == 0) return traits::requires_async();

        auto bufoff = m_info->m_rdpos - m_info->m_bufoff;
        _CharType ch = m_info->m_buffer[bufoff * sizeof(_CharType)];
        return (int_type)ch;
    }

    /// <summary>
    /// Advances the read position, then return the next character without advancing again.
    /// </summary>
    /// <returns>A <c>task</c> that holds the value of the byte, which is EOF if the read fails.</returns>
    virtual pplx::task<int_type> _nextc()
    {
        return m_readOps.enqueue_operation([this]() -> pplx::task<int_type> {
            _seekrdpos_fsb(m_info, m_info->m_rdpos + 1, sizeof(_CharType));
            if (m_info->m_atend) return pplx::task_from_result(basic_file_buffer<_CharType>::traits::eof());
            return this->_getcImpl();
        });
    }

    /// <summary>
    /// Retreats the read position, then return the current character without advancing.
    /// </summary>
    /// <returns>A <c>task</c> that holds the value of the byte. The value is EOF if the read fails,
    /// <c>requires_async</c> if an asynchronous read is required</returns>
    virtual pplx::task<int_type> _ungetc()
    {
        return m_readOps.enqueue_operation([this]() -> pplx::task<int_type> {
            if (m_info->m_rdpos == 0)
                return pplx::task_from_result<int_type>(basic_file_buffer<_CharType>::traits::eof());
            _seekrdpos_fsb(m_info, m_info->m_rdpos - 1, sizeof(_CharType));
            return this->_getcImpl();
        });
    }

    /// <summary>
    /// Reads up to a given number of characters from the stream.
    /// </summary>
    /// <param name="ptr">The address of the target memory area</param>
    /// <param name="count">The maximum number of characters to read</param>
    /// <returns>A <c>task</c> that holds the number of characters read. This number is O if the end of the stream is
    /// reached, EOF if there is some error.</returns>
    virtual pplx::task<size_t> _getn(_Out_writes_(count) _CharType* ptr, _In_ size_t count)
    {
        return m_readOps.enqueue_operation([=]() -> pplx::task<size_t> {
            if (m_info->m_atend || count == 0) return pplx::task_from_result<size_t>(0);

            if (_in_avail_unprot() >= count)
            {
                pplx::extensibility::scoped_recursive_lock_t lck(m_info->m_lock);

                // Check again once the lock is held.

                if (_in_avail_unprot() >= count)
                {
                    auto bufoff = m_info->m_rdpos - m_info->m_bufoff;
                    std::memcpy(
                        (void*)ptr, this->m_info->m_buffer + bufoff * sizeof(_CharType), count * sizeof(_CharType));

                    m_info->m_rdpos += count;
                    return pplx::task_from_result<size_t>(count);
                }
            }

            auto result_tce = pplx::task_completion_event<size_t>();
            auto callback = new _filestream_callback_read(m_info, result_tce);

            size_t read = _getn_fsb(m_info, callback, ptr, count, sizeof(_CharType));

            if (read != 0 && read != size_t(-1))
            {
                delete callback;
                pplx::extensibility::scoped_recursive_lock_t lck(m_info->m_lock);
                m_info->m_rdpos += read / sizeof(_CharType);
                return pplx::task_from_result<size_t>(read / sizeof(_CharType));
            }
            return pplx::create_task(result_tce);
        });
    }

    /// <summary>
    /// Reads up to a given number of characters from the stream.
    /// </summary>
    /// <param name="ptr">The address of the target memory area</param>
    /// <param name="count">The maximum number of characters to read</param>
    /// <returns>The number of characters read. O if the end of the stream is reached or an asynchronous read is
    /// required.</returns> <remarks>This is a synchronous operation, but is guaranteed to never block.</remarks>
    size_t _sgetn(_Out_writes_(count) _CharType* ptr, _In_ size_t count)
    {
        m_readOps.wait();
        if (m_info->m_atend) return 0;

        if (count == 0 || in_avail() == 0) return 0;

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

        size_t available = _in_avail_unprot();
        size_t copy = (count < available) ? count : available;

        auto bufoff = m_info->m_rdpos - m_info->m_bufoff;
        std::memcpy((void*)ptr, this->m_info->m_buffer + bufoff * sizeof(_CharType), copy * sizeof(_CharType));

        m_info->m_rdpos += copy;
        m_info->m_atend = (copy < count);
        return copy;
    }

    /// <summary>
    /// Copies up to a given number of characters from the stream.
    /// </summary>
    /// <param name="ptr">The address of the target memory area</param>
    /// <param name="count">The maximum number of characters to copy</param>
    /// <returns>The number of characters copied. O if the end of the stream is reached or an asynchronous read is
    /// required.</returns> <remarks>This is a synchronous operation, but is guaranteed to never block.</remarks>
    virtual size_t _scopy(_CharType*, size_t) { return 0; }

    /// <summary>
    /// Gets the current read or write position in the stream.
    /// </summary>
    /// <param name="direction">The I/O direction to seek (see remarks)</param>
    /// <returns>The current position. EOF if the operation fails.</returns>
    /// <remarks>Some streams may have separate write and read cursors.
    ///          For such streams, the direction parameter defines whether to move the read or the write
    ///          cursor.</remarks>
    virtual pos_type getpos(std::ios_base::openmode mode) const
    {
        return const_cast<basic_file_buffer*>(this)->seekoff(0, std::ios_base::cur, mode);
    }

    /// <summary>
    /// Seeks to the given position.
    /// </summary>
    /// <param name="pos">The offset from the beginning of the stream</param>
    /// <param name="direction">The I/O direction to seek (see remarks)</param>
    /// <returns>The position. EOF if the operation fails.</returns>
    /// <remarks>Some streams may have separate write and read cursors.
    ///          For such streams, the direction parameter defines whether to move the read or the write
    ///          cursor.</remarks>
    virtual pos_type seekpos(pos_type pos, std::ios_base::openmode mode)
    {
        if (mode == std::ios_base::in)
        {
            m_readOps.wait();
            return (pos_type)_seekrdpos_fsb(m_info, size_t(pos), sizeof(_CharType));
        }
        else if ((m_info->m_mode & std::ios::ios_base::app) == 0)
        {
            return (pos_type)_seekwrpos_fsb(m_info, size_t(pos), sizeof(_CharType));
        }
        return (pos_type)Concurrency::streams::char_traits<_CharType>::eof();
    }

    /// <summary>
    /// Seeks to a position given by a relative offset.
    /// </summary>
    /// <param name="offset">The relative position to seek to</param>
    /// <param name="way">The starting point (beginning, end, current) for the seek.</param>
    /// <param name="mode">The I/O direction to seek (see remarks)</param>
    /// <returns>The position. EOF if the operation fails.</returns>
    /// <remarks>Some streams may have separate write and read cursors.
    ///          For such streams, the mode parameter defines whether to move the read or the write cursor.</remarks>
    virtual pos_type seekoff(off_type offset, std::ios_base::seekdir way, std::ios_base::openmode mode)
    {
        if (mode == std::ios_base::in)
        {
            m_readOps.wait();
            size_t current_pos = static_cast<size_t>(-1);
            switch (way)
            {
                case std::ios_base::beg: return (pos_type)_seekrdpos_fsb(m_info, size_t(offset), sizeof(_CharType));
                case std::ios_base::cur:
                    return (pos_type)_seekrdpos_fsb(m_info, size_t(m_info->m_rdpos + offset), sizeof(_CharType));
                case std::ios_base::end:
                    current_pos = _seekrdtoend_fsb(m_info, int64_t(offset), sizeof(_CharType));
                    if (current_pos == static_cast<size_t>(-1))
                    {
                        return -1;
                    }
                    return (pos_type)current_pos;
                default:
                    // Fail on invalid input (_S_ios_seekdir_end)
                    assert(false);
            }
        }
        else if ((m_info->m_mode & std::ios::ios_base::app) == 0)
        {
            switch (way)
            {
                case std::ios_base::beg: return (pos_type)_seekwrpos_fsb(m_info, size_t(offset), sizeof(_CharType));
                case std::ios_base::cur:
                    return (pos_type)_seekwrpos_fsb(m_info, size_t(m_info->m_wrpos + offset), sizeof(_CharType));
                case std::ios_base::end: return (pos_type)_seekwrpos_fsb(m_info, size_t(-1), sizeof(_CharType));
                default:
                    // Fail on invalid input (_S_ios_seekdir_end)
                    assert(false);
            }
        }
        return (pos_type)traits::eof();
    }

    /// <summary>
    /// For output streams, flush any internally buffered data to the underlying medium.
    /// </summary>
    virtual pplx::task<bool> _sync()
    {
        return flush_internal().then([]() { return true; });
    }

private:
    template<typename _CharType1>
    friend class ::concurrency::streams::file_buffer;

    pplx::task<void> flush_internal()
    {
        pplx::task_completion_event<void> result_tce;
        auto callback = utility::details::make_unique<_filestream_callback_write_b>(m_info, result_tce);

        if (!_sync_fsb(m_info, callback.get()))
        {
            return pplx::task_from_exception<void>(std::runtime_error("failure to flush stream"));
        }
        callback.release();
        return pplx::create_task(result_tce);
    }

    basic_file_buffer(_In_ _file_info* info) : streambuf_state_manager<_CharType>(info->m_mode), m_info(info) {}

#if !defined(__cplusplus_winrt)
    static pplx::task<std::shared_ptr<basic_streambuf<_CharType>>> open(
        const utility::string_t& _Filename,
        std::ios_base::openmode _Mode = std::ios_base::out,
#ifdef _WIN32
        int _Prot = (int)std::ios_base::_Openprot
#else
        int _Prot = 0                                         // unsupported on Linux, for now
#endif
    )
    {
        auto result_tce = pplx::task_completion_event<std::shared_ptr<basic_streambuf<_CharType>>>();
        auto callback = new _filestream_callback_open(result_tce);
        _open_fsb_str(callback, _Filename.c_str(), _Mode, _Prot);
        return pplx::create_task(result_tce);
    }

#else
    static pplx::task<std::shared_ptr<basic_streambuf<_CharType>>> open(
        ::Windows::Storage::StorageFile ^ file, std::ios_base::openmode _Mode = std::ios_base::out)
    {
        auto result_tce = pplx::task_completion_event<std::shared_ptr<basic_streambuf<_CharType>>>();
        auto callback = new _filestream_callback_open(result_tce);
        _open_fsb_stf_str(callback, file, _Mode, 0);
        return pplx::create_task(result_tce);
    }
#endif

    class _filestream_callback_open : public details::_filestream_callback
    {
    public:
        _filestream_callback_open(const pplx::task_completion_event<std::shared_ptr<basic_streambuf<_CharType>>>& op)
            : m_op(op)
        {
        }

        virtual void on_opened(_In_ _file_info* info)
        {
            m_op.set(std::shared_ptr<basic_file_buffer<_CharType>>(new basic_file_buffer<_CharType>(info)));
            delete this;
        }

        virtual void on_error(const std::exception_ptr& e)
        {
            m_op.set_exception(e);
            delete this;
        }

    private:
        pplx::task_completion_event<std::shared_ptr<basic_streambuf<_CharType>>> m_op;
    };

    class _filestream_callback_close : public details::_filestream_callback
    {
    public:
        _filestream_callback_close(const pplx::task_completion_event<void>& op) : m_op(op) {}

        virtual void on_closed()
        {
            m_op.set();
            delete this;
        }

        virtual void on_error(const std::exception_ptr& e)
        {
            m_op.set_exception(e);
            delete this;
        }

    private:
        pplx::task_completion_event<void> m_op;
    };

    template<typename ResultType>
    class _filestream_callback_write : public details::_filestream_callback
    {
    public:
        _filestream_callback_write(_In_ _file_info* info, const pplx::task_completion_event<ResultType>& op)
            : m_info(info), m_op(op)
        {
        }

        virtual void on_completed(size_t result)
        {
            m_op.set((ResultType)result / sizeof(_CharType));
            delete this;
        }

        virtual void on_error(const std::exception_ptr& e)
        {
            m_op.set_exception(e);
            delete this;
        }

    private:
        _file_info* m_info;
        pplx::task_completion_event<ResultType> m_op;
    };

    class _filestream_callback_write_b : public details::_filestream_callback
    {
    public:
        _filestream_callback_write_b(_In_ _file_info* info, const pplx::task_completion_event<void>& op)
            : m_info(info), m_op(op)
        {
        }

        virtual void on_completed(size_t)
        {
            m_op.set();
            delete this;
        }

        virtual void on_error(const std::exception_ptr& e)
        {
            m_op.set_exception(e);
            delete this;
        }

    private:
        _file_info* m_info;
        pplx::task_completion_event<void> m_op;
    };

    class _filestream_callback_read : public details::_filestream_callback
    {
    public:
        _filestream_callback_read(_In_ _file_info* info, const pplx::task_completion_event<size_t>& op)
            : m_info(info), m_op(op)
        {
        }

        virtual void on_completed(size_t result)
        {
            result = result / sizeof(_CharType);
            m_info->m_rdpos += result;
            m_op.set(result);
            delete this;
        }

        virtual void on_error(const std::exception_ptr& e)
        {
            m_op.set_exception(e);
            delete this;
        }

    private:
        _file_info* m_info;
        pplx::task_completion_event<size_t> m_op;
    };

    class _filestream_callback_bumpc : public details::_filestream_callback
    {
    public:
        _filestream_callback_bumpc(_In_ _file_info* info, const pplx::task_completion_event<int_type>& op)
            : m_ch(0), m_info(info), m_op(op)
        {
        }

        virtual void on_completed(size_t result)
        {
            if (result == sizeof(_CharType))
            {
                m_info->m_rdpos += 1;
                m_op.set(m_ch);
            }
            else
            {
                m_op.set(traits::eof());
            }
            delete this;
        }

        virtual void on_error(const std::exception_ptr& e)
        {
            m_op.set_exception(e);
            delete this;
        }

        int_type m_ch;

    private:
        _file_info* m_info;
        pplx::task_completion_event<int_type> m_op;
    };

    class _filestream_callback_getc : public details::_filestream_callback
    {
    public:
        _filestream_callback_getc(_In_ _file_info* info, const pplx::task_completion_event<int_type>& op)
            : m_ch(0), m_info(info), m_op(op)
        {
        }

        virtual void on_completed(size_t result)
        {
            if (result == sizeof(_CharType))
            {
                m_op.set(m_ch);
            }
            else
            {
                m_op.set(traits::eof());
            }
            delete this;
        }

        int_type m_ch;

        virtual void on_error(const std::exception_ptr& e)
        {
            m_op.set_exception(e);
            delete this;
        }

    private:
        _file_info* m_info;
        pplx::task_completion_event<int_type> m_op;
    };

    _file_info* m_info;
    async_operation_queue m_readOps;
};

} // namespace details

/// <summary>
/// Stream buffer for file streams.
/// </summary>
/// <typeparam name="_CharType">
/// The data type of the basic element of the <c>file_buffer</c>.
/// </typeparam>
template<typename _CharType>
class file_buffer
{
public:
#if !defined(__cplusplus_winrt)
    /// <summary>
    /// Open a new stream buffer representing the given file.
    /// </summary>
    /// <param name="file_name">The name of the file</param>
    /// <param name="mode">The opening mode of the file</param>
    /// <param name="prot">The file protection mode</param>
    /// <returns>A <c>task</c> that returns an opened stream buffer on completion.</returns>
    static pplx::task<streambuf<_CharType>> open(const utility::string_t& file_name,
                                                 std::ios_base::openmode mode = std::ios_base::out,
#ifdef _WIN32
                                                 int prot = _SH_DENYRD
#else
                                                 int prot = 0 // unsupported on Linux
#endif
    )
    {
        auto bfb = details::basic_file_buffer<_CharType>::open(file_name, mode, prot);
        return bfb.then(
            [](pplx::task<std::shared_ptr<details::basic_streambuf<_CharType>>> op) -> streambuf<_CharType> {
                return streambuf<_CharType>(op.get());
            });
    }

#else
    /// <summary>
    /// Open a new stream buffer representing the given file.
    /// </summary>
    /// <param name="file">The StorageFile instance</param>
    /// <param name="mode">The opening mode of the file</param>
    /// <param name="prot">The file protection mode</param>
    /// <returns>A <c>task</c> that returns an opened stream buffer on completion.</returns>
    static pplx::task<streambuf<_CharType>> open(::Windows::Storage::StorageFile ^ file,
                                                 std::ios_base::openmode mode = std::ios_base::out)
    {
        auto bfb = details::basic_file_buffer<_CharType>::open(file, mode);
        return bfb.then(
            [](pplx::task<std::shared_ptr<details::basic_streambuf<_CharType>>> op) -> streambuf<_CharType> {
                return streambuf<_CharType>(op.get());
            });
    }
#endif
};

/// <summary>
/// File stream class containing factory functions for file streams.
/// </summary>
/// <typeparam name="_CharType">
/// The data type of the basic element of the <c>file_stream</c>.
/// </typeparam>
template<typename _CharType>
class file_stream