<commit_before>/*
 * Copyright (C) 2015 Cloudius Systems, Ltd.
 */

#pragma once

#include "core/reactor.hh"
#include "core/iostream.hh"
#include "core/distributed.hh"
#include "core/print.hh"
#include "core/sstring.hh"
#include "net/api.hh"
#include "util/serialization.hh"
#include "gms/inet_address.hh"
#include "rpc/rpc.hh"
#include <unordered_map>
#include "db/config.hh"
#include "frozen_mutation.hh"
#include "db/serializer.hh"

namespace net {

/* All verb handler identifiers */
enum class messaging_verb : int32_t {
    MUTATION,
    MUTATION_DONE,
    BINARY, // Deprecated
    READ_REPAIR,
    READ,
    READ_DATA,
    READ_DIGEST,
    REQUEST_RESPONSE, // client-initiated reads and writes
    STREAM_INITIATE, // Deprecated
    STREAM_INITIATE_DONE, // Deprecated
    STREAM_REPLY, // Deprecated
    STREAM_REQUEST, // Deprecated
    RANGE_SLICE,
    BOOTSTRAP_TOKEN, // Deprecated
    TREE_REQUEST, // Deprecated
    TREE_RESPONSE, // Deprecated
    JOIN, // Deprecated
    GOSSIP_DIGEST_SYN,
    GOSSIP_DIGEST_ACK,
    GOSSIP_DIGEST_ACK2,
    DEFINITIONS_ANNOUNCE, // Deprecated
    DEFINITIONS_UPDATE,
    TRUNCATE,
    SCHEMA_CHECK,
    INDEX_SCAN, // Deprecated
    REPLICATION_FINISHED,
    INTERNAL_RESPONSE, // responses to internal calls
    COUNTER_MUTATION,
    STREAMING_REPAIR_REQUEST, // Deprecated
    STREAMING_REPAIR_RESPONSE, // Deprecated
    SNAPSHOT, // Similar to nt snapshot
    MIGRATION_REQUEST,
    GOSSIP_SHUTDOWN,
    _TRACE,
    ECHO,
    REPAIR_MESSAGE,
    PAXOS_PREPARE,
    PAXOS_PROPOSE,
    PAXOS_COMMIT,
    PAGED_RANGE,
    UNUSED_1,
    UNUSED_2,
    UNUSED_3,
    // Used by streaming
    STREAM_INIT_MESSAGE,
    PREPARE_MESSAGE,
    STREAM_MUTATION,
    INCOMING_FILE_MESSAGE,
    OUTGOING_FILE_MESSAGE,
    RECEIVED_MESSAGE,
    RETRY_MESSAGE,
    COMPLETE_MESSAGE,
    SESSION_FAILED_MESSAGE,
    LAST,
};

} // namespace net

namespace std {
template <>
class hash<net::messaging_verb> {
public:
    size_t operator()(const net::messaging_verb& x) const {
        return hash<int32_t>()(int32_t(x));
    }
};
} // namespace std

namespace net {

future<> ser_messaging_verb(output_stream<char>& out, messaging_verb& v);
future<> des_messaging_verb(input_stream<char>& in, messaging_verb& v);
future<> ser_sstring(output_stream<char>& out, sstring& v);
future<> des_sstring(input_stream<char>& in, sstring& v);
future<> ser_frozen_mutation(output_stream<char>& out, const frozen_mutation& v);
future<> des_frozen_mutation(input_stream<char>& in, frozen_mutation& v);

// NOTE: operator(input_stream<char>&, T&) takes a reference to uninitialized
//       T object and should use placement new in case T is non POD
struct serializer {
    // For integer type
    template<typename T>
    inline auto operator()(output_stream<char>& out, T&& v, std::enable_if_t<std::is_integral<std::remove_reference_t<T>>::value, void*> = nullptr) {
        auto v_ = net::hton(v);
        return out.write(reinterpret_cast<const char*>(&v_), sizeof(T));
    }
    template<typename T>
    inline auto operator()(input_stream<char>& in, T& v, std::enable_if_t<std::is_integral<T>::value, void*> = nullptr) {
        return in.read_exactly(sizeof(v)).then([&v] (temporary_buffer<char> buf) mutable {
            if (buf.size() != sizeof(v)) {
                throw rpc::closed_error();
            }
            v = net::ntoh(*reinterpret_cast<const net::packed<T>*>(buf.get()));
        });
    }

