Drools banking skeleton application wired to a kafka topic / published by https://github.com/dacr/code-examples-manager #cf0d98c9-9383-4297-b6f3-ff1fed217e23/aebf4fe5f3fea0ab39eb0a6790d424aa1b025f52

drools-kb-banking-skeleton.sc

// summary : Drools banking skeleton application wired to a kafka topic
// keywords : scala, drools, mvel, ai, knowledgebase, kafka, cep, complex-event-processing
// publish : gist
// authors : David Crosson
// license : Apache License Version 2.0 (https://www.apache.org/licenses/LICENSE-2.0.txt)
// id : cf0d98c9-9383-4297-b6f3-ff1fed217e23
// execution : scala ammonite script (http://ammonite.io/) - run as follow 'amm scriptname.sc'
// created-on : 2019-10-03T08:46:39+02:00
// managed-by : https://github.com/dacr/code-examples-manager

/*
  PLEASE FIRST START A REPL SESSION AND EVAL THE CONTENT OF
  the "drools-kb-banking-events-stream-generator.sc" SCRIPT
  to start an embedded kafka server and get simple way to generate events
  (See : https://gist.github.com/dacr/da70d346f0e477c1a6e78bc15405cfd3)
  then quickly makes such call on THE REPL :
  Generator.operate("send", "aaa-bbb", "bbb-ddd", "Give me cash", 32)
  And see the effect on this script
 */

import $ivy.`fr.janalyse::drools-scripting:1.0.11`
import $ivy.`org.apache.kafka::kafka:2.6.0`
import fr.janalyse.droolscripting._

import org.apache.kafka.clients.consumer.KafkaConsumer

val drl =
  """package banking
    |
    |declare OperationRequest @role(event) @timestamp(timestamp)
    |  timestamp: java.util.Date
    |  operationId: String
    |  fromAccountId: String
    |  toAccountId: String
    |  label:String
    |  amount: double
    |end
    |
    |rule "Log events"
    |when
    |  OperationRequest($label:label, $amount:amount)
    |then
    |  System.out.println(""+$label+" "+$amount);
    |end
    |
    |""".stripMargin

val config = DroolsEngineConfig(
  withDroolsLogging = false,
  equalsWithIdentity = false,
  pseudoClock = false
)
val engine = DroolsEngine(drl,config)

val props = new java.util.Properties()
props.put("bootstrap.servers", "127.0.0.1:4242")
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer")
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer")
props.put("auto.offset.reset", "latest") // latest message; versus "earliest"
props.put("group.id", "consumer-group")

import scala.jdk.CollectionConverters._
import java.time.Duration

val consumer = new KafkaConsumer[String, String](props)
consumer.subscribe(List("banking-operations").asJavaCollection)
while (true) {
  val record = consumer.poll(Duration.ofSeconds(1)).asScala
  for (data <- record.iterator) {
    val json = data.value()
    engine.insertJson(json, "banking.OperationRequest")
  }
  engine.fireAllRules()
}