www.espertech.comDocumentation

Chapter 21. Integration and Extension

21.1. Overview
21.2. Single-Row Function
21.2.1. Implementing a Single-Row Function
21.2.2. Configuring the Single-Row Function Name
21.2.3. Value Cache
21.2.4. Single-Row Functions in Filter Predicate Expressions
21.2.5. Single-Row Functions Taking Events as Parameters
21.2.6. Single-Row Functions Returning Events
21.2.7. Receiving a Context Object
21.2.8. Exception Handling
21.3. Virtual Data Window
21.3.1. How to Use
21.3.2. Implementing the Forge
21.3.3. Implementing the Factory-Factory
21.3.4. Implementing the Factory
21.3.5. Implementing the Virtual Data Window
21.4. Data Window View and Derived-Value View
21.4.1. Implementing a View Forge
21.4.2. Implementing a View Factory
21.4.3. Implementing a View
21.4.4. View Contract
21.4.5. Configuring View Namespace and Name
21.4.6. Requirement for Data Window Views
21.4.7. Requirement for Derived-Value Views
21.5. Aggregation Function
21.5.1. Aggregation Single-Function Development
21.5.2. Aggregation Multi-Function Development
21.6. Pattern Guard
21.6.1. Implementing a Guard Forge
21.6.2. Implementing a Guard Factory
21.6.3. Implementing a Guard Class
21.6.4. Configuring Guard Namespace and Name
21.7. Pattern Observer
21.7.1. Implementing an Observer Forge
21.7.2. Implementing an Observer Factory
21.7.3. Implementing an Observer Class
21.7.4. Configuring Observer Namespace and Name

This chapter summarizes integration and describes in detail each of the extension APIs that allow integrating external data and/or extend runtime functionality.

For information on calling external services via instance method invocation, for instance to integrate with dependency injection frameworks such as Spring or Guice, please see Section 5.17.5, “Class and Event-Type Variables”.

For information on input and output adapters that connect to an event transport and perform event transformation for incoming and outgoing on-the-wire event data, for use with streaming data, please see the EsperIO reference documentation. The data flow instances as described in Chapter 20, EPL Reference: Data Flow are an easy way to plug in operators that perform input and output. Data flows allow providing parameters and managing individual flows independent of runtime lifecycle. Also consider using the Plug-in Loader API for creating a new adapter that starts or stops as part of the CEP runtime initialization and destroy lifecycle, see Section 16.15, “Plug-In Loader”.

To join data that resides in a relational database and that is accessible via JDBC driver and SQL statement the runtime offers syntax for using SQL within EPL, see Section 5.13, “Accessing Relational Data via SQL”. A relational database input and output adapter for streaming input from and output to a relational database also exists (EsperIO).

To join data that resides in a non-relational store the runtime offers a two means: First, the virtual data window, as described below, for transparently integrating the external store as a named window. The second mechanism is a special join syntax based on static method invocation; see Section 5.14, “Accessing Non-Relational Data via Method, Script or UDF Invocation”.

Tip

The best way to test that your extension code works correctly is to write unit tests against a statement that utilizes the extension code. Samples can be obtained from Esper regression test code base.

Note

For all extension code and similar to listeners and subscribers, to send events into the runtime from extension code the routeEvent method should be used (and not sendEvent) to avoid the possibility of stack overflow due to event-callback looping and ensure correct processing of the current and routed event. Note that if outbound-threading is enabled, listeners and subscribers should use sendEvent and not routeEvent.

Note

For all extension code it is not safe to deploy and undeploy within the extension code. For example, it is not safe to implement a data window that deploys compiled modules and that undeploys deployments.

Single-row functions return a single value. They are not expected to aggregate rows but instead should be stateless functions. These functions can appear in any expressions and can be passed any number of parameters.

The following steps are required to develop and use a custom single-row function.

  1. Implement a class providing one or more public static methods accepting the number and type of parameters as required.

  2. Register the single-row function class and method name with the compiler by supplying a function name.

You may not override a built-in function with a single-row function provided by you. The single-row function you register must have a different name then any of the built-in functions.

An example single-row function can also be found in the examples under the runtime configuration example.

Use a virtual data window if you have a (large) external data store that you want to access as a named window. The access is transparent: There is no need to use special syntax or join syntax. All regular queries including subqueries, joins, on-merge, on-select, on-insert, on-delete, on-update and fire-and-forget are supported with virtual data windows.

There is no need to keep any data or events in memory with virtual data windows. The only requirement for virtual data windows is that all data rows returned are EventBean instances.

When implementing a virtual data window it is not necessary to send any events into the runtime or to use insert-into. The event content is simply assumed to exist and accessible to the runtime via the API implementation you provide.

The distribution ships with a sample virtual data window in the examples folder under the name virtualdw. The code snippets below are extracts from the example.

We use the term store here to mean a source set of data that is managed by the virtual data window. We use the term store row or just row to mean a single data item provided by the store. We use the term lookup to mean a read operation against the store returning zero, one or many rows.

Virtual data windows allow high-performance low-latency lookup by exposing all relevant statement access path information. This makes it possible for the virtual data window to choose the desired access method into its store.