    // For vectors
    template<typename T>
    inline auto operator()(output_stream<char>& out, std::vector<T>& v) {
        return operator()(out, v.size()).then([&out, &v, this] {
            return do_for_each(v.begin(), v.end(), [&out, this] (T& e) {
                return operator()(out, e);
            });
        });
    }
    template<typename T>
    inline auto operator()(input_stream<char>& in, std::vector<T>& v) {
        using size_type = typename  std::vector<T>::size_type;
        return in.read_exactly(sizeof(size_type)).then([&v, &in, this] (temporary_buffer<char> buf) {
            if (buf.size() != sizeof(size_type)) {
                throw rpc::closed_error();
            }
            size_type c = net::ntoh(*reinterpret_cast<const net::packed<size_type>*>(buf.get()));
            new (&v) std::vector<T>;
            v.reserve(c);
            union U {
                U(){}
                ~U(){}
                U(U&&) {}
                T v;
            };
            return do_with(U(), [c, &v, &in, this] (U& u) {
                return do_until([c = c] () mutable {return !c--;}, [&v, &in, &u, this] () mutable {
                    return operator()(in, u.v).then([&u, &v] {
                        v.emplace_back(std::move(u.v));
                    });
                });
            });
        });
    }

    // For messaging_verb
    inline auto operator()(output_stream<char>& out, messaging_verb& v) {
        return ser_messaging_verb(out, v);
    }
    inline auto operator()(input_stream<char>& in, messaging_verb& v) {
        return des_messaging_verb(in, v);
    }

