In order to create a CEP Drools runtime, we need to provide a few extra configurations from the default initialization. The first one we need to add is the event processing mode of the Kie Base that we'll use.

The event processing mode will determine the manner in which the new data inserted into the runtime will be processed. The default event processing mode is called the CLOUD mode and basically treats any incoming data the same way, regardless of being events or simple facts. This means that the runtime will not understand the concept of events, so that we cannot use it for CEP.

We will need to configure our Kie Base to use the STREAM event processing mode. This configuration will inform the runtime that it should manage events and keep them internally ordered by their timestamp. Due to this ordering, we are able to run time operations against events and use sliding windows on them.

There are many ways to configure the STREAM event processing mode in a Kie Base. The simplest one is to do it directly in the kmodule.xml as an attribute of the kbase tag:

<kbase name="cepKbase" eventProcessingMode="stream" packages="chapter06.cep">
    <ksession name="cepKsession"/>
</kbase>

In this way, we can later on use the Kie Base or Kie Session directly from the corresponding Kie Container and the configuration for its runtime will be using the STREAM event processing mode. We can see an example of this configuration in the chapter-06/chapter-06-rules/src/main/resources/META-INF/kmodule.xml file.

Another way to configure this event processing mode is programmatically. To do so, we will make use of a KieBaseConfiguration bean and its setOption method, as follows:

KieServices ks = KieServices.Factory.get();
    KieContainer kc = ks.getKieClasspathContainer();
    KieBaseConfiguration kbconf = ks.newKieBaseConfiguration();
    kbconf.setOption(EventProcessingOption.STREAM);
    KieBase kbase = kc.newKieBase(kbconf, null);

In the previous example, we used the Kie classpath container for simplicity, but we could be using any Kie Container to create the Kie Base. It is very useful when defining dynamic knowledge modules.

Once we define a Kie Base with the STREAM processing mode, we will need to understand the different options that we will have to run a KIE Session and manage our CEP scenarios.

The first thing we'll need to understand when running our CEP rules is whether or not we need to run them in a continuous or discrete fashion. The main difference between the two is as follows:

These two ways of firing rules will depend entirely on our case and the situations that might trigger a rule. If we have a scenario where the absence of events will trigger a rule, or to put it in other words, the absence of events could be abstracted into another event, then you should use continuous rule firing. If, on the other hand, the only thing that could trigger new rules is the insertion of new events into the Kie Session, then discrete rule firing will be enough for our case.

Let's discuss a couple of examples to understand these two scenarios.

First, let's discuss a common case for discrete rule firing: fraud detection. Most fraud detection systems will work based on the cumulative information from transactions. Basically, if we have a specific number of transactions with specific parameters, we might consider the possibility of fraud. In this type of scenario, the only way we would trigger a rule is if we insert a new transaction to match the conditions of our rules. For this case, we can just call fireAllRules after every transaction or transaction batch is inserted in our Kie Session. No rules will need to fire if they don't do it immediately after adding the latest data.

In a different scenario, let's imagine that a heart monitor is sending events to our CEP engine. About once every second, we get a heart beat event from an oscilloscope. If we get events too close in time or at an irregular pace, we might detect a stroke or arrhythmia complex events. What would happen if we wanted to detect whether the heart stops beating? This case would be a cardiac arrest event. If we want to detect it, our system will need the ability to fire rules when no events are being inserted. This type of scenario is typical of a continuous rule firing case.

One more useful configuration when creating Kie Sessions to run CEP-based scenarios is the possibility to configure its internal clock. By default, Kie Sessions will understand the passing of time using the clock of the machine on which it is running. However, this is just one of the two available configurations, called runtime clock. The other configuration allows us to define a clock controlled by the application, called pseudo clock.

Both runtime and pseudo session clocks only move in one direction (forward in time). However, the pseudo clock will only do so if you call a specific method on it, called advanceTime. Here's a small example of how you can use the pseudo clock from inside the Kie Session:

SessionPseudoClock clock = ksession.getSessionClock();
clock.advanceTime(2, TimeUnit.HOURS);
clock.advanceTime(5, TimeUnit.MINUTES);

In the previous example, we told the clock to advance two hours and five minutes. These two calls will take only milliseconds, which make this clock an excellent option for testing CEP scenarios. If you had to check the case where two events with default timestamps (the moment they are inserted in the Kie Session) happen apart from each other by two hours, the pseudo clock would let you run this case almost immediately, while the runtime clock would need at least two hours to run.

In order to use the pseudo clock in our Kie Session, we need to provide a specific configuration for it through the kmodule.xml file:

<kbase name="cepKbase" eventProcessingMode="stream" packages="chapter06.cep">
    <ksession name="cepKsession" clockType="pseudo"/>
</kbase>

We can even use it through a KieSessionConfiguration bean:

KieServices ks = KieServices.Factory.get();
KieContainer kc = ks.getKieClasspathContainer();
KieSessionConfiguration ksconf = ks.newKieSessionConfiguration();
ksconf.setOption(ClockTypeOption.get(
ClockType.PSEUDO_CLOCK.getId()));
KieSession ksession = kc.newKieSession(ksconf);

You can see an example of this code running in the chapter-06-tests project of the code bundle.

Even if the most common use for the pseudo clock is to test, another case where it is commonly used is, oddly, distributed production environments. The reason for this is that, for large environments where CEP scenarios might be executed in multiple servers, the pseudo clock is usually used to easily synchronize the clocks of all the sessions in different servers. An extra thread or server can have the responsibility of invoking a ticking mechanism in each server at almost the same time and each server with a Kie Session can advance the time to make sure that they all are operating at almost the same clock values. This is usually simpler than having all the internal clocks of multiple servers synchronized, which is a requirement when rules are in charge of real-time decisions.