Read Blog Post

Read Blog Post

Contributed on November 28, 2022

59

223

Drools Reactive Messaging processing

Blog: Drools & jBPM Blog

The latest Drools 8.31.0.Final comes with a Reactive Messaging example, which demonstrates reactively consuming messages from a Kafka topic, firing rules and then sending result messages to another Kafka topic. You can find it in https://github.com/kiegroup/drools/tree/main/drools-drl-quarkus-extension/drools-drl-quarkus-examples/drools-drl-quarkus-examples-reactive. This blog post explains how it works.

How To Run the Example

Clone drools repository

$ git clone https://github.com/kiegroup/drools.git

Go to the example directory

$ cd drools-drl-quarkus-extension/drools-drl-quarkus-examples/drools-drl-quarkus-examples-reactive/

docker-compose.yml is provided for a quick test with Kafka and Kafdrop.

$ docker-compose up -d

Build and start the application with dev mode.

$ mvn clean compile quarkus:dev

Open another terminal and send a message to a Kafka topic events

$ echo '{"type":"temperature","value":35}' | kafka-console-producer.sh --broker-list localhost:9092 --topic events

You will see STDOUT log in the terminal where the application is running. It means the message is consumed and the rule is fired.

rule IncomingEvent fired : Event [type=temperature, value=35]

If you access Kafdrop http://localhost:9000 with a browser, which is already started by docker-compose, you will find alerts topic. You can confirm the result message sent by the rule.

{"severity":"warning","message":"Event [type=temperature, value=35]"}

To shutdown the app, press Ctrl+C on the terminal.

To shutdown Kafka and Kafdrop,

$ docker-compose down

How it works

Here is the diagram of this example architecture.

The important part is how to connect Drools DataSources with Kafka topics. Thanks to Quarkus’ reactive messaging support, we can achieve it with a very small amount of codes.

In order to get reactive messaging support with Kafka, you just need to have a dependency quarkus-smallrye-reactive-messaging-kafka.

    <dependency>
      <groupId>io.quarkus</groupId>
      <artifactId>quarkus-smallrye-reactive-messaging-kafka</artifactId>
    </dependency>

Take a look at Adaptor class.

@Startup
@ApplicationScoped
public class Adaptor {

    @Inject
    RuleUnit<AlertingUnit> ruleUnit;

    AlertingUnit alertingUnit;
    RuleUnitInstance<AlertingUnit> ruleUnitInstance;

    @Inject
    @Channel("alerts")
    Emitter<Alert> emitter;

    @PostConstruct
    void init() {
        this.alertingUnit = new AlertingUnit();
        this.ruleUnitInstance = ruleUnit.createInstance(alertingUnit);
        alertingUnit.getAlertData().subscribe(DataObserver.of(emitter::send));
    }

    @Incoming("events")
    public void receive(Event event) throws InterruptedException {
        alertingUnit.getEventData().append(event);
        ruleUnitInstance.fire();
    }
}

With @Incoming("events"), you can receive Event object from Kafka topic events. This association is configured in application.properties.

mp.messaging.incoming.events.connector=smallrye-kafka
mp.messaging.incoming.events.topic=events
mp.messaging.incoming.events.value.deserializer=org.drools.quarkus.ruleunit.examples.reactive.EventDeserializer

mp.messaging.outgoing.alerts.connector=smallrye-kafka
mp.messaging.outgoing.alerts.topic=alerts
mp.messaging.outgoing.alerts.value.serializer=io.quarkus.kafka.client.serialization.ObjectMapperSerializer

When "IncomingEvent" rule is fired, an Alert object is appended to DataStream alertData.

rule IncomingEvent
  when
    $e : /eventData [ type == "temperature", value >= 30 ]
  then
    System.out.println("rule IncomingEvent fired : "+ $e);
    Alert alert = new Alert( "warning", $e.toString() );
    alertData.append( alert );
end

As you see in Adaptor.init(), the Alert object will be sent to Kafka topic alerts.

Now you can develop a rule service which consumes messages reactively in a micro service environment.

Going forward

We plan to develop Specialized DataSources to connect out-of-the-box Drools rule units with external frameworks and tools, e.g. reads/writes from a Kafka topic even without a glue code like the Adaptor class. Stay tuned!

The post Drools Reactive Messaging processing appeared first on KIE Community.