Stream.hh

#ifndef avro_Stream_hh__
#define avro_Stream_hh__

#include <memory>
#include <string.h>
#include <stdint.h>
#include "boost/utility.hpp"
#include "Exception.hh"

namespace avro {
class InputStream : boost::noncopyable {
public:
    InputStream() { }
    virtual ~InputStream() { }

    virtual bool next(const uint8_t** data, size_t* len) = 0;

    virtual void backup(size_t len) = 0;

    virtual void skip(size_t len) = 0;

    virtual size_t byteCount() const = 0;
};

class OutputStream : boost::noncopyable {
public:
    OutputStream() { }
    virtual ~OutputStream() { }

    virtual bool next(uint8_t** data, size_t* len) = 0;

    virtual void backup(size_t len) = 0;

    virtual uint64_t byteCount() const = 0;

    virtual void flush() = 0;
};

std::auto_ptr<OutputStream> memoryOutputStream(size_t chunkSize = 4 * 1024);

std::auto_ptr<InputStream> memoryInputStream(const uint8_t* data, size_t len);

std::auto_ptr<InputStream> memoryInputStream(const OutputStream& source);

std::auto_ptr<OutputStream> fileOutputStream(const char* filename,
    size_t bufferSize = 8 * 1024);

std::auto_ptr<InputStream> fileInputStream(const char* filename,
    size_t bufferSize = 8 * 1024);

struct StreamReader {
    InputStream* in_;
    const uint8_t* next_;
    const uint8_t* end_;

    StreamReader() : in_(0), next_(0), end_(0) { }
    StreamReader(InputStream& in) : in_(0), next_(0), end_(0) { reset(in); }

    void reset(InputStream& is) {
        if (in_ != 0) {
            in_->backup(end_ - next_);
        }
        in_ = &is;
        next_ = end_ = 0;
    }

    uint8_t read() {
        if (next_ == end_) {
            more();
        }
        return *next_++;
    }

    void readBytes(uint8_t* b, size_t n) {
        while (n > 0) {
            if (next_ == end_) {
                more();
            }
            size_t q = end_ - next_;
            if (q > n) {
                q = n;
            }
            ::memcpy(b, next_, q);
            next_ += q;
            b += q;
            n -= q;
        }
    }

    void skipBytes(size_t n) {
        if (n > (end_ - next_)) {
            n -= end_ - next_;
            next_ = end_;
            in_->skip(n);
        } else {
            next_ += n;
        }
    }

    bool fill() {
        size_t n = 0;
        while (in_->next(&next_, &n)) {
            if (n != 0) {
                end_ = next_ + n;
                return true;
            }
        }
        return false;
    }

    void more() {
        if (! fill()) {
            throw Exception("EOF reached");
        }
    }

    bool hasMore() {
        return (next_ == end_) ? fill() : true;
    }
};

struct StreamWriter {
    OutputStream* out_;
    uint8_t* next_;
    uint8_t* end_;

    StreamWriter() : out_(0), next_(0), end_(0) { }
    StreamWriter(OutputStream& out) : out_(0), next_(0), end_(0) { reset(out); }

    void reset(OutputStream& os) {
        if (out_ != 0) {
            out_->backup(end_ - next_);
        }
        out_ = &os;
        next_ = end_;
    }

    void write(uint8_t c) {
        if (next_ == end_) {
            more();
        }
        *next_++ = c;
    }

    void writeBytes(const uint8_t* b, size_t n) {
        while (n > 0) {
            if (next_ == end_) {
                more();
            }
            size_t q = end_ - next_;
            if (q > n) {
                q = n;
            }
            ::memcpy(next_, b, q);
            next_ += q;
            b += q;
            n -= q;
        }
    }

    void more() {
        size_t n = 0;
        while (out_->next(&next_, &n)) {
            if (n != 0) {
                end_ = next_ + n;
                return;
            }
        }
        throw Exception("EOF reached");
    }

    void flush() {
        if (next_ != end_) {
            out_->backup(end_ - next_);
            next_ = end_;
        }
        out_->flush();
    }
};

inline void copy(InputStream& in, OutputStream& out)
{
    const uint8_t *p = 0;
    size_t n = 0;
    StreamWriter w(out);
    while (in.next(&p, &n)) {
        w.writeBytes(p, n);
    }
    w.flush();
}

}   // namespace avro
#endif