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 272666 2016-01-11 13:30:16Z rit $
 */

#ifndef _SAG_CONNECTIVITY_PLUGINS_HPP_
#define _SAG_CONNECTIVITY_PLUGINS_HPP_

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

namespace com {
namespace softwareag {
namespace connectivity { 

class Logger
{
public:
        typedef sag_log_level_t level_t;
        
        Logger(const std::string &category): category(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 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...); }

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

private:
        std::string category;
};

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

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:
        AbstractCodec(const std::string &name, const std::string &chainId, const map_t &config)
                : Plugin(name, chainId, config) 
        {}
        virtual void setNextTowardsHost(HostSide::ptr_t &&host)
        {
                hostSide = std::move(host);
        }
        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:
        AbstractTransport(const std::string &name, const std::string &chainId, const map_t &config)
                : Plugin(name, chainId, config) 
        {}
        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) 
        {}
        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) 
        {}
        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) _SAG_DECLARE_CONNECTIVITY_TRANSPORT(Class)

#define SAG_DECLARE_CONNECTIVITY_CODEC(Class) _SAG_DECLARE_CONNECTIVITY_CODEC(Class)

#endif // _SAG_CONNECTIVITY_PLUGINS_HPP_