[KOGITO-9613] Adding Awaitbility and to process definitions addon

apache · Jul 25, 2023 · 5642155 · 5642155
1 parent d57d6ed
commit 5642155
Show file tree

Hide file tree

Showing 5 changed files with 23 additions and 6 deletions.
diff --git a/.../src/main/java/org/kie/kogito/serverless/workflow/executor/StaticWorkflowApplication.java b/.../src/main/java/org/kie/kogito/serverless/workflow/executor/StaticWorkflowApplication.java
@@ -107,7 +107,7 @@ public void afterProcessCompleted(ProcessCompletedEvent event) {
             SynchronousQueue<JsonNodeModel> queue = queues.remove(instance.getId());
             if (queue != null) {
                 try {
-                    queue.offer(new JsonNodeModel(instance.getId(), instance.getVariables().get(SWFConstants.DEFAULT_WORKFLOW_VAR)), 1000L, TimeUnit.SECONDS);
+                    queue.offer(new JsonNodeModel(instance.getId(), instance.getVariables().get(SWFConstants.DEFAULT_WORKFLOW_VAR)), 1L, TimeUnit.SECONDS);
                 } catch (InterruptedException e) {
                     Thread.currentThread().interrupt();
                 }
@@ -173,7 +173,7 @@ public JsonNodeModel execute(Workflow workflow, JsonNode data) {
         return execute(findOrCreate(workflow), data);
     }
 
-    public StaticWorkflowApplication withProcessInstances(ProcessInstancesFactory processInstanceFactory) {
+    public StaticWorkflowApplication processInstancesFactory(ProcessInstancesFactory processInstanceFactory) {
         this.processInstancesFactory = processInstanceFactory;
         return this;
     }

diff --git a/kogito-serverless-workflow/kogito-serverless-workflow-executor-kafka/pom.xml b/kogito-serverless-workflow/kogito-serverless-workflow-executor-kafka/pom.xml
@@ -36,6 +36,11 @@
       <artifactId>junit-jupiter-api</artifactId>
       <scope>test</scope>
     </dependency>
+    <dependency>
+      <groupId>org.awaitility</groupId>
+      <artifactId>awaitility</artifactId>
+      <scope>test</scope>
+    </dependency>
     <dependency>
       <groupId>ch.qos.logback</groupId>
       <artifactId>logback-classic</artifactId>

diff --git a/...rc/test/java/org/kie/kogito/serverless/workflow/executor/WorkflowEventSubscriberTest.java b/...rc/test/java/org/kie/kogito/serverless/workflow/executor/WorkflowEventSubscriberTest.java
@@ -41,6 +41,7 @@
 
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
+import static org.awaitility.Awaitility.await;
 import static org.kie.kogito.serverless.workflow.fluent.ActionBuilder.call;
 import static org.kie.kogito.serverless.workflow.fluent.EventDefBuilder.eventDef;
 import static org.kie.kogito.serverless.workflow.fluent.FunctionBuilder.expr;
@@ -73,16 +74,15 @@ void testCallbackSubscriberWithPersistence(@TempDir Path tempDir) throws Interru
         final String additionalData = "This has been injected by the event";
         Workflow workflow = workflow("testCallback").start(callback(call(expr("concat", "{slogan:.slogan+\"er Beti\"}")), eventDef(eventType))).end().build();
         try (StaticWorkflowApplication application =
-                StaticWorkflowApplication.create().withProcessInstances(new RocksDBProcessInstancesFactory(new Options().setCreateIfMissing(true), tempDir.toString()))) {
+                StaticWorkflowApplication.create().processInstancesFactory(new RocksDBProcessInstancesFactory(new Options().setCreateIfMissing(true), tempDir.toString()))) {
             String id = application.execute(workflow, jsonObject().put("slogan", "Viva ")).getId();
-            assertThat(application.variables(id).orElseThrow().getWorkflowdata().get("slogan").equals(new TextNode("Viva ")));
+            assertThat(application.variables(id).orElseThrow().getWorkflowdata()).doesNotContain(new TextNode(additionalData));
             publish(eventType, buildCloudEvent(eventType, id)
                     .withData(JsonCloudEventData.wrap(jsonObject().put("additionalData", additionalData)))
                     .build());
             assertThat(application.waitForFinish(id, Duration.ofSeconds(20)).orElseThrow().getWorkflowdata())
                     .isEqualTo(jsonObject().put("additionalData", additionalData).put("slogan", "Viva er Beti"));
-            Thread.sleep(10);
-            assertThat(application.variables(id)).isEmpty();
+            await().atMost(Duration.ofSeconds(1)).pollInterval(Duration.ofMillis(50)).until(() -> application.variables(id).isEmpty());
         }
     }
 

diff --git a/quarkus/addons/process-definitions/runtime/pom.xml b/quarkus/addons/process-definitions/runtime/pom.xml
@@ -35,6 +35,11 @@
       <artifactId>quarkus-junit5</artifactId>
       <scope>test</scope>
     </dependency>
+    <dependency>
+      <groupId>org.kie.kogito</groupId>
+      <artifactId>kogito-addons-quarkus-persistence-rocksdb</artifactId>
+      <scope>test</scope>
+    </dependency>
     <dependency>
       <groupId>io.rest-assured</groupId>
       <artifactId>rest-assured</artifactId>

diff --git a/.../runtime/src/main/java/org/kie/kogito/process/definitions/ProcessDefinitionsResource.java b/.../runtime/src/main/java/org/kie/kogito/process/definitions/ProcessDefinitionsResource.java
@@ -20,6 +20,8 @@
 
 import javax.annotation.PostConstruct;
 import javax.annotation.PreDestroy;
+import javax.enterprise.inject.Instance;
+import javax.inject.Inject;
 import javax.ws.rs.Consumes;
 import javax.ws.rs.POST;
 import javax.ws.rs.Path;
@@ -28,6 +30,7 @@
 import javax.ws.rs.core.MediaType;
 import javax.ws.rs.core.Response;
 
+import org.kie.kogito.process.ProcessInstancesFactory;
 import org.kie.kogito.serverless.workflow.executor.StaticWorkflowApplication;
 import org.kie.kogito.serverless.workflow.models.JsonNodeModel;
 import org.kie.kogito.serverless.workflow.models.JsonNodeModelInput;
@@ -41,9 +44,13 @@ public class ProcessDefinitionsResource {
 
     private StaticWorkflowApplication application;
 
+    @Inject
+    Instance<ProcessInstancesFactory> factories;
+
     @PostConstruct
     void init() {
         application = StaticWorkflowApplication.create();
+        factories.stream().findFirst().ifPresent(application::processInstancesFactory);
     }
 
     @PreDestroy