serialization.h

/*
 * Copyright (c) 2008-2025, Hazelcast, Inc. All Rights Reserved.
 *
 * Licensed 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.
 */
#pragma once
 
#include <unordered_map>
#include <unordered_set>
#include <type_traits>
 
#include <boost/any.hpp>
#include <boost/optional.hpp>
#include <boost/optional/optional_io.hpp>
#include <boost/uuid/uuid.hpp>
 
#include "hazelcast/client/hazelcast_json_value.h"
#include "hazelcast/client/serialization/pimpl/data_input.h"
#include "hazelcast/client/serialization/pimpl/data.h"
#include "hazelcast/client/serialization/pimpl/data_output.h"
#include "hazelcast/client/serialization/generic_record.h"
#include "hazelcast/client/serialization_config.h"
#include "hazelcast/client/partition_aware.h"
#include "hazelcast/util/SynchronizedMap.h"
#include "hazelcast/util/Disposable.h"
#include "hazelcast/client/big_decimal.h"
#include "hazelcast/client/local_time.h"
#include "hazelcast/client/local_date.h"
#include "hazelcast/client/local_date_time.h"
#include "hazelcast/client/offset_date_time.h"
 
#if defined(WIN32) || defined(_WIN32) || defined(WIN64) || defined(_WIN64)
#pragma warning(push)
#pragma warning(disable : 4251) // for dll export
#endif
 
namespace hazelcast {
namespace client {
class hazelcast_client;
 
namespace serialization {
class object_data_input;
namespace pimpl {
// forward declarations
class PortableContext;
class ClassDefinitionContext;
class ClassDefinitionWriter;
class DefaultPortableWriter;
class DefaultPortableReader;
class MorphingPortableReader;
class PortableSerializer;
class DataSerializer;
class SerializationService;
class default_compact_writer;
namespace offset_reader {
template<typename OFFSET_TYPE>
int32_t
get_offset(serialization::object_data_input& in,
           uint32_t variable_offsets_pos,
           uint32_t index);
}
enum struct HAZELCAST_API serialization_constants
{
    CONSTANT_TYPE_NULL = 0,
    CONSTANT_TYPE_PORTABLE = -1,
    CONSTANT_TYPE_DATA = -2,
    CONSTANT_TYPE_BYTE = -3,
    CONSTANT_TYPE_BOOLEAN = -4,
    CONSTANT_TYPE_CHAR = -5,
    CONSTANT_TYPE_SHORT = -6,
    CONSTANT_TYPE_INTEGER = -7,
    CONSTANT_TYPE_LONG = -8,
    CONSTANT_TYPE_FLOAT = -9,
    CONSTANT_TYPE_DOUBLE = -10,
    CONSTANT_TYPE_STRING = -11,
    CONSTANT_TYPE_BYTE_ARRAY = -12,
    CONSTANT_TYPE_BOOLEAN_ARRAY = -13,
    CONSTANT_TYPE_CHAR_ARRAY = -14,
    CONSTANT_TYPE_SHORT_ARRAY = -15,
    CONSTANT_TYPE_INTEGER_ARRAY = -16,
    CONSTANT_TYPE_LONG_ARRAY = -17,
    CONSTANT_TYPE_FLOAT_ARRAY = -18,
    CONSTANT_TYPE_DOUBLE_ARRAY = -19,
    CONSTANT_TYPE_STRING_ARRAY = -20,
    CONSTANT_TYPE_UUID = -21,
    CONSTANT_TYPE_COMPACT = -55,
    JAVASCRIPT_JSON_SERIALIZATION_TYPE = -130,
 
    CONSTANT_TYPE_GLOBAL = INT32_MIN
    // ------------------------------------------------------------
};

struct HAZELCAST_API object_type
{
    object_type();
 
    serialization_constants type_id;
    int32_t factory_id;
    int32_t class_id;
};
 
std::ostream HAZELCAST_API&
operator<<(std::ostream& os, const object_type& type);
} // namespace pimpl
} // namespace serialization

class HAZELCAST_API typed_data
{
public:
    typed_data();
 
    typed_data(
      serialization::pimpl::data d,
      serialization::pimpl::SerializationService& serialization_service);

    serialization::pimpl::object_type get_type() const;

    template<typename T>
    boost::optional<T> get() const;

    const serialization::pimpl::data& get_data() const;
 
private:
    serialization::pimpl::data data_;
    serialization::pimpl::SerializationService* ss_;
};
 
bool HAZELCAST_API
operator<(const typed_data& lhs, const typed_data& rhs);
 
namespace serialization {
class object_data_input;
class object_data_output;
class portable_reader;
struct compact_serializer;
 
namespace pimpl {
// forward declarations
class PortableContext;
class ClassDefinitionContext;
class ClassDefinitionWriter;
class DefaultPortableWriter;
class DefaultPortableReader;
class MorphingPortableReader;
class PortableSerializer;
class compact_stream_serializer;
class DataSerializer;
class SerializationService;
} // namespace pimpl
 
template<typename T>
struct hz_serializer
{};
 
struct builtin_serializer
{};
 
struct custom_serializer
{};
 
struct global_serializer
{
    virtual ~global_serializer() = default;
 
    static pimpl::serialization_constants get_type_id()
    {
        return pimpl::serialization_constants::CONSTANT_TYPE_GLOBAL;
    }
 
    virtual void write(const boost::any& object, object_data_output& out) = 0;
 
    virtual boost::any read(object_data_input& in) = 0;
};

struct identified_data_serializer
{};

struct portable_serializer
{};
 
struct versioned_portable_serializer : public portable_serializer
{};
 
template<>
struct HAZELCAST_API hz_serializer<byte> : public builtin_serializer
{
public:
    static inline pimpl::serialization_constants get_type_id()
    {
        return pimpl::serialization_constants::CONSTANT_TYPE_BYTE;
    }
};
 
template<>
struct HAZELCAST_API hz_serializer<bool> : public builtin_serializer
{
public:
    static inline pimpl::serialization_constants get_type_id()
    {
        return pimpl::serialization_constants::CONSTANT_TYPE_BOOLEAN;
    }
};
 
template<>
struct HAZELCAST_API hz_serializer<char> : public builtin_serializer
{
public:
    static inline pimpl::serialization_constants get_type_id()
    {
        return pimpl::serialization_constants::CONSTANT_TYPE_CHAR;
    }
};
 
template<>
struct HAZELCAST_API hz_serializer<char16_t> : public builtin_serializer
{
public:
    static inline pimpl::serialization_constants get_type_id()
    {
        return pimpl::serialization_constants::CONSTANT_TYPE_CHAR;
    }
};
 
template<>
struct HAZELCAST_API hz_serializer<int16_t> : public builtin_serializer
{
public:
    static inline pimpl::serialization_constants get_type_id()
    {
        return pimpl::serialization_constants::CONSTANT_TYPE_SHORT;
    }
};
 
template<>
struct HAZELCAST_API hz_serializer<int32_t> : public builtin_serializer
{
public:
    static inline pimpl::serialization_constants get_type_id()
    {
        return pimpl::serialization_constants::CONSTANT_TYPE_INTEGER;
    }
};
 
template<>
struct HAZELCAST_API hz_serializer<int64_t> : public builtin_serializer
{
public:
    static inline pimpl::serialization_constants get_type_id()
    {
        return pimpl::serialization_constants::CONSTANT_TYPE_LONG;
    }
};
 
template<>
struct HAZELCAST_API hz_serializer<float> : public builtin_serializer
{
public:
    static inline pimpl::serialization_constants get_type_id()
    {
        return pimpl::serialization_constants::CONSTANT_TYPE_FLOAT;
    }
};
 
template<>
struct HAZELCAST_API hz_serializer<double> : public builtin_serializer
{
public:
    static inline pimpl::serialization_constants get_type_id()
    {
        return pimpl::serialization_constants::CONSTANT_TYPE_DOUBLE;
    }
};
 
template<>
struct HAZELCAST_API hz_serializer<std::string> : public builtin_serializer
{
public:
    static inline pimpl::serialization_constants get_type_id()
    {
        return pimpl::serialization_constants::CONSTANT_TYPE_STRING;
    }
};
 
template<>
struct HAZELCAST_API hz_serializer<hazelcast_json_value>
  : public builtin_serializer
{
public:
    static inline pimpl::serialization_constants get_type_id()
    {
        return pimpl::serialization_constants::
          JAVASCRIPT_JSON_SERIALIZATION_TYPE;
    }
};
 
template<>
struct HAZELCAST_API hz_serializer<std::vector<byte>>
  : public builtin_serializer
{
public:
    static inline pimpl::serialization_constants get_type_id()
    {
        return pimpl::serialization_constants::CONSTANT_TYPE_BYTE_ARRAY;
    }
};
 
template<>
struct HAZELCAST_API hz_serializer<std::vector<bool>>
  : public builtin_serializer
{
public:
    static inline pimpl::serialization_constants get_type_id()
    {
        return pimpl::serialization_constants::CONSTANT_TYPE_BOOLEAN_ARRAY;
    }
};
 
template<>
struct HAZELCAST_API hz_serializer<std::vector<char>>
  : public builtin_serializer
{
public:
    static inline pimpl::serialization_constants get_type_id()
    {
        return pimpl::serialization_constants::CONSTANT_TYPE_CHAR_ARRAY;
    }
};
 
template<>
struct HAZELCAST_API hz_serializer<std::vector<int16_t>>
  : public builtin_serializer
{
public:
    static inline pimpl::serialization_constants get_type_id()
    {
        return pimpl::serialization_constants::CONSTANT_TYPE_SHORT_ARRAY;
    }
};
 
template<>
struct HAZELCAST_API hz_serializer<std::vector<int32_t>>
  : public builtin_serializer
{
public:
    static inline pimpl::serialization_constants get_type_id()
    {
        return pimpl::serialization_constants::CONSTANT_TYPE_INTEGER_ARRAY;
    }
};
 
template<>
struct HAZELCAST_API hz_serializer<std::vector<int64_t>>
  : public builtin_serializer
{
public:
    static inline pimpl::serialization_constants get_type_id()
    {
        return pimpl::serialization_constants::CONSTANT_TYPE_LONG_ARRAY;
    }
};
 
template<>
struct HAZELCAST_API hz_serializer<std::vector<float>>
  : public builtin_serializer
{
public:
    static inline pimpl::serialization_constants get_type_id()
    {
        return pimpl::serialization_constants::CONSTANT_TYPE_FLOAT_ARRAY;
    }
};
 
template<>
struct HAZELCAST_API hz_serializer<std::vector<double>>
  : public builtin_serializer
{
public:
    static inline pimpl::serialization_constants get_type_id()
    {
        return pimpl::serialization_constants::CONSTANT_TYPE_DOUBLE_ARRAY;
    }
};
 
template<>
struct HAZELCAST_API hz_serializer<std::vector<std::string>>
  : public builtin_serializer
{
public:
    static inline pimpl::serialization_constants get_type_id()
    {
        return pimpl::serialization_constants::CONSTANT_TYPE_STRING_ARRAY;
    }
};
 
template<>
struct HAZELCAST_API hz_serializer<boost::uuids::uuid>
  : public builtin_serializer
{
public:
    static inline pimpl::serialization_constants get_type_id()
    {
        return pimpl::serialization_constants::CONSTANT_TYPE_UUID;
    }
};
 
enum struct field_type
{
    TYPE_PORTABLE = 0,
    TYPE_BYTE = 1,
    TYPE_BOOLEAN = 2,
    TYPE_CHAR = 3,
    TYPE_SHORT = 4,
    TYPE_INT = 5,
    TYPE_LONG = 6,
    TYPE_FLOAT = 7,
    TYPE_DOUBLE = 8,
    TYPE_STRING = 9,
    TYPE_PORTABLE_ARRAY = 10,
    TYPE_BYTE_ARRAY = 11,
    TYPE_BOOLEAN_ARRAY = 12,
    TYPE_CHAR_ARRAY = 13,
    TYPE_SHORT_ARRAY = 14,
    TYPE_INT_ARRAY = 15,
    TYPE_LONG_ARRAY = 16,
    TYPE_FLOAT_ARRAY = 17,
    TYPE_DOUBLE_ARRAY = 18,
    TYPE_STRING_ARRAY = 19
};

class HAZELCAST_API FieldDefinition
{
public:
    FieldDefinition();

