[incubator-kie-issues-1555] Multiple sub process instances cancelled …

…by the same timer. (#3745)

* [incubator-kie-issues-1555] Multiple sub process instances cancelled by the same timer.

* fix tests

* fix Composite rest

jbpm/jbpm-flow-builder/src/main/java/org/jbpm/test/util/DefaultCountDownProcessEventListener.java

@@ -43,6 +43,20 @@ public boolean waitTillCompleted() {
         return waitTillCompleted(10000);
     }
 
+    public boolean await() {
+        try {
+            latch.await();
+            return true;
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+            logger.error("Interrputed thread while waiting for all triggers", e);
+            return false;
+        } catch (Exception e) {
+            logger.error("Error during waiting state", e);
+            return false;
+        }
+    }
+
     public boolean waitTillCompleted(long timeOut) {
         try {
             return latch.await(timeOut, TimeUnit.MILLISECONDS);

jbpm/jbpm-flow/src/main/java/org/jbpm/process/core/event/EventTypeFilter.java

@@ -19,10 +19,19 @@
 package org.jbpm.process.core.event;
 
 import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
 import java.util.function.Function;
+import java.util.regex.Matcher;
 
 import org.jbpm.process.core.correlation.CorrelationInstance;
 import org.jbpm.process.core.correlation.CorrelationManager;
+import org.jbpm.util.PatternConstants;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -55,37 +64,106 @@ public void setType(String type) {
         this.type = type;
     }
 
+    public void setMessageRef(String messageRef) {
+        this.messageRef = messageRef;
+    }
+
     public String toString() {
         return "Event filter: [" + this.type + "]";
     }
 
     @Override
     public boolean acceptsEvent(String type, Object event, Function<String, Object> resolver) {
-        logger.debug("This event is subscribed to a message type {} with payload {}", type, event);
+        if (this.type == null) {
+            return false;
+        }
+
         if (resolver == null) {
-            return this.type != null && this.type.equals(type);
+            return this.type.equals(type);
         }
 
-        if (this.type != null && this.type.equals(type)) {
-            if (correlationManager != null && correlationManager.isSubscribe(messageRef)) {
-                if (event == null) {
-                    logger.debug("This event is subscribed to a message ref {}", type);
-                    return false;
-                }
-                CorrelationInstance messageCorrelation = correlationManager.computeCorrelationInstance(messageRef, event);
-                CorrelationInstance processCorrelation = correlationManager.computeSubscription(messageRef, resolver);
-                logger.debug("The event type {} is correlated, computing correlations. Message correlation is {}; process correlation is: {} ", type, messageCorrelation, processCorrelation);
-                return messageCorrelation.equals(processCorrelation);
+        if (this.type.equals(type) && correlationManager != null && correlationManager.isSubscribe(messageRef)) {
+            logger.debug("This event is subscribed to a message type {} with payload {}", type, event);
+            if (event == null) {
+                logger.debug("Cannot compute subscription for messageref {} and type {}", messageRef, type);
+                return false;
             }
-            return true;
+            CorrelationInstance messageCorrelation = correlationManager.computeCorrelationInstance(messageRef, event);
+            CorrelationInstance processCorrelation = correlationManager.computeSubscription(messageRef, resolver);
+            logger.debug("The event type {} is correlated, computing correlations. Message correlation is {}; process correlation is: {} ", type, messageCorrelation, processCorrelation);
+            return messageCorrelation.equals(processCorrelation);
         }
 
-        String resolvedType = (String) resolver.apply(this.type);
-        return resolvedType != null && resolvedType.equals(type);
+        return isAccepted(type, resolver);
 
     }
 
-    public void setMessageRef(String messageRef) {
-        this.messageRef = messageRef;
+    public boolean isAccepted(String type, Function<String, Object> resolver) {
+        return resolveVariable(this.type, resolver).contains(type);
     }
+
+    private List<String> resolveVariable(String varExpression, Function<String, Object> resolver) {
+        if (varExpression == null) {
+            return Collections.emptyList();
+        }
+        Map<String, Object[]> replacements = new HashMap<>();
+        Matcher matcher = PatternConstants.PARAMETER_MATCHER.matcher(varExpression);
+        while (matcher.find()) {
+            String paramName = matcher.group(1);
+            Object value = resolver.apply(paramName);
+            if (value == null) {
+                logger.warn("expression {} in dynamic signal {} not resolved", paramName, varExpression);
+                continue;
+            } else if (value instanceof Object[]) {
+                replacements.put(paramName, (Object[]) value);
+            } else {
+                replacements.put(paramName, new Object[] { value });
+            }
+        }
+        List<String> acceptedTypes = new ArrayList<>();
+        List<Map<String, String>> data = generateCombinations(replacements.keySet(), replacements);
+        for (Map<String, String> combination : data) {
+            String tmp = varExpression;
+            for (Map.Entry<String, String> replacement : combination.entrySet()) {
+                tmp = tmp.replace("#{" + replacement.getKey() + "}", replacement.getValue());
+            }
+            acceptedTypes.add(tmp);
+        }
+        if (acceptedTypes.isEmpty()) {
+            acceptedTypes.add(varExpression);
+        }
+        return acceptedTypes;
+    }
+
+    private List<Map<String, String>> generateCombinations(Set<String> keys, Map<String, Object[]> data) {
+        List<Map<String, String>> combinations = new ArrayList<>();
+        for (String key : keys) {
+            Set<String> remaining = new HashSet<>(keys);
+            remaining.remove(key);
+            List<Map<String, String>> subCombinations = generateCombinations(remaining, data);
+            if (subCombinations.isEmpty()) {
+                for (Object value : data.get(key)) {
+                    Map<String, String> combination = new HashMap<>();
+                    combination.put(key, value.toString());
+                    if (!combinations.contains(combination)) {
+                        combinations.add(combination);
+                    }
+                }
+            } else {
+                for (Map<String, String> subCombination : subCombinations) {
+                    for (Object value : data.get(key)) {
+                        Map<String, String> combination = new HashMap<>();
+                        combination.putAll(subCombination);
+                        combination.put(key, value.toString());
+                        if (!combinations.contains(combination)) {
+                            combinations.add(combination);
+                        }
+                    }
+                }
+            }
+        }
+
+        return combinations;
+    }
+
 }

jbpm/jbpm-flow/src/main/java/org/jbpm/process/instance/impl/actions/AbstractNodeInstanceAction.java

@@ -24,7 +24,7 @@
 import org.jbpm.process.instance.impl.Action;
 import org.jbpm.workflow.instance.node.CompositeNodeInstance;
 import org.kie.api.runtime.process.NodeInstance;
-import org.kie.api.runtime.process.WorkflowProcessInstance;
+import org.kie.api.runtime.process.NodeInstanceContainer;
 import org.kie.kogito.internal.process.runtime.KogitoProcessContext;
 
 public abstract class AbstractNodeInstanceAction implements Action, Serializable {
@@ -39,7 +39,7 @@ protected AbstractNodeInstanceAction(String attachedToNodeId) {
 
     @Override
     public void execute(KogitoProcessContext context) throws Exception {
-        WorkflowProcessInstance pi = context.getNodeInstance().getProcessInstance();
+        NodeInstanceContainer pi = context.getNodeInstance().getNodeInstanceContainer();
         NodeInstance nodeInstance = findNodeByUniqueId(pi.getNodeInstances(), attachedToNodeId);
         if (nodeInstance != null) {
             execute(nodeInstance);

jbpm/jbpm-flow/src/main/java/org/jbpm/ruleflow/core/RuleFlowProcessFactory.java

@@ -42,6 +42,7 @@
 import org.jbpm.process.instance.impl.ReturnValueEvaluator;
 import org.jbpm.process.instance.impl.actions.CancelNodeInstanceAction;
 import org.jbpm.process.instance.impl.actions.SignalProcessInstanceAction;
+import org.jbpm.process.instance.impl.util.VariableUtil;
 import org.jbpm.ruleflow.core.validation.RuleFlowProcessValidator;
 import org.jbpm.workflow.core.DroolsAction;
 import org.jbpm.workflow.core.WorkflowModelValidator;
@@ -412,8 +413,10 @@ protected void linkBoundaryErrorEvent(Node node, String attachedTo, Node attache
 
     protected DroolsAction timerAction(String type) {
         DroolsAction signal = new DroolsAction();
-
-        Action action = kcontext -> kcontext.getProcessInstance().signalEvent(type, kcontext.getNodeInstance().getStringId());
+        Action action = kcontext -> {
+            String eventType = VariableUtil.resolveVariable(type, kcontext.getNodeInstance());
+            kcontext.getProcessInstance().signalEvent(eventType, kcontext.getNodeInstance().getStringId());
+        };
         signal.wire(action);
 
         return signal;