Writing EPL plug-ins in C++

Creating a plug-in using C++

The APIs for writing plug-ins in C++ are all documented in the API reference for C++ (Doxygen). The relevant classes are in the com::apama::epl:: namespace, provided in the epl_plugin.hpp header file.

To create a plug-in for EPL in C++, you have to create a class which inherits from EPLPlugin. EPLPlugin is a class template with one template parameter, which should be the derived class which implements your plug-in. Your class must provide a zero-argument constructor and a static initialize method which takes a base_plugin_t::method_data_t & argument. For example:

class MyPlugin: public EPLPlugin<MyPlugin>
{
public:
    MyPlugin(): base_plugin_t("MyPlugin")
    {}
    static void initialize(base_plugin_t::method_data_t &md)
    {
    }
};

base_plugin_t is a convenience typedef to the EPLPlugin base class. The base class constructor takes a single argument of descriptive string which is used for logging purposes. The base class provides two members to derived classes:

A single instance of your class is created when the plug-in is loaded. Functions which you want to expose to EPL are member functions on that instance. To export a function to EPL, you need to declare a member function on the class with argument and return types which can be translated into EPL (see Method signatures). The function is exported into EPL by calling registerMethod on the method_data_t passed to initialize:

static void initialize(base_plugin_t::method_data_t &md)
{
    md.registerMethod<decltype(&MyPlugin::doSomething),
       &MyPlugin::doSomething>("doSomething");
}
void doSomething(int64_t arg1, double arg2)
{
    //...
}

registerMethod is templated over the following:

• the type of the function to export, which can be shortcut with decltype() as shown above, and
• the function to export.

registerMethod takes a mandatory argument, which is the name of the function in EPL, and two optional arguments.

• If the method signatures in EPL are not deducible from the C++ arguments, then you must provide the EPL type signature of the method.
• There is a final Boolean parameter, which defaults to true. It should be true if the plug-in may block (see Blocking behavior of plug-ins) and false otherwise.

md.registerMethod<decltype(&MyPlugin::simpleSignature),
   &MyPlugin::simpleSignature>("simpleMethod");
md.registerMethod<decltype(&MyPlugin::simpleSignatureNonBlocking),
   &MyPlugin::simpleSignatureNonBlocking>("nonBlockingMethod", false);
md.registerMethod<decltype(&MyPlugin::complexSignature),
   &MyPlugin::complexSignature>("complexMethod",
      "action<sequence<string>, integer> returns string");

Finally, to export the plug-in, you have to use the APAMA_DECLARE_EPL_PLUGIN macro. If your class is in a namespace, then you have to put the macro inside the same namespace and give it the class name without qualification as an argument:

APAMA_DECLARE_EPL_PLUGIN(MyClass)

Plug-ins are loaded into the correlator by using the import statement in the monitor or event which wants to use the plug-in. The argument is the name of the library which contains the plug-in.

import "MyPlugin" as plugin;

Complete simple example

This example implements a plug-in which simply keeps a single global count.

#include <epl_plugin.hpp>
using namespace com::apama::epl;

class CountPlugin: public EPLPlugin<MyPlugin>
{
public:
    CountPlugin()
        : base_plugin_t("CountPlugin"),
        count(0)
    {}
    static void initialize(base_plugin_t::method_data_t &md)
    {
        md.registerMethod<decltype(&CountPlugin::incrementCount),
           &CountPlugin::incrementCount>("increment");
        md.registerMethod<decltype(&CountPlugin::getCount),
           &CountPlugin::getCount>("getCount");
    }
    int64_t getCount() { return count; }
    void incrementCount(int64_t n) { count += n; }
    // Note that this is not thread-safe; a real plug-in would
    // want to use some synchronization here
private:
    int64_t count;
};

monitor foo
{
    import "PluginLibrary" as counter;
    action onload()
    {
        on all A() {
            counter.increment(1);
            print "Current count: "+counter.getCount().toString();
        }
    }
}

Method signatures

Only methods which have arguments and return values which can be converted into EPL types may be exported. For some of the types, the equivalent EPL signature can automatically be deduced. For methods which only take and return those types, you do not need to supply the explicit type. For other types, you must provide the EPL signature when registering the function.

EPL method signatures take the following forms:

action<integer>
action<sequence<string>, integer>
action<> returns integer
action<dictionary<string, string>, string> returns string

The following tables lists the EPL types and their equivalent C++ types.