    FieldDefinition(int,
                    const std::string&,
                    field_type const& type,
                    int version);

    FieldDefinition(int index,
                    const std::string& field_name,
                    field_type const& type,
                    int factory_id,
                    int class_id,
                    int version);

    const field_type& get_type() const;

    std::string get_name() const;

    int get_index() const;

    int get_factory_id() const;

    int get_class_id() const;

    void write_data(pimpl::data_output& data_output);

    void read_data(object_data_input& data_input);
 
    bool operator==(const FieldDefinition& rhs) const;
 
    bool operator!=(const FieldDefinition& rhs) const;
 
    friend std::ostream& operator<<(std::ostream& os,
                                    const FieldDefinition& definition);
 
private:
    int index_;
    std::string field_name_;
    field_type type_;
    int class_id_;
    int factory_id_;
    int version_;
};
 
class HAZELCAST_API ClassDefinition
{
public:
    ClassDefinition();

    ClassDefinition(int factory_id, int class_id, int version);

    void add_field_def(FieldDefinition& field_definition);

    bool has_field(const std::string& field_name) const;

    const FieldDefinition& get_field(const std::string& field_name) const;

    field_type get_field_type(const std::string& field_name) const;

    int get_field_count() const;

    int get_factory_id() const;

    int get_class_id() const;

    int get_version() const;

    void set_version_if_not_set(int new_version);

    void write_data(pimpl::data_output& data_output);

    void read_data(object_data_input& data_input);
 
    bool operator==(const ClassDefinition& rhs) const;
 
    bool operator!=(const ClassDefinition& rhs) const;
 
    friend std::ostream& operator<<(std::ostream& os,
                                    const ClassDefinition& definition);
 
private:
    int factory_id_;
    int class_id_;
    int version_;
 
    ClassDefinition(const ClassDefinition&) = delete;
 
    ClassDefinition& operator=(const ClassDefinition& rhs) = delete;
 
    std::unordered_map<std::string, FieldDefinition> field_definitions_map_;
 
    std::unique_ptr<std::vector<byte>> binary_;
};

class HAZELCAST_API ClassDefinitionBuilder
{
public:
    ClassDefinitionBuilder(int factory_id, int class_id, int version);
 
    ClassDefinitionBuilder& add_portable_field(
      const std::string& field_name,
      std::shared_ptr<ClassDefinition> def);
 
    ClassDefinitionBuilder& add_portable_array_field(
      const std::string& field_name,
      std::shared_ptr<ClassDefinition> def);
 
    ClassDefinitionBuilder& add_field(FieldDefinition& field_definition);
 
    void add_field(const std::string& field_name, field_type const& field_type);
 
    std::shared_ptr<ClassDefinition> build();
 
    int get_factory_id();
 
    int get_class_id();
 
    int get_version();
 
private:
    int factory_id_;
    int class_id_;
    int version_;
    int index_;
    bool done_;
 
    std::vector<FieldDefinition> field_definitions_;
 
    void check();
};
 
class PortableVersionHelper
{
public:
    template<typename T>
    static inline typename std::enable_if<
      std::is_base_of<versioned_portable_serializer, hz_serializer<T>>::value,
      int>::type
    get_version(int)
    {
        return hz_serializer<T>::get_class_version();
    }
 
    template<typename T>
    static inline typename std::enable_if<
      !std::is_base_of<versioned_portable_serializer, hz_serializer<T>>::value,
      int>::type
    get_version(int default_version)
    {
        return default_version;
    }
};
 
namespace compact {
class compact_writer;
class compact_reader;
class compact_serializer;
} // namespace compact
 
class HAZELCAST_API object_data_input
  : public pimpl::data_input<std::vector<byte>>
{
    template<typename OFFSET_TYPE>
    friend int32_t pimpl::offset_reader::get_offset(
      object_data_input& in,
      uint32_t variable_offsets_pos,
      uint32_t index);
 
    friend class compact::compact_reader;
    friend class portable_reader;
 
public:
    object_data_input(
      boost::endian::order byte_order,
      const std::vector<byte>& buffer,
      int offset,
      pimpl::PortableSerializer& portable_ser,
      pimpl::compact_stream_serializer& compact_ser,
      pimpl::DataSerializer& data_ser,
      std::shared_ptr<serialization::global_serializer> global_serializer);

    template<typename T>
    typename std::enable_if<
      !(std::is_array<T>::value &&
        std::is_same<typename std::remove_all_extents<T>::type, char>::value),
      boost::optional<T>>::type inline read_object();
 
    template<typename T>
    typename std::enable_if<
      std::is_array<T>::value &&
        std::is_same<typename std::remove_all_extents<T>::type, char>::value,
      boost::optional<std::string>>::type inline read_object();
 
    template<typename T>
    typename std::enable_if<
      std::is_base_of<identified_data_serializer, hz_serializer<T>>::value,
      boost::optional<T>>::type inline read_object(int32_t type_id);
 
    template<typename T>
    typename std::enable_if<
      std::is_base_of<portable_serializer, hz_serializer<T>>::value,
      boost::optional<T>>::type inline read_object(int32_t type_id);
 
    template<typename T>
    typename std::enable_if<
      std::is_base_of<compact::compact_serializer, hz_serializer<T>>::value,
      boost::optional<T>>::type inline read_object(int32_t type_id);
 
    template<typename T>
    typename std::enable_if<
      std::is_base_of<builtin_serializer, hz_serializer<T>>::value,
      boost::optional<T>>::type inline read_object(int32_t type_id);
 
    template<typename T>
    typename std::enable_if<
      std::is_base_of<custom_serializer, hz_serializer<T>>::value,
      boost::optional<T>>::type inline read_object(int32_t type_id);
 
    template<typename T>
    typename std::enable_if<
      std::is_same<generic_record::generic_record, T>::value,
      boost::optional<T>>::type
    read_object(int32_t type_id);

    template<typename T>
    typename std::enable_if<
      !(std::is_base_of<identified_data_serializer, hz_serializer<T>>::value ||
        std::is_base_of<portable_serializer, hz_serializer<T>>::value ||
        std::is_base_of<compact::compact_serializer, hz_serializer<T>>::value ||
        std::is_base_of<builtin_serializer, hz_serializer<T>>::value ||
        std::is_base_of<custom_serializer, hz_serializer<T>>::value ||
        std::is_same<generic_record::generic_record, T>::value),
      boost::optional<T>>::type inline read_object(int32_t type_id);
 
private:
    pimpl::PortableSerializer& portable_serializer_;
    pimpl::compact_stream_serializer& compact_serializer_;
    pimpl::DataSerializer& data_serializer_;
    std::shared_ptr<serialization::global_serializer> global_serializer_;
};
 
class HAZELCAST_API object_data_output : public pimpl::data_output
{
    friend pimpl::DefaultPortableWriter;
    friend pimpl::default_compact_writer;
 
public:
    explicit object_data_output(
      boost::endian::order byte_order,
      bool dont_write = false,
      pimpl::PortableSerializer* portable_ser = nullptr,
      pimpl::compact_stream_serializer* compact_ser = nullptr,
      std::shared_ptr<serialization::global_serializer> global_serializer =
        nullptr);
 
    template<typename T>
    void write_object(const T* object);
 
    /* enable_if needed here since 'boost::optional<char [5]>' can not be
     * composed this template match */
    template<typename T>
    typename std::enable_if<
      !(std::is_array<T>::value &&
        std::is_same<typename std::remove_all_extents<T>::type, char>::value),
      void>::type
    write_object(const boost::optional<T>& object);
 
    template<typename T>
    typename std::enable_if<
      std::is_array<T>::value &&
        std::is_same<typename std::remove_all_extents<T>::type, char>::value,
      void>::type inline write_object(const T& object);
 
    template<typename T>
    typename std::enable_if<
      std::is_base_of<builtin_serializer, hz_serializer<T>>::value,
      void>::type inline write_object(const T& object);
 
