<commit_before>/*
 * #%L
 * %%
 * Copyright (C) 2011 - 2016 BMW Car IT GmbH
 * %%
 * 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.
 * #L%
 */
#include "joynr/Dispatcher.h"

#include <cassert>
#include <cstdint>
#include <chrono>

#include "joynr/serializer/Serializer.h"
#include "joynr/DispatcherUtils.h"
#include "joynr/SubscriptionRequest.h"
#include "joynr/BroadcastSubscriptionRequest.h"
#include "joynr/SubscriptionReply.h"
#include "joynr/SubscriptionPublication.h"
#include "joynr/SubscriptionStop.h"
#include "joynr/MessagingQos.h"
#include "joynr/JoynrMessageSender.h"
#include "joynr/MessagingQos.h"
#include "joynr/IRequestInterpreter.h"
#include "libjoynr/joynr-messaging/dispatcher/ReceivedMessageRunnable.h"
#include "joynr/PublicationInterpreter.h"
#include "joynr/PublicationManager.h"
#include "joynr/ISubscriptionManager.h"
#include "joynr/InterfaceRegistrar.h"
#include "joynr/Request.h"
#include "joynr/exceptions/JoynrException.h"
#include "joynr/exceptions/JoynrExceptionUtil.h"
#include "joynr/TypeUtil.h"
#include "joynr/Reply.h"

