sag_connectivity_plugins.hpp

Go to the documentation of this file.

/*
 * Title:        sag_connectivity_plugins.hpp
 * Description:  C++ API for writing connectivity plugins
 * $Copyright (c) 2015-2024 Cumulocity GmbH, Duesseldorf, Germany and/or its affiliates and/or their licensors.$
 * Use, reproduction, transfer, publication or disclosure is prohibited except as specifically provided for in your License Agreement with Cumulocity GmbH.
 */

#ifndef _SAG_CONNECTIVITY_PLUGINS_HPP_
#define _SAG_CONNECTIVITY_PLUGINS_HPP_

#include <sag_connectivity_cpp.hpp>
#include <sag_plugin_logging.hpp>
#include <memory>
#include <mutex>

// must define __STDC_FORMAT_MACROS before first include of inttypes else printf macros won't be defined
#ifndef __STDC_FORMAT_MACROS
#define __STDC_FORMAT_MACROS 1
#endif
#include <inttypes.h>

namespace com {
namespace softwareag {
namespace _DATAT_INTERNAL_CPP_NAMESPACE { 

namespace
{
        void replace(std::string &input, const std::string &from, const std::string &to)
        {
                size_t pos = 0;
                while((pos = input.find(from, pos)) != std::string::npos)
                {
                        input.replace(pos, from.length(), to);
                        pos += to.length();
                }
        }

        std::string to_string(const Message &m, bool truncate=true)
        {
                std::string payload = to_string(m.getPayload());
                if (truncate && payload.length() > 200) payload = payload.substr(0, 196) + " ...";
                // security sanitization to prevent log message faking
                replace(payload, "\n", "\\n");
                replace(payload, "\r", "\\r");
                return "Message<metadata="+to_string(m.getMetadataMap())+", payload="+payload+">";
        }
}

enum class Direction { 
        TOWARDS_HOST = 1,
        TOWARDS_TRANSPORT = 2

};

class PluginHost {
        friend class Plugin;
public:
        typedef std::unique_ptr<PluginHost> ptr_t;

        void enableReliability(Direction direction) { 
                if (SAG_ERROR_OK != sag_enable_reliability(chain, static_cast<int>(direction))) { 
                        throw std::runtime_error("An error occurred while setting chain reliability"); 
                }
        }

        void enableQueueFlushMessages() {
                if (SAG_ERROR_OK != sag_enable_queue_flush_messages(chain)) {
                        throw std::runtime_error("An error occurred while enabling queue flush messages");
                }
        }
        
        bool isShuttingDown() { 
                bool isShuttingDown = false; 
                if (SAG_ERROR_OK != sag_is_host_shutting_down(chain, isShuttingDown)) {
                        throw std::runtime_error("An error occurred while checking if host is shutting down");
                }
                return isShuttingDown; 
        }
private:
        PluginHost(void* chain = nullptr) :chain(chain) {}
        void* chain;
};

// forward decl for parameters friend
class Plugin;

class PluginConstructorParameters
{
        // for constructor access to connectivityManager
        friend class Plugin;

public:
        class TransportConstructorParameters;

        class CodecConstructorParameters;

        const map_t &getConfig() const { return config; }

        const std::string &getChainId() const { return chainId; }

        const std::string &getPluginName() const { return pluginName; }

protected:
        PluginConstructorParameters(const std::string &pluginName, const std::string &chainId, const map_t &config, void* connectivityManager, void* chain)
                : chainId(chainId), pluginName(pluginName), config(config.copy()), connectivityManager(connectivityManager), chain(chain)
        {}

private:
        std::string chainId;
        std::string pluginName;
        map_t config;
        void* connectivityManager;
        void* chain;
};

class PluginConstructorParameters::TransportConstructorParameters : public PluginConstructorParameters
{
public:
        TransportConstructorParameters(const std::string &pluginName, const std::string &chainId, const map_t &config, void* connectivityManager, void* reserved)
        : PluginConstructorParameters(pluginName, chainId, config, connectivityManager, reserved)
        {}
};

class PluginConstructorParameters::CodecConstructorParameters : public PluginConstructorParameters
{
public:
        CodecConstructorParameters(const std::string &pluginName, const std::string &chainId, const map_t &config, void* connectivityManager, void* reserved)
        : PluginConstructorParameters(pluginName, chainId, config, connectivityManager, reserved)
        {}
        CodecConstructorParameters(const std::string &pluginName)
        : PluginConstructorParameters(pluginName, "", map_t{}, nullptr, nullptr)
        {}
};


class StatusReporter
{
public:

