rawptrstream.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.
*
* ==--==
* =+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+
*
* This file defines a stream buffer that is based on a raw pointer and block size. Unlike a vector-based
* stream buffer, the buffer cannot be expanded or contracted, it has a fixed capacity.
*
* For the latest on this and related APIs, please see http://casablanca.codeplex.com.
*
* =-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-
****/
#pragma once

#ifndef _CASA_RAWPTR_STREAMS_H
#define _CASA_RAWPTR_STREAMS_H

#include <vector>
#include <queue>
#include <algorithm>
#include <iterator>

#include "pplx/pplxtasks.h"
#include "cpprest/astreambuf.h"
#include "cpprest/streams.h"

namespace Concurrency { namespace streams {

    // Forward declarations
    template <typename _CharType> class rawptr_buffer;

    namespace details {

    template<typename _CharType>
    class basic_rawptr_buffer : public streams::details::streambuf_state_manager<_CharType>
    {
    public:
        typedef _CharType char_type;

        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;

        basic_rawptr_buffer()
            : streambuf_state_manager<_CharType>(std::ios_base::in | std::ios_base::out),
              m_data(nullptr),
              m_current_position(0),
              m_size(0)
        {
        }

        virtual ~basic_rawptr_buffer()
        {
            this->_close_read();
            this->_close_write();
        }

    protected:

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

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

        virtual utility::size64_t size() const
        {
            return utility::size64_t(m_size);
        }

        virtual size_t buffer_size(std::ios_base::openmode = std::ios_base::in) const
        {
            return 0;
        }

        virtual void set_buffer_size(size_t , std::ios_base::openmode = std::ios_base::in)
        {
            return;
        }

        virtual size_t in_avail() const
        {
            // See the comment in seek around the restiction that we do not allow read head to
            // seek beyond the current size.
            _ASSERTE(m_current_position <= m_size);

            msl::safeint3::SafeInt<size_t> readhead(m_current_position);
            msl::safeint3::SafeInt<size_t> writeend(m_size);
            return (size_t)(writeend - readhead);
        }

        virtual pplx::task<void> close(std::ios_base::openmode mode)
        {
            if (mode & std::ios_base::in)
            {
                this->_close_read().get(); // Safe to call get() here.
            }

            if (mode & std::ios_base::out)
            {
                this->_close_write().get(); // Safe to call get() here.
            }

            if ( !this->can_read() && !this->can_write() )
            {
                m_data = nullptr;
            }

            // Exceptions will be propagated out of _close_read or _close_write
            return pplx::task_from_result();
        }

        virtual pplx::task<bool> _sync()
        {
            return pplx::task_from_result(true);
        }

        virtual pplx::task<int_type> _putc(_CharType ch)
        {
            if (m_current_position >= m_size)
                return pplx::task_from_result<int_type>(traits::eof());
            int_type retVal = (this->write(&ch, 1) == 1) ? static_cast<int_type>(ch) : traits::eof();
            return pplx::task_from_result<int_type>(retVal);
        }

        virtual pplx::task<size_t> _putn(const _CharType *ptr, size_t count)
        {
            msl::safeint3::SafeInt<size_t> newSize = msl::safeint3::SafeInt<size_t>(count) + m_current_position;
            if ( newSize > m_size )
                return pplx::task_from_exception<size_t>(std::make_exception_ptr(std::runtime_error("Writing past the end of the buffer")));
            return pplx::task_from_result<size_t>(this->write(ptr, count));
        }

        _CharType* _alloc(size_t count)
        {
            if (!this->can_write()) return nullptr;

            msl::safeint3::SafeInt<size_t> readhead(m_current_position);
            msl::safeint3::SafeInt<size_t> writeend(m_size);
            size_t space_left = (size_t)(writeend - readhead);

            if (space_left < count) return nullptr;

            // Let the caller copy the data
            return (_CharType*)(m_data+m_current_position);
        }

        void _commit(size_t actual)
        {
            // Update the write position and satisfy any pending reads
            update_current_position(m_current_position+actual);
        }

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

            if (!this->can_read()) return false;

            count = in_avail();