The following steps are required to develop and use a virtual data window:

Once you have completed above steps, the virtual data window is ready to use in statements.

From a threading perspective, virtual data window implementation classes must be thread-safe if objects are shared between multiple named windows. If no objects are shared between multiple different named windows, thereby each object is only used for the same named window and other named windows receive a separate instance, it is no necessary that the implementation classes are thread-safe.

Your application must first register the virtual data window factory as part of configuration:

Configuration config = new Configuration();
config.getCompiler().addPlugInVirtualDataWindow("sample", "samplevdw", 
    SampleVirtualDataWindowForge.class.getName());

Your application may then create a named window backed by a virtual data window.

For example, assume that the SampleEvent event type is declared as follows:

create schema SampleEvent as (key1 string, key2 string, value1 int, value2 double)

The next statement creates a named window MySampleWindow that provides SampleEvent events and is backed by a virtual data window:

create window MySampleWindow.sample:samplevdw() as SampleEvent

You may then access the named window, same as any other named window, for example by subquery, join, on-action, fire-and-forget query or by consuming its insert and remove stream. While this example uses Map-type events, the example code is the same for POJO or other events.

Your application may obtain a reference to the virtual data window from the runtime context.

This code snippet looks up the virtual data window by the named window name:

try {
  return (VirtualDataWindow) runtime.getContext().lookup("/virtualdw/MySampleWindow");
}
catch (NamingException e) {
  throw new RuntimeException("Failed to look up virtual data window, is it created yet?");
}

When you application registers a subquery, join or on-action query or executes a fire-and-forget query against a virtual data window the runtime interacts with the virtual data window. The interaction is a two-step process.

At time of deployment (once), the runtime uses the information the compiler collected by analyzing the EPL where-clause, if present. It then creates a list of hash-index and binary tree (btree, i.e. sorted) index properties. It passes the property names that are queried as well as the operators (i.e. =, >, range etc.) to the virtual data window. The virtual data window returns a lookup strategy object to the runtime.

At time of statement execution (repeatedly as triggered), the runtime uses that lookup strategy object to execute a lookup. It passes to the lookup all actual key values (hash, btree including ranges) to make fast and efficient lookup achievable.

To explain in detail, assume that your application creates a statement with a subquery as follows:

select (select * from MySampleWindow where key1 = 'A1') from OtherEvent

At the time of compilation of the statement above the compiler analyzes the statement. It determines that the subquery queries a virtual data window. It determines from the where-clause that the lookup uses property key1 and hash-equals semantics. The runtime then provides this information as part of VirtualDataWindowLookupContext passed to the getLookup method. Your application may inspect hash and btree properties and may determine the appropriate store access method to use.

The hash and btree property lookup information is for informational purposes, to enable fast and performant queries that return the smallest number of rows possible. Your implementation classes may use some or none of the information provided and may also instead return some or perhaps even all rows, as is practical to your implementation. The where-clause still remains in effect and gets evaluated on all rows that are returned by the lookup strategy.

Following the above example, the sub-query executes once when a OtherEvent event arrives. At time of execution the runtime delivers the string value A1 to the VirtualDataWindowLookup lookup implementation provided by your application. The lookup object queries the store and returns store rows as EventBean instances.

As a second example, consider an EPL join statement as follows:

select * from MySampleWindow, MyTriggerEvent where key1 = trigger1 and key2 = trigger2

The compiler analyzes the statement and the runtime passes to the virtual data window the information that the lookup occurs on properties key1 and key2 under hash-equals semantics. When a MyTriggerEvent arrives, it passes the actual value of the trigger1 and trigger2 properties of the current MyTriggerEvent to the lookup.

As a last example, consider a fire-and-forget query as follows:

select * from MySampleWindow key1 = 'A2' and value1 between 0 and 1000

The compiler analyzes the statement and the runtime passes to the virtual data window the lookup information. The lookup occurs on property key1 under hash-equals semantics and on property value1 under btree-open-range semantics. When you application executes the fire-and-forget query the runtime passes A2 and the range endpoints 0 and 1000 to the lookup.

For more information, please consult the JavaDoc API documentation for class VirtualDataWindow, VirtualDataWindowLookupContext or VirtualDataWindowLookupFieldDesc.

For each named window that refers to the virtual data window, the runtime instantiates one instance of the forge at compile-time.

A virtual data window forge class is responsible for the following functions:

The compiler instantiates a VirtualDataWindowForge instance for each named window created by create window. The compiler invokes the initialize method once in respect to the named window being created passing a VirtualDataWindowForgeContext context object.

The sample code shown here can be found among the examples in the distribution under virtualdw:

public class SampleVirtualDataWindowForge implements VirtualDataWindowForge {

    public void initialize(VirtualDataWindowForgeContext initializeContext) {
    }

    public VirtualDataWindowFactoryMode getFactoryMode() {
        // The injection strategy defines how to obtain and configure the factory-factory.
        InjectionStrategy injectionStrategy = new InjectionStrategyClassNewInstance(SampleVirtualDataWindowFactoryFactory.class);
        
        // The managed-mode is the default. It uses the provided injection strategy.
        VirtualDataWindowFactoryModeManaged managed = new VirtualDataWindowFactoryModeManaged();
        managed.setInjectionStrategyFactoryFactory(injectionStrategy);
        
        return managed;
    }

