RoutingServiceImpl.hpp

/*
 * (c) Copyright, Real-Time Innovations, 2017.
 * All rights reserved.
 *
 * No duplications, whole or partial, manual or electronic, may be made
 * without express written permission.  Any such copies, or
 * revisions thereof, must display this notice unaltered.
 * This code contains trade secrets of Real-Time Innovations, Inc.
 */
 
#ifndef RTI_ROUTING_ROUTING_SERVICE_IMPL_HPP_
#define RTI_ROUTING_ROUTING_SERVICE_IMPL_HPP_
 
#include <dds/core/refmacros.hpp>
#include <rti/core/detail/SelfReference.hpp>
#include "routingservice/routingservice_adapter.h"
#include "routingservice/routingservice_service.h"
#include "routingservice/routingservice_log.h"
#include "routingservice/routingservice_remote_config.h"
#include "routingservice/ServiceAdmin.h"
 
#include <rti/routing/ServiceProperty.hpp>
#include <rti/routing/adapter/AdapterPlugin.hpp>
#include <rti/routing/transf/TransformationPlugin.hpp>
#include <rti/routing/processor/ProcessorPlugin.hpp>
#include <rti/routing/detail/ShutdownHookForwarder.hpp>
#include <rti/idlgen/ServiceAdmin.hpp>
 
