[KOGITO-6785] Serverless Workflow Correlation (#2160)

* Inserting serverless workflow correlation

api/kogito-api/src/main/java/org/kie/kogito/correlation/CompositeCorrelation.java

@@ -0,0 +1,71 @@
+/*
+ * Copyright 2022 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.correlation;
+
+import java.util.Collections;
+import java.util.Objects;
+import java.util.Set;
+import java.util.StringJoiner;
+import java.util.stream.Collectors;
+
+public class CompositeCorrelation implements Correlation<Set<? extends Correlation<?>>> {
+
+    private String key;
+    private Set<? extends Correlation<?>> correlations;
+
+    public CompositeCorrelation(Set<? extends Correlation<?>> correlations) {
+        this.key = buildKey(correlations);
+        this.correlations = Collections.unmodifiableSet(correlations);
+    }
+
+    private static String buildKey(Set<? extends Correlation<?>> correlations) {
+        return correlations.stream().map(Correlation::getKey).collect(Collectors.joining("|"));
+    }
+
+    @Override
+    public String getKey() {
+        return key;
+    }
+
+    @Override
+    public Set<? extends Correlation<?>> getValue() {
+        return correlations;
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) {
+            return true;
+        }
+        if (!(o instanceof CompositeCorrelation)) {
+            return false;
+        }
+        CompositeCorrelation that = (CompositeCorrelation) o;
+        return Objects.equals(getValue(), that.getValue()) && Objects.equals(getKey(), that.getKey());
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(getValue(), getKey());
+    }
+
+    @Override
+    public String toString() {
+        return new StringJoiner(", ", CompositeCorrelation.class.getSimpleName() + "[", "]")
+                .add("correlations=" + correlations)
+                .toString();
+    }
+}

api/kogito-api/src/main/java/org/kie/kogito/correlation/Correlation.java

@@ -17,25 +17,13 @@
 
 import java.util.Optional;
 
-public class Correlation {
+public interface Correlation<V> {
 
-    private String key;
-    private Object value;
+    String getKey();
 
-    public Correlation(String key, Object value) {
-        this.key = key;
-        this.value = value;
-    }
-
-    public String getKey() {
-        return key;
-    }
-
-    public Object getValue() {
-        return value;
-    }
+    V getValue();
 
-    public String asString() {
-        return Optional.ofNullable(value).map(Object::toString).orElse(null);
+    default String asString() {
+        return Optional.ofNullable(getValue()).map(Object::toString).orElse(null);
     }
 }

api/kogito-api/src/main/java/org/kie/kogito/correlation/CorrelationEncoder.java

@@ -0,0 +1,21 @@
+/*
+ * Copyright 2022 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.correlation;
+
+public interface CorrelationEncoder {
+
+    String encode(Correlation correlation);
+}

api/kogito-api/src/main/java/org/kie/kogito/correlation/CorrelationInstance.java

@@ -0,0 +1,41 @@
+/*
+ * Copyright 2022 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.correlation;
+
+public class CorrelationInstance {
+
+    private String correlationId;// encoded based on correlations
+    private String correlatedId;// == processInstanceId
+    private Correlation<?> correlation;
+
+    public CorrelationInstance(String correlationId, String correlatedId, Correlation<?> correlation) {
+        this.correlationId = correlationId;
+        this.correlatedId = correlatedId;
+        this.correlation = correlation;
+    }
+
+    public String getCorrelationId() {
+        return correlationId;
+    }
+
+    public String getCorrelatedId() {
+        return correlatedId;
+    }
+
+    public Correlation<?> getCorrelation() {
+        return correlation;
+    }
+}

api/kogito-api/src/main/java/org/kie/kogito/correlation/CorrelationResolver.java

@@ -13,11 +13,10 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-
 package org.kie.kogito.correlation;
 
 public interface CorrelationResolver {
 
-    Correlation resolve(Object data);
+    Correlation<?> resolve(Object data);
 
 }

api/kogito-api/src/main/java/org/kie/kogito/correlation/CorrelationService.java

@@ -0,0 +1,30 @@
+/*
+ * Copyright 2022 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.correlation;
+
+import java.util.Optional;
+
+public interface CorrelationService {
+
+    CorrelationInstance create(Correlation correlation, String correlatedId);
+
+    Optional<CorrelationInstance> find(Correlation correlation);
+
+    Optional<CorrelationInstance> findByCorrelatedId(String correlatedId);
+
+    void delete(Correlation correlation);
+
+}

api/kogito-api/src/main/java/org/kie/kogito/correlation/SimpleCorrelation.java

@@ -0,0 +1,67 @@
+/*
+ * Copyright 2022 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.correlation;
+
+import java.util.Objects;
+import java.util.StringJoiner;
+
+public class SimpleCorrelation<T> implements Correlation<T> {
+
+    private String key;
+    private T value;
+
+    public SimpleCorrelation(String key) {
+        this(key, null);
+    }
+
+    public SimpleCorrelation(String key, T value) {
+        this.key = key;
+        this.value = value;
+    }
+
+    public String getKey() {
+        return key;
+    }
+
+    public T getValue() {
+        return value;
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) {
+            return true;
+        }
+        if (!(o instanceof SimpleCorrelation)) {
+            return false;
+        }
+        SimpleCorrelation that = (SimpleCorrelation) o;
+        return Objects.equals(getKey(), that.getKey()) && Objects.equals(getValue(), that.getValue());
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(getKey(), getValue());
+    }
+
+    @Override
+    public String toString() {
+        return new StringJoiner(", ", SimpleCorrelation.class.getSimpleName() + "[", "]")
+                .add("key='" + key + "'")
+                .add("value=" + value)
+                .toString();
+    }
+}

api/kogito-api/src/main/java/org/kie/kogito/internal/process/runtime/KogitoProcessInstance.java

@@ -86,4 +86,8 @@ public interface KogitoProcessInstance extends ProcessInstance, KogitoEventListe
      * @return map with headers
      */
     Map<String, List<String>> getHeaders();
+
+    void wrap(org.kie.kogito.process.ProcessInstance<?> kogitoProcessInstance);
+
+    org.kie.kogito.process.ProcessInstance<?> unwrap();
 }

