Creating Custom C++ Functions

<streambase-configuration>
    <global>
        <plugin file="./nth_value"/>
    </global>
    <server>
        <param name="tcp-port" value="10000"/>
    </server>
</streambase-configuration>

To see what happens when the aggregate function throws an exception, try enqueueing a number to the input_that_will_cause_an_error stream.

To run this sample on a machine where StreamBase is installed:

To see what happens when the aggregate function throws an exception, try enqueueing a number to the input_that_will_cause_an_error stream.

As the simple function sample shows, the basic steps are to writing a simple function are first to include some headers:

#include "StreamBase.hpp"
#include "PluginFunction.hpp"

You need to define some name spaces:

using namespace std;
using namespace sb;

Then define the class name and have it inherit from PluginFunction. For example:

class MySimpleFunction : public PluginFunction

Two virtual methods are required in simple function classes. They are:

Within the simple function class, you must declare the class to StreamBase:

This declaration provides some infrastructure that aides in registering the simple function class for use by StreamBase.

Finally, outside of the simple function class definition, you must "define" the class and what it will be known as in the StreamBase application:

This definition provides a means to register your simple function and the mapping between the simple function class name and what you want to call this function in your StreamBase application.

This section describes guidelines for writing custom C++ aggregate functions.

Creating aggregate functions is very similar to simple functions. Again, you need to include some headers:

And then define name spaces:

Then define the class name and have it inherit from PluginAggregate (instead of PluginFunction):

class MyAggregateFunction : public PluginAggregate

There are four required methods in an aggregate function:

Within the aggregate function class, you must "declare" the class to StreamBase:

Finally, outside of the aggregate function class definition, you must "define" the class and what it will be known as in the StreamBase application:

When you write custom functions by extending the StreamBase C++ API classes, your function object may use internal fields to record the data types with which it was typechecked. Note that functions can be polymorphic (take different argument data types), and they can also take a variable number of arguments.

To specify variable number of arguments, call arg_types.size() in your typecheck method.

If your function is polymorphic, then it is probably necessary to record the type information provided at typecheck time.This recorded information can be used in the implementation of the eval() method, to vary behavior based on the types of the arguments. The object is guaranteed to have been typechecked before eval() is called, and to receive arguments to eval() that match the types passed to typecheck().

The following code example returns the sum of the integer or double arguments, as a double. If there are no arguments, it returns 0 using the setIntValue method. The eval method is used to verify the data type of inputs. It :

class IntDoubleSumFunction : public PluginFunction {
  private:
    unsigned int _arg_count;
  public:
     typecheck(const Schema &argSchema)
    {
        _arg_count = argSchema.getNumFields();
        for (unsigned int i = 0; i < _arg_count; ++i)
            requireType(argSchema, i, DataType::INT, DataType::DOUBLE);
        setReturnType(DataType::DOUBLE);
    }
    virtual void eval(Tuple &retval, ConstTuple &args)
    {
        if (_arg_count == 0)
            retval.setDouble(0, 0);
        else {
            double ret = 0.0;
            for (unsigned int i = 0; i < _arg_count; ++i) {
                DataType dt = args.getSchema().getField(i).getType();

                if(DataType::DOUBLE == dt) {
                    ret += args.getDouble(i);
                } else if(DataType::INT == dt) {
                    ret += args.getInt(i);
                }
            }
            retval.setDouble(0, ret);
        }
    }
    STREAMBASE_DECLARE_PLUGIN_FUNCTION(IntDoubleSumFunction);
};
STREAMBASE_DEFINE_PLUGIN_FUNCTION(IntDoubleSumFunction, "int_double_sum");

Strings passed between the runtime and C++ plug-ins are of fixed size. In your C++ custom function, you can use one of these two methods to control string length:

To understand how to build custom C++ functions on UNIX, please see the Makefile installed in the following StreamBase directories. The default locations are shown here:

/opt/streambase/sample/custom-aggregate-function

/opt/streambase/sample/custom-simple-function

Also refer to the related section, Configuration Steps, in this topic.

This section explains how to build StreamBase custom C++ functions on Windows.

On a Windows machine where StreamBase is installed, if you use Microsoft Visual Studio, configure it as follows:

Also see the Configuration Steps for an important reminder.

See the StreamBase Server Configuration XML topic for details.