Automatically deducible types

Types requiring explicit signatures

Compiling C++ plug-ins

To compile a C++ plug-in, you need to do the following:

1. Add #include <epl_plugin.hpp> to the source file.
2. Put $APAMA_HOME/include on your compiler’s include path.
3. Link the plug-in with apclient.dll (Windows) or libapclient.so (Linux).
4. Put $APAMA_HOME/lib on your compiler’s linker path.
5. Enable C++11 standards mode.
6. Compile your class into a shared library .dll (Windows) or .so (Linux).

Using GCC on Linux this would look like this:

g++ --shared --std=c++0x -I$APAMA_HOME/include -L$APAMA_HOME/lib -lapclient -o libMyPlugin.so MyPlugin.cpp

You can find example makefiles (Linux) and Visual Studio project files (Windows) in the samples directory of your Apama installation.

The C++ EPL plug-in API is implemented using a C-only ABI, even though it has a C++ API. This means that there is no requirement to compile your plug-in library with a specific compiler or compiler version. However, the compiler must support the C++ 11 syntax used in the API header files. We recommend use of the compilers specified in the Supported Platforms document for the current version.

If you are building a shared library to be used by multiple plug-ins and using the plug-in-specific data structures as part of your API between the library and the plug-ins, then you must ensure that the library and all of the plug-ins are compiled using the same version of the Apama header files. This means that if you upgrade Apama and want to recompile one of them, you must recompile all of them. You can choose not to recompile anything and they will still work.

If you compile with headers from multiple service packs of Apama, then you may see errors similar to the following when you try to link them.