namespace rti { namespace routing {
class RoutingServiceImpl
        : public rti::core::detail::RetainableType<RoutingServiceImpl> {
public:
 
    typedef RTI::Service::Admin::CommandRequest CommandRequest;
    typedef RTI::Service::Admin::CommandReply CommandReply;
 
    template <typename HookFunc>
    explicit RoutingServiceImpl(
            const ServiceProperty& property,
            HookFunc& shutdown_hook)
            : native_(RTI_RoutingService_new(&property.native())),
            shutdown_hook_fwd_(
                    new rti::routing::detail::ShutdownHookForwarder<HookFunc>(
                            shutdown_hook))
    {
        rti::core::check_create_entity(native_, "RoutingService");
        RTI_RoutingService_set_remote_shutdown_hook(
                native_,
                shutdown_hook_fwd_->native());
    }
 
    explicit RoutingServiceImpl(const ServiceProperty& property)
            : native_(RTI_RoutingService_new(&property.native())),
              shutdown_hook_fwd_(nullptr)
    {
        rti::core::check_create_entity(native_, "RoutingService");
    }
 
    RoutingServiceImpl(const RTI_RoutingServiceProperty& property)
            : native_(RTI_RoutingService_new(&property)),
            shutdown_hook_fwd_(NULL)
    {
        rti::core::check_create_entity(native_, "RoutingService");
    }
 
    RoutingServiceImpl(RTI_RoutingService *native)
            : native_(native), shutdown_hook_fwd_(NULL)
    {
        rti::core::check_create_entity(native_, "RoutingService");
    }
 
    ~RoutingServiceImpl()
    {
        RTI_RoutingService_delete(native_);
        delete shutdown_hook_fwd_;
    }
 
    void start()
    {
        if (!RTI_RoutingService_start(native_)) {
            throw dds::core::Error("failed to start RoutingService");
        }
    }
 
    void stop()
    {
        if (!RTI_RoutingService_stop(native_)) {
            throw dds::core::Error("failed to stop RoutingService");
        }
    }
 
    void attach_adapter_plugin(
            rti::routing::adapter::AdapterPlugin *adapter_plugin,
            const std::string& registered_name)
    {
        RTI_RoutingServiceAdapterPluginExt *native_plugin =
                adapter::detail::AdapterForwarder::create_plugin(adapter_plugin);
        if (!RTI_RoutingService_attach_adapter_plugin(
                native_,
                native_plugin,
                registered_name.c_str())) {
            throw dds::core::Error("failed to attach native adapter");
        }
    }
 
    void attach_transformation_plugin(
            rti::routing::transf::TransformationPlugin *transformation_plugin,
            const std::string& registered_name)
    {
        RTI_RoutingServiceTransformationPlugin *native_plugin =
                transf::detail::TransformationPluginForwarder::create_plugin(
                        transformation_plugin);
        if (!RTI_RoutingService_attach_transformation_plugin(
                native_,
                native_plugin,
                registered_name.c_str())) {
            throw dds::core::Error("failed to attach native transformation");
        }
    }
 
    void attach_processor_plugin(
            rti::routing::processor::ProcessorPlugin *processor_plugin,
            const std::string& registered_name)
    {
        RTI_RoutingServiceProcessorPlugin *native_plugin =
                processor::detail::ProcessorPluginForwarder::create_plugin(
                        processor_plugin);
        if (!RTI_RoutingService_attach_processor_plugin(
                native_,
                native_plugin,
                registered_name.c_str())) {
            throw dds::core::Error("failed to attach native processor");
        }
    }
 
    CommandReply& execute_command(
            CommandReply& reply,
            const CommandRequest& request)
    {
        RTI_Service_Admin_CommandRequest native_request =
                RTI_Service_Admin_CommandRequest_INITIALIZER;
        RTI_Service_Admin_CommandReply *native_reply_ptr = nullptr;
 
        if (!RTI_Service_Admin_CommandRequest_initialize(&native_request)) {
            RTI_Service_Admin_CommandRequest_finalize(&native_request);
            throw dds::core::Error("failed to initialize CommandRequest");
        }
        std::shared_ptr<RTI_Service_Admin_CommandRequest> request_ptr(
                &native_request,
                [](RTI_Service_Admin_CommandRequest *native_req_ptr) {
                    RTI_Service_Admin_CommandRequest_finalize(native_req_ptr);
                });
 
        to_native(native_request, request);
        RTI_RoutingService_execute_command(
                native_,
                &native_reply_ptr,
                &native_request);
        from_native(reply, *native_reply_ptr);
        RTI_RoutingService_return_reply(native_, native_reply_ptr);
        return reply;
    }
 
    CommandReply execute_command(const CommandRequest& request)
    {
        CommandReply reply;
        return execute_command(reply, request);
    }
 
 
    RTI_RoutingService* native() const
    {
        return native_;
    }
 
private:
 
    RTI_RoutingService *native_;
 
    rti::routing::detail::AbstractShutdownHookForwarder *shutdown_hook_fwd_;
 
    static RTI_Service_Admin_CommandRequest& to_native(
            RTI_Service_Admin_CommandRequest& native,
            const RTI::Service::Admin::CommandRequest& request)
    {
        native.action = static_cast<RTI_Service_Admin_CommandActionKind>(
                request.action());
        if (request.application_name().is_set()) {
            if (DDS_String_replace(
                    &native.application_name,
                    request.application_name().get().c_str()) == NULL) {
                throw dds::core::Error("failed to set application name");
            }
        }
        native.instance_id = request.instance_id();
        if (DDS_String_replace(
                &native.resource_identifier,
                request.resource_identifier().c_str()) == NULL) {
            throw dds::core::Error("failed to set resource identifier");
        }
        if (DDS_String_replace(
                &native.string_body,
                request.string_body().c_str()) == NULL) {
            throw dds::core::Error("failed to set string body");
        }
        if (!DDS_OctetSeq_from_array(
                &native.octet_body,
                (DDS_Octet *) request.octet_body().data(),
                (DDS_Long) request.octet_body().size())) {
            throw dds::core::Error("failed to set octet body");
        }
        return native;
    }
 
    static RTI::Service::Admin::CommandReply& from_native(
            RTI::Service::Admin::CommandReply& reply,
            const RTI_Service_Admin_CommandReply& native)
    {
        using namespace RTI::Service::Admin;
 
        reply.retcode(static_cast<CommandReplyRetcode>(native.retcode));
        reply.native_retcode(native.native_retcode);
        DDS_Octet *native_octet_body = DDS_OctetSeq_get_contiguous_buffer(
                (DDS_OctetSeq *) &native.octet_body);
        if (DDS_OctetSeq_get_length(&native.octet_body) < 0) {
            throw dds::core::Error("invalid octet_body length");
        }
        reply.octet_body().resize(
                (size_t) DDS_OctetSeq_get_length(&native.octet_body));
        std::copy(
                native_octet_body,
                native_octet_body + DDS_OctetSeq_get_length(&native.octet_body),
                reply.octet_body().begin());
        reply.string_body(native.string_body);
 
        return reply;
    }
 
};
 
}}
 
#endif // RTI_ROUTINGv_ROUTING_SERVICE_IMPL_HPP_