    public Set<String> getUniqueKeyPropertyNames() {
        // lets assume there is no unique key property names
        return null;
    }
}

Your forge class must implement the getFactoryMode method which instructs the compiler how to obtain a factory class that returns a factory for creating virtual data window instances (a factory-factory). The class acting as the factory-factory will be SampleVirtualDataWindowFactoryFactory.

For each named window that refers to the virtual data window, the runtime instantiates one instance of the factory.

A virtual data window factory class is responsible for the following functions:

The runtime instantiates a VirtualDataWindowFactory instance for each named window created via create window. The runtime invokes the initialize method once in respect to the named window being created passing a VirtualDataWindowFactoryContext context object.

If not using contexts, the runtime calls the create method once after calling the initialize method. If using contexts, the runtime calls the create method every time it allocates a context partition. If using contexts and your virtual data window implementation operates thread-safe, you may return the same virtual data window implementation object for each context partition. If using contexts and your implementation object is not thread safe, return a separate thread-safe implementation object for each context partition.

The runtime invokes the destroy method once when the named window is undeployed. If not using contexts, the runtime calls the destroy method of the virtual data window implementation object before calling the destroy method on the factory object. If using contexts, the runtime calls the destroy method on each instance associates to a context partition at the time the associated context partition terminates.

The sample code shown here can be found among the examples in the distribution under virtualdw:

public class SampleVirtualDataWindowFactory implements VirtualDataWindowFactory {

    public void initialize(VirtualDataWindowFactoryContext factoryContext) {
    }

    public VirtualDataWindow create(VirtualDataWindowContext context) {
        return new SampleVirtualDataWindow(context);
    }

    public void destroy() {
        // cleanup can be performed here
    }

    public Set<String> getUniqueKeyPropertyNames() {
        // lets assume there is no unique key property names
        return null;
    }
}

Your factory class must implement the create method which receives a VirtualDataWindowContext object. This method is called once for each EPL that creates a virtual data window (see example create window above).

The VirtualDataWindowContext provides to your application:

String namedWindowName;	// Name of named window being created.
Object[] parameters;  // Any optional parameters provided as part of create-window.
EventType eventType;  // The event type of events.
EventBeanFactory eventFactory;  // A factory for creating EventBean instances from store rows.
VirtualDataWindowOutStream outputStream;  // For stream output to consuming statements.
AgentInstanceContext agentInstanceContext;  // Other statement information in statement context.

When using contexts you can decide whether your factory returns a new virtual data window for each context partition or returns the same virtual data window instance for all context partitions. Your extension code may refer to the named window name to identify the named window and may refer to the agent instance context that holds the agent instance id which is the id of the context partition.

A virtual data window implementation is responsible for the following functions:

The sample code shown here can be found among the examples in the distribution under virtualdw.

The implementation class must implement the VirtualDataWindow interface like so:

public class SampleVirtualDataWindow implements VirtualDataWindow {

  private final VirtualDataWindowContext context;
  
  public SampleVirtualDataWindow(VirtualDataWindowContext context) {
    this.context = context;
  } ...

When the compiler compiles a statement and detects a virtual data window, the compiler compiles access path information and the runtime invokes the getLookup method indicating hash and btree access path information by passing a VirtualDataWindowLookupContext context. The lookup method must return a VirtualDataWindowLookup implementation that the statement uses for all lookups until the statement is stopped or destroyed.

The sample implementation does not use the hash and btree access path information and simply returns a lookup object:

public VirtualDataWindowLookup getLookup(VirtualDataWindowLookupContext desc) {

  // Place any code that interrogates the hash-index and btree-index fields here.

  // Return the lookup strategy.
  return new SampleVirtualDataWindowLookup(context);
}

The runtime calls the update method when data changes because of on-merge, on-delete, on-update or insert-into. For example, if you have an on-merge statement that is triggered and that updates the virtual data window, the newData parameter receives the new (updated) event and the oldData parameter receives the event prior to the update. Your code may use these events to update the store or delete from the store, if needed.

If your application plans to consume data from the virtual data window, for example via select * from MySampleWindow, then the code must implement the update method to forward insert and remove stream events, as shown below, to receive the events in consuming statements. To post insert and remove stream data, use the VirtualDataWindowOutStream provided by the context object as follows.

public void update(EventBean[] newData, EventBean[] oldData) {
  // This sample simply posts into the insert and remove stream what is received.
  context.getOutputStream().update(newData, oldData);
}

Your application should not use VirtualDataWindowOutStream to post new events that originate from the store. The object is intended for use with on-action statements. Use insert-into instead for any new events that originate from the store.

Views in EPL are used to derive information from an event stream, and to represent data windows onto an event stream. This chapter describes how to plug-in a new, custom view.

The following steps are required to develop and use a custom view.

  1. Implement a view forge class. View forges are compile-time classes that accept and check view parameters and refer to the appropriate view factory for the runtime.

  2. Implement a view factory class. View factories are classes that instantiate the appropriate view class at runtime.

