<commit_before>#include "ClientConnection.hpp"

#include "ConnectionManager.hpp"

#include <util/Epoch.hpp>
#include <util/ErrorCode.hpp>
#include <util/Logging.hpp>
#include <util/MessageTypes.hpp>
#include <util/MessageWriter.hpp>

namespace tell {
namespace store {

namespace {

/**
 * @brief Reads the snapshot descriptor from the message
 */
SnapshotDescriptor readSnapshot(uint64_t version, BufferReader& request) {
    request.align(sizeof(uint32_t));
    auto descriptorLength = request.read<uint32_t>();
    auto descriptorData = request.read(descriptorLength);

    std::unique_ptr<unsigned char[]> dataBuffer(new unsigned char[descriptorLength]);
    memcpy(dataBuffer.get(), descriptorData, descriptorLength);
    return SnapshotDescriptor(dataBuffer.release(), descriptorLength, version);
}

} // anonymous namespace

ClientConnection::~ClientConnection() {
    boost::system::error_code ec;
    mSocket.close(ec);
    if (ec) {
        // TODO Handle this situation somehow (this should probably not happen at this point)
    }
}

void ClientConnection::init() {
    mSocket.setHandler(this);
}

void ClientConnection::shutdown() {
    boost::system::error_code ec;
    mSocket.disconnect(ec);
    if (ec) {
        LOG_ERROR("Error disconnecting [error = %1% %2%]", ec, ec.message());
        // TODO Handle this situation somehow - Can this even happen?
    }
}

void ClientConnection::onConnected(const boost::system::error_code& ec) {
    if (ec) {
        LOG_ERROR("Failure while establishing client connection [error = %1% %2%]", ec, ec.message());
        mManager.removeConnection(this);
    }
}

void ClientConnection::onReceive(const void* buffer, size_t length, const boost::system::error_code& ec) {
    if (ec) {
        LOG_ERROR("Error receiving message [error = %1% %2%]", ec, ec.message());
        // TODO Handle this situation somehow
        return;
    }

    auto startTime = std::chrono::high_resolution_clock::now();

    BufferReader request(reinterpret_cast<const char*>(buffer), length);
    MessageWriter writer(mSocket);
    while (!request.exhausted()) {
        auto transactionId = request.read<uint64_t>();
        auto requestType = request.read<uint64_t>();
        if (requestType > static_cast<uint64_t>(RequestType::LAST)) {
            requestType = static_cast<uint64_t>(RequestType::UNKOWN);
        }

        LOG_TRACE("TID %1%] Handling request of type %2%", transactionId, requestType);

        switch (static_cast<RequestType>(requestType)) {

        /**
         * The create table request has the following format:
         * - 2 bytes: Length of the table name string
         * - x bytes: The table name string
         * - y bytes: Variable padding to make message 8 byte aligned
         * - 4 bytes: Length of the schema field
         * - x bytes: The table schema
         *
         * The response consists of the following format:
         * - 8 bytes: The table ID of the newly created table or 0 when the table already exists
         */
        case RequestType::CREATE_TABLE: {
            auto tableNameSize = request.read<uint16_t>();
            auto tableNameData = request.read(tableNameSize);
            request.align(sizeof(uint64_t));
            crossbow::string tableName(tableNameData, tableNameSize);

            // TODO Refactor schema serialization
            auto schemaData = request.data();
            auto schemaSize = request.read<uint32_t>();
            request.advance(schemaSize - sizeof(uint32_t));
            Schema schema(schemaData);

            uint64_t tableId = 0;
            auto succeeded = mStorage.createTable(tableName, schema, tableId);
            LOG_ASSERT((tableId == 0) ^ succeeded, "Table ID of 0 does not denote failure");

            size_t messageSize = sizeof(uint64_t);
            boost::system::error_code ec;
            auto response = writer.writeResponse(transactionId, ResponseType::CREATE_TABLE, messageSize, ec);
            if (ec) {
                LOG_ERROR("Error while handling create table request [error = %1% %2%]", ec, ec.message());
                break;
            }
            response.write<uint64_t>(tableId);
        } break;
<commit_msg>Do not expect a table ID of 0 when table already exists<commit_after>#include "ClientConnection.hpp"

#include "ConnectionManager.hpp"

#include <util/Epoch.hpp>
#include <util/ErrorCode.hpp>
#include <util/Logging.hpp>
#include <util/MessageTypes.hpp>
#include <util/MessageWriter.hpp>

namespace tell {
namespace store {

namespace {

/**
 * @brief Reads the snapshot descriptor from the message
 */
SnapshotDescriptor readSnapshot(uint64_t version, BufferReader& request) {
    request.align(sizeof(uint32_t));
    auto descriptorLength = request.read<uint32_t>();
    auto descriptorData = request.read(descriptorLength);

    std::unique_ptr<unsigned char[]> dataBuffer(new unsigned char[descriptorLength]);
    memcpy(dataBuffer.get(), descriptorData, descriptorLength);
    return SnapshotDescriptor(dataBuffer.release(), descriptorLength, version);
}

} // anonymous namespace

ClientConnection::~ClientConnection() {
    boost::system::error_code ec;
    mSocket.close(ec);
    if (ec) {
        // TODO Handle this situation somehow (this should probably not happen at this point)
    }
}

void ClientConnection::init() {
    mSocket.setHandler(this);
}

void ClientConnection::shutdown() {
    boost::system::error_code ec;
    mSocket.disconnect(ec);
    if (ec) {
        LOG_ERROR("Error disconnecting [error = %1% %2%]", ec, ec.message());
        // TODO Handle this situation somehow - Can this even happen?
    }
}

void ClientConnection::onConnected(const boost::system::error_code& ec) {
    if (ec) {
        LOG_ERROR("Failure while establishing client connection [error = %1% %2%]", ec, ec.message());
        mManager.removeConnection(this);
    }
}

void ClientConnection::onReceive(const void* buffer, size_t length, const boost::system::error_code& ec) {
    if (ec) {
        LOG_ERROR("Error receiving message [error = %1% %2%]", ec, ec.message());
        // TODO Handle this situation somehow
        return;
    }

    auto startTime = std::chrono::high_resolution_clock::now();

    BufferReader request(reinterpret_cast<const char*>(buffer), length);
    MessageWriter writer(mSocket);
    while (!request.exhausted()) {
        auto transactionId = request.read<uint64_t>();
        auto requestType = request.read<uint64_t>();
        if (requestType > static_cast<uint64_t>(RequestType::LAST)) {
            requestType = static_cast<uint64_t>(RequestType::UNKOWN);
        }

        LOG_TRACE("TID %1%] Handling request of type %2%", transactionId, requestType);

        switch (static_cast<RequestType>(requestType)) {

        /**
         * The create table request has the following format:
         * - 2 bytes: Length of the table name string
         * - x bytes: The table name string
         * - y bytes: Variable padding to make message 8 byte aligned
         * - 4 bytes: Length of the schema field
         * - x bytes: The table schema
         *
         * The response consists of the following format:
         * - 8 bytes: The table ID of the newly created table or 0 when the table already exists
         */
        case RequestType::CREATE_TABLE: {
            auto tableNameSize = request.read<uint16_t>();
            auto tableNameData = request.read(tableNameSize);
            request.align(sizeof(uint64_t));
            crossbow::string tableName(tableNameData, tableNameSize);

            // TODO Refactor schema serialization
            auto schemaData = request.data();
            auto schemaSize = request.read<uint32_t>();
            request.advance(schemaSize - sizeof(uint32_t));
            Schema schema(schemaData);

            uint64_t tableId = 0;
            auto succeeded = mStorage.createTable(tableName, schema, tableId);
            LOG_ASSERT((tableId != 0) || !succeeded, "Table ID of 0 does not denote failure");

            size_t messageSize = sizeof(uint64_t);
            boost::system::error_code ec;
            auto response = writer.writeResponse(transactionId, ResponseType::CREATE_TABLE, messageSize, ec);
            if (ec) {
                LOG_ERROR("Error while handling create table request [error = %1% %2%]", ec, ec.message());
                break;
            }
            response.write<uint64_t>(succeeded ? tableId : 0x0u);
        } break;

{'fraction_non_alphanumeric': 0.11322538707008628, 'fraction_numerical': 0.011227987235551353, 'mean_word_length': 2.9952785646836637, 'pattern_counts': {'":': 0, '<': 35, '<?xml version=': 0, '>': 37, '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': '6116718', 'n_tokens_mistral': 2364, 'n_tokens_neox': 2287, 'n_words': 639}