<commit_before>/**
 * Copyright (c) 2011-2017 libbitcoin developers (see AUTHORS)
 *
 * This file is part of libbitcoin.
 *
 * This program is free software: you can redistribute it and/or modify
 * it under the terms of the GNU Affero General Public License as published by
 * the Free Software Foundation, either version 3 of the License, or
 * (at your option) any later version.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU Affero General Public License for more details.
 *
 * You should have received a copy of the GNU Affero General Public License
 * along with this program.  If not, see <http://www.gnu.org/licenses/>.
 */
#include <bitcoin/server/services/query_service.hpp>

#include <bitcoin/protocol.hpp>
#include <bitcoin/server/server_node.hpp>
#include <bitcoin/server/settings.hpp>

namespace libbitcoin {
namespace server {

using namespace bc::config;
using namespace bc::protocol;

static const auto domain = "query";
const config::endpoint query_service::public_query("inproc://public_query");
const config::endpoint query_service::secure_query("inproc://secure_query");
const config::endpoint query_service::public_notify("inproc://public_notify");
const config::endpoint query_service::secure_notify("inproc://secure_notify");

query_service::query_service(zmq::authenticator& authenticator,
    server_node& node, bool secure)
  : worker(node.thread_pool()),
    secure_(secure),
    settings_(node.server_settings()),
    authenticator_(authenticator)
{
}

// Implement worker as a broker.
// The dealer blocks until there are available workers.
// The router drops messages for lost peers (clients) and high water.
void query_service::work()
{
    zmq::socket router(authenticator_, zmq::socket::role::router);
    zmq::socket query_dealer(authenticator_, zmq::socket::role::dealer);
    zmq::socket notify_dealer(authenticator_, zmq::socket::role::dealer);

    // Bind sockets to the service and worker endpoints.
    if (!started(bind(router, query_dealer, notify_dealer)))
        return;

    zmq::poller poller;
    poller.add(router);
    poller.add(query_dealer);
    poller.add(notify_dealer);

    while (!poller.terminated() && !stopped())
    {
        const auto signaled = poller.wait();

        if (signaled.contains(router.id()) &&
            !forward(router, query_dealer))
        {
            LOG_WARNING(LOG_SERVER)
                << "Failed to forward from router to query_dealer.";
        }

        if (signaled.contains(query_dealer.id()) &&
            !forward(query_dealer, router))
        {
            LOG_WARNING(LOG_SERVER)
                << "Failed to forward from query_dealer to router.";
        }

        if (signaled.contains(notify_dealer.id()) &&
            !forward(notify_dealer, router))
        {
            LOG_WARNING(LOG_SERVER)
                << "Failed to forward from notify_dealer to router.";
        }
    }

    // Unbind the sockets and exit this thread.
    finished(unbind(router, query_dealer, notify_dealer));
}

// Bind/Unbind.
//-----------------------------------------------------------------------------

bool query_service::bind(zmq::socket& router, zmq::socket& query_dealer,
    zmq::socket& notify_dealer)
{
    const auto security = secure_ ? "secure" : "public";
    const auto& query_worker = secure_ ? secure_query : public_query;
    const auto& notify_worker = secure_ ? secure_notify : public_notify;
    const auto& query_service = secure_ ? settings_.secure_query_endpoint :
        settings_.public_query_endpoint;

    if (!authenticator_.apply(router, domain, secure_))
        return false;

    auto ec = router.bind(query_service);

    if (ec)
    {
        LOG_ERROR(LOG_SERVER)
            << "Failed to bind " << security << " query service to "
            << query_service << " : " << ec.message();
        return false;
    }

    ec = query_dealer.bind(query_worker);

    if (ec)
    {
        LOG_ERROR(LOG_SERVER)
            << "Failed to bind " << security << " query workers to "
            << query_worker << " : " << ec.message();
        return false;
    }

    ec = notify_dealer.bind(notify_worker);

    if (ec)
    {
        LOG_ERROR(LOG_SERVER)
            << "Failed to bind " << security << " notify workers to "
            << notify_worker << " : " << ec.message();
        return false;
    }

    LOG_INFO(LOG_SERVER)
        << "Bound " << security << " query service to " << query_service;
    return true;
}

bool query_service::unbind(zmq::socket& router, zmq::socket& query_dealer,
    zmq::socket& notify_dealer)
{
    // Stop all even if one fails.<commit_msg>Fix text.<commit_after>/**
 * Copyright (c) 2011-2017 libbitcoin developers (see AUTHORS)
 *
 * This file is part of libbitcoin.
 *
 * This program is free software: you can redistribute it and/or modify
 * it under the terms of the GNU Affero General Public License as published by
 * the Free Software Foundation, either version 3 of the License, or
 * (at your option) any later version.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU Affero General Public License for more details.
 *
 * You should have received a copy of the GNU Affero General Public License
 * along with this program.  If not, see <http://www.gnu.org/licenses/>.
 */
#include <bitcoin/server/services/query_service.hpp>

#include <bitcoin/protocol.hpp>
#include <bitcoin/server/server_node.hpp>
#include <bitcoin/server/settings.hpp>

namespace libbitcoin {
namespace server {

using namespace bc::config;
using namespace bc::protocol;

static const auto domain = "query";
const config::endpoint query_service::public_query("inproc://public_query");
const config::endpoint query_service::secure_query("inproc://secure_query");
const config::endpoint query_service::public_notify("inproc://public_notify");
const config::endpoint query_service::secure_notify("inproc://secure_notify");

query_service::query_service(zmq::authenticator& authenticator,
    server_node& node, bool secure)
  : worker(node.thread_pool()),
    secure_(secure),
    settings_(node.server_settings()),
    authenticator_(authenticator)
{
}

// Implement worker as a broker.
// The dealer blocks until there are available workers.
// The router drops messages for lost peers (clients) and high water.
void query_service::work()
{
    zmq::socket router(authenticator_, zmq::socket::role::router);
    zmq::socket query_dealer(authenticator_, zmq::socket::role::dealer);
    zmq::socket notify_dealer(authenticator_, zmq::socket::role::dealer);

    // Bind sockets to the service and worker endpoints.
    if (!started(bind(router, query_dealer, notify_dealer)))
        return;

    zmq::poller poller;
    poller.add(router);
    poller.add(query_dealer);
    poller.add(notify_dealer);

    while (!poller.terminated() && !stopped())
    {
        const auto signaled = poller.wait();

        if (signaled.contains(router.id()) &&
            !forward(router, query_dealer))
        {
            LOG_WARNING(LOG_SERVER)
                << "Failed to forward from router to query_dealer.";
        }

        if (signaled.contains(query_dealer.id()) &&
            !forward(query_dealer, router))
        {
            LOG_WARNING(LOG_SERVER)
                << "Failed to forward from query_dealer to router.";
        }

        if (signaled.contains(notify_dealer.id()) &&
            !forward(notify_dealer, router))
        {
            LOG_WARNING(LOG_SERVER)
                << "Failed to forward from notify_dealer to router.";
        }
    }

    // Unbind the sockets and exit this thread.
    finished(unbind(router, query_dealer, notify_dealer));
}

// Bind/Unbind.
//-----------------------------------------------------------------------------

bool query_service::bind(zmq::socket& router, zmq::socket& query_dealer,
    zmq::socket& notify_dealer)
{
    const auto security = secure_ ? "secure" : "public";
    const auto& query_worker = secure_ ? secure_query : public_query;
    const auto& notify_worker = secure_ ? secure_notify : public_notify;
    const auto& query_service = secure_ ? settings_.secure_query_endpoint :
        settings_.public_query_endpoint;

    if (!authenticator_.apply(router, domain, secure_))
        return false;

    auto ec = router.bind(query_service);

    if (ec)
    {
        LOG_ERROR(LOG_SERVER)
            << "Failed to bind " << security << " query service to "
            << query_service << " : " << ec.message();
        return false;
    }

    ec = query_dealer.bind(query_worker);

    if (ec)
    {
        LOG_ERROR(LOG_SERVER)
            << "Failed to bind " << security << " query workers to "
            << query_worker << " : " << ec.message();
        return false;
    }

    ec = notify_dealer.bind(notify_worker);

    if (ec)
    {
        LOG_ERROR(LOG_SERVER)
            << "Failed to bind " << security << " notify worker to "
            << notify_worker << " : " << ec.message();
        return false;
    }

    LOG_INFO(LOG_SERVER)
        << "Bound " << security << " query service to " << query_service;
    return true;
}

bool query_service::unbind(zmq::socket& router, zmq::socket& query_dealer,
    zmq::socket& notify_dealer)
{
    // Stop all even if one fails.
{'fraction_non_alphanumeric': 0.130298273155416, 'fraction_numerical': 0.0018838304552590266, 'mean_word_length': 2.9438712340074287, 'pattern_counts': {'":': 0, '<': 113, '<?xml version=': 0, '>': 13, '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': '4867903', 'n_tokens_mistral': 2759, 'n_tokens_neox': 2684, 'n_words': 752}