astreambuf.h

/***
* ==++==
*
* 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. This is an extension to the PPL concurrency features and therefore
* lives in the Concurrency namespace.
*
* For the latest on this and related APIs, please see http://casablanca.codeplex.com.
*
* =-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-
****/
#pragma once

#include <ios>
#include <memory>
#include <cstring>
#include <math.h>

#include "pplx/pplxtasks.h"
#include "cpprest/details/basic_types.h"
#include "cpprest/asyncrt_utils.h"

namespace Concurrency
{
namespace streams
{
    template<typename _CharType>
    struct char_traits : std::char_traits<_CharType>
    {
        static typename std::char_traits<_CharType>::int_type requires_async() { return std::char_traits<_CharType>::eof()-1; }
    };
#if !defined(_WIN32)
    template<>
    struct char_traits<unsigned char> : private std::char_traits<char>
    {
    public:
        typedef unsigned char char_type;

        using std::char_traits<char>::eof;
        using std::char_traits<char>::int_type;
        using std::char_traits<char>::off_type;
        using std::char_traits<char>::pos_type;

        static size_t length(const unsigned char* str)
        {
            return std::char_traits<char>::length(reinterpret_cast<const char*>(str));
        }

        static void assign(unsigned char& left, const unsigned char& right) { left = right; }
        static unsigned char* assign(unsigned char* left, size_t n, unsigned char value)
        {
            return reinterpret_cast<unsigned char*>(std::char_traits<char>::assign(reinterpret_cast<char*>(left), n, static_cast<char>(value)));
        }

        static unsigned char* copy(unsigned char* left, const unsigned char* right, size_t n)
        {
            return reinterpret_cast<unsigned char*>(std::char_traits<char>::copy(reinterpret_cast<char*>(left), reinterpret_cast<const char*>(right), n));
        }

        static unsigned char* move(unsigned char* left, const unsigned char* right, size_t n)
        {
            return reinterpret_cast<unsigned char*>(std::char_traits<char>::move(reinterpret_cast<char*>(left), reinterpret_cast<const char*>(right), n));
        }

        static int_type requires_async() { return eof() - 1; }
    };
#endif

    namespace details {

    template<typename _CharType>
    class basic_streambuf
    {
    public:
        typedef _CharType char_type;
        typedef ::concurrency::streams::char_traits<_CharType> traits;

        typedef typename traits::int_type int_type;
        typedef typename traits::pos_type pos_type;
        typedef typename traits::off_type off_type;


        virtual ~basic_streambuf() { }

        virtual bool can_read() const = 0;

        virtual bool can_write() const = 0;

        virtual bool can_seek() const = 0;

        virtual bool has_size() const = 0;

        virtual bool is_eof() const = 0;

        virtual size_t buffer_size(std::ios_base::openmode direction = std::ios_base::in) const = 0;

        virtual void set_buffer_size(size_t size, std::ios_base::openmode direction = std::ios_base::in) = 0;

        virtual size_t in_avail() const = 0;

        virtual bool is_open() const = 0;

        virtual pplx::task<void> close(std::ios_base::openmode mode = (std::ios_base::in | std::ios_base::out)) = 0;

        virtual pplx::task<void> close(std::ios_base::openmode mode, std::exception_ptr eptr) = 0;

        virtual pplx::task<int_type> putc(_CharType ch) = 0;

        virtual pplx::task<size_t> putn(const _CharType *ptr, size_t count) = 0;

        virtual pplx::task<size_t> putn_nocopy(const _CharType *ptr, size_t count) = 0;

        virtual pplx::task<int_type> bumpc() = 0;

        virtual int_type sbumpc() = 0;

        virtual pplx::task<int_type> getc() = 0;

        virtual int_type sgetc() = 0;

        virtual pplx::task<int_type> nextc() = 0;

        virtual pplx::task<int_type> ungetc() = 0;

        virtual pplx::task<size_t> getn(_Out_writes_(count) _CharType *ptr, _In_ size_t count) = 0;

        virtual size_t scopy(_Out_writes_(count) _CharType *ptr, _In_ size_t count) = 0;

        virtual pos_type getpos(std::ios_base::openmode direction) const = 0;

        virtual utility::size64_t size() const = 0;

        virtual pos_type seekpos(pos_type pos, std::ios_base::openmode direction) = 0;

        virtual pos_type seekoff(off_type offset, std::ios_base::seekdir way, std::ios_base::openmode mode) = 0;

        virtual pplx::task<void> sync() = 0;

