Chapter 7. Complex Event Processing

To declare a fact type as an event, assign the @role metadata tag to the fact with the event parameter. The @role metadata tag can accept two possible values:

This example declares that a stock broker application’s StockTick fact type will be handled as an event:

import some.package.StockTick

declare StockTick
  @role( event )
end

Facts can also be declared inline. If StockTick was a fact type declared in the DRL instead of in a pre-existing class, the code would be as follows:

declare StockTick
  @role(event)

  datetime : java.util.Date
  symbol : String
  price : double
end

For more information about type declarations, see Section 8.9, “Type Declaration”.

Every event has associated metadata. Typically, the metadata is automatically added as each event is inserted into working memory. The metadata defaults can be changed on an event-type basis using the metadata tags:

The following examples assume the application domain model includes the following class:

/**
 * A class that represents a voice call in a Telecom domain model.
 */
public class VoiceCall {
  private String  originNumber;
  private String  destinationNumber;
  private Date    callDateTime;
  private long    callDuration;  // in milliseconds

  // Constructors, getters, and setters.
}

Events have strong temporal constraints making it is necessary to use a reference clock. If a rule needs to determine the average price of a given stock over the last sixty minutes, it is necessary to compare the stock price event’s timestamp with the current time. The reference clock provides the current time.

Because the rules engine can simultaneously run an array of different scenarios that require different clocks, multiple clock implementations can be used by the engine.

Scenarios that require different clocks include the following:

JBoss BRMS Complex Event Processing comes equipped with two clock implementations:

For a list of Maven dependencies, see example Embedded jBPM Engine Dependencies. If you use Red Hat JBoss BRMS, see Embedded Drools Engine Dependencies.

Cloud mode is the default operating mode of Red Hat JBoss Business Rules Management System.

Running in Cloud mode, the engine applies a many-to-many pattern matching algorithm, which treats the events as an unordered cloud. Events still have timestamps, but there is no way for the rules engine running in Cloud mode to draw relevance from the timestamp because Cloud mode is unaware of the present time.

This mode uses the rules constraints to find the matching tuples, activate, and fire rules.

Cloud mode does not impose any kind of additional requirements on facts; however, because it has no concept of time, it cannot take advantage of temporal features such as sliding windows or automatic life-cycle management. In Cloud mode, it is necessary to explicitly retract events when they are no longer needed.

Certain requirements that are not imposed include the following:

Cloud mode can be specified either by setting a system property, using configuration property files, or using the API.

The API call follows:

import org.kie.api.conf.EventProcessingOption;
import org.kie.api.KieBaseConfiguration;
import org.kie.api.KieServices.Factory;

KieBaseConfiguration config = KieServices.Factory.get().newKieBaseConfiguration();

config.setOption(EventProcessingOption.CLOUD);

The equivalent property follows:

drools.eventProcessingMode = cloud

For a list of Maven dependencies, see example Embedded jBPM Engine Dependencies. If you use Red Hat JBoss BRMS, see Embedded Drools Engine Dependencies.

Stream mode processes events chronologically as they are inserted into the rules engine. Stream mode uses a session clock that enables the rules engine to process events as they occur in time. The session clock enables processing events as they occur based on the age of the events. Stream mode also synchronizes streams of events (so events in different streams can be processed in chronological order), implements sliding windows of interest, and enables automatic life-cycle management.

The requirements for using stream mode are the following:

Stream mode can be enabled by setting a system property, using configuration property files, or using the API.

The API call follows:

import org.kie.api.conf.EventProcessingOption;
import org.kie.api.KieBaseConfiguration;
import org.kie.api.KieServices.Factory;

KieBaseConfiguration config = KieServices.Factory.get().newKieBaseConfiguration();

config.setOption(EventProcessingOption.STREAM);

The equivalent property follows:

drools.eventProcessingMode = stream

For a list of Maven dependencies, see example Embedded jBPM Engine Dependencies. If you use Red Hat JBoss BRMS, see Embedded Drools Engine Dependencies.

