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

// mcpack2pb - Make protobuf be front-end of mcpack/compack

// Date: Mon Oct 19 17:17:36 CST 2015

#ifndef MCPACK2PB_MCPACK_SERIALIZER_H

#define MCPACK2PB_MCPACK_SERIALIZER_H

#include <limits>

#include <vector>

#include <google/protobuf/io/zero_copy_stream.h>

#include "butil/logging.h"

#include "butil/strings/string_piece.h"

#include "mcpack2pb/field_type.h"

// CAUTION: Methods in this header is not intended to be public to users of

// brpc, and subject to change at any future time.

namespace mcpack2pb {

// Send bytes into ZeroCopyOutputStream

class OutputStream {

public:

    class Area {

    public:

        Area() : _addr1(NULL)

               , _addr2(NULL)

               , _size1(0)

               , _size2(0)

               , _addional_area(NULL) {}

        Area(const butil::LinkerInitialized&) {}

        Area(const Area& rhs);

        Area& operator=(const Area& rhs);

        ~Area();

        void add(void* data, size_t n);

        void assign(const void* data) const;

    private:

        void* _addr1;

        void* _addr2;

        unsigned _size1;

        unsigned _size2;

        std::vector<butil::StringPiece>* _addional_area;

    };

    // TODO(gejun): The zero-copy stream MUST return permanent memory blocks

    // to support reserve(). E.g. StringOutputStream can't be used because

    // the string inside invalidates previous memory blocks after resizing.

    OutputStream(google::protobuf::io::ZeroCopyOutputStream* stream)

        : _good(true)

        , _fullsize(0)

        , _size(0)

        , _data(NULL)

        , _zc_stream(stream)

        , _pushed_bytes(0)

    {}

    ~OutputStream() { done(); }

    // Append n bytes.

    void append(const void* data, int n);

    // Append a pod.

    template <typename T> void append_packed_pod(const T& packed_pod);

    template <typename T> T* append_packed_pod();

    // Append a byte.

    void push_back(char c);

    // If next n bytes in the zero-copy stream is continuous, consume it

    // and return the begining address. NULL otherwise.

    void* skip_continuous(int n);

    // Consume n bytes from the stream and return an object representing the

    // consumed area.

    Area reserve(int n);

    // Change data at the area. `data' must be as long as the reserved area.

    void assign(const Area&, const void* data);

    // Go back for n bytes.

    void backup(int n);

    // Returns bytes pushed and cut since creation of this stream.

    size_t pushed_bytes() const { return _pushed_bytes; }

    // Returns false if error occurred during serialization.

    bool good() { return _good; }

    void set_bad() { _good = false; }

    // Optionally called to backup buffered bytes to zero-copy stream.

    void done();

private:

    bool _good;

    int _fullsize;

    int _size;

    void* _data;

    google::protobuf::io::ZeroCopyOutputStream* _zc_stream;

    size_t _pushed_bytes;

};

// This class is different from butil::StringPiece that it only can be

// constructed from std::string or const char* which are both ended with

// zero, so that in later serialization of the name, we can simply copy one

// extra byte to end the name with zero(required by compack) rather than

// appending the zero in a separate function call which is less efficient.

class StringWrapper {

public:

    StringWrapper(const std::string& str)

        : _data(str.c_str()), _size(str.size()) {}

    StringWrapper(const char* str) {

        if (str) {

            _data = str;

            _size = strlen(str);

        } else {

            _data = "";

            _size = 0;

        }

    }

    ~StringWrapper() { }

    const char* data() const { return _data; }

    size_t size() const { return _size; }

    bool empty() const { return !_size; }

private:

    const char* _data;

    size_t _size;

};

inline std::ostream& operator<<(std::ostream& os, const StringWrapper& sw) {

    return os << butil::StringPiece(sw.data(), sw.size());

}

class Serializer {

public:

    // Serialize into `output'.

    explicit Serializer(OutputStream* stream);

    ~Serializer();

    bool good() const { return _stream->good(); }

    void set_bad() { _stream->set_bad(); }

    size_t pushed_bytes() const { return _stream->pushed_bytes(); }

    // WARNING: Names to all methods cannot be longer that 254 bytes.

    // Append a primitive type with name.

    // Used between begin_object() and end_object().

    void add_int8(const StringWrapper& name, int8_t value);

    void add_int16(const StringWrapper& name, int16_t value);

    void add_int32(const StringWrapper& name, int32_t value);

    void add_int64(const StringWrapper& name, int64_t value);

    void add_uint8(const StringWrapper& name, uint8_t value);