        //
        // Efficient read and write.
        //
        // The following routines are intended to be used for more efficient, copy-free, reading and
        // writing of data from/to the stream. Rather than having the caller provide a buffer into which
        // data is written or from which it is read, the stream buffer provides a pointer directly to the
        // internal data blocks that it is using. Since not all stream buffers use internal data structures
        // to copy data, the functions may not be supported by all. An application that wishes to use this
        // functionality should therefore first try them and check for failure to support. If there is
        // such failure, the application should fall back on the copying interfaces (putn / getn)
        //

        virtual _CharType* alloc(_In_ size_t count) = 0;

        virtual void commit(_In_ size_t count) = 0;

        virtual bool acquire(_Out_ _CharType*& ptr, _Out_ size_t& count) = 0;

        virtual void release(_Out_writes_(count) _CharType *ptr, _In_ size_t count) = 0;

        virtual std::exception_ptr exception() const = 0;
    };


    template<typename _CharType>
    class streambuf_state_manager : public basic_streambuf<_CharType>, public std::enable_shared_from_this<streambuf_state_manager<_CharType>>
    {
    public:
        typedef typename details::basic_streambuf<_CharType>::traits traits;
        typedef typename details::basic_streambuf<_CharType>::int_type int_type;
        typedef typename details::basic_streambuf<_CharType>::pos_type pos_type;
        typedef typename details::basic_streambuf<_CharType>::off_type off_type;

        virtual bool can_read() const
        {
            return m_stream_can_read;
        }

        virtual bool can_write() const
        {
            return m_stream_can_write;
        }

        virtual bool is_open() const
        {
            return can_read() || can_write();
        }

        virtual pplx::task<void> close(std::ios_base::openmode mode = std::ios_base::in | std::ios_base::out)
        {
            pplx::task<void> closeOp = pplx::task_from_result();

            if (mode & std::ios_base::in && can_read()) {
                closeOp = _close_read();
            }

            // After the flush_internal task completed, "this" object may have been destroyed,
            // accessing the members is invalid, use shared_from_this to avoid access violation exception.
            auto this_ptr = std::static_pointer_cast<streambuf_state_manager>(this->shared_from_this());

            if (mode & std::ios_base::out && can_write()) {
                if (closeOp.is_done())
                    closeOp = closeOp && _close_write().then([this_ptr]{}); // passing down exceptions from closeOp
                else
                    closeOp = closeOp.then([this_ptr] { return this_ptr->_close_write().then([this_ptr]{}); });
            }

            return closeOp;
        }

        virtual pplx::task<void> close(std::ios_base::openmode mode, std::exception_ptr eptr)
        {
            if (m_currentException == nullptr)
                m_currentException = eptr;
            return close(mode);
        }

        virtual bool is_eof() const
        {
            return m_stream_read_eof;
        }

        virtual pplx::task<int_type> putc(_CharType ch)
        {
            if (!can_write())
                return create_exception_checked_value_task<int_type>(traits::eof());

            return create_exception_checked_task<int_type>(_putc(ch), [](int_type) {
                return false; // no EOF for write
            });
        }

        CASABLANCA_DEPRECATED("This API in some cases performs a copy. It is deprecated and will be removed in a future release. Use putn_nocopy instead.")
        virtual pplx::task<size_t> putn(const _CharType *ptr, size_t count)
        {
            if (!can_write())
                return create_exception_checked_value_task<size_t>(0);
            if (count == 0)
                return pplx::task_from_result<size_t>(0);

            return create_exception_checked_task<size_t>(_putn(ptr, count, true), [](size_t) {
                return false; // no EOF for write
            });
        }

        virtual pplx::task<size_t> putn_nocopy(const _CharType *ptr, size_t count)
        {
            if (!can_write())
                return create_exception_checked_value_task<size_t>(0);
            if (count == 0)
                return pplx::task_from_result<size_t>(0);

            return create_exception_checked_task<size_t>(_putn(ptr, count), [](size_t) {
                return false; // no EOF for write
            });
        }

        virtual pplx::task<int_type> bumpc()
        {
            if (!can_read())
                return create_exception_checked_value_task<int_type>(streambuf_state_manager<_CharType>::traits::eof());

            return create_exception_checked_task<int_type>(_bumpc(), [](int_type val) {
                return val == streambuf_state_manager<_CharType>::traits::eof();
            });
        }

        virtual int_type sbumpc()
        {
            if ( !(m_currentException == nullptr) )
                std::rethrow_exception(m_currentException);
            if (!can_read())
                return traits::eof();
            return check_sync_read_eof(_sbumpc());
        }

        virtual pplx::task<int_type> getc()
        {
            if (!can_read())
                return create_exception_checked_value_task<int_type>(traits::eof());

            return create_exception_checked_task<int_type>(_getc(), [](int_type val) {
                return val == streambuf_state_manager<_CharType>::traits::eof();
            });
        }