Entry points are declared implicitly by making direct use of them in rules. Referencing an entry point in a rule will make the engine, at compile time, identify and create the proper internal structures to support that entry point.

For example, a banking application that has transactions fed into the engine using streams could have one stream for all of the transactions executed at ATMs. A rule for this scenario could state, "A withdrawal is only allowed if the account balance is greater than the withdrawal amount the customer has requested."

rule "Authorize Withdraw"
when
  WithdrawRequest($ai : accountId, $am : amount) from entry-point "ATM Stream"
  CheckingAccount(accountId == $ai, balance > $am)
then
  // authorize withdraw
end

When the engine compiles this rule, it will identify that the pattern is tied to the entry point ATM Stream. The engine will create all the necessary structures for the rule-base to support the ATM Stream, and this rule will only match WithdrawRequest events coming from the ATM Stream.

Note the ATM example rule joins the event (WithdrawalRequest) from the stream with a fact from the main working memory (CheckingAccount).

The banking application may have a second rule that states, "A fee of $2 must be applied to a withdraw request made using a branch teller."

rule "Apply Fee on Withdraws on Branches"
when
  WithdrawRequest($ai : accountId, processed == true) from entry-point "Branch Stream"
  CheckingAccount(accountId == $ai)
then
  // apply a $2 fee on the account
end

This rule matches events of the same type (WithdrawRequest) as the example ATM rule but from a different stream. Events inserted into the ATM Stream will never match the pattern on the second rule, which is tied to the Branch Stream; accordingly, events inserted into the Branch Stream will never match the pattern on the example ATM rule, which is tied to the ATM Stream.

Declaring the stream in a rule states that the rule is only interested in events coming from that stream.

Events can be inserted manually into an entry point instead of directly into the working memory.

import org.kie.api.runtime.KieSession;
import org.kie.api.runtime.rule.EntryPoint;

// Create your rulebase and your session as usual:
KieSession session = ...

// Get a reference to the entry point:
EntryPoint atmStream = session.getEntryPoint("ATM Stream");

// ...and start inserting your facts into the entry point:
atmStream.insert(aWithdrawRequest);

For a list of Maven dependencies, see example Embedded jBPM Engine Dependencies. If you use Red Hat JBoss BRMS, see Embedded Drools Engine Dependencies.

A negative pattern is concerned with conditions that are not met. Negative patterns make reasoning in the absence of events possible. For instance, a safety system could have a rule that states "If a fire is detected and the sprinkler is not activated, sound the alarm."

In Cloud mode, the engine assumes all facts (regular facts and events) are known in advance and evaluates negative patterns immediately.

rule "Sound the Alarm"
when
  $f : FireDetected()
  not(SprinklerActivated())
then
  // sound the alarm
end

An example in stream mode is displayed below. This rule keeps consistency when dealing with negative patterns and temporal constraints at the same time interval.

rule "Sound the Alarm"
  duration(10s)
when
  $f : FireDetected()
  not(SprinklerActivated(this after[0s,10s] $f))
then
  // sound the alarm
end

In stream mode, negative patterns with temporal constraints may force the engine to wait for a set time before activating a rule. A rule may be written for an alarm system that states, "If a fire is detected and the sprinkler is not activated after 10 seconds, sound the alarm." Unlike the previous stream mode example, this one does not require the user to calculate and write the duration parameter.

rule "Sound the Alarm"
when
  $f : FireDetected()
  not(SprinklerActivated(this after[0s,10s] $f))
then
  // sound the alarm
end

The rule depicted below expects one "Heartbeat" event to occur every 10 seconds; if not, the rule fires. What is special about this rule is that it uses the same type of object in the first pattern and in the negative pattern. The negative pattern has the temporal constraint to wait between 0 to 10 seconds before firing, and it excludes the Heartbeat bound to $h. Excluding the bound Heartbeat is important since the temporal constraint [0s, …​] does not exclude by itself the bound event $h from being matched again, thus preventing the rule to fire.

