PluginAggregate.hpp

//
// Copyright (c) 2004-2017 Cloud Software Group, Inc. All rights reserved.
//

#ifndef STREAMBASE_PLUGIN_AGGREGATE_H
#define STREAMBASE_PLUGIN_AGGREGATE_H

#include "StreamBase.hpp"
#include "StreamBaseVersion.hpp"

#include "Plugin.hpp"
#include "DataType.hpp"
#include "PluginException.hpp"
#include "PluginRegistry.hpp"

SB_NAMESPACE_BEGIN;

class PluginAggregateRep;

/// A plugin aggregate function.
class SB_EXPORT PluginAggregate : public Plugin {
  public:
    /// Virtual destructor.
    virtual ~PluginAggregate() {}

    /// Check to make sure the argument types are valid for this
    /// aggregate function.
    /// @param arg_types The types of the arguments to this function.
    /// @return the type of the return value.
    /// @throw TypecheckException if the argument types are not
    /// appropriate.
    virtual void typecheck(const Schema &arg_types) {}

    /// Initialize any window state. Clear the window values, if any.
    /// Instances of PluginAggregate may be reused for multiple
    /// windows.
    virtual void initialize() = 0;

    /// Add the values passed as parameters to the window state.
    /// initialize() will be called first.
    virtual void increment(const Tuple &args) = 0;

    /// Calculate the value of the aggregate over the values in the window.
    /// initialize() will be called first. No guarantee that increment()
    /// will ever be called.
    virtual void calculate(Tuple &retval) = 0;

    ///
    /// Clear internal state. This method is called after this use of the aggregate
    /// is finished.
    /// 
    virtual void release() { }

    /// Return a binary string containing the checkpoint state of this
    /// object.  This string will be passed to the load() method if a
    /// checkpoint is loaded.
    virtual std::string save() { return std::string(); }

    /// Recover the state of this object from the argument, a binary
    /// string that was returned by a previous invocation of save()
    virtual void load(std::string data) {}
};

typedef PluginRegistryInfo<PluginAggregate> PluginAggregateRegistryInfo;

void SB_EXPORT plugin_aggregate_callback(const PluginAggregateRegistryInfo &info);

#ifndef DOXYGEN_SKIP
template<>
struct PluginRegistryCallback<PluginAggregateRegistryInfo> {
    void operator()(const PluginAggregateRegistryInfo &info)
    {
        plugin_aggregate_callback(info);
    }
};

typedef PluginRegistryCallback<PluginAggregateRegistryInfo> PluginAggregateRegistryCallback;

typedef PluginRegistry<PluginAggregate,
                       PluginAggregateRegistryInfo,
                       PluginAggregateRegistryCallback> PluginAggregateRegistry;

#ifdef WIN32

#ifndef STREAMBASE_PLUGIN_VERSION_DEFINED
#define STREAMBASE_PLUGIN_VERSION_DEFINED
// this function is used to determine what streambase version the plugin was compiled against
extern "C" __declspec( dllexport ) inline const char*streambase_plugin_version() {return ((char *)&StreamBaseVersion::INFO_LINE);}
#endif

// There is a limitation where most stl containers can not be
// exported across a dll boundary. To get plugs - which really need to be dlls - to work, the
// registration mechanism can not require anything with stl containers to be exported.
// The current mechanism exposes PluginRegistry (and maybe others) which uses map.
// So on windows, the registration mechanism is an exported class called PluginRegistryAggregate
// (and one for each type of registred object) that does not directly reference the user
// defined plugin class name.
class SB_EXPORT PluginRegistryAggregate {
    public:
        static bool Add(PluginRegistry<PluginAggregate, PluginRegistryInfo<PluginAggregate>, PluginRegistryCallback< PluginRegistryInfo<PluginAggregate> > >::Info* streambase_plugin_info) {
            return (PluginRegistry<PluginAggregate>::get().add(*streambase_plugin_info), true);
        }
};

#define STREAMBASE_DEFINE_PLUGIN_AGGREGATE(ClassName, FuncName) \
    const SimplePluginRegistryInfo<PluginAggregate, ClassName> ClassName::_streambase_plugin_info(FuncName); \
    bool ClassName::_streambase_plugin_registered = \
    PluginRegistryAggregate::Add((PluginRegistryInfo<PluginAggregate>*)&(ClassName::_streambase_plugin_info));

#define STREAMBASE_DECLARE_PLUGIN_AGGREGATE(ClassName) \
    public: \
        static const SimplePluginRegistryInfo<PluginAggregate, ClassName> _streambase_plugin_info; \
        static bool _streambase_plugin_registered;

#else
#define STREAMBASE_DECLARE_PLUGIN_AGGREGATE(ClassName) \
    STREAMBASE_DECLARE_PLUGIN_REG_CLASS(PluginAggregate, ClassName)

#define STREAMBASE_DEFINE_PLUGIN_AGGREGATE(ClassName, FuncName) \
    STREAMBASE_DEFINE_PLUGIN_REG_CLASS_WITH_KEY(PluginAggregate, ClassName, FuncName)
#endif

#endif // !defined(DOXYGEN_SKIP)

SB_NAMESPACE_END;

#endif //PLUGIN_AGGREGATE_H