        static const char* STATUS_ONLINE() { return "ONLINE"; }
        static const char* STATUS_STARTING() { return "STARTING"; }
        static const char* STATUS_FAILED() { return "FAILED"; }

        class StatusItem
        {
                friend class StatusReporter;
        public:

                ~StatusItem()
                {
                        sag_delete_user_status_item(connectivityManager, underlying);
                        underlying.item = nullptr;
                }

                void setStatus(const std::string &value)
                {
                        std::unique_lock<std::mutex> ul(status_lock);
                        setStatusLocked(value);
                }

                void setStatus(int64_t value)
                {
                        std::unique_lock<std::mutex> ul(status_lock);
                        intValue = value;
                        setStatusLocked(convert_to_details::integerToString(value));
                }

                std::string getStatus() {
                        std::unique_lock<std::mutex> ul(status_lock);
                        return lastValue;
                }

                void increment(int64_t incrementValue = 1)
                {
                        if (incrementValue == 0) return;
                        std::unique_lock<std::mutex> ul(status_lock);
                        intValue += incrementValue;
                        setStatusLocked(convert_to_details::integerToString(intValue));
                }

                const std::string &key() { return mkey; }

        private:
                StatusItem(const StatusItem& other) = delete; // non construction-copyable
                StatusItem& operator=(const StatusItem&) = delete; // non copyable

                StatusItem(void* connectivityManager, const std::string &key, const std::string &initialValue, const int64_t intValue)
                        : intValue(intValue),
                        mkey(key),
                        lastValue(initialValue),
                        connectivityManager(connectivityManager),
                        underlying(sag_create_user_status_item(connectivityManager, key.c_str(), initialValue.c_str()))
                {
                        if (!underlying.item)
                        {
                                std::ostringstream oss;
                                oss << "Failed to create status item '" << key << "' (ensure the key is unique and that this plug-in has not been shutdown already)";
                                throw std::runtime_error(oss.str());
                        }
                }

                void setStatusLocked(const std::string &value)
                {
                        if (value == lastValue) return; // no-op this case
                        lastValue = value;

                        sag_set_user_status_item(underlying, value.c_str());
                }

                int64_t intValue;
                const std::string mkey;
                std::string lastValue;
                void* connectivityManager;
                sag_status_item_t underlying;
                std::mutex status_lock;
        };

        typedef std::unique_ptr<StatusItem> item_ptr;

        item_ptr createStatusItem(const std::string &key, const std::string &initialValue)
        {
                return std::unique_ptr<StatusItem>(new StatusItem(connectivityManager, key, initialValue, 0));
        }

        item_ptr createStatusItem(const std::string &key, int64_t initialValue)
        {
                std::ostringstream oss;
                oss << initialValue;
                return std::unique_ptr<StatusItem>(new StatusItem(connectivityManager, key, oss.str(), initialValue));
        }

        void setStatus(const map_t &statusmap)
        {
                sag_set_user_status_map(connectivityManager, reinterpret_cast<const sag_underlying_map_t&>(statusmap));
                // keep track of what keys we've added so we can automatically remove them when we're destroyed
                for (auto it = statusmap.cbegin(); it != statusmap.cend(); it++)
                        if (it->first.type_tag() == SAG_DATA_STRING) // ignore invalid ones
                        {
                        if (it->second.empty())
                                mapKeysToCleanup.erase(it->first);
                        else
                                mapKeysToCleanup.insert(it->first.copy(), data_t());
                        }
        }



        ~StatusReporter()
        {
                if (!mapKeysToCleanup.empty())
                        sag_set_user_status_map(connectivityManager, reinterpret_cast<const sag_underlying_map_t&>(mapKeysToCleanup));
        }

        explicit StatusReporter(void *connectivityManager) : connectivityManager(connectivityManager), mapKeysToCleanup()
        {
        }

private:
        void* connectivityManager;
        map_t mapKeysToCleanup;


        StatusReporter() = delete;

        // non-copyable, due to the cleanup code we don't want people to shoot themselves in the foot by having multiple copies
        StatusReporter(const StatusReporter& other) = delete; 
        StatusReporter& operator=(const StatusReporter&) = delete; 

};

class Plugin
{
public:
        /* Constructor.
         * @since 9.12.0.1
         */
        Plugin(const PluginConstructorParameters &params)
                : pluginName(params.getPluginName()), chainId(params.getChainId()), config(params.getConfig().copy()),
                host(new PluginHost(params.chain)), logger("connectivity." + pluginName + "." + chainId), 
                statusReporter(new StatusReporter(params.connectivityManager))
        {
        }