rule "Sound the Alarm"
when
  $h: Heartbeat() from entry-point "MonitoringStream"
  not(Heartbeat(this != $h, this after[0s,10s] $h) from entry-point "MonitoringStream")
then
  // sound the alarm
end

Complex Event Processing requires the rules engine to engage in temporal reasoning. Events have strong temporal constraints so it is vital the rules engine can determine and interpret an event’s temporal attributes, both as they relate to other events and the 'flow of time' as it appears to the rules engine. This makes it possible for rules to take time into account; for instance, a rule could state "Calculate the average price of a stock over the last 60 minutes."

JBoss BRMS Complex Event Processing implements the following temporal operators and their logical complements (negation):

The after operator correlates two events and matches when the temporal distance (the time between the two events) from the current event to the event being correlated falls into the distance range declared for the operator.

For example:

$eventA : EventA(this after[3m30s, 4m] $eventB)

This pattern only matches if the temporal distance between the time when $eventB finished and the time when $eventA started is between the lower limit of three minutes and thirty seconds and the upper limit of four minutes.

This can also be represented as follows:

3m30s <= $eventA.startTimestamp - $eventB.endTimeStamp <= 4m

The after operator accepts one or two optional parameters:

The after operator also accepts negative temporal distances.

For example:

$eventA : EventA(this after[-3m30s, -2m] $eventB)

If the first value is greater than the second value, the engine will automatically reverse them.

The following two patterns are equivalent to each other:

$eventA : EventA(this after[-3m30s, -2m] $eventB)
$eventA : EventA(this after[-2m, -3m30s] $eventB)

The before operator correlates two events and matches when the temporal distance (time between the two events) from the event being correlated to the current event falls within the distance range declared for the operator.

For example:

$eventA : EventA(this before[3m30s, 4m] $eventB)

This pattern only matches if the temporal distance between the time when $eventA finished and the time when $eventB started is between the lower limit of three minutes and thirty seconds and the upper limit of four minutes.

This can also be represented as follows:

3m30s <= $eventB.startTimestamp - $eventA.endTimeStamp <= 4m

The before operator accepts one or two optional parameters:

The before operator also accepts negative temporal distances.

For example:

$eventA : EventA(this before[-3m30s, -2m] $eventB)

If the first value is greater than the second value, the engine will automatically reverse them.

The following two patterns are equivalent to each other:

$eventA : EventA(this before[-3m30s, -2m] $eventB)
$eventA : EventA(this before[-2m, -3m30s] $eventB)

The coincides operator correlates two events and matches when both events happen at the same time.

For example:

$eventA : EventA(this coincides $eventB)

This pattern only matches if both the start timestamps of $eventA and $eventB are identical and the end timestamps of both $eventA and $eventB are also identical.

The coincides operator accepts optional thresholds for the distance between the events' start times and the events' end times, so the events do not have to start at exactly the same time or end at exactly the same time, but they need to be within the provided thresholds.

The following rules apply when defining thresholds for the coincides operator:

For example:

$eventA : EventA(this coincides[15s, 10s] $eventB)

This pattern will only match if the following conditions are met:

abs($eventA.startTimestamp - $eventB.startTimestamp) <= 15s
&&
abs($eventA.endTimestamp - $eventB.endTimestamp) <= 10s

The during operator correlates two events and matches when the current event happens during the event being correlated.

For example:

$eventA : EventA(this during $eventB)

This pattern only matches if $eventA starts after $eventB and ends before $eventB ends.

This can also be represented as follows:

$eventB.startTimestamp < $eventA.startTimestamp <= $eventA.endTimestamp < $eventB.endTimestamp

The during operator accepts one, two, or four optional parameters:

The following rules apply when providing parameters for the during operator:

The finishes operator correlates two events and matches when the current event’s start timestamp post-dates the correlated event’s start timestamp and both events end simultaneously.

For example:

$eventA : EventA(this finishes $eventB)

This pattern only matches if $eventA starts after $eventB starts and ends at the same time as $eventB ends.

This can be represented as follows:

$eventB.startTimestamp < $eventA.startTimestamp
&&
$eventA.endTimestamp == $eventB.endTimestamp

