<commit_before>/****************************************************************
* Copyright (C) 2016 <NAME>, <NAME>
*
* The content of this file may not be copied and/or distributed
* without the expressed permission of the copyright owner.
*
****************************************************************/

#include "mocca/net/message/LoopbackConnection.h"

#include "mocca/net/NetworkError.h"

using namespace mocca;
using namespace mocca::net;

LoopbackConnection::LoopbackConnection(std::shared_ptr<MQ> sendQueue, std::shared_ptr<MQ> receiveQueue, std::shared_ptr<SQ> outSignalQueue,
                                       std::shared_ptr<SQ> inSignalQueue, std::shared_ptr<const ConnectionID> connectionID)
    : connectionID_(connectionID)
    , connected_(true)
    , sendQueue_(sendQueue)
    , receiveQueue_(receiveQueue)
    , outSignalQueue_(outSignalQueue)
    , inSignalQueue_(inSignalQueue) {}

LoopbackConnection::~LoopbackConnection() {
    outSignalQueue_->enqueue(Signal::Disconnect);
}

std::shared_ptr<const ConnectionID> LoopbackConnection::connectionID() const {
    return connectionID_;
}

bool LoopbackConnection::isConnected() const {
    auto signal = inSignalQueue_->dequeueNoWait();
    if (!signal.isNull() && signal == Signal::Disconnect) {
        connected_ = false;
    }
    return connected_;
}

void LoopbackConnection::send(Message message) const {
    if (!isConnected()) {
        throw ConnectionClosedError("Connection to peer " + connectionID_->peerEndpoint.toString() + " has been closed", *connectionID_,
                                    __FILE__, __LINE__);
    }
    sendQueue_->enqueue(std::move(message));
}

Message LoopbackConnection::receive() const {
    if (!isConnected()) {
        throw ConnectionClosedError("Connection to peer " + connectionID_->peerEndpoint.toString() + " has been closed", *connectionID_,
                                    __FILE__, __LINE__);
    }
    auto data = receiveQueue_->dequeue(std::chrono::milliseconds(100));
    return data.isNull() ? Message() : data.release();
}<commit_msg>issue: TRI-37 fixed missing reply in combined node under heavy load<commit_after>/****************************************************************
* Copyright (C) 2016 <NAME>, <NAME>
*
* The content of this file may not be copied and/or distributed
* without the expressed permission of the copyright owner.
*
****************************************************************/

#include "mocca/net/message/LoopbackConnection.h"

#include "mocca/net/NetworkError.h"

using namespace mocca;
using namespace mocca::net;

LoopbackConnection::LoopbackConnection(std::shared_ptr<MQ> sendQueue, std::shared_ptr<MQ> receiveQueue, std::shared_ptr<SQ> outSignalQueue,
                                       std::shared_ptr<SQ> inSignalQueue, std::shared_ptr<const ConnectionID> connectionID)
    : connectionID_(connectionID)
    , connected_(true)
    , sendQueue_(sendQueue)
    , receiveQueue_(receiveQueue)
    , outSignalQueue_(outSignalQueue)
    , inSignalQueue_(inSignalQueue) {}

LoopbackConnection::~LoopbackConnection() {
    outSignalQueue_->enqueue(Signal::Disconnect);
}

std::shared_ptr<const ConnectionID> LoopbackConnection::connectionID() const {
    return connectionID_;
}

bool LoopbackConnection::isConnected() const {
    auto signal = inSignalQueue_->dequeueNoWait();
    if (!signal.isNull() && signal == Signal::Disconnect) {
        connected_ = false;
    }
    return connected_;
}

void LoopbackConnection::send(Message message) const {
    if (!isConnected()) {
        throw ConnectionClosedError("Connection to peer " + connectionID_->peerEndpoint.toString() + " has been closed", *connectionID_,
                                    __FILE__, __LINE__);
    }
    sendQueue_->enqueue(std::move(message));
}

Message LoopbackConnection::receive() const {
    if (!isConnected()) {
        throw ConnectionClosedError("Connection to peer " + connectionID_->peerEndpoint.toString() + " has been closed", *connectionID_,
                                    __FILE__, __LINE__);
    }
    auto data = receiveQueue_->dequeue(std::chrono::duration_cast<std::chrono::milliseconds>(std::chrono::seconds(30)));
    return data.isNull() ? Message() : data.release();
}
{'fraction_non_alphanumeric': 0.172502937720329, 'fraction_numerical': 0.0035252643948296123, 'mean_word_length': 4.241379310344827, 'pattern_counts': {'":': 0, '<': 20, '<?xml version=': 0, '>': 32, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 6, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1877420', 'n_tokens_mistral': 1159, 'n_tokens_neox': 1106, 'n_words': 240}