        virtual ~Plugin() {}

        virtual void start() {}

        virtual void hostReady() {}

        virtual void shutdown() {}
        const std::string &getName() const { return pluginName; }
protected:
        const std::string pluginName;
        const std::string chainId;
        map_t config;
        const PluginHost::ptr_t host;
        StatusReporter &getStatusReporter() {
                if (statusReporter) return *statusReporter;
                throw std::runtime_error("Cannot call getStatusReporter when using the legacy constructor");
        }

public:
        Logger logger;
private:
        std::unique_ptr<StatusReporter> statusReporter;
};

class HostSide
{
public:
        virtual ~HostSide() {}
        typedef std::unique_ptr<HostSide> ptr_t;
        virtual void sendBatchTowardsHost(Message *start, Message *end) = 0;

        void sendBatchTowardsHost(Message &&message) {
                sendBatchTowardsHost(&message, &message+1);
        }

        template<typename IT>
        auto sendBatchTowardsHost(const IT &begin, const IT &end) -> typename std::enable_if<
                !std::is_const<ampl::remove_ref_t<decltype(*begin)>>::value &&
                ampl::is_same<Message, ampl::remove_const_t<ampl::remove_ref_t<decltype(*begin)>>>::value
                , void>::type // this ensures we can only pass in non-const iterator pairs to Message
        {
                if(begin == end) sendBatchTowardsHost((Message*) nullptr, (Message*) nullptr);
                else sendBatchTowardsHost(&(*begin), (&(*(end-1)))+1);
        }
};

class RemoteHostSide: public HostSide
{
public:
        RemoteHostSide(sag_plugin_t other, sag_send_fn_t fn): other(other), fn(fn) {}
        virtual ~RemoteHostSide() {}
        virtual void sendBatchTowardsHost(Message *start, Message *end);
private:
        sag_plugin_t other;
        sag_send_fn_t fn;
};

class TransportSide
{
public:
        virtual ~TransportSide() {}
        typedef std::unique_ptr<TransportSide> ptr_t;
        virtual void sendBatchTowardsTransport(Message *start, Message *end) = 0;

        void sendBatchTowardsTransport(Message &&message) {
                sendBatchTowardsTransport(&message, &message+1);
        }

        template<typename IT>
        auto sendBatchTowardsTransport(const IT &begin, const IT &end) -> typename std::enable_if<
                !std::is_const<ampl::remove_ref_t<decltype(*begin)>>::value &&
                ampl::is_same<Message, ampl::remove_const_t<ampl::remove_ref_t<decltype(*begin)>>>::value
                , void>::type // this ensures we can only pass in non-const iterator pairs to Message
        {
                if(begin == end) sendBatchTowardsTransport((Message*) nullptr, (Message*) nullptr);
                else sendBatchTowardsTransport(&(*begin), (&(*(end-1)))+1);
        }
};

class RemoteTransportSide: public TransportSide
{
public:
        RemoteTransportSide(sag_plugin_t other, sag_send_fn_t fn): other(other), fn(fn) {}
        virtual ~RemoteTransportSide() {}
        virtual void sendBatchTowardsTransport(Message *start, Message *end);
private:
        sag_plugin_t other;
        sag_send_fn_t fn;
};


class AbstractCodec: public Plugin, public HostSide, public TransportSide
{
public:
        typedef PluginConstructorParameters::CodecConstructorParameters CodecConstructorParameters;

        AbstractCodec(const PluginConstructorParameters::CodecConstructorParameters &params)
                : Plugin(params)
        {}

        // These methods do not need to show up in doxygen
        /* Called between construction and start() to set the hostSide field  */
        virtual void setNextTowardsHost(HostSide::ptr_t &&host)
        {
                hostSide = std::move(host);
        }
        /* Called between construction and start() to set the transportSide field */
        virtual void setNextTowardsTransport(TransportSide::ptr_t &&transport)
        {
                transportSide = std::move(transport);
        }
protected:
        HostSide::ptr_t hostSide;

        TransportSide::ptr_t transportSide;
};

class AbstractTransport: public Plugin, public TransportSide
{
public:
        typedef PluginConstructorParameters::TransportConstructorParameters TransportConstructorParameters;

        AbstractTransport(const PluginConstructorParameters::TransportConstructorParameters &params)
                : Plugin(params)
        {}