The finishes operator accepts one optional parameter. If defined, the optional parameter sets the maximum time allowed between the end times of the two events.

For example:

$eventA : EventA(this finishes[5s] $eventB)

This pattern matches if these conditions are met:

$eventB.startTimestamp < $eventA.startTimestamp
&&
abs($eventA.endTimestamp - $eventB.endTimestamp) <= 5s

The finishedby operator correlates two events and matches when the current event’s start time predates the correlated event’s start time but both events end simultaneously. finishedby is the symmetrical opposite of the finishes operator.

For example:

$eventA : EventA(this finishedby $eventB)

This pattern only matches if $eventA starts before $eventB starts and ends at the same time as $eventB ends.

This can be represented as follows:

$eventA.startTimestamp < $eventB.startTimestamp
&&
$eventA.endTimestamp == $eventB.endTimestamp

The finishedby operator accepts one optional parameter. If defined, the optional parameter sets the maximum time allowed between the end times of the two events.

$eventA : EventA(this finishedby[5s] $eventB)

This pattern matches if these conditions are met:

$eventA.startTimestamp < $eventB.startTimestamp
&&
abs($eventA.endTimestamp - $eventB.endTimestamp) <= 5s

The includes operator examines two events and matches when the event being correlated happens during the current event. It is the symmetrical opposite of the during operator.

For example:

$eventA : EventA(this includes $eventB)

This pattern only matches if $eventB starts after $eventA and ends before $eventA ends.

This can be represented as follows:

$eventA.startTimestamp < $eventB.startTimestamp <= $eventB.endTimestamp < $eventA.endTimestamp

The includes operator accepts 1, 2 or 4 optional parameters:

The meets operator correlates two events and matches when the current event ends at the same time as the correlated event starts.

For example:

$eventA : EventA(this meets $eventB)

This pattern matches if $eventA ends at the same time as $eventB starts.

This can be represented as follows:

abs($eventB.startTimestamp - $eventA.endTimestamp) == 0

The meets operator accepts one optional parameter. If defined, it determines the maximum time allowed between the end time of the current event and the start time of the correlated event.

For example:

$eventA : EventA(this meets[5s] $eventB)

This pattern matches if these conditions are met:

abs($eventB.startTimestamp - $eventA.endTimestamp) <= 5s

The metby operator correlates two events and matches when the current event starts at the same time as the correlated event ends.

For example:

$eventA : EventA(this metby $eventB)

This pattern matches if $eventA starts at the same time as $eventB ends.

This can be represented as follows:

abs($eventA.startTimestamp - $eventB.endTimestamp) == 0

The metby operator accepts one optional parameter. If defined, it sets the maximum distance between the end time of the correlated event and the start time of the current event.

For example:

$eventA : EventA(this metby[5s] $eventB)

This pattern matches if these conditions are met:

abs($eventA.startTimestamp - $eventB.endTimestamp) <= 5s

The overlaps operator correlates two events and matches when the current event starts before the correlated event starts and ends after the correlated event starts, but it ends before the correlated event ends.

For example:

$eventA : EventA(this overlaps $eventB)

This pattern matches if these conditions are met:

$eventA.startTimestamp < $eventB.startTimestamp < $eventA.endTimestamp < $eventB.endTimestamp

The overlaps operator accepts one or two optional parameters:

The overlappedby operator correlates two events and matches when the correlated event starts before the current event, and the correlated event ends after the current event starts but before the current event ends.

For example:

$eventA : EventA(this overlappedby $eventB)

This pattern matches if these conditions are met:

$eventB.startTimestamp < $eventA.startTimestamp < $eventB.endTimestamp < $eventA.endTimestamp

The overlappedby operator accepts one or two optional parameters:

The starts operator correlates two events and matches when they start at the same time, but the current event ends before the correlated event ends.

For example:

$eventA : EventA(this starts $eventB)

This pattern matches if $eventA and $eventB start at the same time, and $eventA ends before $eventB ends.

This can be represented as follows:

