This is an automated email from the ASF dual-hosted git repository.
tkobayas pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/incubator-kie.git
The following commit(s) were added to refs/heads/main by this push:
new 4c0deeab712 [incubator-kie-7079] RuleUnit DSL: Support STREAM event
processing and CEP temporal operators (#7104)
4c0deeab712 is described below
commit 4c0deeab712226ac2f674dda8a7a831e977ed2fd
Author: Toshiya Kobayashi <[email protected]>
AuthorDate: Mon Sep 14 17:48:24 2026 +0900
[incubator-kie-7079] RuleUnit DSL: Support STREAM event processing and CEP
temporal operators (#7104)
Read @EventProcessing annotation in the DSL provider and pass
EventProcessingOption when creating the KieBase. Add bounded after()
and before() temporal methods to Pattern2Def, rejecting unsupported
calls on aggregate-result builders.
Co-authored-by: Claude Opus 4.6 (1M context) <[email protected]>
---
.../ruleunits/dsl/RuleUnitProviderForDSL.java | 12 +-
.../dsl/accumulate/AccumulatePattern2.java | 6 +
.../ruleunits/dsl/accumulate/GroupByPattern1.java | 6 +
.../drools/ruleunits/dsl/patterns/Pattern2Def.java | 18 +
.../ruleunits/dsl/patterns/Pattern2DefImpl.java | 17 +
.../java/org/drools/ruleunits/dsl/CepTest.java | 545 +++++++++++++++++++++
.../drools/ruleunits/dsl/domain/StockTick.java} | 42 +-
7 files changed, 631 insertions(+), 15 deletions(-)
diff --git
a/drools-ruleunits/drools-ruleunits-dsl/src/main/java/org/drools/ruleunits/dsl/RuleUnitProviderForDSL.java
b/drools-ruleunits/drools-ruleunits-dsl/src/main/java/org/drools/ruleunits/dsl/RuleUnitProviderForDSL.java
index 6b8896244c8..964696c8de0 100644
---
a/drools-ruleunits/drools-ruleunits-dsl/src/main/java/org/drools/ruleunits/dsl/RuleUnitProviderForDSL.java
+++
b/drools-ruleunits/drools-ruleunits-dsl/src/main/java/org/drools/ruleunits/dsl/RuleUnitProviderForDSL.java
@@ -30,12 +30,15 @@ import org.drools.ruleunits.api.DataSource;
import org.drools.ruleunits.api.RuleUnit;
import org.drools.ruleunits.api.RuleUnitData;
import org.drools.ruleunits.api.RuleUnitInstance;
+import org.drools.ruleunits.api.conf.EventProcessing;
+import org.drools.ruleunits.api.conf.EventProcessingType;
import org.drools.ruleunits.api.conf.RuleConfig;
import org.drools.ruleunits.impl.EntryPointDataProcessor;
import org.drools.ruleunits.impl.ReteEvaluatorBasedRuleUnitInstance;
import org.drools.ruleunits.impl.RuleUnitProviderImpl;
import org.drools.ruleunits.impl.factory.AbstractRuleUnit;
import org.drools.ruleunits.impl.sessions.RuleUnitExecutorImpl;
+import org.kie.api.conf.EventProcessingOption;
import org.kie.api.runtime.rule.EntryPoint;
import java.util.Map;
@@ -69,7 +72,14 @@ public class RuleUnitProviderForDSL extends
RuleUnitProviderImpl {
public ModelRuleUnit(Class<T> type, Model model, UnitGlobalsResolver
unitGlobalsResolver) {
super(type);
this.unitGlobalsResolver = unitGlobalsResolver;
- this.ruleBase = KieBaseBuilder.createKieBaseFromModel( model );
+ EventProcessing annotation =
type.getAnnotation(EventProcessing.class);
+ if (annotation == null) {
+ this.ruleBase = KieBaseBuilder.createKieBaseFromModel(model);
+ } else {
+ EventProcessingOption option = annotation.value() ==
EventProcessingType.STREAM
+ ? EventProcessingOption.STREAM :
EventProcessingOption.CLOUD;
+ this.ruleBase = KieBaseBuilder.createKieBaseFromModel(model,
option);
+ }
if (DUMP_GENERATED_RETE) {
ReteDumper.dumpRete(this.ruleBase);
}
diff --git
a/drools-ruleunits/drools-ruleunits-dsl/src/main/java/org/drools/ruleunits/dsl/accumulate/AccumulatePattern2.java
b/drools-ruleunits/drools-ruleunits-dsl/src/main/java/org/drools/ruleunits/dsl/accumulate/AccumulatePattern2.java
index d9e2cb0e296..05605972fff 100644
---
a/drools-ruleunits/drools-ruleunits-dsl/src/main/java/org/drools/ruleunits/dsl/accumulate/AccumulatePattern2.java
+++
b/drools-ruleunits/drools-ruleunits-dsl/src/main/java/org/drools/ruleunits/dsl/accumulate/AccumulatePattern2.java
@@ -18,6 +18,7 @@
*/
package org.drools.ruleunits.dsl.accumulate;
+import org.drools.model.functions.temporal.TemporalPredicate;
import org.drools.model.view.ViewItem;
import org.drools.ruleunits.dsl.patterns.Pattern1DefImpl;
import org.drools.ruleunits.dsl.patterns.Pattern2DefImpl;
@@ -34,6 +35,11 @@ public class AccumulatePattern2<A, B, C> extends
Pattern2DefImpl<A, C> {
this.acc = acc;
}
+ @Override
+ protected Pattern2DefImpl<A, C> addTemporalConstraint(TemporalPredicate
temporalPredicate) {
+ throw new UnsupportedOperationException("Temporal constraints are not
supported on accumulate results");
+ }
+
@Override
public ViewItem toExecModelItem() {
return createAccumulate1Item(patternB, getVariable(), acc);
diff --git
a/drools-ruleunits/drools-ruleunits-dsl/src/main/java/org/drools/ruleunits/dsl/accumulate/GroupByPattern1.java
b/drools-ruleunits/drools-ruleunits-dsl/src/main/java/org/drools/ruleunits/dsl/accumulate/GroupByPattern1.java
index fa49aed51bb..4fc82d8a7e2 100644
---
a/drools-ruleunits/drools-ruleunits-dsl/src/main/java/org/drools/ruleunits/dsl/accumulate/GroupByPattern1.java
+++
b/drools-ruleunits/drools-ruleunits-dsl/src/main/java/org/drools/ruleunits/dsl/accumulate/GroupByPattern1.java
@@ -21,6 +21,7 @@ package org.drools.ruleunits.dsl.accumulate;
import org.drools.model.DSL;
import org.drools.model.Variable;
import org.drools.model.functions.Function1;
+import org.drools.model.functions.temporal.TemporalPredicate;
import org.drools.model.view.ExprViewItem;
import org.drools.model.view.ViewItem;
import org.drools.ruleunits.dsl.patterns.InternalPatternDef;
@@ -45,6 +46,11 @@ public class GroupByPattern1<A, K, V> extends
Pattern2DefImpl<K, V> {
this.acc = acc;
}
+ @Override
+ protected Pattern2DefImpl<K, V> addTemporalConstraint(TemporalPredicate
temporalPredicate) {
+ throw new UnsupportedOperationException("Temporal constraints are not
supported on groupBy results");
+ }
+
@Override
public ViewItem toExecModelItem() {
return createGroupByItem(pattern, groupingFunction, acc,
patternA.getVariable(), patternB.getVariable());
diff --git
a/drools-ruleunits/drools-ruleunits-dsl/src/main/java/org/drools/ruleunits/dsl/patterns/Pattern2Def.java
b/drools-ruleunits/drools-ruleunits-dsl/src/main/java/org/drools/ruleunits/dsl/patterns/Pattern2Def.java
index 38cbd5276b3..c59b14ce290 100644
---
a/drools-ruleunits/drools-ruleunits-dsl/src/main/java/org/drools/ruleunits/dsl/patterns/Pattern2Def.java
+++
b/drools-ruleunits/drools-ruleunits-dsl/src/main/java/org/drools/ruleunits/dsl/patterns/Pattern2Def.java
@@ -18,6 +18,8 @@
*/
package org.drools.ruleunits.dsl.patterns;
+import java.util.concurrent.TimeUnit;
+
import org.drools.model.Index;
import org.drools.model.functions.Block1;
import org.drools.model.functions.Block2;
@@ -39,6 +41,22 @@ public interface Pattern2Def<A, B> extends PatternDef {
<V> Pattern2Def<A, B> filter(String fieldName, Function1<B, V>
leftExtractor, Index.ConstraintType constraintType, Function1<A, V>
rightExtractor);
+ /**
+ * Constrains pattern B to occur after pattern A within inclusive bounds
{@code [min, max]}
+ * converted to the given time unit. The gap is measured from {@code
end(A)} to {@code start(B)}.
+ * Both operands must be events annotated with {@code
@Role(Role.Type.EVENT)}.
+ * Supported time units: MILLISECONDS, SECONDS, MINUTES, HOURS, DAYS.
+ */
+ Pattern2Def<A, B> after(long min, long max, TimeUnit unit);
+
+ /**
+ * Constrains pattern B to occur before pattern A within inclusive bounds
{@code [min, max]}
+ * converted to the given time unit. The gap is measured from {@code
end(B)} to {@code start(A)}.
+ * Both operands must be events annotated with {@code
@Role(Role.Type.EVENT)}.
+ * Supported time units: MILLISECONDS, SECONDS, MINUTES, HOURS, DAYS.
+ */
+ Pattern2Def<A, B> before(long min, long max, TimeUnit unit);
+
<C> Pattern3Def<A, B, C> on(DataSource<C> dataSource);
<C> Pattern3Def<A, B, C> join(Function1<RuleFactory, Pattern1Def<C>>
patternBuilder);
diff --git
a/drools-ruleunits/drools-ruleunits-dsl/src/main/java/org/drools/ruleunits/dsl/patterns/Pattern2DefImpl.java
b/drools-ruleunits/drools-ruleunits-dsl/src/main/java/org/drools/ruleunits/dsl/patterns/Pattern2DefImpl.java
index 9c102d8a708..da7108f82ed 100644
---
a/drools-ruleunits/drools-ruleunits-dsl/src/main/java/org/drools/ruleunits/dsl/patterns/Pattern2DefImpl.java
+++
b/drools-ruleunits/drools-ruleunits-dsl/src/main/java/org/drools/ruleunits/dsl/patterns/Pattern2DefImpl.java
@@ -19,10 +19,12 @@
package org.drools.ruleunits.dsl.patterns;
import java.util.UUID;
+import java.util.concurrent.TimeUnit;
import org.drools.model.Condition;
import org.drools.model.DSL;
import org.drools.model.Index;
+import org.drools.model.functions.temporal.TemporalPredicate;
import org.drools.model.functions.Block1;
import org.drools.model.functions.Block2;
import org.drools.model.functions.Block3;
@@ -72,6 +74,21 @@ public class Pattern2DefImpl<A, B> extends
SinglePatternDef<B> implements Patter
return this;
}
+ @Override
+ public Pattern2DefImpl<A, B> after(long min, long max, TimeUnit unit) {
+ return addTemporalConstraint(DSL.after(min, unit, max, unit));
+ }
+
+ @Override
+ public Pattern2DefImpl<A, B> before(long min, long max, TimeUnit unit) {
+ return addTemporalConstraint(DSL.before(min, unit, max, unit));
+ }
+
+ protected Pattern2DefImpl<A, B> addTemporalConstraint(TemporalPredicate
temporalPredicate) {
+ patternB.constraints.add(patternDef ->
patternDef.expr(UUID.randomUUID().toString(), patternA.variable,
temporalPredicate));
+ return this;
+ }
+
@Override
public <C> Pattern3Def<A, B, C> on(DataSource<C> dataSource) {
return join(rule.on(dataSource));
diff --git
a/drools-ruleunits/drools-ruleunits-dsl/src/test/java/org/drools/ruleunits/dsl/CepTest.java
b/drools-ruleunits/drools-ruleunits-dsl/src/test/java/org/drools/ruleunits/dsl/CepTest.java
new file mode 100644
index 00000000000..10b9253ea79
--- /dev/null
+++
b/drools-ruleunits/drools-ruleunits-dsl/src/test/java/org/drools/ruleunits/dsl/CepTest.java
@@ -0,0 +1,545 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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.drools.ruleunits.dsl;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+
+import org.drools.core.WorkingMemoryEntryPoint;
+import org.drools.core.common.InternalFactHandle;
+import org.drools.core.common.ReteEvaluator;
+import org.drools.model.Index;
+import org.drools.ruleunits.api.DataSource;
+import org.drools.ruleunits.api.DataStream;
+import org.drools.ruleunits.api.RuleUnitInstance;
+import org.drools.ruleunits.api.RuleUnitProvider;
+import org.drools.ruleunits.api.conf.ClockType;
+import org.drools.ruleunits.api.conf.EventProcessing;
+import org.drools.ruleunits.api.conf.EventProcessingType;
+import org.drools.ruleunits.api.conf.RuleConfig;
+import org.drools.ruleunits.dsl.domain.StockTick;
+import org.drools.ruleunits.impl.AbstractRuleUnitInstance;
+import org.junit.jupiter.api.Test;
+import org.kie.api.conf.EventProcessingOption;
+import org.kie.api.runtime.rule.EntryPoint;
+import org.kie.api.runtime.rule.FactHandle;
+import org.kie.api.time.SessionPseudoClock;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.drools.model.Index.ConstraintType.EQUAL;
+import static org.drools.ruleunits.dsl.Accumulators.sum;
+
+public class CepTest {
+
+ // --- Configuration tests ---
+
+ @Test
+ public void streamAnnotationProducesStreamKieBase() {
+ StreamAfterUnit unit = new StreamAfterUnit();
+ RuleConfig config = RuleUnitProvider.get().newRuleConfig();
+ config.setClockType(ClockType.PSEUDO);
+ try (RuleUnitInstance<StreamAfterUnit> instance =
RuleUnitProvider.get().createRuleUnitInstance(unit, config)) {
+ ReteEvaluator evaluator = getEvaluator(instance);
+
assertThat(evaluator.getKnowledgeBase().getRuleBaseConfiguration().getEventProcessingMode())
+ .isEqualTo(EventProcessingOption.STREAM);
+ }
+ }
+
+ @Test
+ public void explicitCloudAnnotationProducesCloudKieBase() {
+ ExplicitCloudUnit unit = new ExplicitCloudUnit();
+ try (RuleUnitInstance<ExplicitCloudUnit> instance =
RuleUnitProvider.get().createRuleUnitInstance(unit)) {
+ ReteEvaluator evaluator = getEvaluator(instance);
+
assertThat(evaluator.getKnowledgeBase().getRuleBaseConfiguration().getEventProcessingMode())
+ .isEqualTo(EventProcessingOption.CLOUD);
+ }
+ }
+
+ @Test
+ public void unannotatedUnitDefaultsToCloud() {
+ UnannotatedUnit unit = new UnannotatedUnit();
+ try (RuleUnitInstance<UnannotatedUnit> instance =
RuleUnitProvider.get().createRuleUnitInstance(unit)) {
+ ReteEvaluator evaluator = getEvaluator(instance);
+
assertThat(evaluator.getKnowledgeBase().getRuleBaseConfiguration().getEventProcessingMode())
+ .isEqualTo(EventProcessingOption.CLOUD);
+ }
+ }
+
+ // --- Temporal constraint: after ---
+
+ @Test
+ public void afterMatchesWithinBounds() {
+ StreamAfterUnit unit = new StreamAfterUnit();
+ RuleConfig config = RuleUnitProvider.get().newRuleConfig();
+ config.setClockType(ClockType.PSEUDO);
+ try (RuleUnitInstance<StreamAfterUnit> instance =
RuleUnitProvider.get().createRuleUnitInstance(unit, config)) {
+ SessionPseudoClock clock = instance.getClock();
+ unit.getStockTicks().append(new StockTick("DROO"));
+ clock.advanceTime(6, TimeUnit.SECONDS);
+ unit.getStockTicks().append(new StockTick("ACME"));
+ instance.fire();
+
+ assertThat(unit.getResults()).hasSize(1);
+
assertThat(unit.getResults().get(0).getCompany()).isEqualTo("ACME");
+ }
+ }
+
+ @Test
+ public void afterDoesNotMatchBelowLowerBound() {
+ StreamAfterUnit unit = new StreamAfterUnit();
+ RuleConfig config = RuleUnitProvider.get().newRuleConfig();
+ config.setClockType(ClockType.PSEUDO);
+ try (RuleUnitInstance<StreamAfterUnit> instance =
RuleUnitProvider.get().createRuleUnitInstance(unit, config)) {
+ SessionPseudoClock clock = instance.getClock();
+ unit.getStockTicks().append(new StockTick("DROO"));
+ clock.advanceTime(4999, TimeUnit.MILLISECONDS);
+ unit.getStockTicks().append(new StockTick("ACME"));
+ instance.fire();
+
+ assertThat(unit.getResults()).isEmpty();
+ }
+ }
+
+ @Test
+ public void afterDoesNotMatchAboveUpperBound() {
+ StreamAfterUnit unit = new StreamAfterUnit();
+ RuleConfig config = RuleUnitProvider.get().newRuleConfig();
+ config.setClockType(ClockType.PSEUDO);
+ try (RuleUnitInstance<StreamAfterUnit> instance =
RuleUnitProvider.get().createRuleUnitInstance(unit, config)) {
+ SessionPseudoClock clock = instance.getClock();
+ unit.getStockTicks().append(new StockTick("DROO"));
+ clock.advanceTime(8001, TimeUnit.MILLISECONDS);
+ unit.getStockTicks().append(new StockTick("ACME"));
+ instance.fire();
+
+ assertThat(unit.getResults()).isEmpty();
+ }
+ }
+
+ @Test
+ public void afterMatchesAtExactLowerBound() {
+ StreamAfterUnit unit = new StreamAfterUnit();
+ RuleConfig config = RuleUnitProvider.get().newRuleConfig();
+ config.setClockType(ClockType.PSEUDO);
+ try (RuleUnitInstance<StreamAfterUnit> instance =
RuleUnitProvider.get().createRuleUnitInstance(unit, config)) {
+ SessionPseudoClock clock = instance.getClock();
+ unit.getStockTicks().append(new StockTick("DROO"));
+ clock.advanceTime(5000, TimeUnit.MILLISECONDS);
+ unit.getStockTicks().append(new StockTick("ACME"));
+ instance.fire();
+
+ assertThat(unit.getResults()).hasSize(1);
+
assertThat(unit.getResults().get(0).getCompany()).isEqualTo("ACME");
+ }
+ }
+
+ @Test
+ public void afterMatchesAtExactUpperBound() {
+ StreamAfterUnit unit = new StreamAfterUnit();
+ RuleConfig config = RuleUnitProvider.get().newRuleConfig();
+ config.setClockType(ClockType.PSEUDO);
+ try (RuleUnitInstance<StreamAfterUnit> instance =
RuleUnitProvider.get().createRuleUnitInstance(unit, config)) {
+ SessionPseudoClock clock = instance.getClock();
+ unit.getStockTicks().append(new StockTick("DROO"));
+ clock.advanceTime(8000, TimeUnit.MILLISECONDS);
+ unit.getStockTicks().append(new StockTick("ACME"));
+ instance.fire();
+
+ assertThat(unit.getResults()).hasSize(1);
+
assertThat(unit.getResults().get(0).getCompany()).isEqualTo("ACME");
+ }
+ }
+
+ @Test
+ public void afterWithNonzeroDurationUsesEndToStart() {
+ StreamAfterUnit unit = new StreamAfterUnit();
+ RuleConfig config = RuleUnitProvider.get().newRuleConfig();
+ config.setClockType(ClockType.PSEUDO);
+ try (RuleUnitInstance<StreamAfterUnit> instance =
RuleUnitProvider.get().createRuleUnitInstance(unit, config)) {
+ SessionPseudoClock clock = instance.getClock();
+ // DROO starts at 0 with 3s duration, so end(DROO) = 3s
+ unit.getStockTicks().append(new StockTick("DROO", 3000));
+ // ACME starts at 9s, gap = start(ACME) - end(DROO) = 9s - 3s =
6s, within [5s, 8s]
+ clock.advanceTime(9, TimeUnit.SECONDS);
+ unit.getStockTicks().append(new StockTick("ACME"));
+ instance.fire();
+
+ assertThat(unit.getResults()).hasSize(1);
+
assertThat(unit.getResults().get(0).getCompany()).isEqualTo("ACME");
+ }
+ }
+
+ @Test
+ public void afterWithMillisecondUnit() {
+ StreamAfterMillisUnit unit = new StreamAfterMillisUnit();
+ RuleConfig config = RuleUnitProvider.get().newRuleConfig();
+ config.setClockType(ClockType.PSEUDO);
+ try (RuleUnitInstance<StreamAfterMillisUnit> instance =
RuleUnitProvider.get().createRuleUnitInstance(unit, config)) {
+ SessionPseudoClock clock = instance.getClock();
+ unit.getStockTicks().append(new StockTick("DROO"));
+ clock.advanceTime(600, TimeUnit.MILLISECONDS);
+ unit.getStockTicks().append(new StockTick("ACME"));
+ instance.fire();
+
+ assertThat(unit.getResults()).hasSize(1);
+
assertThat(unit.getResults().get(0).getCompany()).isEqualTo("ACME");
+ }
+ }
+
+ // --- Temporal constraint: before ---
+
+ @Test
+ public void beforeMatchesWithinBounds() {
+ StreamBeforeUnit unit = new StreamBeforeUnit();
+ RuleConfig config = RuleUnitProvider.get().newRuleConfig();
+ config.setClockType(ClockType.PSEUDO);
+ try (RuleUnitInstance<StreamBeforeUnit> instance =
RuleUnitProvider.get().createRuleUnitInstance(unit, config)) {
+ SessionPseudoClock clock = instance.getClock();
+ // ACME occurs first (the "before" event)
+ unit.getStockTicks().append(new StockTick("ACME"));
+ clock.advanceTime(6, TimeUnit.SECONDS);
+ // DROO occurs second
+ unit.getStockTicks().append(new StockTick("DROO"));
+ instance.fire();
+
+ assertThat(unit.getResults()).hasSize(1);
+
assertThat(unit.getResults().get(0).getCompany()).isEqualTo("ACME");
+ }
+ }
+
+ @Test
+ public void beforeDoesNotMatchOutsideBounds() {
+ StreamBeforeUnit unit = new StreamBeforeUnit();
+ RuleConfig config = RuleUnitProvider.get().newRuleConfig();
+ config.setClockType(ClockType.PSEUDO);
+ try (RuleUnitInstance<StreamBeforeUnit> instance =
RuleUnitProvider.get().createRuleUnitInstance(unit, config)) {
+ SessionPseudoClock clock = instance.getClock();
+ unit.getStockTicks().append(new StockTick("ACME"));
+ clock.advanceTime(4999, TimeUnit.MILLISECONDS);
+ unit.getStockTicks().append(new StockTick("DROO"));
+ instance.fire();
+
+ assertThat(unit.getResults()).isEmpty();
+ }
+ }
+
+ @Test
+ public void beforeWithNonzeroDurationUsesEndToStart() {
+ StreamBeforeUnit unit = new StreamBeforeUnit();
+ RuleConfig config = RuleUnitProvider.get().newRuleConfig();
+ config.setClockType(ClockType.PSEUDO);
+ try (RuleUnitInstance<StreamBeforeUnit> instance =
RuleUnitProvider.get().createRuleUnitInstance(unit, config)) {
+ SessionPseudoClock clock = instance.getClock();
+ // ACME starts at 0 with 3s duration, so end(ACME) = 3s
+ unit.getStockTicks().append(new StockTick("ACME", 3000));
+ // DROO starts at 9s, gap = start(DROO) - end(ACME) = 9s - 3s =
6s, within [5s, 8s]
+ clock.advanceTime(9, TimeUnit.SECONDS);
+ unit.getStockTicks().append(new StockTick("DROO"));
+ instance.fire();
+
+ assertThat(unit.getResults()).hasSize(1);
+
assertThat(unit.getResults().get(0).getCompany()).isEqualTo("ACME");
+ }
+ }
+
+ // --- Explicit expiration ---
+
+ @Test
+ public void expiresRemovesEventAfterDuration() {
+ StreamExpirationUnit unit = new StreamExpirationUnit();
+ RuleConfig config = RuleUnitProvider.get().newRuleConfig();
+ config.setClockType(ClockType.PSEUDO);
+ try (RuleUnitInstance<StreamExpirationUnit> instance =
RuleUnitProvider.get().createRuleUnitInstance(unit, config)) {
+ SessionPseudoClock clock = instance.getClock();
+ unit.getStockTicks().append(new StockTick("DROO"));
+ instance.fire();
+
+ EntryPoint entryPoint = findNonDefaultEntryPoint(instance);
+ assertThat(entryPoint.getFactHandles()).hasSize(1);
+
+ clock.advanceTime(11, TimeUnit.SECONDS);
+ instance.fire();
+
+ assertThat(entryPoint.getFactHandles()).isEmpty();
+ }
+ }
+
+ // --- Combined CEP scenario ---
+
+ @Test
+ public void combinedCepScenario() {
+ StreamAfterUnit unit = new StreamAfterUnit();
+ RuleConfig config = RuleUnitProvider.get().newRuleConfig();
+ config.setClockType(ClockType.PSEUDO);
+ try (RuleUnitInstance<StreamAfterUnit> instance =
RuleUnitProvider.get().createRuleUnitInstance(unit, config)) {
+ SessionPseudoClock clock = instance.getClock();
+
+ // 1. Append a zero-duration DROO event at time zero
+ unit.getStockTicks().append(new StockTick("DROO"));
+
+ // 2. Advance 6 seconds and append ACME
+ clock.advanceTime(6, TimeUnit.SECONDS);
+ unit.getStockTicks().append(new StockTick("ACME"));
+
+ // 3. Fire and assert the after(5, 8, SECONDS) rule records ACME
+ instance.fire();
+ assertThat(unit.getResults()).hasSize(1);
+
assertThat(unit.getResults().get(0).getCompany()).isEqualTo("ACME");
+
+ // 4. Both events should still be in the entry point
+ EntryPoint entryPoint = findNonDefaultEntryPoint(instance);
+ assertThat(entryPoint.getFactHandles()).hasSize(2);
+
+ // 5. Advance another 5 seconds (total = 11s) and fire
+ clock.advanceTime(5, TimeUnit.SECONDS);
+ instance.fire();
+
+ // 6. DROO was inserted at t=0 with @Expires("10s"), so it should
be expired
+ // ACME was inserted at t=6 with @Expires("10s"), so it should
still be present
+ Collection<FactHandle> remaining = entryPoint.getFactHandles();
+ assertThat(remaining).hasSize(1);
+ assertThat(((StockTick) ((InternalFactHandle)
remaining.iterator().next()).getObject()).getCompany())
+ .isEqualTo("ACME");
+ }
+ }
+
+ // --- Unsupported aggregate calls ---
+
+ @Test
+ public void afterOnGroupByThrowsUnsupported() {
+ assertThatThrownBy(() -> {
+ new GroupByTemporalUnit();
+ }).isInstanceOf(UnsupportedOperationException.class)
+ .hasMessageContaining("groupBy");
+ }
+
+ @Test
+ public void afterOnAccumulateThrowsUnsupported() {
+ assertThatThrownBy(() -> {
+ new AccumulateTemporalUnit();
+ }).isInstanceOf(UnsupportedOperationException.class)
+ .hasMessageContaining("accumulate");
+ }
+
+ // --- Helpers ---
+
+ @SuppressWarnings("rawtypes")
+ private ReteEvaluator getEvaluator(RuleUnitInstance<?> instance) {
+ return (ReteEvaluator) ((AbstractRuleUnitInstance)
instance).getEvaluator();
+ }
+
+ private EntryPoint findNonDefaultEntryPoint(RuleUnitInstance<?> instance) {
+ ReteEvaluator evaluator = getEvaluator(instance);
+ for (EntryPoint ep : evaluator.getEntryPoints()) {
+ if (!"DEFAULT".equals(ep.getEntryPointId())) {
+ return ep;
+ }
+ }
+ throw new IllegalStateException("No non-DEFAULT entry point found");
+ }
+
+ // --- Unit definitions ---
+
+ @EventProcessing(EventProcessingType.STREAM)
+ public static class StreamAfterUnit implements RuleUnitDefinition {
+ private final DataStream<StockTick> stockTicks;
+ private final List<StockTick> results = new ArrayList<>();
+
+ public StreamAfterUnit() {
+ this.stockTicks = DataSource.createStream();
+ }
+
+ public DataStream<StockTick> getStockTicks() { return stockTicks; }
+ public List<StockTick> getResults() { return results; }
+
+ @Override
+ public void defineRules(RulesFactory rulesFactory) {
+ // A=$a : /stockTicks [ company == "DROO" ]
+ // B=$b : /stockTicks [ company == "ACME", this after[5s,8s] $a ]
+ // → B after A: gap = start(B) - end(A) must be in [5s, 8s]
+ rulesFactory.rule("ACME after DROO")
+ .on(stockTicks) //
pattern A
+ .filter(StockTick::getCompany, EQUAL, "DROO")
+ .join(rule -> rule.on(stockTicks) //
pattern B
+ .filter(StockTick::getCompany, EQUAL, "ACME"))
+ .after(5, 8, TimeUnit.SECONDS) // B
after A
+ .execute(results, (r, droo, acme) -> r.add(acme));
+ }
+ }
+
+ @EventProcessing(EventProcessingType.STREAM)
+ public static class StreamAfterMillisUnit implements RuleUnitDefinition {
+ private final DataStream<StockTick> stockTicks;
+ private final List<StockTick> results = new ArrayList<>();
+
+ public StreamAfterMillisUnit() {
+ this.stockTicks = DataSource.createStream();
+ }
+
+ public DataStream<StockTick> getStockTicks() { return stockTicks; }
+ public List<StockTick> getResults() { return results; }
+
+ @Override
+ public void defineRules(RulesFactory rulesFactory) {
+ // Same as StreamAfterUnit but with millisecond bounds [500ms,
1000ms]
+ rulesFactory.rule("ACME after DROO millis")
+ .on(stockTicks) //
pattern A
+ .filter(StockTick::getCompany, EQUAL, "DROO")
+ .join(rule -> rule.on(stockTicks) //
pattern B
+ .filter(StockTick::getCompany, EQUAL, "ACME"))
+ .after(500, 1000, TimeUnit.MILLISECONDS) // B
after A
+ .execute(results, (r, droo, acme) -> r.add(acme));
+ }
+ }
+
+ @EventProcessing(EventProcessingType.STREAM)
+ public static class StreamBeforeUnit implements RuleUnitDefinition {
+ private final DataStream<StockTick> stockTicks;
+ private final List<StockTick> results = new ArrayList<>();
+
+ public StreamBeforeUnit() {
+ this.stockTicks = DataSource.createStream();
+ }
+
+ public DataStream<StockTick> getStockTicks() { return stockTicks; }
+ public List<StockTick> getResults() { return results; }
+
+ @Override
+ public void defineRules(RulesFactory rulesFactory) {
+ // A=$a : /stockTicks [ company == "DROO" ]
+ // B=$b : /stockTicks [ company == "ACME", this before[5s,8s] $a ]
+ // → B before A: gap = start(A) - end(B) must be in [5s, 8s]
+ rulesFactory.rule("ACME before DROO")
+ .on(stockTicks) //
pattern A
+ .filter(StockTick::getCompany, EQUAL, "DROO")
+ .join(rule -> rule.on(stockTicks) //
pattern B
+ .filter(StockTick::getCompany, EQUAL, "ACME"))
+ .before(5, 8, TimeUnit.SECONDS) // B
before A
+ .execute(results, (r, droo, acme) -> r.add(acme));
+ }
+ }
+
+ @EventProcessing(EventProcessingType.STREAM)
+ public static class StreamExpirationUnit implements RuleUnitDefinition {
+ private final DataStream<StockTick> stockTicks;
+ private final List<StockTick> results = new ArrayList<>();
+
+ public StreamExpirationUnit() {
+ this.stockTicks = DataSource.createStream();
+ }
+
+ public DataStream<StockTick> getStockTicks() { return stockTicks; }
+ public List<StockTick> getResults() { return results; }
+
+ @Override
+ public void defineRules(RulesFactory rulesFactory) {
+ rulesFactory.rule("collect all")
+ .on(stockTicks)
+ .execute(results, (r, tick) -> r.add(tick));
+ }
+ }
+
+ @EventProcessing(EventProcessingType.CLOUD)
+ public static class ExplicitCloudUnit implements RuleUnitDefinition {
+ private final DataStream<StockTick> stockTicks;
+ private final List<StockTick> results = new ArrayList<>();
+
+ public ExplicitCloudUnit() {
+ this.stockTicks = DataSource.createStream();
+ }
+
+ public DataStream<StockTick> getStockTicks() { return stockTicks; }
+ public List<StockTick> getResults() { return results; }
+
+ @Override
+ public void defineRules(RulesFactory rulesFactory) {
+ rulesFactory.rule("collect all")
+ .on(stockTicks)
+ .execute(results, (r, tick) -> r.add(tick));
+ }
+ }
+
+ public static class UnannotatedUnit implements RuleUnitDefinition {
+ private final DataStream<StockTick> stockTicks;
+ private final List<StockTick> results = new ArrayList<>();
+
+ public UnannotatedUnit() {
+ this.stockTicks = DataSource.createStream();
+ }
+
+ public DataStream<StockTick> getStockTicks() { return stockTicks; }
+ public List<StockTick> getResults() { return results; }
+
+ @Override
+ public void defineRules(RulesFactory rulesFactory) {
+ rulesFactory.rule("collect all")
+ .on(stockTicks)
+ .execute(results, (r, tick) -> r.add(tick));
+ }
+ }
+
+ @EventProcessing(EventProcessingType.STREAM)
+ public static class GroupByTemporalUnit implements RuleUnitDefinition {
+ private final DataStream<StockTick> stockTicks;
+ private final List<Integer> results = new ArrayList<>();
+
+ public GroupByTemporalUnit() {
+ this.stockTicks = DataSource.createStream();
+ defineRules(new RulesFactory(this));
+ }
+
+ public DataStream<StockTick> getStockTicks() { return stockTicks; }
+ public List<Integer> getResults() { return results; }
+
+ @Override
+ public void defineRules(RulesFactory rulesFactory) {
+ rulesFactory.rule("group temporal")
+ .groupBy(rule -> rule.on(stockTicks),
+ (StockTick t) -> t.getCompany(),
+ sum(StockTick::getDuration))
+ .after(5, 8, TimeUnit.SECONDS);
+ }
+ }
+
+ @EventProcessing(EventProcessingType.STREAM)
+ public static class AccumulateTemporalUnit implements RuleUnitDefinition {
+ private final DataStream<StockTick> stockTicks;
+ private final List<Integer> results = new ArrayList<>();
+
+ public AccumulateTemporalUnit() {
+ this.stockTicks = DataSource.createStream();
+ defineRules(new RulesFactory(this));
+ }
+
+ public DataStream<StockTick> getStockTicks() { return stockTicks; }
+ public List<Integer> getResults() { return results; }
+
+ @Override
+ public void defineRules(RulesFactory rulesFactory) {
+ rulesFactory.rule("accumulate temporal")
+ .on(stockTicks)
+ .accumulate(rule -> rule.on(stockTicks),
+ sum(StockTick::getDuration))
+ .after(5, 8, TimeUnit.SECONDS);
+ }
+ }
+}
diff --git
a/drools-ruleunits/drools-ruleunits-dsl/src/main/java/org/drools/ruleunits/dsl/accumulate/AccumulatePattern2.java
b/drools-ruleunits/drools-ruleunits-dsl/src/test/java/org/drools/ruleunits/dsl/domain/StockTick.java
similarity index 51%
copy from
drools-ruleunits/drools-ruleunits-dsl/src/main/java/org/drools/ruleunits/dsl/accumulate/AccumulatePattern2.java
copy to
drools-ruleunits/drools-ruleunits-dsl/src/test/java/org/drools/ruleunits/dsl/domain/StockTick.java
index d9e2cb0e296..a72c6f19bdd 100644
---
a/drools-ruleunits/drools-ruleunits-dsl/src/main/java/org/drools/ruleunits/dsl/accumulate/AccumulatePattern2.java
+++
b/drools-ruleunits/drools-ruleunits-dsl/src/test/java/org/drools/ruleunits/dsl/domain/StockTick.java
@@ -16,26 +16,40 @@
* specific language governing permissions and limitations
* under the License.
*/
-package org.drools.ruleunits.dsl.accumulate;
+package org.drools.ruleunits.dsl.domain;
-import org.drools.model.view.ViewItem;
-import org.drools.ruleunits.dsl.patterns.Pattern1DefImpl;
-import org.drools.ruleunits.dsl.patterns.Pattern2DefImpl;
-import org.drools.ruleunits.dsl.util.RuleDefinition;
+import org.kie.api.definition.type.Duration;
+import org.kie.api.definition.type.Expires;
+import org.kie.api.definition.type.Role;
-import static
org.drools.ruleunits.dsl.accumulate.AccumulatePattern1.createAccumulate1Item;
+@Role(Role.Type.EVENT)
+@Duration("duration")
+@Expires("10s")
+public class StockTick {
-public class AccumulatePattern2<A, B, C> extends Pattern2DefImpl<A, C> {
+ private final String company;
+ private final long duration;
- private final Accumulator1<B, C> acc;
+ public StockTick(String company) {
+ this.company = company;
+ this.duration = 0;
+ }
+
+ public StockTick(String company, long duration) {
+ this.company = company;
+ this.duration = duration;
+ }
+
+ public String getCompany() {
+ return company;
+ }
- public AccumulatePattern2(RuleDefinition rule, Pattern1DefImpl<A>
patternA, Pattern1DefImpl<C> patternC, Accumulator1<B, C> acc) {
- super(rule, patternA, patternC);
- this.acc = acc;
+ public long getDuration() {
+ return duration;
}
@Override
- public ViewItem toExecModelItem() {
- return createAccumulate1Item(patternB, getVariable(), acc);
+ public String toString() {
+ return "StockTick [company=" + company + ", duration=" + duration +
"]";
}
-}
\ No newline at end of file
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
