Distributed Cache and CEP with Storm - Real-time Analytics with Storm and Cassandra - page 243

EPStatementcepStatement = cepAdm.createEPL("select *
from " + "CasinoEvent(symbol='Roulette').win:length(2) "
+ "having avg(prizeAmount) > 10000.0");
cepStatement.addListener(new CEPListener());
// We generate a few ticks...
for (inti = 0; i < 5; i++) {
GenerateRandomCasinoWinEvent(cepRT);
}
}
}
Here is the snippet of the output:
In the preceding snippet, CEPListener is the implementation of updateListener
(which listens for the arrival of the event), newData has the stream of one or more new
arriving events, and oldData has the previous state of the stream, that is, before the ar-
rival of the current trigger to the listener.
In the main method, we can load the Esper configuration or, as shown in our preceding
case, create a default configuration. Then, we create an Esper runtime engine instance and
bind the EQL query to it.
If you look at the cepStatement.addListener(new CEPListener()) state-
ment in the preceding code, you will see that we are also binding the listener to the state-
ment, thus wiring all the pieces together.