        virtual int_type sgetc()
        {
            if ( !(m_currentException == nullptr) )
                std::rethrow_exception(m_currentException);
            if (!can_read())
                return traits::eof();
            return check_sync_read_eof(_sgetc());
        }

        virtual pplx::task<int_type> nextc()
        {
            if (!can_read())
                return create_exception_checked_value_task<int_type>(traits::eof());

            return create_exception_checked_task<int_type>(_nextc(), [](int_type val) {
                return val == streambuf_state_manager<_CharType>::traits::eof();
            });
        }

        virtual pplx::task<int_type> ungetc()
        {
            if (!can_read())
                return create_exception_checked_value_task<int_type>(traits::eof());

            return create_exception_checked_task<int_type>(_ungetc(), [](int_type) {
                return false;
            });
        }

        virtual pplx::task<size_t> getn(_Out_writes_(count) _CharType *ptr, _In_ size_t count)
        {
            if (!can_read())
                return create_exception_checked_value_task<size_t>(0);
            if (count == 0)
                return pplx::task_from_result<size_t>(0);

            return create_exception_checked_task<size_t>(_getn(ptr, count), [](size_t val) {
                return val == 0;
            });
        }

        virtual size_t scopy(_Out_writes_(count) _CharType *ptr, _In_ size_t count)
        {
            if ( !(m_currentException == nullptr) )
                std::rethrow_exception(m_currentException);
            if (!can_read())
                return 0;

            return _scopy(ptr, count);
        }

        virtual pplx::task<void> sync()
        {
            if (!can_write())
            {
                if (m_currentException == nullptr)
                    return pplx::task_from_result();
                else
                    return pplx::task_from_exception<void>(m_currentException);
            }
            return create_exception_checked_task<bool>(_sync(), [](bool) {
                return false;
            }).then([](bool){});
        }

        virtual std::exception_ptr exception() const
        {
            return m_currentException;
        }

        _CharType* alloc(size_t count)
        {
            if (m_alloced)
                throw std::logic_error("The buffer is already allocated, this maybe caused by overlap of stream read or write");

            _CharType* alloc_result = _alloc(count);

            if (alloc_result)
                m_alloced = true;

            return alloc_result;
        }

        void commit(size_t count)
        {
            if (!m_alloced)
                throw std::logic_error("The buffer needs to allocate first");

            _commit(count);
            m_alloced = false;
        }

    public:
        virtual bool can_seek() const = 0;
        virtual bool has_size() const = 0;
        virtual utility::size64_t size() const { return 0; }
        virtual size_t buffer_size(std::ios_base::openmode direction = std::ios_base::in) const = 0;
        virtual void set_buffer_size(size_t size, std::ios_base::openmode direction = std::ios_base::in) = 0;
        virtual size_t in_avail() const = 0;
        virtual pos_type getpos(std::ios_base::openmode direction) const = 0;
        virtual pos_type seekpos(pos_type pos, std::ios_base::openmode direction) = 0;
        virtual pos_type seekoff(off_type offset, std::ios_base::seekdir way, std::ios_base::openmode mode) = 0;
        virtual bool acquire(_Out_writes_(count) _CharType*& ptr, _In_ size_t& count) = 0;
        virtual void release(_Out_writes_(count) _CharType *ptr, _In_ size_t count) = 0;
    protected:
        virtual pplx::task<int_type> _putc(_CharType ch) = 0;

        // This API is only needed for file streams and until we remove the deprecated stream buffer putn overload.
        virtual pplx::task<size_t> _putn(const _CharType *ptr, size_t count, bool)
        {
            // Default to no copy, only the file streams API overloads and performs a copy.
            return _putn(ptr, count);
        }
        virtual pplx::task<size_t> _putn(const _CharType *ptr, size_t count) = 0;

        virtual pplx::task<int_type> _bumpc() = 0;
        virtual int_type _sbumpc() = 0;
        virtual pplx::task<int_type> _getc() = 0;
        virtual int_type _sgetc() = 0;
        virtual pplx::task<int_type> _nextc() = 0;
        virtual pplx::task<int_type> _ungetc() = 0;
        virtual pplx::task<size_t> _getn(_Out_writes_(count) _CharType *ptr, _In_ size_t count) = 0;
        virtual size_t _scopy(_Out_writes_(count) _CharType *ptr, _In_ size_t count) = 0;
        virtual pplx::task<bool> _sync() = 0;
        virtual _CharType* _alloc(size_t count) = 0;
        virtual void _commit(size_t count) = 0;

