Mule Drools CEP Example

broker.drl

package org.drools.examples.broker

import org.mule.MessageExchangePattern;
import org.mule.example.cep.Company
import org.mule.example.cep.StockTick

global org.mule.module.bpm.MessageService mule;

# default dialect for the semantic code will be MVEL
dialect "mvel"

# tells the engine that a StockTick instance will assume the
# role (semantics) of events and that the default retention
# policy will be 2 minutes
declare StockTick
    @role( event )
    @expires( 2m )
end

# One can even declare helper facts to make rules easier and
# simpler to write and maintain
declare Statistics
    symbol : String @key()
    average : double
end

# Default alert threshold, this can be updated at run-time
rule "Setup alert threshold"
when
   not( Double() )
then
   Double t = new Double(0.07);
   insert ( t );
end

rule "Setup statistics"
when
   $c : Company( $s : symbol )
   not( Statistics( symbol == $s ) )
then
   Statistics s = new Statistics();
   s.symbol = $s;
   insert( s );
end

# a simple rule to show that it is possible to join
# events from an entry-point (stream) with facts
# present in the working memory
rule "Update stock price"
    lock-on-active

when
    $cp : Company( $sb : symbol )
    $st : StockTick( symbol == $sb, $pr : price ) from entry-point "StockTick stream"
then
    // This shows an update on working memory facts with data from joined events
    modify( $cp ) { currentPrice = $pr }

    // Although events are considered immutable, a common pattern is to use a class
    // to represent an event and enrich that event instance with data derived from other facts/events.
    // Bellow we "enrich" the event instance with the percentual change in the price,
    // based on the previous price
    modify( $st ) { delta = Math.abs($cp.delta) }
end

# this rule shows a trick to get the last available event as well as
# how to call global services from the consequence
rule "sudden drop"
    salience -10
when
    $threshold : Double()
    $st : StockTick( $sb : symbol, $ts : timestamp, $pr : price, $dt : delta >= $threshold ) from entry-point "StockTick stream"
    not( StockTick( symbol == $sb, timestamp > $ts ) from entry-point "StockTick stream" )
then
    msg = "delta > " + StockTick.percent($threshold) + " : " + $sb + "  $" + $pr + " (" + StockTick.percent($dt) + ")";
    mule.generateMessage("alerts", msg, null, MessageExchangePattern.ONE_WAY);
end