    template<typename T>
    typename std::enable_if<
      std::is_base_of<identified_data_serializer, hz_serializer<T>>::value,
      void>::type inline write_object(const T& object);
 
    template<typename T>
    typename std::enable_if<
      std::is_base_of<portable_serializer, hz_serializer<T>>::value,
      void>::type inline write_object(const T& object);
 
    template<typename T>
    typename std::enable_if<
      std::is_same<generic_record::generic_record, T>::value>::type
    write_object(const T& object);
 
    template<typename T>
    typename std::enable_if<
      std::is_base_of<compact::compact_serializer, hz_serializer<T>>::value,
      void>::type inline write_object(const T& object);
 
    template<typename T>
    typename std::enable_if<
      std::is_base_of<custom_serializer, hz_serializer<T>>::value,
      void>::type inline write_object(const T& object);
 
    template<typename T>
    typename std::enable_if<
      !(std::is_base_of<builtin_serializer, hz_serializer<T>>::value ||
        std::is_base_of<identified_data_serializer, hz_serializer<T>>::value ||
        std::is_base_of<portable_serializer, hz_serializer<T>>::value ||
        std::is_base_of<compact::compact_serializer, hz_serializer<T>>::value ||
        std::is_base_of<custom_serializer, hz_serializer<T>>::value ||
        std::is_same<generic_record::generic_record, T>::value ||
        (std::is_array<T>::value &&
         std::is_same<typename std::remove_all_extents<T>::type, char>::value)),
      void>::type inline write_object(const T object);
 
    void write_objects() {}
 
    template<typename FirstObjectType, typename... OtherObjects>
    inline void write_objects(const FirstObjectType& object,
                              const OtherObjects&... objects)
    {
        write_object(object);
        write_objects(objects...);
    }
 
    template<typename T>
    inline void write_bytes(const T& s)
    {
        for (auto c : s) {
            output_stream_.push_back(c);
        }
    }
 
private:
    using schemas_t = std::vector<pimpl::schema>;
 
    friend class pimpl::compact_stream_serializer;
    friend class pimpl::SerializationService;
 
    schemas_t schemas_will_be_replicated_;
    pimpl::PortableSerializer* portable_serializer_;
    pimpl::compact_stream_serializer* compact_serializer_;
    std::shared_ptr<serialization::global_serializer> global_serializer_;
};
 
template<>
void HAZELCAST_API
object_data_output::write_object(const char* object);
 
namespace pimpl {
class HAZELCAST_API serialization_util
{
public:
    template<typename T>
    static typename std::enable_if<
      std::is_same<boost::multiprecision::cpp_int,
                   typename std::remove_cv<T>::type>::value,
      T>::type
    read(client::serialization::object_data_input& object_data_input)
    {
        int32_t size = object_data_input.read<int32_t>();
        std::vector<int8_t> bytes(size);
        object_data_input.read_fully(bytes);
        return client::pimpl::from_bytes(std::move(bytes));
    }
 
    template<typename T>
    static typename std::enable_if<
      std::is_same<boost::multiprecision::cpp_int,
                   typename std::remove_cv<T>::type>::value,
      void>::type
    write(client::serialization::object_data_output& object_data_output,
          const T& value)
    {
        auto v = hazelcast::client::pimpl::to_bytes(value);
        object_data_output.write(v);
    }
 
    template<typename T>
    static typename std::enable_if<
      std::is_same<client::big_decimal,
                   typename std::remove_cv<T>::type>::value,
      T>::type
    read(client::serialization::object_data_input& object_data_input)
    {
        auto cpp_int = read<boost::multiprecision::cpp_int>(object_data_input);
        int32_t scale = object_data_input.read<int32_t>();
        return client::big_decimal{ std::move(cpp_int), scale };
    }
 
    template<typename T>
    static typename std::enable_if<
      std::is_same<client::big_decimal,
                   typename std::remove_cv<T>::type>::value,
      void>::type
    write(client::serialization::object_data_output& object_data_output,
          const T& value)
    {
        write(object_data_output, value.unscaled);
        object_data_output.write(value.scale);
    }
 
    template<typename T>
    static typename std::enable_if<
      std::is_same<client::local_time, typename std::remove_cv<T>::type>::value,
      T>::type
    read(client::serialization::object_data_input& object_data_input)
    {
        byte hour = object_data_input.read<byte>();
        byte minute = object_data_input.read<byte>();
        byte second = object_data_input.read<byte>();
        int32_t nano = object_data_input.read<int32_t>();
        return client::local_time{ hour, minute, second, nano };
    }
 
    template<typename T>
    static typename std::enable_if<
      std::is_same<client::local_time, typename std::remove_cv<T>::type>::value,
      void>::type
    write(client::serialization::object_data_output& object_data_output,
          const T& value)
    {
        object_data_output.write<byte>(value.hours);
        object_data_output.write<byte>(value.minutes);
        object_data_output.write<byte>(value.seconds);
        object_data_output.write<int32_t>(value.nanos);
    }
 
    template<typename T>
    static typename std::enable_if<
      std::is_same<client::local_date, typename std::remove_cv<T>::type>::value,
      T>::type
    read(client::serialization::object_data_input& object_data_input)
    {
        int32_t year = object_data_input.read<int32_t>();
        byte month = object_data_input.read<byte>();
        byte dayOfMonth = object_data_input.read<byte>();
        return client::local_date{ year, month, dayOfMonth };
    }
 
    template<typename T>
    static typename std::enable_if<
      std::is_same<client::local_date, typename std::remove_cv<T>::type>::value,
      void>::type
    write(client::serialization::object_data_output& object_data_output,
          const T& value)
    {
        object_data_output.write<int32_t>(value.year);
        object_data_output.write<byte>(value.month);
        object_data_output.write<byte>(value.day_of_month);
    }
 
    template<typename T>
    static typename std::enable_if<
      std::is_same<client::local_date_time,
                   typename std::remove_cv<T>::type>::value,
      T>::type
    read(client::serialization::object_data_input& object_data_input)
    {
        auto date = read<client::local_date>(object_data_input);
        auto time = read<client::local_time>(object_data_input);
        return client::local_date_time{ date, time };
    }
 
    template<typename T>
    static typename std::enable_if<
      std::is_same<client::local_date_time,
                   typename std::remove_cv<T>::type>::value,
      void>::type
    write(client::serialization::object_data_output& object_data_output,
          const T& value)
    {
        write(object_data_output, value.date);
        write(object_data_output, value.time);
    }
 
    template<typename T>
    static typename std::enable_if<
      std::is_same<client::offset_date_time,
                   typename std::remove_cv<T>::type>::value,
      T>::type
    read(client::serialization::object_data_input& object_data_input)
    {
        auto local_date_time = read<client::local_date_time>(object_data_input);
        int32_t zoneTotalSeconds = object_data_input.read<int32_t>();
        return client::offset_date_time{ local_date_time, zoneTotalSeconds };
    }
 
    template<typename T>
    static typename std::enable_if<
      std::is_same<client::offset_date_time,
                   typename std::remove_cv<T>::type>::value,
      void>::type
    write(client::serialization::object_data_output& object_data_output,
          const T& value)
    {
        write(object_data_output, value.date_time);
        object_data_output.write<int32_t>(value.zone_offset_in_seconds);
    }
};
 
class HAZELCAST_API PortableContext
{
public:
    PortableContext(const serialization_config& serialization_conf);
 
    int get_class_version(int factory_id, int class_id);
 
    void set_class_version(int factory_id, int class_id, int version);
 
    std::shared_ptr<ClassDefinition> lookup_class_definition(int factory_id,
                                                             int class_id,
                                                             int version);
 
    std::shared_ptr<ClassDefinition> register_class_definition(
      std::shared_ptr<ClassDefinition>);
 
    template<typename T>
    std::shared_ptr<ClassDefinition> lookup_or_register_class_definition(
      const T& portable);
 
    int get_version();
 
    std::shared_ptr<ClassDefinition> read_class_definition(
      object_data_input& input,
      int id,
      int class_id,
      int version);
 
    const serialization_config& get_serialization_config() const;
 
    template<typename T>
    typename std::enable_if<
      std::is_same<byte, typename std::remove_cv<T>::type>::value,
      field_type>::type static get_type()
    {
        return field_type::TYPE_BYTE;
    }
 
    template<typename T>
    typename std::enable_if<
      std::is_same<char, typename std::remove_cv<T>::type>::value,
      field_type>::type static get_type()
    {
        return field_type::TYPE_CHAR;
    }
 
    template<typename T>
    typename std::enable_if<
      std::is_same<char16_t, typename std::remove_cv<T>::type>::value,
      field_type>::type static get_type()
    {
        return field_type::TYPE_CHAR;
    }
 
    template<typename T>
    typename std::enable_if<
      std::is_same<bool, typename std::remove_cv<T>::type>::value,
      field_type>::type static get_type()
    {
        return field_type::TYPE_BOOLEAN;
    }
 
    template<typename T>
    typename std::enable_if<
      std::is_same<int16_t, typename std::remove_cv<T>::type>::value,
      field_type>::type static get_type()
    {
        return field_type::TYPE_SHORT;
    }
 
    template<typename T>
    typename std::enable_if<
      std::is_same<int32_t, typename std::remove_cv<T>::type>::value,
      field_type>::type static get_type()
    {
        return field_type::TYPE_INT;
    }
 
    template<typename T>
    typename std::enable_if<
      std::is_same<int64_t, typename std::remove_cv<T>::type>::value,
      field_type>::type static get_type()
    {
        return field_type::TYPE_LONG;
    }
 
    template<typename T>
    typename std::enable_if<
      std::is_same<float, typename std::remove_cv<T>::type>::value,
      field_type>::type static get_type()
    {
        return field_type::TYPE_FLOAT;
    }
 
    template<typename T>
    typename std::enable_if<
      std::is_same<double, typename std::remove_cv<T>::type>::value,
      field_type>::type static get_type()
    {
        return field_type::TYPE_DOUBLE;
    }
 
    template<typename T>
    typename std::enable_if<
      std::is_same<std::string, typename std::remove_cv<T>::type>::value,
      field_type>::type static get_type()
    {
        return field_type::TYPE_STRING;
    }
 
