KOGITO-5533: implement evaluation of rules via CloudEvents

addons/common/events/pom.xml

@@ -13,6 +13,7 @@
 
   <modules>
     <module>decisions</module>
+    <module>rules</module>
   </modules>
 
   <profiles>

addons/common/events/rules/pom.xml

@@ -0,0 +1,47 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+  <modelVersion>4.0.0</modelVersion>
+  <parent>
+    <groupId>org.kie.kogito</groupId>
+    <artifactId>kogito-addons-events-parent</artifactId>
+    <version>2.0.0-SNAPSHOT</version>
+  </parent>
+
+  <artifactId>kogito-addons-events-rules</artifactId>
+  <name>Kogito :: Add-Ons :: Events :: Event-Driven Rules</name>
+  <description>Trigger evaluation of rule models via events</description>
+
+  <dependencies>
+    <dependency>
+      <groupId>org.kie.kogito</groupId>
+      <artifactId>kogito-api</artifactId>
+    </dependency>
+    <dependency>
+      <groupId>org.kie.kogito</groupId>
+      <artifactId>kogito-events-api</artifactId>
+    </dependency>
+    <dependency>
+      <groupId>org.kie.kogito</groupId>
+      <artifactId>kogito-addons-cloudevents-utils</artifactId>
+    </dependency>
+
+    <dependency>
+      <groupId>org.junit.jupiter</groupId>
+      <artifactId>junit-jupiter-engine</artifactId>
+      <scope>test</scope>
+    </dependency>
+    <dependency>
+      <groupId>org.mockito</groupId>
+      <artifactId>mockito-core</artifactId>
+      <scope>test</scope>
+    </dependency>
+    <dependency>
+      <groupId>org.slf4j</groupId>
+      <artifactId>slf4j-simple</artifactId>
+      <scope>test</scope>
+    </dependency>
+  </dependencies>
+
+</project>

addons/common/events/rules/src/main/java/org/kie/kogito/eventdriven/rules/AbstractEventDrivenQueryExecutor.java

