An event is a fact (object) that has a few distinguishing characteristics:
- Usually immutable: since, by the previously discussed definition, events are a record of a state change in the application domain, i.e., a record of something that already happened, and the past can not be “changed”, events are logically immutable.
- Strong temporal constraints: rules involving events usually require the correlation of multiple events, specially temporal correlations where events are said to happen at some point in time relative to other events. For example, event A happens before or after event B or time difference between event A and B.
- Managed lifecycle: Due to their immutable nature and temporal constraints, events will match other events and facts during a limited window of time. After that time the event is garbage collected.
Complex events processing, or CEP, can be seen as a system able to consume events from multiple streaming sources, correlate them considering both temporal constraints and application logic and then apply consequence logic. The special part is temporal constraints, which sub-divide to 2 kinds:
- Absolute constraints: specify when the event becomes active or expired regardless of other events
- Relative constraints: specify correlation between multiple events based on temporal distance between them
There are 2 ways to create rules based on temporal constraints:
- Using sliding windows
- Using temporal operators
Sliding Windows are a way to scope the events of interest by defining a window that is constantly moving.
Sliding Time Windows: allow the user to write rules that will only match events occurring in the last X time units.
Aggregating values over time windows:
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 ) from entry-point SRStream, average( $temp ) ) then // sound the alarm insertLogical(new Alarm()) end
The engine will automatically disregard any SensorReading older than 10 minutes and keep the calculated average consistent.
Sliding Length Windows: work the same way as Time Windows, but consider events based on order of their insertion into the session and their count.
Aggregating values over length windows:
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 ) from entry-point SRStream, average( $temp ) ) then // sound the alarm insertLogical(new Alarm()) end
The engine will keep only consider the last 100 readings to calculate the average temperature.
Every event in CEP system happened at some point of time, so it has a timestamp. Drools implements 13 temporal operators and their negation that allow us to compare event timestamps in different ways. Consider the following rule:
rule "landing system" when $flight : FlightStatus() from entry-point "flight-landing" FlightStatus(this!=$flight, this before[0m, 3m] $flight) from entry-point "flight-landing" then insertLogical(new KeepFlying($flight)); end
The rule controls the flights that will be landing, and redirect flights ready to land when there is another already landing.
The pattern in which we are interested is the second one, which makes a correlation between the flight already landing and a new one that wants to land. But this new incoming flight is correlated with the first one to not allow more than one landing in a period of three minutes.
These events are correlated using the pattern constraint
this before[0m, 3m] $flight, where both are being referenced, using the
this keyword and the
$flight variable, and the distance range is declared with the
[0m, 3m] value. Being even more descriptive, the pattern used to correlate these events can be translated to a simpler equation:
0m <= IncomingFlight.startTimestamp - LandingFlight.endTimeStamp <= 3m