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-2016 Software AG, Darmstadt, Germany and/or Software AG USA Inc., Reston, VA, USA, and/or its subsidiaries 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 Software AG
 * @Version:     $Id: sag_connectivity_plugins.hpp 291193 2016-09-21 11:56:23Z bsp $
 */

#ifndef _SAG_CONNECTIVITY_PLUGINS_HPP_
#define _SAG_CONNECTIVITY_PLUGINS_HPP_

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

namespace com {
namespace softwareag {
namespace connectivity { 

class CacheOwner
{
        friend class CacheForThisThread;
public:
        CacheOwner();
        ~CacheOwner();
private:
        void *cache;
        bool inuse;
};

class CacheForThisThread
{
public:
        CacheForThisThread(CacheOwner &owner);
        ~CacheForThisThread();
private:
        CacheOwner &owner;
};



class Logger
{
public:
        typedef sag_log_level_t level_t;
        
        Logger(const std::string &category): category("AP_LOG_"+category), prefix(category) {}
        template<level_t LEVEL, typename... ARGS>
        void log(const char *format, ARGS... args) const;

        template<typename... ARGS>
        void log(level_t LEVEL, const char *format, ARGS... args) const;
        template<level_t LEVEL, typename... ARGS>
        void log(const std::string &format, ARGS... args) const;

        template<typename... ARGS>
        void log(level_t LEVEL, const std::string &format, ARGS... args) const;


        bool isEnabled(level_t LEVEL) const
        {
                return LEVEL <= getLevel();
        }

        level_t getLevel() const;

        bool isCritEnabled() const { return isEnabled(SAG_LOG_CRIT); }
        bool isFatalEnabled() const { return isEnabled(SAG_LOG_FATAL); }
        bool isErrorEnabled() const { return isEnabled(SAG_LOG_ERROR); }
        bool isWarnEnabled() const { return isEnabled(SAG_LOG_WARN); }
        bool isInfoEnabled() const { return isEnabled(SAG_LOG_INFO); }
        bool isDebugEnabled() const { return isEnabled(SAG_LOG_DEBUG); }
        bool isTraceEnabled() const { return isEnabled(SAG_LOG_TRACE); }

        template<typename... ARGS> void crit(const char *format, ARGS... args) const { log<SAG_LOG_CRIT>(format, args...); }
        template<typename... ARGS> void fatal(const char *format, ARGS... args) const { log<SAG_LOG_FATAL>(format, args...); }
        template<typename... ARGS> void error(const char *format, ARGS... args) const { log<SAG_LOG_ERROR>(format, args...); }
        template<typename... ARGS> void warn(const char *format, ARGS... args) const { log<SAG_LOG_WARN>(format, args...); }
        template<typename... ARGS> void info(const char *format, ARGS... args) const { log<SAG_LOG_INFO>(format, args...); }
        template<typename... ARGS> void debug(const char *format, ARGS... args) const { log<SAG_LOG_DEBUG>(format, args...); }
        template<typename... ARGS> void trace(const char *format, ARGS... args) const { log<SAG_LOG_TRACE>(format, args...); }

        template<typename... ARGS> void crit(const std::string &format, ARGS... args) const { log<SAG_LOG_CRIT>(format, args...); }
        template<typename... ARGS> void fatal(const std::string &format, ARGS... args) const { log<SAG_LOG_FATAL>(format, args...); }
        template<typename... ARGS> void error(const std::string &format, ARGS... args) const { log<SAG_LOG_ERROR>(format, args...); }
        template<typename... ARGS> void warn(const std::string &format, ARGS... args) const { log<SAG_LOG_WARN>(format, args...); }
        template<typename... ARGS> void info(const std::string &format, ARGS... args) const { log<SAG_LOG_INFO>(format, args...); }
        template<typename... ARGS> void debug(const std::string &format, ARGS... args) const { log<SAG_LOG_DEBUG>(format, args...); }
        template<typename... ARGS> void trace(const std::string &format, ARGS... args) const { log<SAG_LOG_TRACE>(format, args...); }


private:
        std::string category;
        std::string prefix;
};



class PluginConstructorParameters
{
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* reserved)
                : chainId(chainId), pluginName(pluginName), config(config.copy()), connectivityManager(connectivityManager), reserved(reserved)
        {}

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

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)
        {}
};

class Plugin
{
public:
        /* Legacy constructor. 
        */
        Plugin(const std::string &pluginName, const std::string &chainId, const map_t &config)
                : name(pluginName), chainId(chainId), config(config.copy()), LOGGER(name + "." + chainId), logger(name + "." + chainId)
        {
                logger.info("The %s connectivity plug-in in chain '%s' is using the legacy (std::string&,std::string&,map_t&) constructor which will be deprecated in a future release. We recommend changing to use the recommended (PluginConstructorParameters&) constructor instead.", 
                        pluginName.c_str(), chainId.c_str());
        }

        /* Constructor.
        * @since 9.12.0.1
        */
        Plugin(const PluginConstructorParameters &params)
                : name(params.getPluginName()), chainId(params.getChainId()), config(params.getConfig().copy()), LOGGER(params.getPluginName() + "." + params.getChainId()), logger(name + "." + chainId)
        {}

        virtual ~Plugin() {}

        virtual void start() {}

        virtual void hostReady() {}

        virtual void shutdown() {}
        const std::string &getName() const { return name; }
protected:
        std::string name;
        std::string chainId;
        map_t config;
public:
        Logger LOGGER;

        Logger logger;
};

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

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::auto_ptr<TransportSide> ptr_t;
        virtual void sendBatchTowardsTransport(Message *start, Message *end) = 0;
};

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 std::string &pluginName, const std::string &chainId, const map_t &config)
                : Plugin(pluginName, chainId, config)
        {}

        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 std::string &name, const std::string &chainId, const map_t &config)
                : Plugin(name, chainId, config) 
        {}

        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 std::string &name, const std::string &chainId, const map_t &config)
                : AbstractTransport(name, chainId, config)
        {}

        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("Caught exception delivering message: %s", e.what());
                } catch (...) {
                        logger.warn("Caught unknown exception delivering message: %s");
                }
        }
};

class AbstractSimpleCodec: public AbstractCodec
{
public:
        AbstractSimpleCodec(const std::string &name, const std::string &chainId, const map_t &config)
                : AbstractCodec(name, chainId, config) 
        {}

        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("Caught exception transforming message: %s", e.what());
                } catch (...) {
                        logger.warn("Caught unknown exception transforming message");
                }
                return false;
        }
};

}}} // 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>
#include <sag_internal/logging.hpp>

#define SAG_DECLARE_CONNECTIVITY_TRANSPORT_CLASS(Class) _SAG_DECLARE_CONNECTIVITY_TRANSPORT_CLASS(Class)

#define SAG_DECLARE_CONNECTIVITY_TRANSPORT(Class) _SAG_DECLARE_CONNECTIVITY_TRANSPORT_LEGACY(Class)

#define SAG_DECLARE_CONNECTIVITY_CODEC_CLASS(Class) _SAG_DECLARE_CONNECTIVITY_CODEC_CLASS(Class)

#define SAG_DECLARE_CONNECTIVITY_CODEC(Class) _SAG_DECLARE_CONNECTIVITY_CODEC_LEGACY(Class)

#endif // _SAG_CONNECTIVITY_PLUGINS_HPP_