    template<typename T>
    typename std::enable_if<
      std::is_same<std::vector<byte>, typename std::remove_cv<T>::type>::value,
      field_type>::type static get_type()
    {
        return field_type::TYPE_BYTE_ARRAY;
    }
 
    template<typename T>
    typename std::enable_if<
      std::is_same<std::vector<char>, typename std::remove_cv<T>::type>::value,
      field_type>::type static get_type()
    {
        return field_type::TYPE_CHAR_ARRAY;
    }
 
    template<typename T>
    typename std::enable_if<
      std::is_same<std::vector<bool>, typename std::remove_cv<T>::type>::value,
      field_type>::type static get_type()
    {
        return field_type::TYPE_BOOLEAN_ARRAY;
    }
 
    template<typename T>
    typename std::enable_if<
      std::is_same<std::vector<int16_t>,
                   typename std::remove_cv<T>::type>::value,
      field_type>::type static get_type()
    {
        return field_type::TYPE_SHORT_ARRAY;
    }
 
    template<typename T>
    typename std::enable_if<
      std::is_same<std::vector<int32_t>,
                   typename std::remove_cv<T>::type>::value,
      field_type>::type static get_type()
    {
        return field_type::TYPE_INT_ARRAY;
    }
 
    template<typename T>
    typename std::enable_if<
      std::is_same<std::vector<int64_t>,
                   typename std::remove_cv<T>::type>::value,
      field_type>::type static get_type()
    {
        return field_type::TYPE_LONG_ARRAY;
    }
 
    template<typename T>
    typename std::enable_if<
      std::is_same<std::vector<float>, typename std::remove_cv<T>::type>::value,
      field_type>::type static get_type()
    {
        return field_type::TYPE_FLOAT_ARRAY;
    }
 
    template<typename T>
    typename std::enable_if<
      std::is_same<std::vector<double>,
                   typename std::remove_cv<T>::type>::value,
      field_type>::type static get_type()
    {
        return field_type::TYPE_DOUBLE_ARRAY;
    }
 
    template<typename T>
    typename std::enable_if<
      std::is_same<std::vector<std::string>,
                   typename std::remove_cv<T>::type>::value,
      field_type>::type static get_type()
    {
        return field_type::TYPE_STRING_ARRAY;
    }
 
private:
    PortableContext(const PortableContext&) = delete;
 
    ClassDefinitionContext& get_class_definition_context(int factory_id);
 
    void operator=(const PortableContext&) = delete;
 
    util::SynchronizedMap<int, ClassDefinitionContext> class_def_context_map_;
    const serialization_config& serialization_config_;
};
 
class ClassDefinitionContext
{
public:
    ClassDefinitionContext(int portable_context, PortableContext* p_context);
 
    int get_class_version(int class_id);
 
    void set_class_version(int class_id, int version);
 
    std::shared_ptr<ClassDefinition> lookup(int, int);
 
    std::shared_ptr<ClassDefinition> register_class_definition(
      std::shared_ptr<ClassDefinition>);
 
private:
    int64_t combine_to_long(int x, int y) const;
 
    const int factory_id_;
    util::SynchronizedMap<long long, ClassDefinition> versioned_definitions_;
    util::SynchronizedMap<int, int> current_class_versions_;
    PortableContext* portable_context_;
};
 
class HAZELCAST_API ClassDefinitionWriter
{
public:
    ClassDefinitionWriter(PortableContext& portable_context,
                          ClassDefinitionBuilder& builder);
 
    template<typename T>
    void write(const std::string& field_name, T value)
    {
        typedef typename std::remove_pointer<typename std::remove_reference<
          typename std::remove_cv<T>::type>::type>::type value_type;
        builder_.add_field(field_name, PortableContext::get_type<value_type>());
    }
 
    template<typename T>
    void write_null_portable(const std::string& field_name)
    {
        T portable;
        int32_t factoryId = hz_serializer<T>::get_factory_id();
        int32_t classId = hz_serializer<T>::get_class_id();
        std::shared_ptr<ClassDefinition> nestedClassDef =
          context_.lookup_class_definition(
            factoryId, classId, context_.get_version());
        if (!nestedClassDef) {
            BOOST_THROW_EXCEPTION(exception::hazelcast_serialization(
              "ClassDefWriter::write_null_portable",
              "Cannot write null portable without explicitly registering class "
              "definition!"));
        }
        builder_.add_portable_field(field_name, nestedClassDef);
    }
 
    template<typename T>
    void write_portable(const std::string& field_name, const T* portable)
    {
        if (NULL == portable) {
            BOOST_THROW_EXCEPTION(exception::hazelcast_serialization(
              "ClassDefinitionWriter::write_portable",
              "Cannot write null portable without explicitly registering class "
              "definition!"));
        }
 
        std::shared_ptr<ClassDefinition> nestedClassDef =
          create_nested_class_def(*portable);
        builder_.add_portable_field(field_name, nestedClassDef);
    }
 
    template<typename T>
    void write_portable_array(const std::string& field_name,
                              const std::vector<T>* portables)
    {
        if (NULL == portables || portables->size() == 0) {
            BOOST_THROW_EXCEPTION(exception::hazelcast_serialization(
              "ClassDefinitionWriter::write_portableArray",
              "Cannot write null portable array without explicitly registering "
              "class definition!"));
        }
        std::shared_ptr<ClassDefinition> nestedClassDef =
          create_nested_class_def((*portables)[0]);
        builder_.add_portable_array_field(field_name, nestedClassDef);
    }
 
    std::shared_ptr<ClassDefinition> register_and_get();
 
    object_data_output& get_raw_data_output();
 
    void end();
 
private:
    template<typename T>
    std::shared_ptr<ClassDefinition> create_nested_class_def(const T& portable);
 
    ClassDefinitionBuilder& builder_;
    PortableContext& context_;
    object_data_output empty_data_output_;
};
 
class HAZELCAST_API PortableReaderBase
{
public:
    PortableReaderBase(PortableSerializer& portable_ser,
                       object_data_input& input,
                       std::shared_ptr<ClassDefinition> cd);
 
    template<typename T>
    typename std::enable_if<
      std::is_same<byte, typename std::remove_cv<T>::type>::value ||
        std::is_same<char, typename std::remove_cv<T>::type>::value ||
        std::is_same<char16_t, typename std::remove_cv<T>::type>::value ||
        std::is_same<bool, typename std::remove_cv<T>::type>::value ||
        std::is_same<int16_t, typename std::remove_cv<T>::type>::value ||
        std::is_same<int32_t, typename std::remove_cv<T>::type>::value ||
        std::is_same<int64_t, typename std::remove_cv<T>::type>::value ||
        std::is_same<float, typename std::remove_cv<T>::type>::value ||
        std::is_same<double, typename std::remove_cv<T>::type>::value ||
        std::is_same<std::string, typename std::remove_cv<T>::type>::value,
      T>::type
    read(const std::string& field_name)
    {
        set_position(field_name, PortableContext::get_type<T>());
        return data_input_->read<T>();
    }
 
    template<typename T>
    typename std::enable_if<
      std::is_same<boost::optional<std::string>,
                   typename std::remove_cv<T>::type>::value,
      T>::type
    read(const std::string& field_name)
    {
        set_position(field_name, PortableContext::get_type<T>());
        return data_input_->read<T>();
    }
 
    template<typename T>
    typename std::enable_if<
      std::is_same<std::vector<byte>,
                   typename std::remove_cv<T>::type>::value ||
        std::is_same<std::vector<char>,
                     typename std::remove_cv<T>::type>::value ||
        std::is_same<std::vector<bool>,
                     typename std::remove_cv<T>::type>::value ||
        std::is_same<std::vector<int16_t>,
                     typename std::remove_cv<T>::type>::value ||
        std::is_same<std::vector<int32_t>,
                     typename std::remove_cv<T>::type>::value ||
        std::is_same<std::vector<int64_t>,
                     typename std::remove_cv<T>::type>::value ||
        std::is_same<std::vector<float>,
                     typename std::remove_cv<T>::type>::value ||
        std::is_same<std::vector<double>,
                     typename std::remove_cv<T>::type>::value ||
        std::is_same<std::vector<std::string>,
                     typename std::remove_cv<T>::type>::value,
      boost::optional<T>>::type
    read(const std::string& field_name)
    {
        set_position(field_name, PortableContext::get_type<T>());
        return data_input_->read<T>();
    }
 
    object_data_input& get_raw_data_input();
 
    void end();
 
protected:
    void set_position(const std::string& field_name,
                      field_type const& field_type);
 
    void check_factory_and_class(FieldDefinition fd,
                                 int factory_id,
                                 int class_id) const;
 
    template<typename T>
    boost::optional<T> get_portable_instance(const std::string& field_name);
 
    std::shared_ptr<ClassDefinition> cd_;
    object_data_input* data_input_;
    PortableSerializer* portable_serializer_;
 
private:
    int final_position_;
    int offset_;
    bool raw_;
};
 
class HAZELCAST_API DefaultPortableReader : public PortableReaderBase
{
public:
    DefaultPortableReader(PortableSerializer& portable_ser,
                          object_data_input& input,
                          std::shared_ptr<ClassDefinition> cd);
 
    template<typename T>
    boost::optional<T> read_portable(const std::string& field_name);
 
    template<typename T>
    boost::optional<std::vector<T>> read_portable_array(
      const std::string& field_name);
};
 
class HAZELCAST_API MorphingPortableReader : public PortableReaderBase
{
public:
    MorphingPortableReader(PortableSerializer& portable_ser,
                           object_data_input& input,
                           std::shared_ptr<ClassDefinition> cd);
 
    template<typename T>
    typename std::enable_if<
      std::is_same<int16_t, typename std::remove_cv<T>::type>::value ||
        std::is_same<int32_t, typename std::remove_cv<T>::type>::value ||
        std::is_same<int64_t, typename std::remove_cv<T>::type>::value ||
        std::is_same<float, typename std::remove_cv<T>::type>::value ||
        std::is_same<double, typename std::remove_cv<T>::type>::value,
      T>::type
    read(const std::string& field_name)
    {
        if (!cd_->has_field(field_name)) {
            return 0;
        }
        const field_type& currentFieldType = cd_->get_field_type(field_name);
        return read_morphing<T>(currentFieldType, field_name);
    }
 