    void add_uint16(const StringWrapper& name, uint16_t value);

    void add_uint32(const StringWrapper& name, uint32_t value);

    void add_uint64(const StringWrapper& name, uint64_t value);

    void add_bool(const StringWrapper& name, bool value);

    void add_float(const StringWrapper& name, float value);

    void add_double(const StringWrapper& name, double value);

    // Add a primitive type without name.

    // Used between begin_xxx_array() and end_xxx_array().

    void add_int8(int8_t value);

    void add_int16(int16_t value);

    void add_int32(int32_t value);

    void add_int64(int64_t value);

    void add_uint8(uint8_t value);

    void add_uint16(uint16_t value);

    void add_uint32(uint32_t value);

    void add_uint64(uint64_t value);

    void add_bool(bool value);

    void add_float(float value);

    void add_double(double value);

    // Add multiple primitive types in one call.

    void add_multiple_int8(const int8_t* values, size_t count);

    void add_multiple_int8(const uint8_t* values, size_t count);

    void add_multiple_int16(const int16_t* values, size_t count);

    void add_multiple_int16(const uint16_t* values, size_t count);

    void add_multiple_int32(const int32_t* values, size_t count);

    void add_multiple_int32(const uint32_t* values, size_t count);

    void add_multiple_int64(const int64_t* values, size_t count);

    void add_multiple_int64(const uint64_t* values, size_t count);

    void add_multiple_uint8(const uint8_t* values, size_t count);

    void add_multiple_uint8(const int8_t* values, size_t count);

    void add_multiple_uint16(const uint16_t* values, size_t count);

    void add_multiple_uint16(const int16_t* values, size_t count);

    void add_multiple_uint32(const uint32_t* values, size_t count);

    void add_multiple_uint32(const int32_t* values, size_t count);

    void add_multiple_uint64(const uint64_t* values, size_t count);

    void add_multiple_uint64(const int64_t* values, size_t count);

    void add_multiple_bool(const bool* values, size_t count);

    void add_multiple_float(const float* values, size_t count);

    void add_multiple_double(const double* values, size_t count);

    // Append a string.

    // The serialized value ends with 0 and the length counts 0.

    void add_string(const StringWrapper& name, const StringWrapper& str);

    void add_string(const StringWrapper& str);

    // Append binary data.

    void add_binary(const StringWrapper& name, const std::string& data);

    void add_binary(const StringWrapper& name, const void* data, size_t n);

    void add_binary(const std::string& data);

    void add_binary(const void* data, size_t n);

    // Append a null.

    void add_null(const StringWrapper& name);

    void add_null();

    // Append an empty array.

    void add_empty_array(const StringWrapper& name);

    void add_empty_array();

    // Begin/end an array.

    // All items inside the array must be of same type. This is not a

    // restriction of mcpack, but we want to align storage model with protobuf

    // as much as possible.

    // If too many levels of array/object reached, this function fails and

    // leaves the output unchanged.

    void begin_mcpack_array(const StringWrapper& name, FieldType item_type);

    void begin_mcpack_array(FieldType item_type);

    void begin_compack_array(const StringWrapper& name, FieldType item_type);

    void begin_compack_array(FieldType item_type);

    // End any kind of array.

    void end_array();

    // Begin/end an object.

    // If too many levels of array/object reached, this function fails and

    // leaves the output unchanged.

    void begin_object(const StringWrapper& name);

    void begin_object();

    void end_object();

    // TODO(gejun): move to begin_object

    void end_object_iso();

public:

    struct GroupInfo {

        uint32_t item_count;

        bool isomorphic;

        uint8_t item_type;

        uint8_t type;

        uint8_t name_size;

        size_t output_offset;

        int pending_null_count;

        OutputStream::Area head_area;

        OutputStream::Area items_head_area;

        void print(std::ostream&) const;

    };

private:

    DISALLOW_COPY_AND_ASSIGN(Serializer);

    GroupInfo* push_group_info();

    GroupInfo & peek_group_info();

    void begin_object_internal(const StringWrapper& name);

    void begin_object_internal();

    void end_object_internal(bool objectisoarray);

    void begin_array_internal(FieldType item_type, bool compack);

    void begin_array_internal(const StringWrapper& name,

                              FieldType item_type, bool compack);

    OutputStream* _stream;

    int _ndepth;

    GroupInfo _group_info_fast[15];

    GroupInfo *_group_info_more;

};

}  // namespace mcpack2pb

#include "mcpack2pb/serializer-inl.h"

#endif  // MCPACK2PB_MCPACK_SERIALIZER_H