        virtual pplx::task<void> _close_read()
        {
            m_stream_can_read = false;
            return pplx::task_from_result();
        }

        virtual pplx::task<void> _close_write()
        {
            m_stream_can_write = false;
            return pplx::task_from_result();
        }

    protected:
        streambuf_state_manager(std::ios_base::openmode mode)
        {
            m_stream_can_read = (mode & std::ios_base::in) != 0;
            m_stream_can_write = (mode & std::ios_base::out) != 0;
            m_stream_read_eof = false;
            m_alloced = false;
        }

        std::exception_ptr m_currentException;
        // The in/out mode for the buffer
        bool m_stream_can_read, m_stream_can_write, m_stream_read_eof, m_alloced;


    private:
        template<typename _CharType1>
        pplx::task<_CharType1> create_exception_checked_value_task(const _CharType1 &val) const
        {
            if (this->exception() == nullptr)
                return pplx::task_from_result<_CharType1>(static_cast<_CharType1>(val));
            else
                return pplx::task_from_exception<_CharType1>(this->exception());
        }

        // Set exception and eof states for async read
        template<typename _CharType1>
        pplx::task<_CharType1> create_exception_checked_task(pplx::task<_CharType1> result, std::function<bool(_CharType1)> eof_test, std::ios_base::openmode mode = std::ios_base::in | std::ios_base::out)
        {
            auto thisPointer = this->shared_from_this();

            auto func1 = [=](pplx::task<_CharType1> t1) -> pplx::task<_CharType1> {
                try {
                    thisPointer->m_stream_read_eof = eof_test(t1.get());
                } catch (...) {
                    thisPointer->close(mode, std::current_exception()).get();
                    return pplx::task_from_exception<_CharType1>(thisPointer->exception(), pplx::task_options());
                }
                if (thisPointer->m_stream_read_eof && !(thisPointer->exception() == nullptr))
                    return pplx::task_from_exception<_CharType1>(thisPointer->exception(), pplx::task_options());
                return t1;
            };

            if ( result.is_done() )
            {
                // If the data is already available, we should avoid scheduling a continuation, so we do it inline.
                return func1(result);
            }
            else
            {
                return result.then(func1);
            }
        }

        // Set eof states for sync read
        int_type check_sync_read_eof(int_type ch)
        {
            m_stream_read_eof = ch == traits::eof();
            return ch;
        }

    };

    } // namespace details

    // Forward declarations
    template<typename _CharType> class basic_istream;
    template<typename _CharType> class basic_ostream;

    template<typename _CharType>
    class streambuf : public details::basic_streambuf<_CharType>
    {
    public:
        typedef typename details::basic_streambuf<_CharType>::traits    traits;
        typedef typename details::basic_streambuf<_CharType>::int_type  int_type;
        typedef typename details::basic_streambuf<_CharType>::pos_type  pos_type;
        typedef typename details::basic_streambuf<_CharType>::off_type  off_type;
        typedef typename details::basic_streambuf<_CharType>::char_type char_type;

        template <typename _CharType2> friend class streambuf;

        streambuf(_In_ const std::shared_ptr<details::basic_streambuf<_CharType>> &ptr) : m_buffer(ptr) {}

        streambuf() { }

        template <typename AlterCharType>
        streambuf(const streambuf<AlterCharType> &other) :
            m_buffer(std::static_pointer_cast<details::basic_streambuf<_CharType>>(std::static_pointer_cast<void>(other.m_buffer)))
        {
            static_assert(std::is_same<pos_type, typename details::basic_streambuf<AlterCharType>::pos_type>::value
                && std::is_same<off_type, typename details::basic_streambuf<AlterCharType>::off_type>::value
                && std::is_integral<_CharType>::value && std::is_integral<AlterCharType>::value
                && std::is_integral<int_type>::value && std::is_integral<typename details::basic_streambuf<AlterCharType>::int_type>::value
                && sizeof(_CharType) == sizeof(AlterCharType)
                && sizeof(int_type) == sizeof(typename details::basic_streambuf<AlterCharType>::int_type),
                "incompatible stream character types");
        }

        concurrency::streams::basic_istream<_CharType> create_istream() const
        {
            if (!can_read()) throw std::runtime_error("stream buffer not set up for input of data");
            return concurrency::streams::basic_istream<_CharType>(*this);
        }

        concurrency::streams::basic_ostream<_CharType> create_ostream() const
        {
            if (!can_write()) throw std::runtime_error("stream buffer not set up for output of data");
            return concurrency::streams::basic_ostream<_CharType>(*this);
        }

        operator bool() const { return (bool)m_buffer; }

        virtual ~streambuf() { }