    template<typename T>
    typename std::enable_if<
      std::is_same<byte, typename std::remove_cv<T>::type>::value ||
        std::is_same<char, typename std::remove_cv<T>::type>::value ||
        std::is_same<char16_t, typename std::remove_cv<T>::type>::value ||
        std::is_same<bool, typename std::remove_cv<T>::type>::value,
      T>::type
    read(const std::string& field_name)
    {
        if (!cd_->has_field(field_name)) {
            return 0;
        }
        return PortableReaderBase::read<T>(field_name);
    }
 
    template<typename T>
    typename std::enable_if<
      std::is_same<std::string, typename std::remove_cv<T>::type>::value,
      T>::type
    read(const std::string& field_name)
    {
        if (!cd_->has_field(field_name)) {
            return std::string();
        }
        return PortableReaderBase::read<T>(field_name);
    }
 
    template<typename T>
    typename std::enable_if<
      std::is_same<boost::optional<std::string>,
                   typename std::remove_cv<T>::type>::value,
      T>::type
    read(const std::string& field_name)
    {
        if (!cd_->has_field(field_name)) {
            return boost::none;
        }
        return PortableReaderBase::read<T>(field_name);
    }
 
    template<typename T>
    typename std::enable_if<
      std::is_same<std::vector<byte>,
                   typename std::remove_cv<T>::type>::value ||
        std::is_same<std::vector<char>,
                     typename std::remove_cv<T>::type>::value ||
        std::is_same<std::vector<bool>,
                     typename std::remove_cv<T>::type>::value ||
        std::is_same<std::vector<int16_t>,
                     typename std::remove_cv<T>::type>::value ||
        std::is_same<std::vector<int32_t>,
                     typename std::remove_cv<T>::type>::value ||
        std::is_same<std::vector<int64_t>,
                     typename std::remove_cv<T>::type>::value ||
        std::is_same<std::vector<float>,
                     typename std::remove_cv<T>::type>::value ||
        std::is_same<std::vector<double>,
                     typename std::remove_cv<T>::type>::value ||
        std::is_same<std::vector<std::string>,
                     typename std::remove_cv<T>::type>::value,
      boost::optional<T>>::type
    read(const std::string& field_name)
    {
        if (!cd_->has_field(field_name)) {
            return boost::none;
        }
        return PortableReaderBase::read<T>(field_name);
    }
 
    template<typename T>
    boost::optional<T> read_portable(const std::string& field_name);
 
    template<typename T>
    boost::optional<std::vector<T>> read_portable_array(
      const std::string& field_name);
 
private:
    template<typename T>
    typename std::enable_if<
      std::is_same<int16_t, typename std::remove_cv<T>::type>::value,
      T>::type
    read_morphing(field_type current_field_type, const std::string& field_name)
    {
        switch (current_field_type) {
            case field_type::TYPE_BYTE:
                return PortableReaderBase::read<byte>(field_name);
            case field_type::TYPE_SHORT:
                return PortableReaderBase::read<int16_t>(field_name);
            default:
                BOOST_THROW_EXCEPTION(exception::hazelcast_serialization(
                  "MorphingPortableReader::*", "IncompatibleClassChangeError"));
        }
    }
 
    template<typename T>
    typename std::enable_if<
      std::is_same<int32_t, typename std::remove_cv<T>::type>::value,
      T>::type
    read_morphing(field_type current_field_type, const std::string& field_name)
    {
        switch (current_field_type) {
            case field_type::TYPE_INT:
                return PortableReaderBase::read<int32_t>(field_name);
            case field_type::TYPE_CHAR:
                return PortableReaderBase::read<char>(field_name);
            default:
                return read_morphing<int16_t>(current_field_type, field_name);
        }
    }
 
    template<typename T>
    typename std::enable_if<
      std::is_same<int64_t, typename std::remove_cv<T>::type>::value,
      T>::type
    read_morphing(field_type current_field_type, const std::string& field_name)
    {
        switch (current_field_type) {
            case field_type::TYPE_LONG:
                return PortableReaderBase::read<int64_t>(field_name);
            default:
                return read_morphing<int32_t>(current_field_type, field_name);
        }
    }
 
    template<typename T>
    typename std::enable_if<
      std::is_same<float, typename std::remove_cv<T>::type>::value,
      T>::type
    read_morphing(field_type current_field_type, const std::string& field_name)
    {
        switch (current_field_type) {
            case field_type::TYPE_FLOAT:
                return PortableReaderBase::read<float>(field_name);
            default:
                return static_cast<float>(
                  read_morphing<int32_t>(current_field_type, field_name));
        }
    }
 
    template<typename T>
    typename std::enable_if<
      std::is_same<double, typename std::remove_cv<T>::type>::value,
      T>::type
    read_morphing(field_type current_field_type, const std::string& field_name)
    {
        switch (current_field_type) {
            case field_type::TYPE_DOUBLE:
                return PortableReaderBase::read<double>(field_name);
            case field_type::TYPE_FLOAT:
                return PortableReaderBase::read<float>(field_name);
            default:
                return static_cast<double>(
                  read_morphing<int64_t>(current_field_type, field_name));
        }
    }
};
 
class DefaultPortableWriter;
class HAZELCAST_API PortableSerializer
{
    friend DefaultPortableWriter;
 
public:
    PortableSerializer(PortableContext& portable_context);
 
    template<typename T>
    T read_object(object_data_input& in);
 
    template<typename T>
    T read(object_data_input& in, int32_t factory_id, int32_t class_id);
 
    template<typename T>
    void write(const T& object, object_data_output& out);
 
private:
    PortableContext& context_;
 
    template<typename T>
    int find_portable_version(int factory_id, int class_id) const;
 
    portable_reader create_reader(object_data_input& input,
                                  int factory_id,
                                  int class_id,
                                  int version,
                                  int portable_version);
 
    int32_t read_int(object_data_input& in) const;
 
    template<typename T>
    void write_internal(const T& object, object_data_output& out);
 
    template<typename T>
    void write_internal(const T& object,
                        std::shared_ptr<ClassDefinition>& cd,
                        object_data_output& out);
 
    template<typename T>
    std::shared_ptr<ClassDefinition> lookup_or_register_class_definition(
      const T& portable);
};
 
class HAZELCAST_API DataSerializer
{
public:
    template<typename T>
    static boost::optional<T> read_object(object_data_input& in)
    {
        bool identified = in.read<bool>();
        if (!identified) {
            BOOST_THROW_EXCEPTION(exception::hazelcast_serialization(
              "object_data_input::read_object<identified_data_serializer>",
              "Received data is not identified data serialized."));
        }
 
        int32_t expectedFactoryId = hz_serializer<T>::get_factory_id();
        int32_t expectedClassId = hz_serializer<T>::get_class_id();
        int32_t factoryId = in.read<int32_t>();
        int32_t classId = in.read<int32_t>();
        if (expectedFactoryId != factoryId || expectedClassId != classId) {
            BOOST_THROW_EXCEPTION(exception::hazelcast_serialization(
              "object_data_input::read_object<identified_data_serializer>",
              (boost::format("Factory id %1% and class id %2% of data do not "
                             "match expected "
                             "factory id %3% and class id %4%!") %
               factoryId % classId % expectedFactoryId % expectedClassId)
                .str()));
        }
 
        return boost::make_optional(hz_serializer<T>::read_data(in));
    }
 
    template<typename T>
    static void write(const T& object, object_data_output& out);
 
private:
    int32_t read_int(object_data_input& in) const;
};
 
class HAZELCAST_API DefaultPortableWriter
{
public:
    DefaultPortableWriter(PortableSerializer& portable_ser,
                          std::shared_ptr<ClassDefinition> cd,
                          object_data_output& output);
 
    object_data_output& get_raw_data_output();
 
    template<typename T>
    void write(const std::string& field_name, T value)
    {
        typedef typename std::remove_pointer<typename std::remove_reference<
          typename std::remove_cv<T>::type>::type>::type value_type;
        set_position(field_name, PortableContext::get_type<value_type>());
        object_data_output_.write(value);
    }
 
    void end();
 
    template<typename T>
    void write_null_portable(const std::string& field_name);
 
    template<typename T>
    void write_portable(const std::string& field_name, const T* portable);
 
    template<typename T>
    void write_portable_array(const std::string& field_name,
                              const std::vector<T>* values);
 
private:
    FieldDefinition const& set_position(const std::string& field_name,
                                        field_type field_type);
 
    template<typename T>
    void check_portable_attributes(const FieldDefinition& fd);
 
    bool raw_;
    PortableSerializer& portable_serializer_;
    object_data_output& object_data_output_;
    size_t begin_;
    size_t offset_;
    std::unordered_set<std::string> written_fields_;
    std::shared_ptr<ClassDefinition> cd_;
};
} // namespace pimpl
} // namespace serialization
} // namespace client
} // namespace hazelcast
 
#include "hazelcast/client/serialization/pimpl/compact/compact.h"
 
namespace hazelcast {
namespace client {
namespace serialization {
namespace pimpl {
class HAZELCAST_API SerializationService : public util::Disposable
{
public:
    SerializationService(const serialization_config& config,
                         default_schema_service&);
 
    PortableSerializer& get_portable_serializer();
 
    compact_stream_serializer& get_compact_serializer();
 
    DataSerializer& get_data_serializer();
 
    template<typename T>
    inline data to_data(const T* object)
    {
        object_data_output output(
          serialization_config_.get_byte_order(),
          false,
          &portable_serializer_,
          &compact_serializer_,
          serialization_config_.get_global_serializer());
 
        write_hash<T>(object, output);
 
        output.write_object<T>(object);
 
        return { std::move(output).to_byte_array(),
                 std::move(output.schemas_will_be_replicated_) };
    }
 