• Linux :

   undefined reference to `Foo::test(com::softwareag::connectivity10_5_3::data_t const&)'
  

• Windows:

   testlib2.obj : error LNK2019: unresolved external symbol "public: void __cdecl Foo::test(class com::softwareag::connectivity10_5_3::data_t const &)" (?test@Foo@@QEAAXAEBVdata_t@connectivity10_5_3@softwareag@com@@@Z) referenced in function "public: void __cdecl Bar::test(class com::softwareag::connectivity10_5_3::data_t const &)" (?test@Bar@@QEAAXAEBVdata_t@connectivity10_5_3@softwareag@com@@@Z)
  
  testlib2.dll : fatal error LNK1120: 1 unresolved externals
  

If you encounter a similar error, try recompiling all your components with the same version of the headers.

If you are compiling a single plug-in, or multiple completely independent plug-ins, you can recompile them in any combination at any time.

Exceptions

Plug-ins can throw exceptions from plug-in methods. They must be derived from std::exception as normal.

If a plug-in throws an exception, this will be turned into an exception in EPL which may be caught with the EPL try { } catch syntax.

If you do not catch the exception, then the calling monitor instance will be terminated with the message in the exception thrown from the plug-in.

Writing to the correlator log file

The EPLPlugin base class provides a member called logger to plug-ins. This can be used to write messages to the correlator’s log file.

Log messages are prefixed with the string plugins.*PluginName*, which is also the category that can be used to control the log level for this plug-in via the correlatorLogging section in the YAML configuration file (see Setting correlator and plug-in log files and log levels in a YAML configuration file).

void logMessage(const char *msg) {
    logger.info("Message is: %s", msg);
}

Documentation about the available functions and log levels can be found in the API reference for C++ (Doxygen).

Storing data using chunks

By themselves plug-ins can only store global state which must be protected by threading primitives from access from multiple EPL contexts simultaneously. It is possible for plug-ins to store state in an opaque (to EPL) object. This can be stored specific to a single monitor instance and then passed back into the plug-in for further processing. In EPL, this is represented using a chunk object. Chunk objects are specific to a single plug-in, and an attempt to pass them to other plug-ins will throw an exception.

You can store any C++ type from your plug-in in a chunk, provided that it is copyable using the default copy constructor. If you have multiple different types from a single plug-in stored in EPL chunks, then you must have them share a class hierarchy so that you can distinguish them when they are passed back from EPL.

The argument and return type that you must use to pass chunks is custom_t<T>. custom_t is templated over the type that you are storing in the chunk. Chunks can only be created by your plug-in and returned from a function to EPL. When they are passed back to your plug-in, you can modify the referred to class, but not change which object it points to. Chunk objects are garbage collected by the EPL runtime for you and will be deleted when they go out of scope. You do not retain ownership of them.

If you need lifetime shared between EPL and the plug-in (or another chunk object), then indirectly access the shared object via std::shared_ptr. The custom_t would then point to a std::shared_ptr or a class containing a std::shared_ptr, and the plug-in follows the shared_ptr to access the shared object. The plug-in can also have a std::shared_ptr to the shared object. The object will not be destroyed until both EPL has garbage collected the chunk object and the plug-in has destroyed its shared_ptr to the object.

class EPLData
{
public:
    EPLData(int64_t value): value(value) {}
    // default copy constructor and destructor
    int64_t value;
};

custom_t<EPLData> createChunk(int64_t value)
{
    return custom_t<EPLData>(new EPLData(value));
}
void logChunkValue(const custom_t<EPLData> &chunk)
{
    logger.info("Chunk value is: %" PRId64, chunk->value);
}

monitor m {
    import "DataPlugin" as plugin;
    action onload() {
        chunk c := plugin.createChunk(42);
        plugin.logChunkValue(c);
    }
}

Chunk objects can be copied in EPL. This occurs when the explicit clone() method is called on a chunk or an event containing the chunk, or if the monitor spawns. When a chunk object is copied, the value in the custom_t is copied, using the default copy constructor of the template parameter for custom_t. Thus, each instance will only be accessible from a single monitor instance and thus only from one thread at a time (though copies may be accessed concurrently to the original object).

If you have multiple different chunk types, then use the base class as the template parameter for custom_t and then use dynamic_cast or virtual dispatch to distinguish between them. When returning a custom_t value, you must use the derived type as the template parameter, because that is the type used to create a copy of the object, if needed.

class ChunkBase
{
    virtual void doSomething() = 0;
};
class DerivedChunk: public ChunkBase
{
};
void modifyChunk(const custom_t<ChunkBase> &chunk)
{
    chunk->doSomething(); // virtual dispatch
    auto derived = dynamic_cast<DerivedChunk*>(chunk.get());
    if (derived) {
        // derived-specific methods
    }
};

Sending events

Method calls on plug-ins are synchronous and generally should be written not to take too long or hold up processing in the correlator (see Blocking behavior of plug-ins). One technique to enable asynchronous behavior in the plug-in is to interact with EPL by sending events which can be handled asynchronously, potentially from a background thread which is processing events as well as from methods themselves.

The CorrelatorInterface returned from getCorrelator() contains several methods for sending events into the correlator. You can send events either as the string representation of the event in Apama’s internal string format (for example, MyEvent(1.3, true)) or as a map_t type where the keys are strings corresponding to the field names in the event, and the values are the values for those fields, in the appropriate type/format for the type of the field. In the latter case, you also need to supply the name of the event type to parse the map as.

You can select the destination of the event in several ways:

• Named channel. The preferred method is to deliver the event to a specific named channel. This will go to everything which has subscribed to that named channel. Subscribers can either be context, external receivers, connectivity chains or other plug-ins which are subscribed to receive events on that channel (see Receiving events).

  Send as string:

  getCorrelator().sendEventTo("MyEvent(42)", "channelName");
  

  Send as object:

  map_t event;
  event.insert(data_t("number"), data_t(int64_t(42));
  
  getCorrelator().sendEventTo("MyEvent", std::move(event), "channelName");
  

• Context by ID. You can deliver the event to a specific context by context ID. Context IDs can either be passed into a plug-in from an EPL context() object, or they can be obtained with the CorrelatorInterface.getCurrentContextId() method.

  Send as string:

  getCorrelator().sendEventTo("MyEvent(42)", ctxId);
  

  Send as object:

  map_t event;
  event.insert(data_t("number"), data_t(int64_t(42));
  
  getCorrelator().sendEventTo("MyEvent", std::move(event), ctxId);
  

You can send events from within a method called from EPL, from any callback handler method, or from threads spawned by the plug-in itself.

Receiving events

Plug-ins can subscribe to receive events which are delivered to certain named channels. They will receive events sent to those channels from EPL, from external senders, from connectivity chains or from other plug-ins (see Sending events). To subscribe to events, you need to subclass the EventHandler class and implement the handleEvent() virtual method. Then you need to call registerEventHandler on the CorrelatorInterface returned by getCorrelator() with an instance of your EventHandler. registerEventHandler takes ownership of the handler object.

class MyHandler: public EventHandler
{
    virtual void handleEvent(const char *type, data_t &&event,
      const char *channel)
    {
        // do something with event
    }
};
MyPlugin(): base_plugin_t("MyPlugin") {
    EventHandler::subscription_t handler = getCorrelator().registerEventHandler(
      MyHandler::ptr_t(new MyHandler()), "channelName",
      /*mode=*/MAP_MODE, /*blocking=*/true);
}

registerEventHandler() parameters

registerEventHandler() has mandatory and optional parameters:

handleEvent() parameters

handleEvent() is called with the following parameters:

EventHandler::subscription_t methods

registerEventHandler() returns an object of type EventHandler::subscription_t, which can be used to add channels to a subscription, remove channels from a subscription, or remove the handler entirely. You must not call either of the remove methods from inside the handler object.

EventHandler::subscription_t has the following methods:

Blocking behavior of plug-ins

The correlator system assumes that all functions return in a reasonable amount of time, and do not do potentially blocking operations such as file-system operations or remote calls. For code written in EPL, the correlator can enforce this. For code provided in plug-ins, the correlator cannot enforce this or know whether the method may block. Therefore, by default, the correlator assumes that any plug-in method may block indefinitely. This may cause the correlator to create new operating system threads to service incoming events on other contexts.

If you know that your plug-in does not do anything long-running or potentially blocking, then you may declare the method as “non-blocking” at the point it is registered in the initialize function:

static void initialize(base_plugin_t::method_data_t &md)
{
   md.registerMethod<decltype(&MyPlugin::nonBlockingMethod),
      &MyPlugin::nonBlockingMethod>("methodName", /*blocking=*/false);
}

For event handlers, this is done with the call to registerEventHandler (see Receiving events).

If you do this, then the correlator assumes that the method will return soon and not spawn additional threads. This can avoid extra overhead of starting and stopping operating system threads.

If you have a method which is normally non-blocking, but may sometimes block, you can declare it as non-blocking initially and then, when it encounters a condition which is blocking, notify the correlator that this call is blocked to allow the correlator to spawn additional threads.

int64_t get(int64_t key) {
    if (local(key)) {
        return local(key);
    } else {
        getCorrelator().pluginMethodBlocking();
        return remote(key);
    }
}

Load-time or unload-time code

If you need to execute code at the time the plug-in is loaded, then you should put it in the plug-in class constructor.

If you need to execute code at the time the plug-in is unloaded, then you should put it in the plug-in class destructor.

Typically, you will register any callbacks at load time in the constructor.

class MyPlugin: public EPLPlugin<MyPlugin>
{
    MyPlugin(): base_plugin_t("MyPlugin") {
        // load-time code here
    }
    ~MyPlugin() {
        // unload-time code here
    }
};

Handling thread-specific data in plug-ins

The correlator executes plug-ins on a number of internal threads which may change over time, including spawning new threads or destroying old threads. If a plug-in needs to store any state in thread-local variables, then - as well as cleaning up this state when the plug-in unloads - it also needs to clean it up when a thread is destroyed. To enable this, plug-ins can register to receive a callback whenever the correlator destroys a thread. This is done by subclassing ThreadEndedHandler with an implementation of threadEnded() and then passing an instance of your subclass to the receiveThreadEndedCallbacks() method on CorrelatorInterface. receiveThreadEndedCallbacks takes ownership of the handler and may only be called once per plug-in.

class MyThreadHandler: public ThreadEndedHandler {
public:
  virtual void threadEnded() {
      // cleanup code here
  }
};
MyPlugin(): base_plugin_t("MyPlugin")
{
  getCorrelator().receiveThreadEndedCallbacks(
    MyThreadHandler::ptr_t(new MyThreadHandler()));
}

Using plug-ins written in C++

The plug-in library you compiled must be available on the correlator’s PATH (Windows) or LD_LIBRARY_PATH (Linux). By default, the $APAMA_WORK/lib directory is included on the appropriate path, and we recommend that you use this location for your own plug-ins.

Plug-ins are loaded using the import statement in EPL:

monitor m {
    import "MyPlugin" as plugin;
}

This statement can be made at the top-level of either monitors or events.

The correlator then attempts to load MyPlugin.dll (Windows) or libMyPlugin.so (Linux) from the path. The correlator reads the list of exported methods from the plug-in. These methods are then available on the name provided in the as statement in the rest of that monitor or event:

event E {
    import "MyPlugin" as plugin;
    action increment(integer i) {
        plugin.increment(i);
    }
    action get() returns integer {
        return plugin.getCount();
    }
}