  3. Implement a view class. A view class commonly represents a data window or derives new information from a stream at runtime.

  4. Configure the view factory class supplying a view namespace and name in the compiler configuration.

The example view factory and view class that are used in this chapter can be found in the examples source folder in the OHLC (open-high-low-close) example. The class names are OHLCBarPlugInViewForge, OHLCBarPlugInViewFactory and OHLCBarPlugInView.

Views can make use of the runtime services available via StatementContext, for example:

  • The SchedulingService interface allows views to schedule timer callbacks to a view

Section 21.4.4, “View Contract” outlines the requirements for correct behavior of your custom view within the runtime.

Note that custom views may use runtime services and APIs that can be subject to change between major releases. The runtime services discussed above and view APIs are considered part of the runtime internal API and are only limited stable. Please also consider contributing your custom view to the project by submitting the view code.

A view forge class is a compile-time class and is responsible for the following functions:

View forge classes must implement the ViewFactoryForge interface. Additionally a view forge class must implement the DataWindowViewForge interface if the view is a data window (retains events provided to it).

public class OHLCBarPlugInViewForge implements ViewFactoryForge { ...

Your view forge class must implement the setViewParameters method to accept view parameters and the attach method to attach the view to a stream:

public class OHLCBarPlugInViewForge implements ViewFactoryForge {
    private List<ExprNode> viewParameters;
    private ExprNode timestampExpression;
    private ExprNode valueExpression;
    private EventType eventType;

    public void setViewParameters(List<ExprNode> parameters, ViewForgeEnv viewForgeEnv, int streamNumber) throws ViewParameterException {
        this.viewParameters = parameters;
    }