    template<typename T>
    inline data to_data(const T& object)
    {
        object_data_output output(
          serialization_config_.get_byte_order(),
          false,
          &portable_serializer_,
          &compact_serializer_,
          serialization_config_.get_global_serializer());
 
        write_hash<T>(&object, output);
 
        output.write_object<T>(object);
 
        return { std::move(output).to_byte_array(),
                 std::move(output.schemas_will_be_replicated_) };
    }
 
    template<typename T>
    inline std::shared_ptr<data> to_shared_data(const T* object)
    {
        if (NULL == object) {
            return std::shared_ptr<data>();
        }
        return std::shared_ptr<data>(new data(to_data<T>(object)));
    }
 
    template<typename T>
    inline boost::optional<T> to_object(const data* data)
    {
        if (!data) {
            return boost::none;
        }
        return to_object<T>(*data);
    }
 
    template<typename T>
    typename std::enable_if<
      !(std::is_same<T, const char*>::value ||
        std::is_same<T, const char*>::value ||
        std::is_same<T, typed_data>::value),
      boost::optional<T>>::type inline to_object(const data& data)
    {
        if (is_null_data(data)) {
            return boost::none;
        }
 
        int32_t typeId = data.get_type();
 
        // Constant 8 is Data::DATA_OFFSET. Windows DLL export does not
        // let usage of static member.
        object_data_input objectDataInput(
          serialization_config_.get_byte_order(),
          data.to_byte_array(),
          8,
          portable_serializer_,
          compact_serializer_,
          data_serializer_,
          serialization_config_.get_global_serializer());
        return objectDataInput.read_object<T>(typeId);
    }
 
    template<typename T>
    typename std::enable_if<
      std::is_same<T, typed_data>::value,
      boost::optional<T>>::type inline to_object(const data& d)
    {
        return boost::make_optional(typed_data(data(d), *this));
    }
 
    template<typename T>
    typename std::enable_if<
      std::is_same<T, const char*>::value,
      boost::optional<std::string>>::type inline to_object(const data& data)
    {
        return to_object<std::string>(data);
    }
 
    template<typename T>
    typename std::enable_if<
      std::is_array<T>::value &&
        std::is_same<typename std::remove_all_extents<T>::type, char>::value,
      boost::optional<std::string>>::type inline to_object(const data& data)
    {
        return to_object<std::string>(data);
    }
 
    template<typename T>
    inline std::shared_ptr<data> to_shared_object(
      const std::shared_ptr<data>& data)
    {
        return data;
    }
 
    byte get_version() const;
 
    object_type get_object_type(const data* data);

    void dispose() override;
 
    object_data_output new_output_stream();
 
private:
    SerializationService(const SerializationService&) = delete;
 
    SerializationService& operator=(const SerializationService&) = delete;
 
    const serialization_config& serialization_config_;
    PortableContext portable_context_;
    serialization::pimpl::PortableSerializer portable_serializer_;
    serialization::pimpl::compact_stream_serializer compact_serializer_;
    serialization::pimpl::DataSerializer data_serializer_;
 
    static bool is_null_data(const data& data);
 
    template<typename T>
    void write_hash(const partition_aware_marker* obj, data_output& out)
    {
        typedef typename T::KEY_TYPE PK_TYPE;
        const partition_aware<PK_TYPE>* partitionAwareObj =
          static_cast<const partition_aware<PK_TYPE>*>(obj);
        const PK_TYPE* pk = partitionAwareObj->get_partition_key();
        if (pk != NULL) {
            data partitionKey = to_data<PK_TYPE>(pk);
            out.write<int32_t>(partitionKey.get_partition_hash());
        }
    }
 
    template<typename T>
    void write_hash(const void* /* obj */, data_output& out)
    {
        out.write(0, boost::endian::order::big);
    }
};
 
template<>
data HAZELCAST_API
SerializationService::to_data(const char* object);
} // namespace pimpl

class HAZELCAST_API portable_reader
{
public:
    portable_reader(pimpl::PortableSerializer& portable_ser,
                    object_data_input& data_input,
                    const std::shared_ptr<ClassDefinition>& cd,
                    bool is_default_reader);

    template<typename T>
    typename std::enable_if<
      std::is_same<int16_t, typename std::remove_cv<T>::type>::value ||
        std::is_same<int32_t, typename std::remove_cv<T>::type>::value ||
        std::is_same<int64_t, typename std::remove_cv<T>::type>::value ||
        std::is_same<float, typename std::remove_cv<T>::type>::value ||
        std::is_same<double, typename std::remove_cv<T>::type>::value ||
        std::is_same<byte, typename std::remove_cv<T>::type>::value ||
        std::is_same<char, typename std::remove_cv<T>::type>::value ||
        std::is_same<char16_t, typename std::remove_cv<T>::type>::value ||
        std::is_same<bool, typename std::remove_cv<T>::type>::value ||
        std::is_same<std::string, typename std::remove_cv<T>::type>::value,
      T>::type
    read(const std::string& field_name)
    {
        if (is_default_reader_)
            return default_portable_reader_->read<T>(field_name);
        return morphing_portable_reader_->read<T>(field_name);
    }

    template<typename T>
    typename std::enable_if<
      std::is_same<std::vector<byte>,
                   typename std::remove_cv<T>::type>::value ||
        std::is_same<std::vector<char>,
                     typename std::remove_cv<T>::type>::value ||
        std::is_same<std::vector<bool>,
                     typename std::remove_cv<T>::type>::value ||
        std::is_same<std::vector<int16_t>,
                     typename std::remove_cv<T>::type>::value ||
        std::is_same<std::vector<int32_t>,
                     typename std::remove_cv<T>::type>::value ||
        std::is_same<std::vector<int64_t>,
                     typename std::remove_cv<T>::type>::value ||
        std::is_same<std::vector<float>,
                     typename std::remove_cv<T>::type>::value ||
        std::is_same<std::vector<double>,
                     typename std::remove_cv<T>::type>::value ||
        std::is_same<std::vector<std::string>,
                     typename std::remove_cv<T>::type>::value,
      boost::optional<T>>::type
    read(const std::string& field_name)
    {
        if (is_default_reader_)
            return default_portable_reader_->read<T>(field_name);
        return morphing_portable_reader_->read<T>(field_name);
    }

    template<typename T>
    boost::optional<T> read_portable(const std::string& field_name);

    template<typename T>
    boost::optional<std::vector<T>> read_portable_array(
      const std::string& field_name);

    object_data_input& get_raw_data_input();

    void end();
 
private:
    bool is_default_reader_;
    boost::optional<pimpl::DefaultPortableReader> default_portable_reader_;
    boost::optional<pimpl::MorphingPortableReader> morphing_portable_reader_;
};

class HAZELCAST_API portable_writer
{
public:
    portable_writer(pimpl::DefaultPortableWriter* default_portable_writer);

    portable_writer(pimpl::ClassDefinitionWriter* class_definition_writer);
 
    template<typename T>
    void write(const std::string& field_name, T value)
    {
        if (is_default_writer_) {
            default_portable_writer_->write(field_name, value);
        } else {
            class_definition_writer_->write(field_name, value);
        }
    }

    void end();

    template<typename T>
    void write_null_portable(const std::string& field_name);

    template<typename T>
    void write_portable(const std::string& field_name, const T* portable);

    template<typename T>
    void write_portable_array(const std::string& field_name,
                              const std::vector<T>* values);