$eventA.startTimestamp == $eventB.startTimestamp
&&
$eventA.endTimestamp < $eventB.endTimestamp

The starts operator accepts one optional parameter. If defined, it determines the maximum distance between the start times of events in order for the operator to still match:

$eventA : EventA(this starts[5s] $eventB)

This pattern matches if these conditions are met:

abs($eventA.startTimestamp - $eventB.startTimestamp) <= 5s
&&
$eventA.endTimestamp < $eventB.endTimestamp

The startedby operator correlates two events. It matches when both events start at the same time and the correlating event ends before the current event.

For example:

$eventA : EventA(this startedby $eventB)

This pattern matches if $eventA and $eventB start at the same time, and $eventB ends before $eventA ends.

This can be represented as follows:

$eventA.startTimestamp == $eventB.startTimestamp
&&
$eventA.endTimestamp > $eventB.endTimestamp

The startedby operator accepts one optional parameter. If defined, it sets the maximum distance between the start time of the two events in order for the operator to still match:

$eventA : EventA( this starts[5s] $eventB)

This pattern matches if these conditions are met:

abs( $eventA.startTimestamp - $eventB.startTimestamp ) <= 5s
&&
$eventA.endTimestamp > $eventB.endTimestamp

Stream mode allows events to be matched over a sliding time window. A sliding window is a time period that stretches back in time from the present. For instance, a sliding window of two minutes includes any events that have occurred in the past two minutes. As events fall out of the sliding time window (in this case because they occurred more than two minutes ago), they will no longer match against rules using this particular sliding window.

For example:

StockTick() over window:time(2m)

JBoss BRMS Complex Event Processing uses the over keyword to associate windows with patterns.

Sliding time windows can also be used to calculate averages and over time. For instance, a rule could be written that states "If the average temperature reading for the last ten minutes goes above a certain point, sound the alarm."

rule "Sound the Alarm in Case Temperature Rises Above Threshold"
when
  TemperatureThreshold($max : max)
  Number(doubleValue > $max) from accumulate(
    SensorReading($temp : temperature) over window:time(10m),
    average($temp))
then
  // sound the alarm
end

The engine will automatically discard any SensorReading more than ten minutes old and keep re-calculating the average.

Similar to Time Windows, Sliding Length Windows work in the same manner; however, they consider events based on order of their insertion into the session instead of flow of time.

The pattern below demonstrates this order by only considering the last 10 RHT Stock Ticks independent of how old they are. Unlike the previous StockTick from the Sliding Time Windows pattern, this pattern uses window:length.

StockTick(company == "RHT") over window:length(10)

The example below portrays window length instead of window time; that is, it allows the user to sound an alarm in case the average temperature over the last 100 readings from a sensor is above the threshold value.

rule "Sound the Alarm in Case Temperature Rises Above Threshold"
when
  TemperatureThreshold($max : max)
  Number(doubleValue > $max) from accumulate(
    SensorReading($temp : temperature) over window:length(100),
    average($temp))
then
  // sound the alarm
end

Explicit expiration is set with a declare statement and the metadata @expires tag.

For example:

declare StockTick
  @expires(30m)
end

Declaring expiration against an event-type will, in the above example StockTick events, remove any StockTick events from the session automatically after the defined expiration time if no rules still need the events.

The rules engine can calculate the expiration offset for a given event implicitly by analyzing the temporal constraints in the rules.

For example:

rule "correlate orders"
when
  $bo : BuyOrder($id : id)
  $ae : AckOrder(id == $id, this after[0,10s] $bo)
then
  // do something
end

For the example rule, the rules engine automatically calculates that whenever a BuyOrder event occurs it needs to store the event for up to ten seconds to wait for the matching AckOrder event, making the implicit expiration offset for BuyOrder events ten seconds. An AckOrder event can only match an existing BuyOrder event making its implicit expiration offset zero seconds.

The engine analyzes the entire rule-base to find the offset for every event-type. Whenever an implicit expiration clashes with an explicit expiration the engine uses the greater value of the two.