        const std::shared_ptr<details::basic_streambuf<_CharType>> & get_base() const
        {
            if (!m_buffer)
            {
                throw std::invalid_argument("Invalid streambuf object");
            }

            return m_buffer;
        }

        virtual bool can_read() const { return get_base()->can_read(); }

        virtual bool can_write() const { return  get_base()->can_write(); }

        virtual bool can_seek() const { return get_base()->can_seek(); }

        virtual bool has_size() const { return get_base()->has_size(); }

        virtual utility::size64_t size() const { return get_base()->size(); }

        virtual size_t buffer_size(std::ios_base::openmode direction = std::ios_base::in) const { return get_base()->buffer_size(direction); }

        virtual void set_buffer_size(size_t size, std::ios_base::openmode direction = std::ios_base::in) { get_base()->set_buffer_size(size,direction); }

        virtual size_t in_avail() const { return get_base()->in_avail(); }

        virtual bool is_open() const { return get_base()->is_open(); }

        virtual bool is_eof() const { return get_base()->is_eof(); }

        virtual pplx::task<void> close(std::ios_base::openmode mode = (std::ios_base::in | std::ios_base::out))
        {
            // We preserve the check here to workaround a Dev10 compiler crash
            auto buffer = get_base();
            return buffer ? buffer->close(mode) : pplx::task_from_result();
        }

        virtual pplx::task<void> close(std::ios_base::openmode mode, std::exception_ptr eptr)
        {
            // We preserve the check here to workaround a Dev10 compiler crash
            auto buffer = get_base();
            return buffer ? buffer->close(mode, eptr) : pplx::task_from_result();
        }

        virtual pplx::task<int_type> putc(_CharType ch)
        {
            return get_base()->putc(ch);
        }

        virtual _CharType* alloc(size_t count)
        {
            return get_base()->alloc(count);
        }

        virtual void commit(size_t count)
        {
            get_base()->commit(count);
        }

        virtual bool acquire(_Out_ _CharType*& ptr, _Out_ size_t& count)
        {
            ptr = nullptr;
            count = 0;
            return get_base()->acquire(ptr, count);
        }

        virtual void release(_Out_writes_(count) _CharType *ptr, _In_ size_t count)
        {
            get_base()->release(ptr, count);
        }

        CASABLANCA_DEPRECATED("This API in some cases performs a copy. It is deprecated and will be removed in a future release. Use putn_nocopy instead.")
        virtual pplx::task<size_t> putn(const _CharType *ptr, size_t count)
        {
            return get_base()->putn(ptr, count);
        }

        virtual pplx::task<size_t> putn_nocopy(const _CharType *ptr, size_t count)
        {
            return get_base()->putn_nocopy(ptr, count);
        }

        virtual pplx::task<int_type> bumpc()
        {
            return get_base()->bumpc();
        }

        virtual typename details::basic_streambuf<_CharType>::int_type sbumpc()
        {
            return get_base()->sbumpc();
        }

        virtual pplx::task<int_type> getc()
        {
            return get_base()->getc();
        }

        virtual typename details::basic_streambuf<_CharType>::int_type sgetc()
        {
            return get_base()->sgetc();
        }

        pplx::task<int_type> nextc()
        {
            return get_base()->nextc();
        }

        pplx::task<int_type> ungetc()
        {
            return get_base()->ungetc();
        }

        virtual pplx::task<size_t> getn(_Out_writes_(count) _CharType *ptr, _In_ size_t count)
        {
            return get_base()->getn(ptr, count);
        }

        virtual size_t scopy(_Out_writes_(count) _CharType *ptr, _In_ size_t count)
        {
            return get_base()->scopy(ptr, count);
        }

        virtual typename details::basic_streambuf<_CharType>::pos_type getpos(std::ios_base::openmode direction) const
        {
            return get_base()->getpos(direction);
        }

        virtual typename details::basic_streambuf<_CharType>::pos_type seekpos(typename details::basic_streambuf<_CharType>::pos_type pos, std::ios_base::openmode direction)
        {
            return get_base()->seekpos(pos, direction);
        }

        virtual typename details::basic_streambuf<_CharType>::pos_type seekoff(typename details::basic_streambuf<_CharType>::off_type offset, std::ios_base::seekdir way, std::ios_base::openmode mode)
        {
            return get_base()->seekoff(offset, way, mode);
        }

        virtual pplx::task<void> sync()
        {
            return get_base()->sync();
        }

        virtual std::exception_ptr exception() const
        {
            return get_base()->exception();
        }

    private:
        std::shared_ptr<details::basic_streambuf<_CharType>> m_buffer;

    };

}}