<commit_before>//  Copyright 2010 <NAME>
//
//  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.

#include <zippylog/request_processor.hpp>

#include <zippylog/message.pb.h>
#include <zippylog/protocol.pb.h>
#include <zippylog/protocol/request.pb.h>
#include <zippylog/protocol/response.pb.h>
#include <zippylog/request_processor.pb.h>
#include <zippylog/zeromq.hpp>

namespace zippylog {

#define LOG_MESSAGE(msgvar, socketvar) { \
    msgvar.set_id(this->id); \
    Envelope logenvelope = Envelope(); \
    msgvar.add_to_envelope(&logenvelope); \
    zeromq::send_envelope(socketvar, logenvelope); \
}

/// This handy macro obtains the index'th message of an envelope and stores
/// stores it in a variable, varname. If we can't obtain the message, we record
/// an error in the output vector<Envelope> variable and goto LOG_END.
#define OBTAIN_MESSAGE(type, varname, envelope, index) \
    type * varname = (type *)envelope.GetMessage(index); \
    if (!varname) { \
        ::zippylog::request_processor::EnvelopeEmbeddedMessageParseFailure log = ::zippylog::request_processor::EnvelopeEmbeddedMessageParseFailure(); \
        LOG_MESSAGE(log, this->logger_sock); \
        \
        this->PopulateErrorResponse( \
            protocol::response::UNKNOWN_REQUEST_TYPE, \
            "error parsing message in envelope", \
            output \
        ); \
        goto LOG_END; \
    }

using ::std::invalid_argument;
using ::std::string;
using ::std::vector;
using ::zmq::message_t;
using ::zmq::socket_t;

RequestProcessor::RequestProcessor(RequestProcessorStartParams &params) :
    ctx(params.ctx),
    store_path(params.store_path),
    logger_endpoint(params.logger_endpoint),
    client_endpoint(params.client_endpoint),
    logger_sock(NULL),
    socket(NULL),
    store(NULL),
    active(params.active),
    get_stream_max_bytes(params.get_stream_max_bytes),
    get_stream_max_envelopes(params.get_stream_max_envelopes)
{
    if (!this->active) {
        throw invalid_argument("active parameter cannot be NULL");
    }

    if (!this->ctx) {
        throw invalid_argument("ctx parameter cannot be NULL");
    }

    this->store = Store::CreateStore(params.store_path);

    platform::UUID uuid;
    platform::CreateUUID(uuid);
    this->id = string((const char *)&uuid, sizeof(uuid));

    this->logger_sock = new socket_t(*this->ctx, ZMQ_PUSH);
    this->logger_sock->connect(this->logger_endpoint.c_str());

    this->socket = new socket_t(*this->ctx, ZMQ_XREP);
    this->socket->connect(this->client_endpoint.c_str());

    ::zippylog::request_processor::Create log = ::zippylog::request_processor::Create();
    LOG_MESSAGE(log, this->logger_sock);
}

RequestProcessor::~RequestProcessor()
{
    ::zippylog::request_processor::Destroy log = ::zippylog::request_processor::Destroy();
    LOG_MESSAGE(log, this->logger_sock);

    if (this->logger_sock) delete this->logger_sock;
    if (this->socket) delete this->socket;
    if (this->store) delete this->store;
}

void RequestProcessor::Run()
{
    {
        ::zippylog::request_processor::RunStart log;
        LOG_MESSAGE(log, this->logger_sock);
    }

    while (*this->active) {
        this->Pump(250000);
    }

    ::zippylog::request_processor::RunStop log;
    LOG_MESSAGE(log, this->logger_sock);

    return;
}