            if (count > 0)
            {
                ptr = (_CharType*)(m_data+m_current_position);
                return true;
            }
            else
            {
                ptr = nullptr;

                // Can only be open for read OR write, not both. If there is no data then
                // we have reached the end of the stream so indicate such with true.
                return true;
            }
        }

        virtual void release(_Out_writes_opt_ (count) _CharType *ptr, _In_ size_t count)
        {
            if (ptr != nullptr)
                update_current_position(m_current_position + count);
        }

        virtual pplx::task<size_t> _getn(_Out_writes_ (count) _CharType *ptr, _In_ size_t count)
        {
            return pplx::task_from_result(this->read(ptr, count));
        }

        size_t _sgetn(_Out_writes_ (count) _CharType *ptr, _In_ size_t count)
        {
            return this->read(ptr, count);
        }

        virtual size_t _scopy(_Out_writes_ (count) _CharType *ptr, _In_ size_t count)
        {
            return this->read(ptr, count, false);
        }

        virtual pplx::task<int_type> _bumpc()
        {
            return pplx::task_from_result(this->read_byte(true));
        }

        virtual int_type _sbumpc()
        {
            return this->read_byte(true);
        }

        virtual pplx::task<int_type> _getc()
        {
            return pplx::task_from_result(this->read_byte(false));
        }

        int_type _sgetc()
        {
            return this->read_byte(false);
        }

        virtual pplx::task<int_type> _nextc()
        {
            if (m_current_position >= m_size-1)
                return pplx::task_from_result(basic_streambuf<_CharType>::traits::eof());

            this->read_byte(true);
            return pplx::task_from_result(this->read_byte(false));
        }

        virtual pplx::task<int_type> _ungetc()
        {
            auto pos = seekoff(-1, std::ios_base::cur, std::ios_base::in);
            if ( pos == (pos_type)traits::eof())
                return pplx::task_from_result(traits::eof());
            return this->getc();
        }

        virtual pos_type getpos(std::ios_base::openmode mode) const
        {
            if ( ((mode & std::ios_base::in) && !this->can_read()) ||
                    ((mode & std::ios_base::out) && !this->can_write()))
                    return static_cast<pos_type>(traits::eof());

            if (mode == std::ios_base::in)
                return (pos_type)m_current_position;
            else if (mode == std::ios_base::out)
                return (pos_type)m_current_position;
            else
                return (pos_type)traits::eof();
        }

        virtual pos_type seekpos(pos_type position, std::ios_base::openmode mode)
        {
            pos_type beg(0);
            pos_type end(m_size);

            if (position >= beg)
            {
                auto pos = static_cast<size_t>(position);

                // Read head
                if ((mode & std::ios_base::in) && this->can_read())
                {
                    if (position <= end)
                    {
                        // We do not allow reads to seek beyond the end or before the start position.
                        update_current_position(pos);
                        return static_cast<pos_type>(m_current_position);
                    }
                }

                // Write head
                if ((mode & std::ios_base::out) && this->can_write())
                {
                    // Update write head and satisfy read requests if any
                    update_current_position(pos);

                    return static_cast<pos_type>(m_current_position);
                }
            }

            return static_cast<pos_type>(traits::eof());
        }

        virtual pos_type seekoff(off_type offset, std::ios_base::seekdir way, std::ios_base::openmode mode)
        {
            pos_type beg = 0;
            pos_type cur = static_cast<pos_type>(m_current_position);
            pos_type end = static_cast<pos_type>(m_size);

            switch ( way )
            {
            case std::ios_base::beg:
                return seekpos(beg + offset, mode);

            case std::ios_base::cur:
                return seekpos(cur + offset, mode);

            case std::ios_base::end:
                return seekpos(end + offset, mode);

            default:
                return static_cast<pos_type>(traits::eof());
            }
        }

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

        basic_rawptr_buffer(const _CharType* data, size_t size)
            : streambuf_state_manager<_CharType>(std::ios_base::in),
              m_data(const_cast<_CharType*>(data)),
              m_size(size),
              m_current_position(0)
        {
            validate_mode(std::ios_base::in);
        }

        basic_rawptr_buffer(_CharType* data, size_t size, std::ios_base::openmode mode)
            : streambuf_state_manager<_CharType>(mode),
              m_data(data),
              m_size(size),
              m_current_position(0)
        {
            validate_mode(mode);
        }

