BufferReader.hh

Go to the documentation of this file.

#ifndef avro_BufferReader_hh__
#define avro_BufferReader_hh__

#include "Buffer.hh"

namespace avro {

class 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)
    { }

    size_type bytesRemaining() const {
        return bytesRemaining_;
    }

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

    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;
    }

    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;
    }


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

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

    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 boost::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;
    }

    template<typename T>
    bool read(T &val, const boost::false_type&)
    {
        BOOST_STATIC_ASSERT(sizeof(T)==0);
        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