DataFile.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_DataFile_hh__
#define avro_DataFile_hh__

#include "Config.hh"
#include "Encoder.hh"
#include "buffer/Buffer.hh"
#include "ValidSchema.hh"
#include "Specific.hh"
#include "Stream.hh"

#include <map>
#include <string>
#include <vector>

#include "boost/array.hpp"
#include "boost/utility.hpp"
#include <boost/iostreams/filtering_stream.hpp>
#include <boost/scoped_ptr.hpp>

namespace avro {

enum Codec {
  NULL_CODEC,
  DEFLATE_CODEC
};

typedef boost::array<uint8_t, 16> DataFileSync;

class AVRO_DECL DataFileWriterBase : boost::noncopyable {
    const std::string filename_;
    const ValidSchema schema_;
    const EncoderPtr encoderPtr_;
    const size_t syncInterval_;
    Codec codec_;

    std::auto_ptr<OutputStream> stream_;
    std::auto_ptr<OutputStream> buffer_;
    const DataFileSync sync_;
    int64_t objectCount_;

    typedef std::map<std::string, std::vector<uint8_t> > Metadata;

    Metadata metadata_;

    static std::auto_ptr<OutputStream> makeStream(const char* filename);
    static DataFileSync makeSync();

    void writeHeader();
    void setMetadata(const std::string& key, const std::string& value);

    void sync();

public:
    Encoder& encoder() const { return *encoderPtr_; }

    void syncIfNeeded();

    void incr() {
        ++objectCount_;
    }
    DataFileWriterBase(const char* filename, const ValidSchema& schema,
        size_t syncInterval, Codec codec = NULL_CODEC);

    ~DataFileWriterBase();
    void close();

    const ValidSchema& schema() const { return schema_; }

    void flush();
};

template <typename T>
class DataFileWriter : boost::noncopyable {
    std::auto_ptr<DataFileWriterBase> base_;
public:
    DataFileWriter(const char* filename, const ValidSchema& schema,
        size_t syncInterval = 16 * 1024, Codec codec = NULL_CODEC) :
        base_(new DataFileWriterBase(filename, schema, syncInterval, codec)) { }

    void write(const T& datum) {
        base_->syncIfNeeded();
        avro::encode(base_->encoder(), datum);
        base_->incr();
    }

    void close() { base_->close(); }

    const ValidSchema& schema() const { return base_->schema(); }

    void flush() { base_->flush(); }
};

class AVRO_DECL DataFileReaderBase : boost::noncopyable {
    const std::string filename_;
    const std::auto_ptr<InputStream> stream_;
    const DecoderPtr decoder_;
    int64_t objectCount_;
    bool eof_;
    Codec codec_;

    ValidSchema readerSchema_;
    ValidSchema dataSchema_;
    DecoderPtr dataDecoder_;
    std::auto_ptr<InputStream> dataStream_;
    typedef std::map<std::string, std::vector<uint8_t> > Metadata;

    Metadata metadata_;
    DataFileSync sync_;

    // for compressed buffer
    boost::scoped_ptr<boost::iostreams::filtering_istream> os_;
    std::vector<char> compressed_;

    void readHeader();

    bool readDataBlock();
public:
    Decoder& decoder() { return *dataDecoder_; }

    bool hasMore();

    void decr() { --objectCount_; }

    DataFileReaderBase(const char* filename);

    void init();

    void init(const ValidSchema& readerSchema);

    const ValidSchema& readerSchema() { return readerSchema_; }

    const ValidSchema& dataSchema() { return dataSchema_; }

    void close();
};

template <typename T>
class DataFileReader : boost::noncopyable {
    std::auto_ptr<DataFileReaderBase> base_;
public:
    DataFileReader(const char* filename, const ValidSchema& readerSchema) :
        base_(new DataFileReaderBase(filename)) {
        base_->init(readerSchema);
    }

    DataFileReader(const char* filename) :
        base_(new DataFileReaderBase(filename)) {
        base_->init();
    }


    DataFileReader(std::auto_ptr<DataFileReaderBase> base) : base_(base) {
        base_->init();
    }

    DataFileReader(std::auto_ptr<DataFileReaderBase> base,
        const ValidSchema& readerSchema) : base_(base) {
        base_->init(readerSchema);
    }

    bool read(T& datum) {
        if (base_->hasMore()) {
            base_->decr();
            avro::decode(base_->decoder(), datum);
            return true;
        }
        return false;
    }

    const ValidSchema& readerSchema() { return base_->readerSchema(); }

    const ValidSchema& dataSchema() { return base_->dataSchema(); }

    void close() { return base_->close(); }
};

}   // namespace avro
#endif