int RequestProcessor::Pump(long wait)
{
    zmq::pollitem_t pollitems[1];
    pollitems[0].events = ZMQ_POLLIN;
    pollitems[0].fd = 0;
    pollitems[0].revents = 0;
    pollitems[0].socket = *this->socket;

    if (!zmq::poll(&pollitems[0], 1, wait)) {
        return 0;
    }

    // we have a message available, so we receive it
    vector<message_t *> msgs;
    this->current_request_identities.clear();
    if (!zeromq::receive_multipart_message(this->socket, this->current_request_identities, msgs)) {
        ::zippylog::request_processor::FailReceiveMessage log;
        LOG_MESSAGE(log, this->logger_sock);

        // TODO shouldn't this be part of the receive_multipart_message API?
        vector<message_t *>::iterator iter = msgs.begin();
        for (; iter != msgs.end(); iter++) {
            delete *iter;
        }

        // TODO we used to rebuild socket here. is this a good recovery strategy?
        // this exception is here so we can see if this actually happens
        // I suspect it doesn't
        throw Exception("error receiving multipart message from worker sock");
    }

    vector<Envelope> response_envelopes;
    this->ProcessMessages(this->current_request_identities, msgs, response_envelopes);

    // input messages aren't needed any more, so we destroy them immediately
    vector<message_t *>::iterator iter = msgs.begin();
    for (; iter != msgs.end(); iter++) {
        delete *iter;
    }

    if (response_envelopes.size()) {
        if (!zeromq::send_envelopes(this->socket, this->current_request_identities, response_envelopes)) {
            throw Exception("TODO log failure to send response envelopes");
        }
    }

    return 1;
}