@@ -0,0 +1,75 @@
+/*
+ * Copyright 2021 Red Hat, Inc. and/or its affiliates.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *       http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.kie.kogito.eventdriven.rules;
+
+import org.kie.kogito.cloudevents.CloudEventUtils;
+import org.kie.kogito.rules.RuleUnit;
+import org.kie.kogito.rules.RuleUnitData;
+import org.kie.kogito.rules.RuleUnitInstance;
+import org.kie.kogito.rules.RuleUnitQuery;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+
+import io.cloudevents.CloudEvent;
+
+public abstract class AbstractEventDrivenQueryExecutor<D extends RuleUnitData, R> implements EventDrivenQueryExecutor {
+
+    private RuleUnit<D> ruleUnit;
+    private String queryName;
+    private Class<? extends RuleUnitQuery<R>> queryClass;
+    private Class<D> dataClass;
+
+    protected AbstractEventDrivenQueryExecutor() {
+    }
+
+    protected AbstractEventDrivenQueryExecutor(RuleUnit<D> ruleUnit, String queryName, Class<? extends RuleUnitQuery<R>> queryClass, Class<D> dataClass) {
+        this.ruleUnit = ruleUnit;
+        this.queryName = queryName;
+        this.queryClass = queryClass;
+        this.dataClass = dataClass;
+    }
+
+    protected void setup(RuleUnit<D> ruleUnit, String queryName, Class<? extends RuleUnitQuery<R>> queryClass, Class<D> dataClass) {
+        this.ruleUnit = ruleUnit;
+        this.queryName = queryName;
+        this.queryClass = queryClass;
+        this.dataClass = dataClass;
+    }
+
+    @Override
+    public String getRuleUnitId() {
+        return ruleUnit.id();
+    }
+
+    @Override
+    public String getQueryName() {
+        return queryName;
+    }
+
+    @Override
+    public Object executeQuery(CloudEvent input, ObjectMapper mapper) {
+        return CloudEventUtils.decodeData(input, dataClass, mapper)
+                .map(this::executeQuery)
+                .orElseThrow(IllegalArgumentException::new);
+    }
+
+    private R executeQuery(D input) {
+        RuleUnitInstance<D> instance = ruleUnit.createInstance(input);
+        R response = instance.executeQuery(queryClass);
+        instance.dispose();
+        return response;
+    }
+}

addons/common/events/rules/src/main/java/org/kie/kogito/eventdriven/rules/EventDrivenQueryExecutor.java

@@ -0,0 +1,30 @@
+/*
+ * Copyright 2021 Red Hat, Inc. and/or its affiliates.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *       http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.kie.kogito.eventdriven.rules;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+
+import io.cloudevents.CloudEvent;
+
+public interface EventDrivenQueryExecutor {
+
+    String getRuleUnitId();
+
+    String getQueryName();
+
+    Object executeQuery(CloudEvent input, ObjectMapper mapper);
+
+}

addons/common/events/rules/src/main/java/org/kie/kogito/eventdriven/rules/EventDrivenRulesController.java

@@ -0,0 +1,233 @@
+/*
+ * Copyright 2021 Red Hat, Inc. and/or its affiliates.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *       http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.kie.kogito.eventdriven.rules;
+
+import java.net.URI;
+import java.util.Map;
+import java.util.Optional;
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionStage;
+import java.util.stream.Collectors;
+import java.util.stream.StreamSupport;
+
+import org.kie.kogito.cloudevents.CloudEventUtils;
+import org.kie.kogito.cloudevents.extension.KogitoExtension;
+import org.kie.kogito.conf.ConfigBean;
+import org.kie.kogito.event.EventEmitter;
+import org.kie.kogito.event.EventReceiver;
+import org.kie.kogito.event.SubscriptionInfo;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+
+import io.cloudevents.CloudEvent;
+import io.cloudevents.core.provider.ExtensionProvider;
+
+/**
+ * This class must always have exact FQCN as <code>org.kie.kogito.eventdriven.rules.EventDrivenRulesController</code>
+ * for code generation plugins to correctly detect if this addon is enabled.
+ */
+public class EventDrivenRulesController {
+
+    public static final String REQUEST_EVENT_TYPE = "RulesRequest";
+    public static final String RESPONSE_EVENT_TYPE = "RulesResponse";
+    public static final String RESPONSE_ERROR_EVENT_TYPE = "RulesResponseError";
+
+    private static final Logger LOG = LoggerFactory.getLogger(EventDrivenRulesController.class);
+
+    private Map<String, EventDrivenQueryExecutor> executors;
+    private ConfigBean config;
+    private EventEmitter eventEmitter;
+    private EventReceiver eventReceiver;
+    private ObjectMapper mapper;
+
+    protected EventDrivenRulesController() {
+    }
+
+    protected EventDrivenRulesController(Iterable<EventDrivenQueryExecutor> executors, ConfigBean config, EventEmitter eventEmitter, EventReceiver eventReceiver, ObjectMapper mapper) {
+        this.executors = buildExecutorsMap(executors);
+        this.config = config;
+        this.eventEmitter = eventEmitter;
+        this.eventReceiver = eventReceiver;
+        this.mapper = mapper;
+    }
+
+    protected void setup(Iterable<EventDrivenQueryExecutor> executors, ConfigBean config, EventEmitter eventEmitter, EventReceiver eventReceiver, ObjectMapper mapper) {
+        this.executors = buildExecutorsMap(executors);
+        this.config = config;
+        this.eventEmitter = eventEmitter;
+        this.eventReceiver = eventReceiver;
+        this.mapper = mapper;
+        setup();
+    }
+
+    protected void setup() {
+        eventReceiver.subscribe(this::handleRequest,
+                new SubscriptionInfo<>(CloudEventUtils.Mapper.mapper()::readValue, CloudEvent.class));
+    }
+
+    private CompletionStage<Void> handleRequest(CloudEvent event) {
+        validateRequest(event)
+                .flatMap(this::buildEvaluationContext)
+                .map(this::processRequest)
+                .flatMap(this::buildResponseCloudEvent)
+                .flatMap(CloudEventUtils::toDataEvent)
+                .ifPresent(e -> eventEmitter.emit(e, (String) e.get("type"), Optional.empty()));
+        return CompletableFuture.completedFuture(null);
+    }
+
+    private Optional<CloudEvent> validateRequest(CloudEvent event) {
+        return Optional.ofNullable(event).filter(e -> REQUEST_EVENT_TYPE.equals(e.getType()));
+    }
+
+    private Optional<EvaluationContext> buildEvaluationContext(CloudEvent event) {
+        KogitoExtension kogitoExtension = ExtensionProvider.getInstance().parseExtension(KogitoExtension.class, event);
+        Map<String, Object> data = CloudEventUtils.decodeMapData(event, String.class, Object.class).orElse(null);
+
+        if (kogitoExtension == null) {
+            LOG.warn("Received CloudEvent(id={} source={} type={}) with null Kogito extension", event.getId(), event.getSource(), event.getType());
+        }
+
+        if (data == null) {
+            LOG.warn("Received CloudEvent(id={} source={} type={}) with null data", event.getId(), event.getSource(), event.getType());
+        }
+
+        return Optional.of(new EvaluationContext(event, kogitoExtension));
+    }
+
+    private EvaluationContext processRequest(EvaluationContext ctx) {
+        if (!ctx.isValidRequest()) {
+            ctx.setResponseError(RulesResponseError.BAD_REQUEST);
+            return ctx;
+        }
+
+        Optional<EventDrivenQueryExecutor> optExecutor = getExecutor(ctx.getRuleUnitId(), ctx.getQueryName());
+        if (!optExecutor.isPresent()) {
+            ctx.setResponseError(RulesResponseError.QUERY_NOT_FOUND);
+            return ctx;
+        }
+
+        EventDrivenQueryExecutor executor = optExecutor.get();
+        try {
+            Object result = executor.executeQuery(ctx.getRequestCloudEvent(), mapper);
+            ctx.setQueryResult(result);
+        } catch (RuntimeException e) {
+            ctx.setResponseError(RulesResponseError.INTERNAL_EXECUTION_ERROR);
+        }
+
+        return ctx;
+    }
+
+    private Optional<EventDrivenQueryExecutor> getExecutor(String ruleUnitId, String queryName) {
+        return Optional.ofNullable(executors.get(buildExecutorId(ruleUnitId, queryName)));
+    }
+
+    private Optional<CloudEvent> buildResponseCloudEvent(EvaluationContext ctx) {
+        String id = UUID.randomUUID().toString();
+        URI source = buildResponseCloudEventSource(ctx);
+        String subject = ctx.getRequestCloudEvent().getSubject();
+
+        KogitoExtension kogitoExtension = new KogitoExtension();
+        kogitoExtension.setRuleUnitId(ctx.getRuleUnitId());
+        kogitoExtension.setRuleUnitQuery(ctx.getQueryName());
+
+        if (ctx.isResponseError()) {
+            String data = Optional.ofNullable(ctx.getResponseError()).map(RulesResponseError::name).orElse(null);
+            return CloudEventUtils.build(id, source, RESPONSE_ERROR_EVENT_TYPE, subject, data, kogitoExtension);
+        }
+
+        return CloudEventUtils.build(id, source, RESPONSE_EVENT_TYPE, subject, ctx.getQueryResult(), kogitoExtension);
+    }
+
+    private URI buildResponseCloudEventSource(EvaluationContext ctx) {
+        return CloudEventUtils.buildDecisionSource(config.getServiceUrl(), ctx.getQueryName());
+    }
+
+    private static String buildExecutorId(String ruleUnitId, String queryName) {
+        return String.format("%s#%s", ruleUnitId, queryName);
+    }
+
+    private static Map<String, EventDrivenQueryExecutor> buildExecutorsMap(Iterable<EventDrivenQueryExecutor> iterable) {
+        return StreamSupport.stream(iterable.spliterator(), false)
+                .collect(Collectors.toMap(e -> buildExecutorId(e.getRuleUnitId(), e.getQueryName()), e -> e));
+    }
+
+    private static class EvaluationContext {
+
+        private final CloudEvent requestCloudEvent;
+        private final String ruleUnitId;
+        private final String queryName;
+        private final boolean validRequest;
+
+        private RulesResponseError responseError;
+        private Object queryResult;
+
+        public EvaluationContext(CloudEvent requestCloudEvent, KogitoExtension requestKogitoExtension) {
+            this.requestCloudEvent = requestCloudEvent;
+
+            this.ruleUnitId = Optional.ofNullable(requestKogitoExtension)
+                    .map(KogitoExtension::getRuleUnitId)
+                    .orElse(null);
+            this.queryName = Optional.ofNullable(requestKogitoExtension)
+                    .map(KogitoExtension::getRuleUnitQuery)
+                    .orElse(null);
+
+            this.validRequest = requestCloudEvent != null
+                    && requestKogitoExtension != null
+                    && ruleUnitId != null && !ruleUnitId.isEmpty()
+                    && queryName != null && !queryName.isEmpty();
+        }
+
+        public CloudEvent getRequestCloudEvent() {
+            return requestCloudEvent;
+        }
+
+        public String getRuleUnitId() {
+            return ruleUnitId;
+        }
+
+        public String getQueryName() {
+            return queryName;
+        }
+
+        public boolean isValidRequest() {
+            return validRequest;
+        }
+
+        boolean isResponseError() {
+            return queryResult == null;
+        }
+
+        public RulesResponseError getResponseError() {
+            return responseError;
+        }
+
+        public void setResponseError(RulesResponseError responseError) {
+            this.responseError = responseError;
+        }
+
+        public Object getQueryResult() {
+            return queryResult;
+        }
+
+        public void setQueryResult(Object queryResult) {
+            this.queryResult = queryResult;
+        }
+    }
+
+}