namespace joynr
{

INIT_LOGGER(Dispatcher);

Dispatcher::Dispatcher(JoynrMessageSender* messageSender,
                       boost::asio::io_service& ioService,
                       int maxThreads)
        : messageSender(messageSender),
          requestCallerDirectory("Dispatcher-RequestCallerDirectory", ioService),
          replyCallerDirectory("Dispatcher-ReplyCallerDirectory", ioService),
          publicationManager(nullptr),
          subscriptionManager(nullptr),
          handleReceivedMessageThreadPool("Dispatcher", maxThreads),
          subscriptionHandlingMutex()
{
}

Dispatcher::~Dispatcher()
{
    JOYNR_LOG_DEBUG(logger, "Destructing Dispatcher");
    handleReceivedMessageThreadPool.shutdown();
    delete publicationManager;
    delete subscriptionManager;
    publicationManager = nullptr;
    subscriptionManager = nullptr;
    JOYNR_LOG_DEBUG(logger, "Destructing finished");
}

void Dispatcher::addRequestCaller(const std::string& participantId,
                                  std::shared_ptr<RequestCaller> requestCaller)
{
    std::lock_guard<std::mutex> lock(subscriptionHandlingMutex);
    JOYNR_LOG_DEBUG(logger, "addRequestCaller id= {}", participantId);
    requestCallerDirectory.add(participantId, requestCaller);

    if (publicationManager != nullptr) {
        // publication manager queues received subscription requests, that are
        // received before the corresponding request caller is added
        publicationManager->restore(participantId, requestCaller, messageSender);
    } else {
        JOYNR_LOG_DEBUG(logger, "No publication manager available!");
    }
}

void Dispatcher::removeRequestCaller(const std::string& participantId)
{
    std::lock_guard<std::mutex> lock(subscriptionHandlingMutex);
    JOYNR_LOG_DEBUG(logger, "removeRequestCaller id= {}", participantId);
    // TODO if a provider is removed, all publication runnables are stopped
    // the subscription request is deleted,
    // Q: Should it be restored once the provider is registered again?
    publicationManager->removeAllSubscriptions(participantId);
    requestCallerDirectory.remove(participantId);
}

void Dispatcher::addReplyCaller(const std::string& requestReplyId,
                                std::shared_ptr<IReplyCaller> replyCaller,
                                const MessagingQos& qosSettings)
{
    JOYNR_LOG_DEBUG(logger, "addReplyCaller id= {}", requestReplyId);
    // add the callback to the registry that is responsible for reply messages
    replyCallerDirectory.add(requestReplyId, replyCaller, qosSettings.getTtl());
}

void Dispatcher::removeReplyCaller(const std::string& requestReplyId)
{
    JOYNR_LOG_DEBUG(logger, "removeReplyCaller id= {}", requestReplyId);
    replyCallerDirectory.remove(requestReplyId);
}

void Dispatcher::receive(const JoynrMessage& message)
{
    JOYNR_LOG_DEBUG(logger, "receive(message). Message payload: {}", message.getPayload());
    ReceivedMessageRunnable* receivedMessageRunnable = new ReceivedMessageRunnable(message, *this);
    handleReceivedMessageThreadPool.execute(receivedMessageRunnable);
}

void Dispatcher::handleRequestReceived(const JoynrMessage& message)
{
    std::string senderId = message.getHeaderFrom();
    std::string receiverId = message.getHeaderTo();

    // lookup necessary data
    std::shared_ptr<RequestCaller> caller = requestCallerDirectory.lookup(receiverId);
    if (caller == nullptr) {
        JOYNR_LOG_ERROR(
                logger,
                "caller not found in the RequestCallerDirectory for receiverId {}, ignoring",
                receiverId);
        return;
    }
    std::string interfaceName = caller->getInterfaceName();

    // Get the request interpreter that has been registered with this interface name
    std::shared_ptr<IRequestInterpreter> requestInterpreter =
            InterfaceRegistrar::instance().getRequestInterpreter(interfaceName);

    // deserialize Request
    try {
        Request request;
        joynr::serializer::deserializeFromJson(request, message.getPayload());
        const std::string& requestReplyId = request.getRequestReplyId();
        JoynrTimePoint requestExpiryDate = message.getHeaderExpiryDate();

        auto onSuccess =
                [requestReplyId, requestExpiryDate, this, senderId, receiverId](Reply&& reply) {
            reply.setRequestReplyId(requestReplyId);
            // reply.setResponseVariant(std::move(returnValueVar));
            // send reply back to the original sender (ie. sender and receiver ids are reversed
            // on
            // purpose)
            JOYNR_LOG_DEBUG(logger,
                            "Got reply from RequestInterpreter for requestReplyId {}",
                            requestReplyId);
            JoynrTimePoint now = std::chrono::time_point_cast<std::chrono::milliseconds>(
                    std::chrono::system_clock::now());
            std::int64_t ttl = std::chrono::duration_cast<std::chrono::milliseconds>(
                                       requestExpiryDate - now).count();
            messageSender->sendReply(receiverId, // receiver of the request is sender of reply
                                     senderId,   // sender of request is receiver of reply
                                     MessagingQos(ttl),
                                     std::move(reply));
        };

        auto onError = [requestReplyId, requestExpiryDate, this, senderId, receiverId](
                const std::shared_ptr<exceptions::JoynrException>& exception) {
            Reply reply;
            reply.setRequestReplyId(requestReplyId);
            reply.setError(exception);
            JOYNR_LOG_DEBUG(logger,
                            "Got error reply from RequestInterpreter for requestReplyId {}",
                            requestReplyId);
            JoynrTimePoint now = std::chrono::time_point_cast<std::chrono::milliseconds>(
                    std::chrono::system_clock::now());<commit_msg>[C++] cleaned up Dispatcher<commit_after>/*
 * #%L
 * %%
 * Copyright (C) 2011 - 2016 BMW Car IT GmbH
 * %%
 * 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.
 * #L%
 */
#include "joynr/Dispatcher.h"

#include <cassert>
#include <cstdint>
#include <chrono>

#include "joynr/serializer/Serializer.h"
#include "joynr/DispatcherUtils.h"
#include "joynr/SubscriptionRequest.h"
#include "joynr/BroadcastSubscriptionRequest.h"
#include "joynr/SubscriptionReply.h"
#include "joynr/SubscriptionPublication.h"
#include "joynr/SubscriptionStop.h"
#include "joynr/MessagingQos.h"
#include "joynr/JoynrMessageSender.h"
#include "joynr/MessagingQos.h"
#include "joynr/IRequestInterpreter.h"
#include "libjoynr/joynr-messaging/dispatcher/ReceivedMessageRunnable.h"
#include "joynr/PublicationInterpreter.h"
#include "joynr/PublicationManager.h"
#include "joynr/ISubscriptionManager.h"
#include "joynr/InterfaceRegistrar.h"
#include "joynr/Request.h"
#include "joynr/exceptions/JoynrException.h"
#include "joynr/exceptions/JoynrExceptionUtil.h"
#include "joynr/TypeUtil.h"
#include "joynr/Reply.h"

namespace joynr
{

INIT_LOGGER(Dispatcher);

Dispatcher::Dispatcher(JoynrMessageSender* messageSender,
                       boost::asio::io_service& ioService,
                       int maxThreads)
        : messageSender(messageSender),
          requestCallerDirectory("Dispatcher-RequestCallerDirectory", ioService),
          replyCallerDirectory("Dispatcher-ReplyCallerDirectory", ioService),
          publicationManager(nullptr),
          subscriptionManager(nullptr),
          handleReceivedMessageThreadPool("Dispatcher", maxThreads),
          subscriptionHandlingMutex()
{
}

Dispatcher::~Dispatcher()
{
    JOYNR_LOG_DEBUG(logger, "Destructing Dispatcher");
    handleReceivedMessageThreadPool.shutdown();
    delete publicationManager;
    delete subscriptionManager;
    publicationManager = nullptr;
    subscriptionManager = nullptr;
    JOYNR_LOG_DEBUG(logger, "Destructing finished");
}

void Dispatcher::addRequestCaller(const std::string& participantId,
                                  std::shared_ptr<RequestCaller> requestCaller)
{
    std::lock_guard<std::mutex> lock(subscriptionHandlingMutex);
    JOYNR_LOG_DEBUG(logger, "addRequestCaller id= {}", participantId);
    requestCallerDirectory.add(participantId, requestCaller);

    if (publicationManager != nullptr) {
        // publication manager queues received subscription requests, that are
        // received before the corresponding request caller is added
        publicationManager->restore(participantId, requestCaller, messageSender);
    } else {
        JOYNR_LOG_DEBUG(logger, "No publication manager available!");
    }
}

void Dispatcher::removeRequestCaller(const std::string& participantId)
{
    std::lock_guard<std::mutex> lock(subscriptionHandlingMutex);
    JOYNR_LOG_DEBUG(logger, "removeRequestCaller id= {}", participantId);
    // TODO if a provider is removed, all publication runnables are stopped
    // the subscription request is deleted,
    // Q: Should it be restored once the provider is registered again?
    publicationManager->removeAllSubscriptions(participantId);
    requestCallerDirectory.remove(participantId);
}

void Dispatcher::addReplyCaller(const std::string& requestReplyId,
                                std::shared_ptr<IReplyCaller> replyCaller,
                                const MessagingQos& qosSettings)
{
    JOYNR_LOG_DEBUG(logger, "addReplyCaller id= {}", requestReplyId);
    // add the callback to the registry that is responsible for reply messages
    replyCallerDirectory.add(requestReplyId, replyCaller, qosSettings.getTtl());
}

void Dispatcher::removeReplyCaller(const std::string& requestReplyId)
{
    JOYNR_LOG_DEBUG(logger, "removeReplyCaller id= {}", requestReplyId);
    replyCallerDirectory.remove(requestReplyId);
}

void Dispatcher::receive(const JoynrMessage& message)
{
    JOYNR_LOG_DEBUG(logger, "receive(message). Message payload: {}", message.getPayload());
    ReceivedMessageRunnable* receivedMessageRunnable = new ReceivedMessageRunnable(message, *this);
    handleReceivedMessageThreadPool.execute(receivedMessageRunnable);
}

void Dispatcher::handleRequestReceived(const JoynrMessage& message)
{
    std::string senderId = message.getHeaderFrom();
    std::string receiverId = message.getHeaderTo();

    // lookup necessary data
    std::shared_ptr<RequestCaller> caller = requestCallerDirectory.lookup(receiverId);
    if (caller == nullptr) {
        JOYNR_LOG_ERROR(
                logger,
                "caller not found in the RequestCallerDirectory for receiverId {}, ignoring",
                receiverId);
        return;
    }
    std::string interfaceName = caller->getInterfaceName();

    // Get the request interpreter that has been registered with this interface name
    std::shared_ptr<IRequestInterpreter> requestInterpreter =
            InterfaceRegistrar::instance().getRequestInterpreter(interfaceName);

    // deserialize Request
    try {
        Request request;
        joynr::serializer::deserializeFromJson(request, message.getPayload());
        const std::string& requestReplyId = request.getRequestReplyId();
        JoynrTimePoint requestExpiryDate = message.getHeaderExpiryDate();

        auto onSuccess =
                [requestReplyId, requestExpiryDate, this, senderId, receiverId](Reply&& reply) {
            reply.setRequestReplyId(requestReplyId);
            // send reply back to the original sender (ie. sender and receiver ids are reversed
            // on purpose)
            JOYNR_LOG_DEBUG(logger,
                            "Got reply from RequestInterpreter for requestReplyId {}",
                            requestReplyId);
            JoynrTimePoint now = std::chrono::time_point_cast<std::chrono::milliseconds>(
                    std::chrono::system_clock::now());
            std::int64_t ttl = std::chrono::duration_cast<std::chrono::milliseconds>(
                                       requestExpiryDate - now).count();
            messageSender->sendReply(receiverId, // receiver of the request is sender of reply
                                     senderId,   // sender of request is receiver of reply
                                     MessagingQos(ttl),
                                     std::move(reply));
        };

        auto onError = [requestReplyId, requestExpiryDate, this, senderId, receiverId](
                const std::shared_ptr<exceptions::JoynrException>& exception) {
            Reply reply;
            reply.setRequestReplyId(requestReplyId);
            reply.setError(exception);
            JOYNR_LOG_DEBUG(logger,
                            "Got error reply from RequestInterpreter for requestReplyId {}",
                            requestReplyId);
            JoynrTimePoint now = std::chrono::time_point_cast<std::chrono::milliseconds>(
                    std::chrono::system_clock::now());
{'fraction_non_alphanumeric': 0.0967548477377224, 'fraction_numerical': 0.001865795961884454, 'mean_word_length': 3.32507204610951, 'pattern_counts': {'":': 0, '<': 29, '<?xml version=': 0, '>': 37, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 10, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7655776', 'n_tokens_mistral': 4090, 'n_tokens_neox': 3993, 'n_words': 921}