path: root/contrib/libs/apache/avro/api/buffer/BufferReader.hh

/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you 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
 *
 *     https://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.
 */

#ifndef avro_BufferReader_hh__
#define avro_BufferReader_hh__

#include <type_traits>
#include "Buffer.hh"

#ifdef min
#undef min
#endif
/**
 * \file BufferReader.hh
 *
 * \brief Helper class for reading bytes from buffer in a streaming manner,
 * without the overhead of istreams.
 *
 **/

namespace avro {

/**
 * Helper class for reading bytes from buffer without worrying about
 * chunk boundaries.  May read from an InputBuffer or OutputBuffer.
 *
 **/
class AVRO_DECL BufferReader : private boost::noncopyable
{

  public:

    typedef detail::data_type data_type;
    typedef detail::size_type size_type;

  private:

    size_type chunkRemaining() const {
        return iter_->dataSize() - chunkPos_;
    }

    void incrementChunk(size_type howmuch) {
        bytesRemaining_ -= howmuch;
        chunkPos_ += howmuch;
        if(chunkPos_ == iter_->dataSize()) {
            chunkPos_ = 0;
            ++iter_;
        }
    }

    void rewind() {
        iter_ = bufferImpl_->beginRead();
        bytesRemaining_ = bytes_;
        chunkPos_ = 0;
    }

    const data_type *addr() const {
        return iter_->tellReadPos() + chunkPos_;
    }

  public:

    BufferReader(const InputBuffer &buf) :
        bufferImpl_(buf.pimpl_),
        iter_(bufferImpl_->beginRead()),
        bytes_(bufferImpl_->size()),
        bytesRemaining_(bytes_),
        chunkPos_(0)
    { }

    BufferReader(const OutputBuffer &buf) :
        bufferImpl_(buf.pimpl_),
        iter_(bufferImpl_->beginRead()),
        bytes_(bufferImpl_->size()),
        bytesRemaining_(bytes_),
        chunkPos_(0)
    { }

    /**
     * How many bytes are still not read from this buffer.
     **/

    size_type bytesRemaining() const {
        return bytesRemaining_;
    }

    /**
     * Read a block of data from the front of the buffer.
     **/

    size_type bytesRead() const {
        return bytes_ - bytesRemaining_;
    }

    /**
     * Read a block of data from the buffer.
     **/

    size_type read(data_type *data, size_type size) {

        if(size > bytesRemaining_) {
            size = bytesRemaining_;
        }
        size_type sizeToRead = size;

        while(sizeToRead) {
            const size_type toRead = std::min(sizeToRead, chunkRemaining());
            memcpy(data, addr(), toRead);
            sizeToRead -= toRead;
            data += toRead;
            incrementChunk(toRead);
        }

        return size;
    }

    /**
     * Read a block of data from the buffer.
     **/

    bool read(std::string &str, size_type size) {
        if(size > bytesRemaining_) {
            return false;
        }

        if(size <= chunkRemaining()) {
            fastStringRead(str, size);
        }
        else {
            slowStringRead(str, size);
        }

        return true;
    }


    /**
     * Read a single value from the buffer.  The value must be a "fundamental"
     * type, e.g. int, float, etc.  (otherwise use the other writeTo tests).
     *
     **/

    template<typename T>
    bool read(T &val)  {
        return read(val, std::is_fundamental<T>());
    }

    /**
     * Skips a block of data from the buffer.
     **/

    bool skip(size_type bytes) {
        bool skipped = false;
        if(bytes <= bytesRemaining_) {
            doSkip(bytes);
            skipped = true;
        }
        return skipped;
    }

    /**
     * Seek to a position in the buffer.
     **/

    bool seek(size_type pos) {
        if(pos > bytes_) {
            return false;
        }

        size_type toSkip = pos;
        size_type curPos = bytesRead();
        // if the seek position is ahead, we can use skip to get there
        if(pos >= curPos) {
            toSkip -= curPos;
        }
        // if the seek position is ahead of the start of the chunk we can back up to
        // start of the chunk
        else if(pos >= (curPos - chunkPos_)) {
            curPos -= chunkPos_;
            bytesRemaining_ += chunkPos_;
            chunkPos_ = 0;
            toSkip -= curPos;
        }
        else {
            rewind();
        }
        doSkip(toSkip);
        return true;
    }

    bool peek(char &val) {
        bool ret = (bytesRemaining_ > 0);
        if(ret) {
            val = *(addr());
        }
        return ret;
    }

    InputBuffer copyData(size_type bytes) {
        if(bytes > bytesRemaining_) {
            // force no copy
            bytes = 0;
        }
        detail::BufferImpl::SharedPtr newImpl(new detail::BufferImpl);
        if(bytes) {
            bufferImpl_->copyData(*newImpl, iter_, chunkPos_, bytes);
            doSkip(bytes);
        }
        return InputBuffer(newImpl);
    }

  private:

    void doSkip(size_type sizeToSkip) {

        while(sizeToSkip) {
            const size_type toSkip = std::min(sizeToSkip, chunkRemaining());
            sizeToSkip -= toSkip;
            incrementChunk(toSkip);
        }
    }

    template<typename T>
    bool read(T &val, const std::true_type&)
    {
        if(sizeof(T) > bytesRemaining_) {
            return false;
        }

        if (sizeof(T) <= chunkRemaining()) {
            val = *(reinterpret_cast<const T*> (addr()));
            incrementChunk(sizeof(T));
        }
        else {
            read(reinterpret_cast<data_type *>(&val), sizeof(T));
        }
        return true;
    }

    /// An uninstantiable function, that is if boost::is_fundamental check fails
    template<typename T>
    bool read(T &, const std::false_type&) 
    {
        static_assert(sizeof(T) == 0, "Not a valid type to read");
        return false;
    }

    void fastStringRead(std::string &str, size_type sizeToCopy) {
        str.assign(addr(), sizeToCopy);
        incrementChunk(sizeToCopy);
    }

    void slowStringRead(std::string &str, size_type sizeToCopy) {
        str.clear();
        str.reserve(sizeToCopy);
        while(sizeToCopy) {
            const size_type toCopy = std::min(sizeToCopy, chunkRemaining());
            str.append(addr(), toCopy);
            sizeToCopy -= toCopy;
            incrementChunk(toCopy);
        }
    }

    detail::BufferImpl::ConstSharedPtr bufferImpl_;
    detail::BufferImpl::ChunkList::const_iterator iter_;
    size_type bytes_;
    size_type bytesRemaining_;
    size_type chunkPos_;
};


} // namespace

#endif