introducing Intermediate class

apache · Dec 9, 2024 · 4295790 · 4295790
1 parent 63a31f1
commit 4295790
Show file tree

Hide file tree

Showing 3 changed files with 234 additions and 183 deletions.
diff --git a/...rkus-devui-runtime/src/main/java/org/jbpm/quarkus/devui/runtime/rpc/DataIndexCounter.java b/...rkus-devui-runtime/src/main/java/org/jbpm/quarkus/devui/runtime/rpc/DataIndexCounter.java
@@ -0,0 +1,80 @@
+/*
+ * 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.jbpm.quarkus.devui.runtime.rpc;
+
+import io.smallrye.mutiny.Multi;
+import io.smallrye.mutiny.operators.multi.processors.BroadcastProcessor;
+import io.vertx.core.Future;
+import io.vertx.core.Vertx;
+import io.vertx.core.json.JsonObject;
+import io.vertx.ext.web.client.WebClient;
+import jakarta.enterprise.context.ApplicationScoped;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+@ApplicationScoped
+public class DataIndexCounter {
+    private static final String DATA_INDEX_URL = "kogito.data-index.url";
+    private static final Logger LOGGER = LoggerFactory.getLogger(DataIndexCounter.class);
+
+    private Multi<String> multi;
+    private WebClient dataIndexWebClient;
+
+    public DataIndexCounter(String query, String graphname, BroadcastProcessor<String> stream,WebClient dataIndexWebClient, JBPMDevUIEventPublisher eventPublisher) {
+        this.dataIndexWebClient = dataIndexWebClient;
+
+        Vertx vertx = Vertx.vertx();
+        this.multi = Multi.createFrom().emitter(emitter -> {
+            vertx.setTimer(1000, id -> emitter.emit("Initial data emitted"));
+        });
+
+
+        vertx.setTimer(1000, id -> refreshData(stream,query,graphname)); 
+    }
+
+    private void refreshData(BroadcastProcessor<String> stream, String query, String graphname ) {
+        LOGGER.info("Refreshing data for query: {}", query);
+
+        doQuery(query, graphname).toCompletionStage()
+                 .thenAccept(result -> {
+            stream.onNext(result);
+                 });
+    }
+
+       private Future<String> doQuery(String query, String graphModelName) {
+        if(dataIndexWebClient == null) {
+            LOGGER.warn("Cannot perform '{}' query, dataIndexWebClient couldn't be set. Is DataIndex correctly? Please verify '{}' value", graphModelName, DATA_INDEX_URL);
+             return Future.succeededFuture("-");
+         }
+         return this.dataIndexWebClient.post("/graphql")
+                 .putHeader("content-type", "application/json")
+                 .sendJson(new JsonObject(query))
+                 .map(response -> {
+                    if(response.statusCode() == 200) {
+                         JsonObject responseData = response.bodyAsJsonObject().getJsonObject("data");
+                         return String.valueOf(responseData.getJsonArray(graphModelName).size());
+                     }
+                     return "-";
+                 });
+     }
+
+    public Multi<String> getMulti() {
+        return multi;
+    }
+}
diff --git a/...vui-runtime/src/main/java/org/jbpm/quarkus/devui/runtime/rpc/JBPMDevUIEventPublisher.java b/...vui-runtime/src/main/java/org/jbpm/quarkus/devui/runtime/rpc/JBPMDevUIEventPublisher.java
@@ -17,60 +17,64 @@
  * under the License.
  */
 
-package org.jbpm.quarkus.devui.runtime.rpc;
-import jakarta.enterprise.context.ApplicationScoped;
-import org.kie.kogito.event.DataEvent;
-import org.kie.kogito.event.EventPublisher;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import java.util.Collection;
-import java.util.Objects;
-
-@ApplicationScoped
-public class JBPMDevUIEventPublisher implements EventPublisher {
-
-    private static final Logger LOGGER = LoggerFactory.getLogger(JBPMDevUIEventPublisher.class);
-    private Runnable onProcessEvent;
-    private Runnable onTaskEvent;
-    private Runnable onJobEvent;
-
-    @Override
-    public void publish(DataEvent<?> event) {
-        switch (event.getType()) {
-            case "ProcessInstanceStateDataEvent":
-                maybeRun(onProcessEvent);
-                break;
-            case "UserTaskInstanceStateDataEvent":
-                maybeRun(onTaskEvent);
-                break;
-            case "JobEvent":
-                maybeRun(onJobEvent);
-                break;
-            default:
-                LOGGER.debug("Unknown type of event '{}', ignoring for this publisher", event.getType());
-        }
-    }
-
-    @Override
-    public void publish(Collection<DataEvent<?>> events) {
-        events.forEach(this::publish);
-    }
-
-    private void maybeRun(Runnable runnable) {
-        if(Objects.nonNull(runnable)) {
-            runnable.run();
-        }
-    }
-
-    public void setOnProcessEventListener(Runnable onProcessEvent) {
-        this.onProcessEvent = onProcessEvent;
-    }
-
-    public void setOnTaskEventListener(Runnable onTaskEvent) {
-        this.onTaskEvent = onTaskEvent;
-    }
-
-    public void setOnJobEventListener(Runnable onJobEvent) {
-        this.onJobEvent = onJobEvent;
-    }
-}
+ package org.jbpm.quarkus.devui.runtime.rpc;
+ import jakarta.enterprise.context.ApplicationScoped;
+ import org.kie.kogito.event.DataEvent;
+ import org.kie.kogito.event.EventPublisher;
+ import org.slf4j.Logger;
+ import org.slf4j.LoggerFactory;
+
+ import io.quarkus.arc.profile.IfBuildProfile;
+
+ import java.util.Collection;
+ import java.util.Objects;
+
+ @ApplicationScoped
+ @IfBuildProfile("dev")
+ public class JBPMDevUIEventPublisher implements EventPublisher {
+
+     private static final Logger LOGGER = LoggerFactory.getLogger(JBPMDevUIEventPublisher.class);
+     private Runnable onProcessEvent;
+     private Runnable onTaskEvent;
+     private Runnable onJobEvent;
+
+     @Override
+     public void publish(DataEvent<?> event) {
+         switch (event.getType()) {
+             case "ProcessInstanceStateDataEvent":
+                 maybeRun(onProcessEvent);
+                 break;
+             case "UserTaskInstanceStateDataEvent":
+                 maybeRun(onTaskEvent);
+                 break;
+             case "JobEvent":
+                 maybeRun(onJobEvent);
+                 break;
+             default:
+                 LOGGER.debug("Unknown type of event '{}', ignoring for this publisher", event.getType());
+         }
+     }
+
+     @Override
+     public void publish(Collection<DataEvent<?>> events) {
+         events.forEach(this::publish);
+     }
+
+     private void maybeRun(Runnable runnable) {
+         if(Objects.nonNull(runnable)) {
+             runnable.run();
+         }
+     }
+
+     public void setOnProcessEventListener(Runnable onProcessEvent) {
+         this.onProcessEvent = onProcessEvent;
+     }
+
+     public void setOnTaskEventListener(Runnable onTaskEvent) {
+         this.onTaskEvent = onTaskEvent;
+     }
+
+     public void setOnJobEventListener(Runnable onJobEvent) {
+         this.onJobEvent = onJobEvent;
+     }
+ }