<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 "fuse_messages.pb.h"
#include "logging.h"
#include "make_unique.h"
#include "veilErrors.h"
#include "remote_file_management.pb.h"

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

#include <cassert>
#include <unordered_set>

namespace veil
{
namespace communication
{

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

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
    static std::function<void()> deregisterPushChannel;
    if(deregisterPushChannel)
        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;

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

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

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

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

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

    // First remove the previous Ack message
    static std::function<void()> removeHandshakeAck;
    if(removeHandshakeAck)
        removeHandshakeAck();

    const auto fuseId = m_fuseId;

    // Build HandshakeAck message
    auto handshake = [this, fuseId]{
        LOG(INFO) << "Sending HandshakeAck with fuseId: '" << fuseId << "'";
        protocol::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);
    removeHandshakeAck = [=]{ removeMeta(); removeData(); };
}

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

void Communicator::reply(const Answer &originalMsg, const std::string &module,
                         const google::protobuf::Message &msg)
{
    auto cmsg = createMessage(module, false,
                              veil::protocol::communication_protocol::Atom::default_instance(),
                              msg);
    m_communicationHandler->reply(originalMsg, *cmsg, poolType(msg));
}

void Communicator::send(const std::string &module,
                        const google::protobuf::Message &msg)
{
    auto cmsg = createMessage(module, false,
                              veil::protocol::communication_protocol::Atom::default_instance(),
                              msg);
    m_communicationHandler->send(*cmsg, poolType(msg));
}

std::future<std::unique_ptr<Communicator::Answer>>
Communicator::communicateAsync(const std::string &module,
                               const google::protobuf::Message &msg,
                               const google::protobuf::Message &ans)
{
    auto cmsg = createMessage(module, false, ans, msg);
    return m_communicationHandler->communicate(*cmsg, poolType(msg));
}<commit_msg>VFS-740 Automatize setting up handshakeAck.<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 "fuse_messages.pb.h"
#include "logging.h"
#include "make_unique.h"
#include "veilErrors.h"
#include "remote_file_management.pb.h"

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

#include <cassert>
#include <unordered_set>

namespace veil
{
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
    static std::function<void()> deregisterPushChannel;
    if(deregisterPushChannel)
        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;

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

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

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

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

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

    // First remove the previous Ack message
    static std::function<void()> removeHandshakeAck;
    if(removeHandshakeAck)
        removeHandshakeAck();

    const auto fuseId = m_fuseId;

    // Build HandshakeAck message
    auto handshake = [this, fuseId]{
        LOG(INFO) << "Sending HandshakeAck with fuseId: '" << fuseId << "'";
        protocol::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);
    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 std::string &module,
                         const google::protobuf::Message &msg)
{
    auto cmsg = createMessage(module, false,
                              veil::protocol::communication_protocol::Atom::default_instance(),
                              msg);
    m_communicationHandler->reply(originalMsg, *cmsg, poolType(msg));
}

void Communicator::send(const std::string &module,
                        const google::protobuf::Message &msg)
{
    auto cmsg = createMessage(module, false,
                              veil::protocol::communication_protocol::Atom::default_instance(),
                              msg);
    m_communicationHandler->send(*cmsg, poolType(msg));
}

std::future<std::unique_ptr<Communicator::Answer>>
Communicator::communicateAsync(const std::string &module,
                               const google::protobuf::Message &msg,
                               const google::protobuf::Message &ans)
{
    auto cmsg = createMessage(module, false, ans, msg);
    return m_communicationHandler->communicate(*cmsg, poolType(msg));
}
{'fraction_non_alphanumeric': 0.1262516325642142, 'fraction_numerical': 0.0014148889856334349, 'mean_word_length': 3.2561371005094952, 'pattern_counts': {'":': 0, '<': 99, '<?xml version=': 0, '>': 39, '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': '6119633', 'n_tokens_mistral': 2673, 'n_tokens_neox': 2541, 'n_words': 554}