        static void validate_mode(std::ios_base::openmode mode)
        {
            // Disallow simultaneous use of the stream buffer for writing and reading.
            if ((mode & std::ios_base::in) && (mode & std::ios_base::out))
                throw std::invalid_argument("this combination of modes on raw pointer stream not supported");
        }

        bool can_satisfy(size_t) const
        {
            // We can always satisfy a read, at least partially, unless the
            // read position is at the very end of the buffer.
            return (in_avail() > 0);
        }

        int_type read_byte(bool advance = true)
        {
            _CharType value;
            auto read_size = this->read(&value, 1, advance);
            return read_size == 1 ? static_cast<int_type>(value) : traits::eof();
        }

        size_t read(_Out_writes_ (count) _CharType *ptr, _In_ size_t count, bool advance = true)
        {
            if (!can_satisfy(count))
                return 0;

            msl::safeint3::SafeInt<size_t> request_size(count);
            msl::safeint3::SafeInt<size_t> read_size = request_size.Min(in_avail());

            size_t newPos = m_current_position + read_size;

            auto readBegin = m_data + m_current_position;
            auto readEnd = m_data + newPos;

#ifdef _WIN32
            // Avoid warning C4996: Use checked iterators under SECURE_SCL
            std::copy(readBegin, readEnd, stdext::checked_array_iterator<_CharType *>(ptr, count));
#else
            std::copy(readBegin, readEnd, ptr);
#endif // _WIN32

            if (advance)
            {
                update_current_position(newPos);
            }

            return (size_t) read_size;
        }

        size_t write(const _CharType *ptr, size_t count)
        {
            if (!this->can_write() || (count == 0)) return 0;

            msl::safeint3::SafeInt<size_t> newSize = msl::safeint3::SafeInt<size_t>(count) +m_current_position;

            if ( newSize > m_size )
                throw std::runtime_error("Writing past the end of the buffer");

            // Copy the data
#ifdef _WIN32
            // Avoid warning C4996: Use checked iterators under SECURE_SCL
            std::copy(ptr, ptr + count, stdext::checked_array_iterator<_CharType *>(m_data, m_size, m_current_position));
#else
            std::copy(ptr, ptr + count, m_data+m_current_position);
#endif // _WIN32

            // Update write head and satisfy pending reads if any
            update_current_position(newSize);

            return count;
        }

        void update_current_position(size_t newPos)
        {
            // The new write head
            m_current_position = newPos;

            _ASSERTE(m_current_position <= m_size);
        }

        // The actual memory block
        _CharType* m_data;

        // The size of the memory block
        size_t m_size;

        // Read/write head
        size_t m_current_position;
    };

    } // namespace details

    template<typename _CharType>
    class rawptr_buffer : public streambuf<_CharType>
    {
    public:
        typedef _CharType char_type;

        rawptr_buffer(const char_type* data, size_t size)
            : streambuf<char_type>(std::shared_ptr<details::basic_rawptr_buffer<char_type>>(new details::basic_rawptr_buffer<char_type>(data, size)))
        {
        }

        rawptr_buffer(char_type* data, size_t size, std::ios_base::openmode mode = std::ios::out)
            : streambuf<char_type>(std::shared_ptr<details::basic_rawptr_buffer<char_type>>(new details::basic_rawptr_buffer<char_type>(data, size, mode)))
        {
        }

        rawptr_buffer()
        {
        }
    };

    template<typename _CharType>
    class rawptr_stream
    {
    public:
        typedef _CharType char_type;
        typedef rawptr_buffer<_CharType>  buffer_type;

        static concurrency::streams::basic_istream<char_type> open_istream(const char_type* data, size_t size)
        {
            return concurrency::streams::basic_istream<char_type>(buffer_type(data, size));
        }

        static concurrency::streams::basic_istream<char_type> open_istream(char_type* data, size_t size)
        {
            return concurrency::streams::basic_istream<char_type>(buffer_type(data, size, std::ios::in));
        }

        static concurrency::streams::basic_ostream<char_type> open_ostream(char_type* data, size_t size)
        {
            return concurrency::streams::basic_ostream<char_type>(buffer_type(data, size, std::ios::out));
        }
    };

}} // namespaces

#endif