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 enqueuing 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 enqueuing 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 aids 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 namespaces:

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:

On a supported UNIX machine where StreamBase is installed, use the sb-config utility to set up the environment and define the compiler to use when compiling your program. For example:

CXX=`sb-config --cxx`
$CXX MyClient.cpp `sb-config --cflags` -c -o MyFunction.o
$CXX MyClient.o `sb-config --libs` -o MyFunction

Substitute the name of your client for MyFunction.

Use the Makefiles in the following StreamBase samples as a guide to setting up your projects:

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

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

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

To build your custom C++ function on Windows, you must configure Microsoft Visual Studio as described in Configuring Visual C++.

Also see the Configuration Steps for an important reminder.

See the StreamBase Server Configuration XML topic for details.