    public void attach(EventType parentEventType, int streamNumber, ViewForgeEnv env) throws ViewParameterException {
        if (viewParameters.size() != 2) {
            throw new ViewParameterException("View requires a two parameters: the expression returning timestamps and the expression supplying OHLC data points");
        }
        ExprNode[] validatedNodes = ViewForgeSupport.validate("OHLC view", parentEventType, viewParameters, false, env, streamNumber);

        timestampExpression = validatedNodes[0];
        valueExpression = validatedNodes[1];

        if ((timestampExpression.getForge().getEvaluationType() != long.class) && (timestampExpression.getForge().getEvaluationType() != Long.class)) {
            throw new ViewParameterException("View requires long-typed timestamp values in parameter 1");
        }
        if ((valueExpression.getForge().getEvaluationType() != double.class) && (valueExpression.getForge().getEvaluationType() != Double.class)) {
            throw new ViewParameterException("View requires double-typed values for in parameter 2");
        }
        ....

After the compiler supplied view parameters to the forge, the compiler will ask the view to attach to its parent and validate any parameter expressions against the parent view's event type. If the view will be generating events of a different type then the events generated by the parent view, then the view factory can allocate the new event type.

Finally, the compiler asks the view forge to generate code that initializes the view factory:

public CodegenExpression make(CodegenMethodScope parent, SAIFFInitializeSymbol symbols, CodegenClassScope classScope) {
    return new SAIFFInitializeBuilder(OHLCBarPlugInViewFactory.class, this.getClass(), "factory", parent, symbols, classScope)
                .exprnode("timestampExpression", timestampExpression)
                .exprnode("valueExpression", valueExpression)
                .build();
}

Use the internal SAIFFInitializeBuilder to build your view factory providing it the expressions and other values it needs.

The update method must adhere to the following conventions, to prevent memory leaks and to enable correct behavior within the runtime:

Your view implementation must implement the AgentInstanceStopCallback interface to receive a callback when the view gets destroyed.

Please refer to the sample views for a code sample on how to implement the iterator method.

In terms of multiple threads accessing view state, there is no need for your custom view factory or view implementation to perform any synchronization to protect internal state. The iterator of the custom view implementation does also not need to be thread-safe. The runtime ensures the custom view executes in the context of a single thread at a time. If your view uses shared external state, such external state must be still considered for synchronization when using multiple threads.

Aggregation functions are stateful functions that aggregate events, event property values or expression results. Examples for built-in aggregation functions are count(*), sum(price * volume), window(*) or maxby(volume).

EPL allows two different ways for your application to provide aggregation functions. We use the name aggregation single-function and aggregation multi-function for the two independent extension APIs for aggregation functions.

The aggregation single-function API is simple to use however it imposes certain restrictions on how expressions that contain aggregation functions share state and how they are evaluated.

The aggregation multi-function API is more powerful and provides control over how expressions that contain aggregation functions share state and are evaluated.

The next table compares the two aggregation function extension API's:


The following sections discuss developing an aggregation single-function first, followed by the subject of developing an aggregation multi-function.

Note

The aggregation multi-function API is a powerful and lower-level API to extend the runtime. Any classes that are not part of the client package should be considered unstable and are subject to change between minor and major releases.

This section describes the aggregation single-function extension API for providing aggregation functions.

The following steps are required to develop and use a custom aggregation single-function.

Custom aggregation functions can also be passed multiple parameters, as further described in Section 21.5.1.5, “Aggregation Single-Function: Accepting Multiple Parameters”. In the example below the aggregation function accepts a single parameter.

The code for the example aggregation function as shown in this chapter can be found in the runtime configuration example in the package com.espertech.esper.example.runtimeconfig by the name MyConcatAggregationFunction. The sample function simply concatenates string-type values.

An aggregation function forge class is only used at compile-time and is responsible for the following functions:

Aggregation forge classes implement the interface AggregationFunctionForge:

public class MyConcatAggregationFunctionForge implements AggregationFunctionForge { ...

The compiler constructs one instance of the aggregation function forge class for each time the function is listed in a statement, however the compiler may decide to reduce the number of aggregation forge instances if it finds equivalent aggregations.

The aggregation function forge instance receives the aggregation function name via set setFunctionName method.

The sample concatenation function forge provides an empty setFunctionName method:

public void setFunctionName(String functionName) {
  // no action taken
}

An aggregation function forge must provide an implementation of the validate method that is passed a AggregationFunctionValidationContext validation context object. Within the validation context you find the result type of each of the parameters expressions to the aggregation function as well as information about constant values and data window use. Please see the JavaDoc API documentation for a comprehensive list of validation context information.

Since the example concatenation function requires string types it implements a type check:

public void validate(AggregationValidationContext validationContext) {
  if ((validationContext.getParameterTypes().length != 1) ||
    (validationContext.getParameterTypes()[0] != String.class)) {
    throw new IllegalArgumentException("Concat aggregation requires a single parameter of type String");
  }
}

In order for the compiler to validate the type returned by the aggregation function against the types expected by enclosing expressions, the getValueType must return the result type of any values produced by the aggregation function:

public Class getValueType() {
  return String.class;
}

Finally the forge implementation must provide a getAggregationFunctionMode method that returns information about the factory. The compiler uses this information to build the aggregation function factory.

public AggregationFunctionMode getAggregationFunctionMode() {
    // Inject a factory by using "new"
    InjectionStrategy injectionStrategy = new InjectionStrategyClassNewInstance(MyConcatAggregationFunctionFactory.class);
    
    // The managed mode means there is no need to write code that generates code
    AggregationFunctionModeManaged mode = new AggregationFunctionModeManaged();
    mode.setInjectionStrategyAggregationFunctionFactory(injectionStrategy);
        
    return mode;
}

An aggregation function class is responsible for the following functions:

Aggregation function classes implement the interface AggregationFunction:

public class MyConcatAggregationFunction implements AggregationFunction { ...

The class that provides the aggregation and implements AggregationFunction does not have to be threadsafe.

The constructor initializes the aggregation function:

public class MyConcatAggregationFunction implements AggregationFunction {
  private final static char DELIMITER = ' ';
  private StringBuilder builder;
  private String delimiter;

  public MyConcatAggregationFunction() {
    builder = new StringBuilder();
    delimiter = "";
  }
  ...

The enter method adds a datapoint to the current aggregation value. The example enter method shown below adds a delimiter and the string value to a string buffer:

public void enter(Object value) {
  if (value != null) {
    builder.append(delimiter);
    builder.append(value.toString());
    delimiter = String.valueOf(DELIMITER);
  }
}

Conversly, the leave method removes a datapoint from the current aggregation value. The example leave method removes from the string buffer:

public void leave(Object value) {
  if (value != null) {
    builder.delete(0, value.toString().length() + 1);
  }
}

Finally, the runtime obtains the current aggregation value by means of the getValue method:

public Object getValue() {
  return builder.toString();
}

For on-demand queries the aggregation function must support resetting its value to empty or start values. Implement the clear function to reset the value as shown below:

public void clear() {
  builder = new StringBuilder();
  delimiter = "";
}

Your plug-in aggregation function may accept multiple parameters. You must provide a different mode however:

    public AggregationFunctionMode getAggregationFunctionMode() {
        InjectionStrategy injectionStrategy = new InjectionStrategyClassNewInstance(SupportCountBackAggregationFunctionFactory.class);

        AggregationFunctionModeMultiParam multiParam = new AggregationFunctionModeMultiParam();
        multiParam.setInjectionStrategyAggregationFunctionFactory(injectionStrategy);
        
        return multiParam;
    }

For instance, assume an aggregation function rangeCount that counts all values that fall into a range of values. The EPL that calls this function and provides a lower and upper bounds of 1 and 10 is:

select rangeCount(1, 10, myValue) from MyEvent

The enter method of the plug-in aggregation function may look as follows:

public void enter(Object value)  {
  Object[] params = (Object[]) value;
  int lower = (Integer) params[0];
  int upper = (Integer) params[1];
  int val = (Integer) params[2];
  if ((val >= lower) && (val <= upper)) {
    count++;
  }
}

Your plug-in aggregation function may want to validate parameter types or may want to know which parameters are constant-value expressions. Constant-value expressions are evaluated only once by the runtime and could therefore be cached by your aggregation function for performance reasons. The runtime provides constant-value information as part of the AggregationValidationContext passed to the validate method.

This section introduces the aggregation multi-function API. Please refer to the JavaDoc for more complete class and method-level documentation.

Among the examples is an example use of the aggregation multi-function API in the example by name Cycle-Detect. Cycle-Detect takes incoming transaction events that have from-account and to-account fields. The example detects a cycle in the transactions between accounts in order to detect a possible transaction fraud. Please note that the graph and cycle detection logic of the example is not part of the distribution: The example utilizes the jgrapht library.

In the Cycle-Detect example, the vertices of a graph are the account numbers. For example the account numbers Acct-1, Acct-2 and Acct-3. In the graph the edges are transaction events that identify a from-account and a to-account. An example edge is {from:Acct-1, to:Acct-2}. An example cycle is therefore in the three transactions {from:Acct-1, to:Acct-2}, {from:Acct-2, to:Acct-3} and {from:Acct-3, to:Acct-1}.

The code for the example aggregation multi-function as shown in this chapter can be found in the Cycle-Detect example in the package com.espertech.esper.example.cycledetect. The example provides two aggregation functions named cycledetected and cycleoutput:

In the Cycle-Detect example, the following statement utilizes the two functions cycledetected and cycleoutput that share the same graph state to detect a cycle among the last 1000 events:

@Name('CycleDetector') select cycleoutput() as cyclevertices
from TransactionEvent#length(1000)
having cycledetected(fromAcct, toAcct)

If instead the goal is to run graph cycle detection every 1 second (and not upon arrival of a new event), this sample statement uses a pattern to trigger cycle detection:

@Name('CycleDetector')
select (select cycleoutput(fromAcct, toAcct) from TransactionEvent#length(1000)) as cyclevertices
from pattern [every timer:interval(1)]

The following steps are required to develop and use a custom aggregation multi-function.

  1. Implement an aggregation multi-function forge by implementing the interface com.espertech.esper.common.client.hook.aggmultifunc.AggregationMultiFunctionForge.

  2. Implement one or more handlers for aggregation functions by implementing the interface com.espertech.esper.common.client.hook.aggmultifunc.AggregationMultiFunctionHandler.

  3. Implement an aggregation state key by implementing the interface com.espertech.esper.common.client.hook.aggmultifunc.AggregationMultiFunctionStateKey.

  4. Implement an aggregation state factory by implementing the interface com.espertech.esper.common.client.hook.aggmultifunc.AggregationMultiFunctionStateFactory.

  5. Implement an aggregation state holder by implementing the interface com.espertech.esper.common.client.hook.aggmultifunc.AggregationMultiFunctionState.

  6. Implement a state accessor factory by implementing the interface com.espertech.esper.common.client.hook.aggmultifunc.AggregationMultiFunctionAccessorFactory.

  7. Implement a state accessor by implementing the interface com.espertech.esper.common.client.hook.aggmultifunc.AggregationMultiFunctionAccessor.

  8. For use with tables, implement an agent factory by implementing the interface com.espertech.esper.common.client.hook.aggmultifunc.AggregationMultiFunctionAgentFactory.

  9. For use with tables, implement an agent by implementing the interface com.espertech.esper.common.client.hook.aggmultifunc.AggregationMultiFunctionAgent.

  10. For use with aggregation methods, implement an aggregation method factory by implementing the interface com.espertech.esper.common.client.hook.aggmultifunc.AggregationMultiFunctionAggregationMethodFactory.

  11. For use with aggregation methods, implement an aggregation method by implementing the interface com.espertech.esper.common.client.hook.aggmultifunc.AggregationMultiFunctionAggregationMethod.

  12. Register the aggregation multi-function forge class with the compiler by supplying one or more function names, via the compiler configuration file or the runtime and static configuration API.

An aggregation multi-function forge class is a compile-time class responsible for the following functions:

Aggregation multi-function factory classes implement the interface AggregationMultiFunctionForge:

public class CycleDetectorAggregationForge implements AggregationMultiFunctionForge { ...

The compiler constructs a single instance of the aggregation multi-function forge class that is shared for all aggregation function expressions in a statement that have one of the function names provided in the configuration object.

The compiler invokes the addAggregationFunction method at the time it compiles a statement. The method receives a declaration-time context object that provides the function name as well as additional information.

The sample Cycle-Detect factory class provides an empty addAggregationFunction method:

public void addAggregationFunction(AggregationMultiFunctionDeclarationContext declarationContext) {
    // provides an opportunity to inspect where used
}

The compiler invokes the validateGetHandler method at the time of expression validation. It passes a AggregationMultiFunctionValidationContext validation context object that contains actual parameters expressions. Please see the JavaDoc API documentation for a comprehensive list of validation context information.

The validateGetHandler method must return a handler object the implements the AggregationMultiFunctionHandler interface. Return a handler object for each aggregation function expression according to the aggregation function name and its parameters that are provided in the validation context.

The example cycledetect function takes two parameters that provide the cycle edge (from-account and to-account):

public AggregationMultiFunctionHandler validateGetHandler(AggregationMultiFunctionValidationContext validationContext) {
  if (validationContext.getParameterExpressions().length == 2) {
    fromExpression = validationContext.getParameterExpressions()[0];
    toExpression = validationContext.getParameterExpressions()[1];
  }
  return new CycleDetectorAggregationHandler(this, validationContext);
}

An aggregation multi-function handler class is a compile-time class that must implement the AggregationMultiFunctionHandler interface and is responsible for the following functions:

In the Cycle-Detect example, the class CycleDetectorAggregationHandler is the handler for all aggregation functions.

public class CycleDetectorAggregationHandler implements AggregationMultiFunctionHandler { ...

The getReturnType method provided by the handler instructs the compiler about the return type of each aggregation accessor. The class EPType holds return type information.

In the Cycle-Detect example the cycledetected function returns a single boolean value. The cycleoutput returns a collection of vertices:

public EPType getReturnType() {
    if (validationContext.getFunctionName().toLowerCase(Locale.ENGLISH).equals(CycleDetectorConstant.CYCLEOUTPUT_NAME)) {
        return EPTypeHelper.collectionOfSingleValue(forge.getFromExpression().getForge().getEvaluationType());
    }
    return EPTypeHelper.singleValue(Boolean.class);
}

The compiler invokes the getAggregationStateUniqueKey method to determine whether multiple aggregation function expressions in the same statement can share the same aggregation state or should receive different aggregation state instances.

The getAggregationStateUniqueKey method must return an instance of AggregationMultiFunctionStateKey. The compiler uses equals-semantics (the hashCode and equals methods) to determine whether multiple aggregation function share the state object. If the key object returned for each aggregation function by the handler is an equal key object then the compiler shares aggregation state between such aggregation functions for the same statement and context partition.

In the Cycle-Detect example the state is shared, which it achieves by simply returning the same key instance:

private static final AggregationMultiFunctionStateKey CYCLE_KEY = new AggregationMultiFunctionStateKey() {};

public AggregationMultiFunctionStateKey getAggregationStateUniqueKey() {
    return CYCLE_KEY;
}

The compiler invokes the getStateMode method to obtain an instance of AggregationMultiFunctionStateMode. The state mode is responsible to obtaining and configuring an aggregation state factory instance at time of deployment.

In the Cycle-Detect example the method passes the expression evaluators providing the from-account and to-account expressions to the state factory:

public AggregationMultiFunctionStateMode getStateMode() {
    AggregationMultiFunctionStateModeManaged managed = new AggregationMultiFunctionStateModeManaged();
    InjectionStrategyClassNewInstance injection = new InjectionStrategyClassNewInstance(CycleDetectorAggregationStateFactory.class);
    injection.addExpression("from", forge.getFromExpression());
    injection.addExpression("to", forge.getToExpression());
    managed.setInjectionStrategyAggregationStateFactory(injection);
    return managed;
}

The compiler invokes the getAccessorMode method to obtain an instance of AggregationMultiFunctionAccessorMode. The accessor mode is responsible to obtaining and configuring an accessor factory instance at time of deployment.

The getAccessorMode method provides information about the accessor factories according to whether the aggregation function name is cycledetected or cycleoutput:

public AggregationMultiFunctionAccessorMode getAccessorMode() {
    Class accessor;
    if (validationContext.getFunctionName().toLowerCase(Locale.ENGLISH).equals(CycleDetectorConstant.CYCLEOUTPUT_NAME)) {
        accessor = CycleDetectorAggregationAccessorOutputFactory.class;
    }
    else {
        accessor = CycleDetectorAggregationAccessorDetectFactory.class;
    }
    AggregationMultiFunctionAccessorModeManaged managed = new AggregationMultiFunctionAccessorModeManaged();
    InjectionStrategyClassNewInstance injection = new InjectionStrategyClassNewInstance(accessor);
    managed.setInjectionStrategyAggregationAccessorFactory(injection);
    return managed;
}

Pattern guards are pattern objects that control the lifecycle of the guarded sub-expression, and can filter the events fired by the subexpression.

The following steps are required to develop and use a custom guard object.

  1. Implement a guard forge class, responsible for compile-time guard information.

  2. Implement a guard factory class, responsible for creating guard object instances at runtime.

  3. Implement a guard class (used at runtime).

  4. Register the guard forge class with the compiler by supplying a namespace and name, via the compiler configuration.

The code for the example guard object as shown in this chapter can be found in the test source folder in the package com.espertech.esper.regressionlib.support.extend.pattern by the name MyCountToPatternGuardForge. The sample guard discussed here counts the number of events occurring up to a maximum number of events, and end the sub-expression when that maximum is reached.

Some of the APIs that you use to implement a pattern guard are internal APIs and are not stable and may change between releases. The client package contains all the stable interface classes.

A guard forge class is only used by the compiler and is responsible for the following functions:

Guard forge classes implement the GuardForge:

public class MyCountToPatternGuardForge implements GuardForge { ...

The compiler constructs one instance of the guard forge class for each time the guard is listed in a statement.

The guard forge class implements the setGuardParameters method that is passed the parameters to the guard as supplied by the statement. It verifies the guard parameters, similar to the code snippet shown next. Our example counter guard takes a single numeric parameter:

public void setGuardParameters(List<ExprNode> guardParameters, MatchedEventConvertorForge convertor, StatementCompileTimeServices services) throws GuardParameterException {
    String message = "Count-to guard takes a single integer-value expression as parameter";
    if (guardParameters.size() != 1) {
        throw new GuardParameterException(message);
    }

    Class paramType = guardParameters.get(0).getForge().getEvaluationType();
    if (paramType != Integer.class && paramType != int.class) {
        throw new GuardParameterException(message);
    }
        
    this.numCountToExpr = guardParameters.get(0);
    this.convertor = convertor;
}

The makeCodegen method is called by the compiler to receive the code that builds a guard factory. Use the SAIFFInitializeBuilder to build factory initialization code:

public CodegenExpression makeCodegen(CodegenMethodScope parent, SAIFFInitializeSymbol symbols, CodegenClassScope classScope) {
    SAIFFInitializeBuilder builder = new SAIFFInitializeBuilder(MyCountToPatternGuardFactory.class, this.getClass(), "guardFactory", parent, symbols, classScope);
    return builder.exprnode("numCountToExpr", numCountToExpr)
                .expression("convertor", convertor.makeAnonymous(builder.getMethod(), classScope))
                .build();
}

Pattern observers are pattern objects that are executed as part of a pattern expression and can observe events or test conditions. Examples for built-in observers are timer:at and timer:interval. Some suggested uses of observer objects are:

  • Implement custom scheduling logic using the runtime's own scheduling and timer services

  • Test conditions related to prior events matching an expression

The following steps are required to develop and use a custom observer object within pattern statements:

  1. Implement an observer forge class, which is used by the compiler only and is responsible for validating parameters and for initializing an observer factory.

  2. Implement an observer factory class, responsible for creating observer object instances.

  3. Implement an observer class.

  4. Register an observer factory class with the compiler by supplying a namespace and name, via the compiler configuration file or the configuration API.

The code for the example observer object as shown in this chapter can be found in the test source folder in package com.espertech.esper.regression.client by the name MyFileExistsObserver. The sample observer discussed here very simply checks if a file exists, using the filename supplied by the pattern statement, and via the java.io.File class.

Some of the APIs that you use to implement a pattern observer are internal APIs and are not stable and may change between releases. The client package contains all the stable interface classes.

An observer forge class is responsible for the following functions:

Observer forge classes implement the ObserverForge interface:

public class MyFileExistsObserverForge implements ObserverForge { ...

The compiler constructs one instance of the observer forge class for each time the observer is listed in a statement.

The observer forge class implements the setObserverParameters method that is passed the parameters to the observer as supplied by the statement. It verifies the observer parameters, similar to the code snippet shown next. Our example file-exists observer takes a single string parameter:

public void setObserverParameters(List<ExprNode> observerParameters, MatchedEventConvertorForge convertor, ExprValidationContext validationContext) throws ObserverParameterException {
    String message = "File exists observer takes a single string filename parameter";
    if (observerParameters.size() != 1) {
        throw new ObserverParameterException(message);
    }
    if (!(observerParameters.get(0).getForge().getEvaluationType() == String.class)) {
        throw new ObserverParameterException(message);
    }

    this.filenameExpression = observerParameters.get(0);
    this.convertor = convertor;
}

The compiler calls the makeCodegen method to provide code that initializes the observer factory at time of deployment. It uses the SAIFFInitializeBuilder to build the code.

public CodegenExpression makeCodegen(CodegenMethodScope parent, SAIFFInitializeSymbol symbols, CodegenClassScope classScope) {
    SAIFFInitializeBuilder builder = new SAIFFInitializeBuilder(MyFileExistsObserverFactory.class, this.getClass(), "observerFactory", parent, symbols, classScope);
    return builder.exprnode("filenameExpression", filenameExpression)
            .expression("convertor", convertor.makeAnonymous(builder.getMethod(), classScope))
            .build();
}

An observer factory class is responsible for the following functions:

Observer factory classes implement the ObserverFactory:

public class MyFileExistsObserverFactory implements ObserverFactory { ...

The runtime obtains an instance of the observer factory class at time of deployment.

The runtime calls the makeObserver method to create a new observer instance. The example makeObserver method shown below passes parameters to the observer instance:

public EventObserver makeObserver(PatternAgentInstanceContext context, MatchedEventMap beginState, ObserverEventEvaluator observerEventEvaluator, Object observerState, boolean isFilterChildNonQuitting) {
    EventBean[] events = convertor == null ? null : convertor.convert(beginState);
    Object filename = PatternExpressionUtil.evaluateChecked("File-exists observer ", filenameExpression, events, context.getAgentInstanceContext());
    if (filename == null) {
        throw new EPException("Filename evaluated to null");
    }
    return new MyFileExistsObserver(beginState, observerEventEvaluator, filename.toString());
}

The ObserverEventEvaluator parameter allows an observer to indicate events, and to indicate change of truth value to permanently false. Use this interface to indicate when your observer has received or witnessed an event, or changed it's truth value to true or permanently false.

The MatchedEventMap parameter provides a Map of all matching events for the expression prior to the observer's start. For example, consider a pattern as below:

a=MyEvent -> myplugin:my_observer(...)

The above pattern tagged the MyEvent instance with the tag "a". The runtime starts an instance of my_observer when it receives the first MyEvent. The observer can query the MatchedEventMap using "a" as a key and obtain the tagged event.