    // For sstring
    inline auto operator()(output_stream<char>& out, sstring& v) {<commit_msg>messaging_service: Extract integral reading logic<commit_after>/*
 * Copyright (C) 2015 Cloudius Systems, Ltd.
 */

#pragma once

#include "core/reactor.hh"
#include "core/iostream.hh"
#include "core/distributed.hh"
#include "core/print.hh"
#include "core/sstring.hh"
#include "net/api.hh"
#include "util/serialization.hh"
#include "gms/inet_address.hh"
#include "rpc/rpc.hh"
#include <unordered_map>
#include "db/config.hh"
#include "frozen_mutation.hh"
#include "db/serializer.hh"

namespace net {

/* All verb handler identifiers */
enum class messaging_verb : int32_t {
    MUTATION,
    MUTATION_DONE,
    BINARY, // Deprecated
    READ_REPAIR,
    READ,
    READ_DATA,
    READ_DIGEST,
    REQUEST_RESPONSE, // client-initiated reads and writes
    STREAM_INITIATE, // Deprecated
    STREAM_INITIATE_DONE, // Deprecated
    STREAM_REPLY, // Deprecated
    STREAM_REQUEST, // Deprecated
    RANGE_SLICE,
    BOOTSTRAP_TOKEN, // Deprecated
    TREE_REQUEST, // Deprecated
    TREE_RESPONSE, // Deprecated
    JOIN, // Deprecated
    GOSSIP_DIGEST_SYN,
    GOSSIP_DIGEST_ACK,
    GOSSIP_DIGEST_ACK2,
    DEFINITIONS_ANNOUNCE, // Deprecated
    DEFINITIONS_UPDATE,
    TRUNCATE,
    SCHEMA_CHECK,
    INDEX_SCAN, // Deprecated
    REPLICATION_FINISHED,
    INTERNAL_RESPONSE, // responses to internal calls
    COUNTER_MUTATION,
    STREAMING_REPAIR_REQUEST, // Deprecated
    STREAMING_REPAIR_RESPONSE, // Deprecated
    SNAPSHOT, // Similar to nt snapshot
    MIGRATION_REQUEST,
    GOSSIP_SHUTDOWN,
    _TRACE,
    ECHO,
    REPAIR_MESSAGE,
    PAXOS_PREPARE,
    PAXOS_PROPOSE,
    PAXOS_COMMIT,
    PAGED_RANGE,
    UNUSED_1,
    UNUSED_2,
    UNUSED_3,
    // Used by streaming
    STREAM_INIT_MESSAGE,
    PREPARE_MESSAGE,
    STREAM_MUTATION,
    INCOMING_FILE_MESSAGE,
    OUTGOING_FILE_MESSAGE,
    RECEIVED_MESSAGE,
    RETRY_MESSAGE,
    COMPLETE_MESSAGE,
    SESSION_FAILED_MESSAGE,
    LAST,
};

} // namespace net

namespace std {
template <>
class hash<net::messaging_verb> {
public:
    size_t operator()(const net::messaging_verb& x) const {
        return hash<int32_t>()(int32_t(x));
    }
};
} // namespace std

namespace net {

future<> ser_messaging_verb(output_stream<char>& out, messaging_verb& v);
future<> des_messaging_verb(input_stream<char>& in, messaging_verb& v);
future<> ser_sstring(output_stream<char>& out, sstring& v);
future<> des_sstring(input_stream<char>& in, sstring& v);
future<> ser_frozen_mutation(output_stream<char>& out, const frozen_mutation& v);
future<> des_frozen_mutation(input_stream<char>& in, frozen_mutation& v);

// NOTE: operator(input_stream<char>&, T&) takes a reference to uninitialized
//       T object and should use placement new in case T is non POD
struct serializer {
    template<typename T>
    inline future<T> read_integral(input_stream<char>& in) {
        static_assert(std::is_integral<T>::value, "T should be integral");

        return in.read_exactly(sizeof(T)).then([] (temporary_buffer<char> buf) {
            if (buf.size() != sizeof(T)) {
                throw rpc::closed_error();
            }
            return make_ready_future<T>(net::ntoh(*unaligned_cast<T*>(buf.get())));
        });
    }

    // For integer type
    template<typename T>
    inline auto operator()(output_stream<char>& out, T&& v, std::enable_if_t<std::is_integral<std::remove_reference_t<T>>::value, void*> = nullptr) {
        auto v_ = net::hton(v);
        return out.write(reinterpret_cast<const char*>(&v_), sizeof(T));
    }
    template<typename T>
    inline auto operator()(input_stream<char>& in, T& v, std::enable_if_t<std::is_integral<T>::value, void*> = nullptr) {
        return in.read_exactly(sizeof(v)).then([&v] (temporary_buffer<char> buf) mutable {
            if (buf.size() != sizeof(v)) {
                throw rpc::closed_error();
            }
            v = net::ntoh(*reinterpret_cast<const net::packed<T>*>(buf.get()));
        });
    }

    // For vectors
    template<typename T>
    inline auto operator()(output_stream<char>& out, std::vector<T>& v) {
        return operator()(out, v.size()).then([&out, &v, this] {
            return do_for_each(v.begin(), v.end(), [&out, this] (T& e) {
                return operator()(out, e);
            });
        });
    }
    template<typename T>
    inline auto operator()(input_stream<char>& in, std::vector<T>& v) {
        using size_type = typename  std::vector<T>::size_type;
        return read_integral<size_type>(in).then([&v, &in, this] (size_type c) {
            new (&v) std::vector<T>;
            v.reserve(c);
            union U {
                U(){}
                ~U(){}
                U(U&&) {}
                T v;
            };
            return do_with(U(), [c, &v, &in, this] (U& u) {
                return do_until([c = c] () mutable {return !c--;}, [&v, &in, &u, this] () mutable {
                    return operator()(in, u.v).then([&u, &v] {
                        v.emplace_back(std::move(u.v));
                    });
                });
            });
        });
    }

    // For messaging_verb
    inline auto operator()(output_stream<char>& out, messaging_verb& v) {
        return ser_messaging_verb(out, v);
    }
    inline auto operator()(input_stream<char>& in, messaging_verb& v) {
        return des_messaging_verb(in, v);
    }

    // For sstring
    inline auto operator()(output_stream<char>& out, sstring& v) {
{'fraction_non_alphanumeric': 0.13712501182927983, 'fraction_numerical': 0.0026497586826913977, 'mean_word_length': 2.8401162790697674, 'pattern_counts': {'":': 0, '<': 96, '<?xml version=': 0, '>': 96, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '95281', 'n_tokens_mistral': 3716, 'n_tokens_neox': 3576, 'n_words': 735}