Stream.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
 *
 *     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.
 */

#ifndef avro_Stream_hh__
#define avro_Stream_hh__

#include <memory>
#include <string.h>
#include <stdint.h>

#include "boost/utility.hpp"

#include "Config.hh"
#include "Exception.hh"

namespace avro {

class AVRO_DECL InputStream : boost::noncopyable {
protected:

    InputStream() { }

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

typedef std::unique_ptr<InputStream> InputStreamPtr;

class AVRO_DECL SeekableInputStream : public InputStream {
protected:

    SeekableInputStream() { }

public:
    virtual ~SeekableInputStream() { }

    virtual void seek(int64_t position) = 0;
};

typedef std::unique_ptr<SeekableInputStream> SeekableInputStreamPtr;

class AVRO_DECL OutputStream : boost::noncopyable {
protected:

    OutputStream() { }
public:

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

typedef std::unique_ptr<OutputStream> OutputStreamPtr;

AVRO_DECL OutputStreamPtr memoryOutputStream(size_t chunkSize = 4 * 1024);

AVRO_DECL InputStreamPtr memoryInputStream(const uint8_t* data, size_t len);

AVRO_DECL InputStreamPtr memoryInputStream(const OutputStream& source);

AVRO_DECL std::shared_ptr<std::vector<uint8_t> > snapshot(const OutputStream& source);

AVRO_DECL OutputStreamPtr fileOutputStream(const char* filename,
    size_t bufferSize = 8 * 1024);

AVRO_DECL InputStreamPtr fileInputStream(
    const char *filename, size_t bufferSize = 8 * 1024);
AVRO_DECL SeekableInputStreamPtr fileSeekableInputStream(
    const char *filename, size_t bufferSize = 8 * 1024);

AVRO_DECL OutputStreamPtr ostreamOutputStream(std::ostream& os,
    size_t bufferSize = 8 * 1024);

AVRO_DECL InputStreamPtr istreamInputStream(
    std::istream &in, size_t bufferSize = 8 * 1024);

AVRO_DECL InputStreamPtr nonSeekableIstreamInputStream(
    std::istream& is, 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 && end_ != next_) {
            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 > static_cast<size_t>(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;
    }

    void drain(bool unRead) {
        if (unRead) {
            --next_;
        }
        in_->backup(end_ - next_);
        end_ = next_;
    }
};

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 && end_ != next_) {
            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 flush() {
        if (next_ != end_) {
            out_->backup(end_ - next_);
            next_ = end_;
        }
        out_->flush();
    }

    int64_t byteCount() const {
        return out_->byteCount();
    }

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

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