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
 *
 *     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_Stream_hh__
#define avro_Stream_hh__
 
#include <cstdint>
#include <cstring>
#include <memory>
 
#include "boost/utility.hpp"
 
#include "Config.hh"
#include "Exception.hh"
 
namespace avro {
 
class AVRO_DECL InputStream : boost::noncopyable {
protected:
    InputStream() = default;
 
public:
    virtual ~InputStream() = default;
 
    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() = default;
 
public:
    ~SeekableInputStream() override = default;
 
    virtual void seek(int64_t position) = 0;
};
 
typedef std::unique_ptr<SeekableInputStream> SeekableInputStreamPtr;
 
class AVRO_DECL OutputStream : boost::noncopyable {
protected:
    OutputStream() = default;
 
public:
    virtual ~OutputStream() = default;
 
    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_(nullptr), next_(nullptr), end_(nullptr) {}
 
    explicit StreamReader(InputStream &in) : in_(nullptr), next_(nullptr), end_(nullptr) { reset(in); }
 
    void reset(InputStream &is) {
        if (in_ != nullptr && end_ != next_) {
            in_->backup(end_ - next_);
        }
        in_ = &is;
        next_ = end_ = nullptr;
    }
 
    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();
    }
 
    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_(nullptr), next_(nullptr), end_(nullptr) {}
 
    explicit StreamWriter(OutputStream &out) : out_(nullptr), next_(nullptr), end_(nullptr) { reset(out); }
 
    void reset(OutputStream &os) {
        if (out_ != nullptr && 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 = nullptr;
    size_t n = 0;
    StreamWriter w(out);
    while (in.next(&p, &n)) {
        w.writeBytes(p, n);
    }
    w.flush();
}
 
} // namespace avro
#endif