epl_plugin.hpp

Go to the documentation of this file.

/*
 * Title:        epl_plugin.hpp
 * Description:  C++ API for writing EPL plug-ins
 * $Copyright (c) 2015-2017, 2020 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: epl_plugin.hpp 377142 2020-07-16 10:18:39Z matj $
 */

#ifndef _APAMA_EPL_PLUGINS_HPP_
#define _APAMA_EPL_PLUGINS_HPP_

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

#include <sag_plugin_logging.hpp>
#include <vector>

namespace com {
namespace apama {
namespace correlator {
class CorrelatorInterface;
} // com.apama.correlator

namespace _EPLPLUGIN_INTERNAL_CPP_NAMESPACE { 

// import these types into this namespace for convenience

using com::softwareag::connectivity::decimal_t;
using com::softwareag::connectivity::data_t;
using com::softwareag::connectivity::get;
using com::softwareag::connectivity::list_t;
using com::softwareag::connectivity::map_t;
using com::softwareag::connectivity::buffer_t;
using com::softwareag::connectivity::custom_t;
using com::softwareag::connectivity::visitor;
using com::softwareag::connectivity::const_visitor;
using com::softwareag::connectivity::apply_visitor;
using com::softwareag::connectivity::Logger;

}}} // com.apama.epl


#include <sag_internal/epl_plugin_wrap.hpp>

namespace com {
namespace apama {
namespace _EPLPLUGIN_INTERNAL_CPP_NAMESPACE { 

template<typename DERIVED>
class MethodData
{
public:
        MethodData() {}
        size_t getNumMethods() { return methods.size(); }
        // @internal
        ~MethodData();
        template <typename FN_T, FN_T fn>
        void registerMethod(const char *name, bool blocking=true);
        template <typename FN_T, FN_T fn>
        void registerMethod(const char *name, const char *signature, bool blocking=true);
        ApamaEPLMethodData *getMethods();
private:
        std::vector<ApamaEPLMethodData> methods;
        MethodData(const MethodData &) = delete;
        MethodData &operator=(const MethodData &) = delete;
};

class EventHandlerSubscription
{
        friend class CorrelatorInterface;
public:
        EventHandlerSubscription(): internal(0), obj(0) {}
        EventHandlerSubscription(EventHandlerSubscription &&other)
                : internal(other.internal), obj(other.internal)
        {
                other.internal = 0;
                other.obj = 0;
        }
        EventHandlerSubscription &operator=(EventHandlerSubscription &&other)
        {
                internal = other.internal;
                obj = other.obj;
                other.internal = 0;
                other.obj = 0;
                return *this;
        }
        void addChannel(const char *channel);
        bool removeChannel(const char *channel);
        void removeAllChannels();
private:
        EventHandlerSubscription(void *internal, void *obj): internal(internal), obj(obj) {}
        void *internal;
        void *obj;
        EventHandlerSubscription(const EventHandlerSubscription&) = delete;
        EventHandlerSubscription &operator=(const EventHandlerSubscription &) = delete;
};

class EventHandler
{
public:
        typedef EventHandlerSubscription subscription_t;
        typedef std::unique_ptr<EventHandler> ptr_t;
        virtual ~EventHandler() {}
        virtual void handleEvent(const char *type, data_t &&event, const char *channel) = 0;
};

class ThreadEndedHandler
{
public:
        typedef std::unique_ptr<ThreadEndedHandler> ptr_t;
        virtual ~ThreadEndedHandler() {}
        virtual void threadEnded() = 0;
};

class CorrelatorInterface
{
public:
        CorrelatorInterface(void *internal): internal(internal) {}
        CorrelatorInterface(): internal(0) {}

        enum event_mode_t
        { 
                STRING_MODE, 
                MAP_MODE
        };

        void receiveThreadEndedCallbacks(std::unique_ptr<ThreadEndedHandler> &&handler) const;

        EventHandler::subscription_t registerEventHandler(std::unique_ptr<EventHandler> &&handler, const char *channel, event_mode_t mode=MAP_MODE, bool blocking=true) const;
        void sendEventTo(const std::string &event, uint64_t ctx) const
        {
                sendEventTo(event.c_str(), ctx);
        }
        void sendEventTo(const char *event, uint64_t ctx) const;
        void sendEventTo(const char *type, data_t &&event, uint64_t ctx) const;
        void sendEventTo(const std::string &event, const std::string &channel) const
        {
                sendEventTo(event.c_str(), channel.c_str());
        }
        void sendEventTo(const char *event, const char *channel) const;
        void sendEventTo(const char *type, data_t &&event, const char *channel) const;
        void pluginMethodBlocking() const;
        int64_t getCurrentContextId() const;
protected:
        void *internal;
private:
        static bool callEventHandler(void *handler, const char *type, sag_underlying_data_t event, const char *channel);
        static void deleteEventHandler(void *handler);
        static bool callThreadEndedHandler(void *handler);
        static void deleteThreadEndedHandler(void *handler);

};

template<typename DERIVED>
class EPLPlugin
{
private:
        typedef data_t (*call_method_fn_t)(DERIVED *self, const list_t &args);
public:
        typedef EPLPlugin base_plugin_t;
        typedef MethodData<DERIVED> method_data_t;

        EPLPlugin(const std::string &name):
                logger("plugins."+name)
        {}
        static CorrelatorInterface &getCorrelator() { return correlator; }
        static void setCorrelator(const CorrelatorInterface &corr) { correlator = corr; }
        Logger logger;

        static data_t callMethod(void *self, void *fn, const list_t &args);
private:
        DERIVED &derived() { return *static_cast<DERIVED*>(this); }
        static CorrelatorInterface correlator;
};

}
namespace epl { using namespace _EPLPLUGIN_INTERNAL_CPP_NAMESPACE; }

}} // com.apama.epl

// internal implementation included from these files
#include <sag_internal/epl_plugin_functions.hpp>
#include <sag_internal/epl_plugin_methods.hpp>
#include <sag_internal/exception.hpp>
#include <sag_internal/epl_plugin_macros.hpp>

#define APAMA_DECLARE_EPL_PLUGIN(Class) _APAMA_DECLARE_EPL_PLUGIN(Class)

#endif // _APAMA_EPL_PLUGINS_HPP_