api/kogito-api/src/main/java/org/kie/kogito/process/Process.java

@@ -19,6 +19,8 @@
 import java.util.function.Predicate;
 
 import org.kie.kogito.Model;
+import org.kie.kogito.correlation.CompositeCorrelation;
+import org.kie.kogito.correlation.CorrelationService;
 import org.kie.kogito.internal.process.runtime.KogitoNode;
 
 public interface Process<T> {
@@ -27,10 +29,14 @@ public interface Process<T> {
 
     ProcessInstance<T> createInstance(String businessKey, T workingMemory);
 
+    ProcessInstance<T> createInstance(String businessKey, CompositeCorrelation correlation, T workingMemory);
+
     ProcessInstances<T> instances();
 
     Collection<KogitoNode> findNodes(Predicate<KogitoNode> filter);
 
+    CorrelationService correlations();
+
     <S> void send(Signal<S> sig);
 
     T createModel();

api/kogito-api/src/main/java/org/kie/kogito/process/ProcessInstance.java

@@ -24,6 +24,7 @@
 import java.util.function.Function;
 import java.util.function.Predicate;
 
+import org.kie.kogito.correlation.Correlation;
 import org.kie.kogito.internal.process.runtime.KogitoNodeInstance;
 import org.kie.kogito.internal.process.runtime.KogitoWorkItem;
 import org.kie.kogito.process.flexible.AdHocFragment;
@@ -257,4 +258,5 @@ default ProcessInstance<T> checkError() {
 
     long version();
 
+    Optional<? extends Correlation<?>> correlation();
 }

api/kogito-api/src/main/java/org/kie/kogito/process/ProcessService.java

@@ -26,6 +26,7 @@
 import org.kie.kogito.MappableToModel;
 import org.kie.kogito.Model;
 import org.kie.kogito.auth.SecurityPolicy;
+import org.kie.kogito.correlation.CompositeCorrelation;
 import org.kie.kogito.process.workitem.Attachment;
 import org.kie.kogito.process.workitem.AttachmentInfo;
 import org.kie.kogito.process.workitem.Comment;
@@ -45,7 +46,8 @@ <T extends Model> ProcessInstance<T> createProcessInstance(Process<T> process, S
     <T extends Model> ProcessInstance<T> createProcessInstance(Process<T> process, String businessKey, T model,
             String startFromNodeId,
             String trigger,
-            String kogitoReferenceId);
+            String kogitoReferenceId,
+            CompositeCorrelation correlation);
 
     <T extends MappableToModel<R>, R> List<R> getProcessInstanceOutput(Process<T> process);