void RequestProcessor::ProcessMessages(vector<string> &, vector<message_t *> &input, vector<Envelope> &output)
{
    // we need to declare all our variables first b/c we use goto

    // always clear the output list so we don't accidentally send leftovers
    output.clear();

    char msg_version = 0;
    bool have_request_envelope = false;
    Envelope request_envelope;
    ResponseStatus result;
    bool successful_process = false;

    // TODO should we log?
    if (!input.size()) return;

    // we expect the first message to have an envelope of some kind
    if (!input[0]->size()) {
        ::zippylog::request_processor::ReceiveEmptyMessage log;
        LOG_MESSAGE(log, this->logger_sock);

        this->PopulateErrorResponse(<commit_msg>use new SendEnvelope() API and handle error conditions<commit_after>//  Copyright 2010 <NAME>
//
//  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.

#include <zippylog/request_processor.hpp>

#include <zippylog/message.pb.h>
#include <zippylog/protocol.pb.h>
#include <zippylog/protocol/request.pb.h>
#include <zippylog/protocol/response.pb.h>
#include <zippylog/request_processor.pb.h>
#include <zippylog/zeromq.hpp>

namespace zippylog {

#define LOG_MESSAGE(msgvar, socketvar) { \
    msgvar.set_id(this->id); \
    Envelope logenvelope = Envelope(); \
    msgvar.add_to_envelope(&logenvelope); \
    zeromq::send_envelope(socketvar, logenvelope); \
}

/// This handy macro obtains the index'th message of an envelope and stores
/// stores it in a variable, varname. If we can't obtain the message, we record
/// an error in the output vector<Envelope> variable and goto LOG_END.
#define OBTAIN_MESSAGE(type, varname, envelope, index) \
    type * varname = (type *)envelope.GetMessage(index); \
    if (!varname) { \
        ::zippylog::request_processor::EnvelopeEmbeddedMessageParseFailure log = ::zippylog::request_processor::EnvelopeEmbeddedMessageParseFailure(); \
        LOG_MESSAGE(log, this->logger_sock); \
        \
        this->PopulateErrorResponse( \
            protocol::response::UNKNOWN_REQUEST_TYPE, \
            "error parsing message in envelope", \
            output \
        ); \
        goto LOG_END; \
    }

using ::std::invalid_argument;
using ::std::string;
using ::std::vector;
using ::zmq::message_t;
using ::zmq::socket_t;

RequestProcessor::RequestProcessor(RequestProcessorStartParams &params) :
    ctx(params.ctx),
    store_path(params.store_path),
    logger_endpoint(params.logger_endpoint),
    client_endpoint(params.client_endpoint),
    logger_sock(NULL),
    socket(NULL),
    store(NULL),
    active(params.active),
    get_stream_max_bytes(params.get_stream_max_bytes),
    get_stream_max_envelopes(params.get_stream_max_envelopes)
{
    if (!this->active) {
        throw invalid_argument("active parameter cannot be NULL");
    }

    if (!this->ctx) {
        throw invalid_argument("ctx parameter cannot be NULL");
    }

    this->store = Store::CreateStore(params.store_path);

    platform::UUID uuid;
    platform::CreateUUID(uuid);
    this->id = string((const char *)&uuid, sizeof(uuid));

    this->logger_sock = new socket_t(*this->ctx, ZMQ_PUSH);
    this->logger_sock->connect(this->logger_endpoint.c_str());

    this->socket = new socket_t(*this->ctx, ZMQ_XREP);
    this->socket->connect(this->client_endpoint.c_str());

    ::zippylog::request_processor::Create log = ::zippylog::request_processor::Create();
    LOG_MESSAGE(log, this->logger_sock);
}

RequestProcessor::~RequestProcessor()
{
    ::zippylog::request_processor::Destroy log = ::zippylog::request_processor::Destroy();
    LOG_MESSAGE(log, this->logger_sock);

    if (this->logger_sock) delete this->logger_sock;
    if (this->socket) delete this->socket;
    if (this->store) delete this->store;
}

void RequestProcessor::Run()
{
    {
        ::zippylog::request_processor::RunStart log;
        LOG_MESSAGE(log, this->logger_sock);
    }

    while (*this->active) {
        this->Pump(250000);
    }

    ::zippylog::request_processor::RunStop log;
    LOG_MESSAGE(log, this->logger_sock);

    return;
}

int RequestProcessor::Pump(long wait)
{
    zmq::pollitem_t pollitems[1];
    pollitems[0].events = ZMQ_POLLIN;
    pollitems[0].fd = 0;
    pollitems[0].revents = 0;
    pollitems[0].socket = *this->socket;

    if (!zmq::poll(&pollitems[0], 1, wait)) {
        return 0;
    }

    // we have a message available, so we receive it
    vector<message_t *> msgs;
    this->current_request_identities.clear();
    if (!zeromq::receive_multipart_message(this->socket, this->current_request_identities, msgs)) {
        ::zippylog::request_processor::FailReceiveMessage log;
        LOG_MESSAGE(log, this->logger_sock);

        // TODO shouldn't this be part of the receive_multipart_message API?
        vector<message_t *>::iterator iter = msgs.begin();
        for (; iter != msgs.end(); iter++) {
            delete *iter;
        }

        // TODO we used to rebuild socket here. is this a good recovery strategy?
        // this exception is here so we can see if this actually happens
        // I suspect it doesn't
        throw Exception("error receiving multipart message from worker sock");
    }

    vector<Envelope> response_envelopes;
    this->ProcessMessages(this->current_request_identities, msgs, response_envelopes);

    // input messages aren't needed any more, so we destroy them immediately
    vector<message_t *>::iterator iter = msgs.begin();
    for (; iter != msgs.end(); iter++) {
        delete *iter;
    }

    if (response_envelopes.size()) {
        int flags = response_envelopes.size() > 1 ? ZMQ_SNDMORE : 0;

        int result = zeromq::SendEnvelope(*this->socket, this->current_request_identities, response_envelopes[0], true, 0);
        if (result == -1) {
            throw Exception("Serialization error in response envelope");
        }
        else if (!result) {
            throw Exception("Send failure for response");
        }

        if (flags) {
            vector<Envelope>::iterator itor = response_envelopes.begin()++;
            while(itor != response_envelopes.end()) {
                result = zeromq::SendEnvelope(*this->socket, *itor, true, ++itor == response_envelopes.end() ? 0 : ZMQ_SNDMORE);
                if (result == -1) {
                    throw Exception("Serialization error in response envelope");
                }
                else if (!result) {
                    throw Exception("Send failure for response envelope");
                }
            }
        }
    }

    return 1;
}

void RequestProcessor::ProcessMessages(vector<string> &, vector<message_t *> &input, vector<Envelope> &output)
{
    // we need to declare all our variables first b/c we use goto

    // always clear the output list so we don't accidentally send leftovers
    output.clear();

    char msg_version = 0;
    bool have_request_envelope = false;
    Envelope request_envelope;
    ResponseStatus result;
    bool successful_process = false;

    // TODO should we log?
    if (!input.size()) return;

    // we expect the first message to have an envelope of some kind
    if (!input[0]->size()) {
        ::zippylog::request_processor::ReceiveEmptyMessage log;
        LOG_MESSAGE(log, this->logger_sock);

        this->PopulateErrorResponse(
{'fraction_non_alphanumeric': 0.12374305868227525, 'fraction_numerical': 0.004577517634699085, 'mean_word_length': 3.328353361481, 'pattern_counts': {'":': 0, '<': 36, '<?xml version=': 0, '>': 120, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, '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': '6325806', 'n_tokens_mistral': 4045, 'n_tokens_neox': 3962, 'n_words': 1003}