<commit_before>/**
 * @file communicator.cc
 * @author <NAME>
 * @copyright (C) 2014 ACK CYFRONET AGH
 * @copyright This software is released under the MIT license cited in 'LICENSE.txt'
 */

#include "communication/communicator.h"

#include "communication/connection.h"
#include "communication/exception.h"
#include "communication/websocketConnectionPool.h"
#include "fuse_messages.pb.h"
#include "logging.h"
#include "oneErrors.h"
#include "remote_file_management.pb.h"
#include "scheduler.h"

#include <boost/algorithm/string.hpp>
#include <google/protobuf/descriptor.h>

#include <cassert>
#include <unordered_set>

namespace one
{
namespace communication
{

Communicator::Communicator(std::unique_ptr<CommunicationHandler> communicationHandler)
    : m_communicationHandler{std::move(communicationHandler)}
{
    setupHandshakeAck();
}

void Communicator::setupPushChannels(std::function<void(const Answer&)> callback)
{
    LOG(INFO) << "Setting up push channels with fuseId: '" << m_fuseId << "'";

    // First close the push channel that was already opened on the connection
    if(m_deregisterPushChannel)
        m_deregisterPushChannel();

    // Subscribe for push messages (negative message id)
    auto pred = [](const Answer &ans){ return ans.message_id() < 0; };
    auto unsubscribe = m_communicationHandler->subscribe({std::move(pred),
                                                          std::move(callback)});

    const auto fuseId = m_fuseId;
    const auto fslogic = toString(ServerModule::FSLOGIC);

    auto handshake = [=]{
        LOG(INFO) << "Opening a push channel with fuseId: '" << fuseId << "'";
        clproto::fuse_messages::ChannelRegistration reg;
        reg.set_fuse_id(fuseId);
        return createMessage(fslogic, true, Atom::default_instance(), reg);
    };

    auto goodbye = [=]{
        LOG(INFO) << "Closing the push channel with fuseId: '" << fuseId << "'";
        clproto::fuse_messages::ChannelClose close;
        close.set_fuse_id(fuseId);
        return createMessage(fslogic, true, Atom::default_instance(), close);
    };

    auto remove = m_communicationHandler->addHandshake(std::move(handshake),
                                                       std::move(goodbye),
                                                       CommunicationHandler::Pool::META);

    m_deregisterPushChannel = [=]{ unsubscribe(); remove(); };
}

void Communicator::setupHandshakeAck()
{
    LOG(INFO) << "Setting up HandshakeAck with fuseId: '" << m_fuseId << "'";

    // First remove the previous Ack message
    if(m_removeHandshakeAck)
        m_removeHandshakeAck();

    const auto fuseId = m_fuseId;

    // Build HandshakeAck message
    auto handshake = [this, fuseId]{
        LOG(INFO) << "Sending HandshakeAck with fuseId: '" << fuseId << "'";
        clproto::fuse_messages::HandshakeAck ack;
        ack.set_fuse_id(fuseId);
        return createMessage("", true, Atom::default_instance(), ack);
    };

    auto removeMeta = m_communicationHandler->addHandshake(handshake, CommunicationHandler::Pool::META);
    auto removeData = m_communicationHandler->addHandshake(handshake, CommunicationHandler::Pool::DATA);
    m_removeHandshakeAck = [=]{ removeMeta(); removeData(); };
}

void Communicator::setFuseId(std::string fuseId)
{
    LOG(INFO) << "Setting fuseId for communication: '" << fuseId << '"';
    m_fuseId = std::move(fuseId);
    setupHandshakeAck();
}

void Communicator::reply(const Answer &originalMsg, const ServerModule module,
                         const google::protobuf::Message &msg,
                         const unsigned int retries)
{
    auto cmsg = createMessage(toString(module), false,
                              one::clproto::communication_protocol::Atom::default_instance(),
                              msg);

    m_communicationHandler->reply(originalMsg, *cmsg, poolType(msg), retries);
}

void Communicator::send(const ServerModule module,
                        const google::protobuf::Message &msg,
                        const unsigned int retries)
{
    auto cmsg = createMessage(toString(module), false,
                              one::clproto::communication_protocol::Atom::default_instance(),
                              msg);
    m_communicationHandler->send(*cmsg, poolType(msg), retries);
}

void Communicator::recreate()
{
    m_communicationHandler->recreate();
}

std::future<std::unique_ptr<Communicator::Answer>>
Communicator::communicateAsync(const std::string &module,
                               const google::protobuf::Message &msg,
                               const google::protobuf::Message &ans,
                               const unsigned int retries)
{
    auto cmsg = createMessage(module, false, ans, msg);
    return m_communicationHandler->communicate(*cmsg, poolType(msg), retries);
}

std::unique_ptr<Communicator::Answer>
Communicator::communicate(const std::string &module,
                          const google::protobuf::Message &msg,
                          const google::protobuf::Message &ans,
                          const unsigned int retries,
                          const std::chrono::milliseconds timeout)
{
    auto cmsg = createMessage(module, true, ans, msg);
    auto future = m_communicationHandler->communicate(*cmsg, poolType(msg), retries);

    if(future.wait_for(timeout) != std::future_status::ready)
        throw ReceiveError{"timeout of " + std::to_string(timeout.count()) +<commit_msg>bump minimal timeout value v2<commit_after>/**
 * @file communicator.cc
 * @author <NAME>
 * @copyright (C) 2014 ACK CYFRONET AGH
 * @copyright This software is released under the MIT license cited in 'LICENSE.txt'
 */

#include "communication/communicator.h"

#include "communication/connection.h"
#include "communication/exception.h"
#include "communication/websocketConnectionPool.h"
#include "fuse_messages.pb.h"
#include "logging.h"
#include "oneErrors.h"
#include "remote_file_management.pb.h"
#include "scheduler.h"

#include <boost/algorithm/string.hpp>
#include <google/protobuf/descriptor.h>

#include <cassert>
#include <unordered_set>
#include <curses.h>

namespace one
{
namespace communication
{

Communicator::Communicator(std::unique_ptr<CommunicationHandler> communicationHandler)
    : m_communicationHandler{std::move(communicationHandler)}
{
    setupHandshakeAck();
}

void Communicator::setupPushChannels(std::function<void(const Answer&)> callback)
{
    LOG(INFO) << "Setting up push channels with fuseId: '" << m_fuseId << "'";

    // First close the push channel that was already opened on the connection
    if(m_deregisterPushChannel)
        m_deregisterPushChannel();

    // Subscribe for push messages (negative message id)
    auto pred = [](const Answer &ans){ return ans.message_id() < 0; };
    auto unsubscribe = m_communicationHandler->subscribe({std::move(pred),
                                                          std::move(callback)});

    const auto fuseId = m_fuseId;
    const auto fslogic = toString(ServerModule::FSLOGIC);

    auto handshake = [=]{
        LOG(INFO) << "Opening a push channel with fuseId: '" << fuseId << "'";
        clproto::fuse_messages::ChannelRegistration reg;
        reg.set_fuse_id(fuseId);
        return createMessage(fslogic, true, Atom::default_instance(), reg);
    };

    auto goodbye = [=]{
        LOG(INFO) << "Closing the push channel with fuseId: '" << fuseId << "'";
        clproto::fuse_messages::ChannelClose close;
        close.set_fuse_id(fuseId);
        return createMessage(fslogic, true, Atom::default_instance(), close);
    };

    auto remove = m_communicationHandler->addHandshake(std::move(handshake),
                                                       std::move(goodbye),
                                                       CommunicationHandler::Pool::META);

    m_deregisterPushChannel = [=]{ unsubscribe(); remove(); };
}

void Communicator::setupHandshakeAck()
{
    LOG(INFO) << "Setting up HandshakeAck with fuseId: '" << m_fuseId << "'";

    // First remove the previous Ack message
    if(m_removeHandshakeAck)
        m_removeHandshakeAck();

    const auto fuseId = m_fuseId;

    // Build HandshakeAck message
    auto handshake = [this, fuseId]{
        LOG(INFO) << "Sending HandshakeAck with fuseId: '" << fuseId << "'";
        clproto::fuse_messages::HandshakeAck ack;
        ack.set_fuse_id(fuseId);
        return createMessage("", true, Atom::default_instance(), ack);
    };

    auto removeMeta = m_communicationHandler->addHandshake(handshake, CommunicationHandler::Pool::META);
    auto removeData = m_communicationHandler->addHandshake(handshake, CommunicationHandler::Pool::DATA);
    m_removeHandshakeAck = [=]{ removeMeta(); removeData(); };
}

void Communicator::setFuseId(std::string fuseId)
{
    LOG(INFO) << "Setting fuseId for communication: '" << fuseId << '"';
    m_fuseId = std::move(fuseId);
    setupHandshakeAck();
}

void Communicator::reply(const Answer &originalMsg, const ServerModule module,
                         const google::protobuf::Message &msg,
                         const unsigned int retries)
{
    auto cmsg = createMessage(toString(module), false,
                              one::clproto::communication_protocol::Atom::default_instance(),
                              msg);

    m_communicationHandler->reply(originalMsg, *cmsg, poolType(msg), retries);
}

void Communicator::send(const ServerModule module,
                        const google::protobuf::Message &msg,
                        const unsigned int retries)
{
    auto cmsg = createMessage(toString(module), false,
                              one::clproto::communication_protocol::Atom::default_instance(),
                              msg);
    m_communicationHandler->send(*cmsg, poolType(msg), retries);
}

void Communicator::recreate()
{
    m_communicationHandler->recreate();
}

std::future<std::unique_ptr<Communicator::Answer>>
Communicator::communicateAsync(const std::string &module,
                               const google::protobuf::Message &msg,
                               const google::protobuf::Message &ans,
                               const unsigned int retries)
{
    auto cmsg = createMessage(module, false, ans, msg);
    return m_communicationHandler->communicate(*cmsg, poolType(msg), retries);
}

std::unique_ptr<Communicator::Answer>
Communicator::communicate(const std::string &module,
                          const google::protobuf::Message &msg,
                          const google::protobuf::Message &ans,
                          const unsigned int retries,
                          const std::chrono::milliseconds timeout)
{
    if(timeout < 1500ms)
        timeout = 1500ms;
    auto cmsg = createMessage(module, true, ans, msg);
    auto future = m_communicationHandler->communicate(*cmsg, poolType(msg), retries);

    if(future.wait_for(timeout) != std::future_status::ready)
        throw ReceiveError{"timeout of " + std::to_string(timeout.count()) +
{'fraction_non_alphanumeric': 0.12179196517638524, 'fraction_numerical': 0.001723043438831958, 'mean_word_length': 3.030701754385965, 'pattern_counts': {'":': 0, '<': 99, '<?xml version=': 0, '>': 42, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 12, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7568617', 'n_tokens_mistral': 3156, 'n_tokens_neox': 2999, 'n_words': 658}