    object_data_output& get_raw_data_output();
 
private:
    pimpl::DefaultPortableWriter* default_portable_writer_;
    pimpl::ClassDefinitionWriter* class_definition_writer_;
    bool is_default_writer_;
};
 
template<typename T>
boost::optional<T>
portable_reader::read_portable(const std::string& field_name)
{
    if (is_default_reader_)
        return default_portable_reader_->read_portable<T>(field_name);
    return morphing_portable_reader_->read_portable<T>(field_name);
}

template<typename T>
boost::optional<std::vector<T>>
portable_reader::read_portable_array(const std::string& field_name)
{
    if (is_default_reader_)
        return default_portable_reader_->read_portable_array<T>(field_name);
    return morphing_portable_reader_->read_portable_array<T>(field_name);
};
 
template<typename T>
void
portable_writer::write_null_portable(const std::string& field_name)
{
    if (is_default_writer_)
        return default_portable_writer_->write_null_portable<T>(field_name);
    return class_definition_writer_->write_null_portable<T>(field_name);
}

template<typename T>
void
portable_writer::write_portable(const std::string& field_name,
                                const T* portable)
{
    if (is_default_writer_)
        return default_portable_writer_->write_portable(field_name, portable);
    return class_definition_writer_->write_portable(field_name, portable);
}

template<typename T>
void
portable_writer::write_portable_array(const std::string& field_name,
                                      const std::vector<T>* values)
{
    if (is_default_writer_)
        return default_portable_writer_->write_portable_array(field_name,
                                                              values);
    return class_definition_writer_->write_portable_array(field_name, values);
}
 
template<typename T>
void
object_data_output::write_object(const T* object)
{
    if (is_no_write_) {
        return;
    }
    if (!object) {
        write(static_cast<int32_t>(
                pimpl::serialization_constants::CONSTANT_TYPE_NULL),
              boost::endian::order::big);
        return;
    }
 
    write_object<T>(*object);
}
 
template<typename T>
typename std::enable_if<
  !(std::is_array<T>::value &&
    std::is_same<typename std::remove_all_extents<T>::type, char>::value),
  void>::type
object_data_output::write_object(const boost::optional<T>& object)
{
    if (is_no_write_) {
        return;
    }
    if (!object) {
        write(static_cast<int32_t>(
                pimpl::serialization_constants::CONSTANT_TYPE_NULL),
              boost::endian::order::big);
        return;
    }
 
    write_object<T>(object.value());
}
 
template<typename T>
typename std::enable_if<
  std::is_base_of<identified_data_serializer, hz_serializer<T>>::value,
  void>::type inline object_data_output::write_object(const T& object)
{
    if (is_no_write_) {
        return;
    }
    write(
      static_cast<int32_t>(pimpl::serialization_constants::CONSTANT_TYPE_DATA),
      boost::endian::order::big);
    pimpl::DataSerializer::write<T>(object, *this);
}
 
template<typename T>
typename std::enable_if<
  std::is_base_of<portable_serializer, hz_serializer<T>>::value,
  void>::type inline object_data_output::write_object(const T& object)
{
    if (is_no_write_) {
        return;
    }
    write(static_cast<int32_t>(
            pimpl::serialization_constants::CONSTANT_TYPE_PORTABLE),
          boost::endian::order::big);
    portable_serializer_->write<T>(object, *this);
}
 
template<typename T>
typename std::enable_if<
  std::is_base_of<compact::compact_serializer, hz_serializer<T>>::value,
  void>::type inline object_data_output::write_object(const T& object)
{
    if (is_no_write_) {
        return;
    }
    write(static_cast<int32_t>(
            pimpl::serialization_constants::CONSTANT_TYPE_COMPACT),
          boost::endian::order::big);
    compact_serializer_->write<T>(object, *this);
}
 
template<typename T>
typename std::enable_if<
  std::is_base_of<builtin_serializer, hz_serializer<T>>::value,
  void>::type inline object_data_output::write_object(const T& object)
{
    if (is_no_write_) {
        return;
    }
    write(static_cast<int32_t>((hz_serializer<T>::get_type_id())),
          boost::endian::order::big);
    write<T>(object);
}
 
template<typename T>
typename std::enable_if<
  std::is_array<T>::value &&
    std::is_same<typename std::remove_all_extents<T>::type, char>::value,
  void>::type inline object_data_output::write_object(const T& object)
{
    write_object(std::string(object));
}
 
template<typename T>
typename std::enable_if<
  std::is_base_of<custom_serializer, hz_serializer<T>>::value,
  void>::type inline object_data_output::write_object(const T& object)
{
    if (is_no_write_) {
        return;
    }
    static_assert(hz_serializer<T>::get_type_id() > 0,
                  "Custom serializer type id can not be negative!");
    write(hz_serializer<T>::get_type_id(), boost::endian::order::big);
    hz_serializer<T>::write(object, *this);
}
 
template<typename T>
typename std::enable_if<
  std::is_same<generic_record::generic_record, T>::value>::type
object_data_output::write_object(const T& object)
{
    if (is_no_write_) {
        return;
    }
 
    const auto& record =
      static_cast<const generic_record::generic_record&>(object);
 
    write(static_cast<int32_t>(
            pimpl::serialization_constants::CONSTANT_TYPE_COMPACT),
          boost::endian::order::big);
    compact_serializer_->write_generic_record(record, *this);
}

template<typename T>
typename std::enable_if<
  !(std::is_base_of<builtin_serializer, hz_serializer<T>>::value ||
    std::is_base_of<identified_data_serializer, hz_serializer<T>>::value ||
    std::is_base_of<portable_serializer, hz_serializer<T>>::value ||
    std::is_base_of<compact::compact_serializer, hz_serializer<T>>::value ||
    std::is_base_of<custom_serializer, hz_serializer<T>>::value ||
    std::is_same<generic_record::generic_record, T>::value ||
    (std::is_array<T>::value &&
     std::is_same<typename std::remove_all_extents<T>::type, char>::value)),
  void>::type inline object_data_output::write_object(const T object)
{
    if (!global_serializer_) {
        throw exception::hazelcast_serialization(
          "object_data_output::write_object",
          (boost::format("No serializer found for type(%1%).") %
           typeid(T).name())
            .str());
    }
    if (is_no_write_) {
        return;
    }
    write(static_cast<int32_t>(global_serializer::get_type_id()),
          boost::endian::order::big);
    global_serializer_->write(boost::any(std::move(object)), *this);
}
 
template<typename T>
typename std::enable_if<
  !(std::is_array<T>::value &&
    std::is_same<typename std::remove_all_extents<T>::type, char>::value),
  boost::optional<T>>::type inline object_data_input::read_object()
{
    int32_t typeId = read(boost::endian::order::big);
    if (static_cast<int32_t>(
          pimpl::serialization_constants::CONSTANT_TYPE_NULL) == typeId) {
        return boost::none;
    }
    return read_object<T>(typeId);
}
 
template<typename T>
typename std::enable_if<
  std::is_array<T>::value &&
    std::is_same<typename std::remove_all_extents<T>::type, char>::value,
  boost::optional<std::string>>::type inline object_data_input::read_object()
{
    return read_object<std::string>();
}
 
template<typename T>
typename std::enable_if<
  std::is_base_of<identified_data_serializer, hz_serializer<T>>::value,
  boost::optional<T>>::type inline object_data_input::read_object(int32_t
                                                                    type_id)
{
    if (type_id != static_cast<int32_t>(
                     pimpl::serialization_constants::CONSTANT_TYPE_DATA)) {
        BOOST_THROW_EXCEPTION(exception::hazelcast_serialization(
          "object_data_input::read_object<identified_data_serializer>",
          (boost::format("The associated serializer Serializer<T> is "
                         "identified_data_serializer "
                         "but received data type id is %1%") %
           type_id)
            .str()));
    }
 
    return data_serializer_.read_object<T>(*this);
}
 
template<typename T>
typename std::enable_if<
  std::is_base_of<portable_serializer, hz_serializer<T>>::value,
  boost::optional<T>>::type inline object_data_input::read_object(int32_t
                                                                    type_id)
{
    if (type_id != static_cast<int32_t>(
                     pimpl::serialization_constants::CONSTANT_TYPE_PORTABLE)) {
        BOOST_THROW_EXCEPTION(exception::hazelcast_serialization(
          "object_data_input::read_object<portable_serializer>",
          (boost::format(
             "The associated serializer Serializer<T> is portable_serializer "
             "but received data type id is %1%") %
           type_id)
            .str()));
    }
 
    return portable_serializer_.read_object<T>(*this);
}
 
template<typename T>
typename std::enable_if<
  std::is_base_of<compact::compact_serializer, hz_serializer<T>>::value,
  boost::optional<T>>::type inline object_data_input::read_object(int32_t
                                                                    type_id)
{
    return compact_serializer_.template read<T>(*this);
}
 
template<typename T>
typename std::enable_if<
  std::is_base_of<custom_serializer, hz_serializer<T>>::value,
  boost::optional<T>>::type inline object_data_input::read_object(int32_t
                                                                    type_id)
{
    if (type_id != hz_serializer<T>::get_type_id()) {
        BOOST_THROW_EXCEPTION(exception::hazelcast_serialization(
          "object_data_input::read_object<>",
          (boost::format("The associated serializer Serializer<T> type id %1% "
                         "does not match "
                         "received data type id is %2%") %
           hz_serializer<T>::get_type_id() % type_id)
            .str()));
    }
 
    return boost::optional<T>(hz_serializer<T>::read(*this));
}
 
template<typename T>
typename std::enable_if<
  std::is_base_of<builtin_serializer, hz_serializer<T>>::value,
  boost::optional<T>>::type inline object_data_input::read_object(int32_t
                                                                    type_id)
{
    assert(type_id == static_cast<int32_t>(hz_serializer<T>::get_type_id()));
    (void)type_id;
 
    return boost::optional<T>(read<T>());
}
 
template<typename T>
typename std::enable_if<std::is_same<generic_record::generic_record, T>::value,
                        boost::optional<T>>::type
object_data_input::read_object(int32_t type_id)
{
    return compact_serializer_.read_generic_record(*this);
}
 
template<typename T>
typename std::enable_if<
  !(std::is_base_of<identified_data_serializer, hz_serializer<T>>::value ||
    std::is_base_of<portable_serializer, hz_serializer<T>>::value ||
    std::is_base_of<compact::compact_serializer, hz_serializer<T>>::value ||
    std::is_base_of<builtin_serializer, hz_serializer<T>>::value ||
    std::is_base_of<custom_serializer, hz_serializer<T>>::value ||
    std::is_same<generic_record::generic_record, T>::value),
  boost::optional<T>>::type inline object_data_input::read_object(int32_t
                                                                    type_id)
{
    if (!global_serializer_) {
        throw exception::hazelcast_serialization(
          "object_data_input::read_object",
          (boost::format("No serializer found for type %1%.") %
           typeid(T).name())
            .str());
    }
 
    if (type_id != static_cast<int32_t>(global_serializer_->get_type_id())) {
        BOOST_THROW_EXCEPTION(exception::hazelcast_serialization(
          "object_data_input::read_object<>",
          (boost::format("The global serializer type id %1% does not match "
                         "received data type id is %2%") %
           static_cast<int32_t>(global_serializer_->get_type_id()) % type_id)
            .str()));
    }
 
    return boost::optional<T>(
      boost::any_cast<T>(std::move(global_serializer_->read(*this))));
}
 
namespace pimpl {
template<>
data
SerializationService::to_data(const typed_data* object);
 
template<typename T>
boost::optional<T>
DefaultPortableReader::read_portable(const std::string& field_name)
{
    return get_portable_instance<T>(field_name);
}
 
template<typename T>
boost::optional<std::vector<T>>
DefaultPortableReader::read_portable_array(const std::string& field_name)
{
    PortableReaderBase::set_position(field_name,
                                     field_type::TYPE_PORTABLE_ARRAY);
 
    data_input_->read<int32_t>();
    std::vector<T> portables;
 
    set_position(field_name, field_type::TYPE_PORTABLE_ARRAY);
 
    int32_t len = data_input_->read<int32_t>();
    if (len == util::Bits::NULL_ARRAY) {
        return boost::none;
    }
    int32_t factoryId = data_input_->read<int32_t>();
    int32_t classId = data_input_->read<int32_t>();
 
    check_factory_and_class(cd_->get_field(field_name), factoryId, classId);
 
    if (len > 0) {
        int offset = data_input_->position();
        for (int i = 0; i < len; i++) {
            data_input_->position(offset + i * util::Bits::INT_SIZE_IN_BYTES);
            int32_t start = data_input_->read<int32_t>();
            data_input_->position(start);
 
            portables.push_back(
              portable_serializer_->read<T>(*data_input_, factoryId, classId));
        }
    }
    return portables;
}
 
template<typename T>
boost::optional<T>
MorphingPortableReader::read_portable(const std::string& field_name)
{
    return get_portable_instance<T>(field_name);
}
 
template<typename T>
boost::optional<std::vector<T>>
MorphingPortableReader::read_portable_array(const std::string& field_name)
{
    PortableReaderBase::set_position(field_name,
                                     field_type::TYPE_PORTABLE_ARRAY);
 
    data_input_->read<int32_t>();
    std::vector<T> portables;
 
    set_position(field_name, field_type::TYPE_PORTABLE_ARRAY);
 
    int32_t len = data_input_->read<int32_t>();
    if (len == util::Bits::NULL_ARRAY) {
        return boost::none;
    }
    int32_t factoryId = data_input_->read<int32_t>();
    int32_t classId = data_input_->read<int32_t>();
 
    check_factory_and_class(cd_->get_field(field_name), factoryId, classId);
 
    if (len > 0) {
        portables.reserve(static_cast<size_t>(len));
        int offset = data_input_->position();
        for (int i = 0; i < len; i++) {
            data_input_->position(offset + i * util::Bits::INT_SIZE_IN_BYTES);
            int32_t start = data_input_->read<int32_t>();
            data_input_->position(start);
 
            portables.emplace_back(
              portable_serializer_->read<T>(*data_input_, factoryId, classId));
        }
    }
 
    return boost::make_optional(std::move(portables));
}
 
template<typename T>
T
PortableSerializer::read_object(object_data_input& in)
{
    int32_t factoryId = read_int(in);
    int32_t classId = read_int(in);
 
    if (factoryId != hz_serializer<T>::get_factory_id() ||
        classId != hz_serializer<T>::get_class_id()) {
        BOOST_THROW_EXCEPTION(exception::hazelcast_serialization(
          "PortableSerializer::read_object",
          (boost::format("Received data (factory-class id)=(%1%, %2%) does not "
                         "match expected factory-class id (%3%, %4%)") %
           factoryId % classId % hz_serializer<T>::get_factory_id() %
           hz_serializer<T>::get_class_id())
            .str()));
    }
    return read<T>(in, factoryId, classId);
}
 
template<typename T>
T
PortableSerializer::read(object_data_input& in,
                         int32_t factory_id,
                         int32_t class_id)
{
    int version = in.read<int32_t>();
 
    int portableVersion = find_portable_version<T>(factory_id, class_id);
 
    portable_reader reader =
      create_reader(in, factory_id, class_id, version, portableVersion);
    T result = hz_serializer<T>::read_portable(reader);
    reader.end();
    return result;
}
 
template<typename T>
void
PortableSerializer::write(const T& object, object_data_output& out)
{
    out.write<int32_t>(hz_serializer<T>::get_factory_id());
    out.write<int32_t>(hz_serializer<T>::get_class_id());
 
    write_internal(object, out);
}
 
template<typename T>
void
PortableSerializer::write_internal(const T& object, object_data_output& out)
{
    auto cd = context_.lookup_or_register_class_definition<T>(object);
    write_internal(object, cd, out);
}
 
template<typename T>
void
PortableSerializer::write_internal(const T& object,
                                   std::shared_ptr<ClassDefinition>& cd,
                                   object_data_output& out)
{
    out.write<int32_t>(cd->get_version());
 
    DefaultPortableWriter dpw(*this, cd, out);
    portable_writer portableWriter(&dpw);
    hz_serializer<T>::write_portable(object, portableWriter);
    portableWriter.end();
}
 
template<typename T>
std::shared_ptr<ClassDefinition>
PortableSerializer::lookup_or_register_class_definition(const T& portable)
{
    return context_.lookup_or_register_class_definition<T>(portable);
}
 
template<typename T>
int
PortableSerializer::find_portable_version(int factory_id, int class_id) const
{
    int currentVersion = context_.get_class_version(factory_id, class_id);
    if (currentVersion < 0) {
        currentVersion =
          PortableVersionHelper::get_version<T>(context_.get_version());
        if (currentVersion > 0) {
            context_.set_class_version(factory_id, class_id, currentVersion);
        }
    }
    return currentVersion;
}
 
template<typename T>
void
DataSerializer::write(const T& object, object_data_output& out)
{
    out.write<bool>(true);
    out.write<int32_t>(hz_serializer<T>::get_factory_id());
    out.write<int32_t>(hz_serializer<T>::get_class_id());
    hz_serializer<T>::write_data(object, out);
}
 
template<typename T>
std::shared_ptr<ClassDefinition>
PortableContext::lookup_or_register_class_definition(const T& portable)
{
    int portableVersion = PortableVersionHelper::get_version<T>(
      serialization_config_.get_portable_version());
    std::shared_ptr<ClassDefinition> cd =
      lookup_class_definition(hz_serializer<T>::get_factory_id(),
                              hz_serializer<T>::get_class_id(),
                              portableVersion);
    if (cd.get() == NULL) {
        ClassDefinitionBuilder classDefinitionBuilder(
          hz_serializer<T>::get_factory_id(),
          hz_serializer<T>::get_class_id(),
          portableVersion);
        ClassDefinitionWriter cdw(*this, classDefinitionBuilder);
        portable_writer portableWriter(&cdw);
        hz_serializer<T>::write_portable(portable, portableWriter);
        cd = cdw.register_and_get();
    }
    return cd;
}
 
template<typename T>
boost::optional<T>
PortableReaderBase::get_portable_instance(const std::string& field_name)
{
    set_position(field_name, field_type::TYPE_PORTABLE);
 
    bool isNull = data_input_->read<bool>();
    int32_t factoryId = data_input_->read<int32_t>();
    int32_t classId = data_input_->read<int32_t>();
 
    check_factory_and_class(cd_->get_field(field_name), factoryId, classId);
 
    if (isNull) {
        return boost::none;
    } else {
        return portable_serializer_->read<T>(*data_input_, factoryId, classId);
    }
}
 
template<typename T>
void
DefaultPortableWriter::write_null_portable(const std::string& field_name)
{
    set_position(field_name, field_type::TYPE_PORTABLE);
    object_data_output_.write<bool>(true);
    object_data_output_.write<int32_t>(hz_serializer<T>::get_factory_id());
    object_data_output_.write<int32_t>(hz_serializer<T>::get_class_id());
}
 
template<typename T>
void
DefaultPortableWriter::write_portable(const std::string& field_name,
                                      const T* portable)
{
    FieldDefinition const& fd =
      set_position(field_name, field_type::TYPE_PORTABLE);
    bool isNull = (nullptr == portable);
    object_data_output_.write<bool>(isNull);
 
    object_data_output_.write<int32_t>(hz_serializer<T>::get_factory_id());
    object_data_output_.write<int32_t>(hz_serializer<T>::get_class_id());
 
    if (!isNull) {
        check_portable_attributes<T>(fd);
        portable_serializer_.write_internal(*portable, object_data_output_);
    }
 
    portable_serializer_.write(*portable, object_data_output_);
}
 
template<typename T>
void
DefaultPortableWriter::write_portable_array(const std::string& field_name,
                                            const std::vector<T>* values)
{
    FieldDefinition const& fd =
      set_position(field_name, field_type::TYPE_PORTABLE_ARRAY);
    check_portable_attributes<T>(fd);
 
    int32_t len =
      (values ? static_cast<int32_t>(values->size()) : util::Bits::NULL_ARRAY);
    object_data_output_.write<int32_t>(len);
 
    object_data_output_.write<int32_t>(fd.get_factory_id());
    object_data_output_.write<int32_t>(fd.get_class_id());
 
    if (len > 0) {
        std::shared_ptr<ClassDefinition> classDefinition =
          portable_serializer_.lookup_or_register_class_definition<T>(
            (*values)[0]);
        size_t currentOffset = object_data_output_.position();
        object_data_output_.position(currentOffset +
                                     len * util::Bits::INT_SIZE_IN_BYTES);
        for (int32_t i = 0; i < len; i++) {
            size_t position = object_data_output_.position();
            object_data_output_.write_at(currentOffset +
                                           i * util::Bits::INT_SIZE_IN_BYTES,
                                         static_cast<int32_t>(position));
            portable_serializer_.write_internal(
              (*values)[i], classDefinition, object_data_output_);
        }
    }
}
 
template<typename T>
void
DefaultPortableWriter::check_portable_attributes(const FieldDefinition& fd)
{
    if (fd.get_factory_id() != hz_serializer<T>::get_factory_id()) {
        BOOST_THROW_EXCEPTION(exception::hazelcast_serialization(
          "DefaultPortableWriter::::checkPortableAttributes",
          (boost::format("Wrong Portable type! Expected factory-id: %1%, "
                         "Actual factory-id: %2%") %
           fd.get_factory_id() % hz_serializer<T>::get_factory_id())
            .str()));
    }
    if (fd.get_class_id() != hz_serializer<T>::get_class_id()) {
        BOOST_THROW_EXCEPTION(exception::hazelcast_serialization(
          "DefaultPortableWriter::::checkPortableAttributes",
          (boost::format("Wrong Portable type! Expected class-id: %1%, Actual "
                         "class-id: %2%") %
           fd.get_class_id() % hz_serializer<T>::get_class_id())
            .str()));
    }
}
 
template<typename T>
std::shared_ptr<ClassDefinition>
ClassDefinitionWriter::create_nested_class_def(const T& portable)
{
    int version = PortableVersionHelper::get_version<T>(context_.get_version());
    ClassDefinitionBuilder definitionBuilder(hz_serializer<T>::get_factory_id(),
                                             hz_serializer<T>::get_class_id(),
                                             version);
 
    ClassDefinitionWriter nestedWriter(context_, definitionBuilder);
    portable_writer portableWriter(&nestedWriter);
    hz_serializer<T>::write_portable(portable, portableWriter);
    return context_.register_class_definition(definitionBuilder.build());
}
} // namespace pimpl
} // namespace serialization
 
template<typename T>
boost::optional<T>
typed_data::get() const
{
    return ss_->to_object<T>(data_);
}
} // namespace client
} // namespace hazelcast
 
#include "hazelcast/client/serialization/pimpl/compact/compact_impl.h"
 
#if defined(WIN32) || defined(_WIN32) || defined(WIN64) || defined(_WIN64)
#pragma warning(pop)
#endif