        // This method does not need to show up in doxygen
        /* Called between construction and start() to set the hostSide field */
        virtual void setNextTowardsHost(HostSide::ptr_t &&host)
        {
                hostSide = std::move(host);
        }

protected:
        HostSide::ptr_t hostSide;
};

class AbstractSimpleTransport: public AbstractTransport
{
public:

        AbstractSimpleTransport(const PluginConstructorParameters::TransportConstructorParameters &params)
                : AbstractTransport(params)
        {}


        virtual void sendBatchTowardsTransport(Message *start, Message *end)
        {
                for (Message *it = start; it != end; ++it) {
                        try {
                                if (it->getPayload().empty()) {
                                        deliverNullPayloadTowardsTransport(*it);
                                } else {
                                        deliverMessageTowardsTransport(*it);
                                }
                        } catch (...) {
                                handleException(*it);
                        }
                }
        }
        virtual void deliverMessageTowardsTransport(Message &msg) = 0;
        virtual void deliverNullPayloadTowardsTransport(Message &msg) 
        {
                // do nothing
        }

        virtual void handleException(Message &m)
        {
                try {
                        throw;
                } catch (const std::exception &e) {
                        logger.warn("Error while delivering message: %s; %s will be dropped.", e.what(), to_string(m).c_str());
                } catch (...) {
                        logger.warn("Unknown error delivering message: %s", to_string(m).c_str());
                }
        }
};

class AbstractSimpleCodec: public AbstractCodec
{
public:

        AbstractSimpleCodec(const PluginConstructorParameters::CodecConstructorParameters &params)
                : AbstractCodec(params)
        {}

        virtual void sendBatchTowardsHost(Message *start, Message *end)
        {
                Message *curr = start;
                for (Message *it = start; it != end; ++it) {
                        bool rv;
                        try {
                                if (it->getPayload().empty()) {
                                        rv = transformNullPayloadTowardsHost(*it);
                                } else {
                                        rv = transformMessageTowardsHost(*it);
                                }
                        } catch (...) {
                                rv = handleException(*it, false);
                        }
                        // if we keep it (and it didn't throw) swap it with the accumulator
                        if (rv) {
                                if (it != curr) it->swap(std::move(*curr)); // don't self-swap
                                ++curr;
                        }
                }
                if (hostSide.get()) hostSide->sendBatchTowardsHost(start, curr);
        }
        virtual void sendBatchTowardsTransport(Message *start, Message *end)
        {
                Message *curr = start;
                for (Message *it = start; it != end; ++it) {
                        bool rv;
                        try {
                                // process the message
                                if (it->getPayload().empty()) {
                                        rv = transformNullPayloadTowardsTransport(*it);
                                } else {
                                        rv = transformMessageTowardsTransport(*it);
                                }
                        } catch (...) {
                                rv = handleException(*it, true);
                        }
                        // if we keep it (and it didn't throw) swap it with the accumulator
                        if (rv) {
                                if (it != curr) it->swap(std::move(*curr)); // don't self-swap
                                ++curr;
                        }
                }
                if (transportSide.get()) transportSide->sendBatchTowardsTransport(start, curr);
        }

        virtual bool transformMessageTowardsHost(Message &msg) = 0;
        virtual bool transformMessageTowardsTransport(Message &msg) = 0;
        virtual bool transformNullPayloadTowardsHost(Message &msg)
        {
                // do nothing
                return true;
        }
        virtual bool transformNullPayloadTowardsTransport(Message &msg)
        {
                // do nothing
                return true;
        }
        virtual bool handleException(Message &m, bool towardsTransport)
        {
                try {
                        throw;
                } catch (const std::exception &e) {
                        logger.warn("Error while transforming message: %s; %s will be dropped.", e.what(), to_string(m).c_str());
                } catch (...) {
                        logger.warn("Unknown error transforming message: %s", to_string(m).c_str());
                }
                return false;
        }
};

}

namespace connectivity { using namespace _DATAT_INTERNAL_CPP_NAMESPACE; }

}} // com.softwareag.connectivity

// internal implementation included from these files
#include <sag_internal/exception.hpp>
#include <sag_internal/remote_plugins.hpp>
#include <sag_internal/plugin_macros.hpp>

#define SAG_DECLARE_CONNECTIVITY_TRANSPORT_CLASS(Class) _SAG_DECLARE_CONNECTIVITY_TRANSPORT_CLASS(Class)

#define SAG_DECLARE_CONNECTIVITY_CODEC_CLASS(Class) _SAG_DECLARE_CONNECTIVITY_CODEC_CLASS(Class)

#endif // _SAG_CONNECTIVITY_PLUGINS_HPP_