Issue 32595 - Kafka Streams Dev UI migration to v2

extensions/kafka-streams/deployment/src/main/java/io/quarkus/kafka/streams/deployment/devui/KafkaStreamsDevUIProcessor.java

@@ -0,0 +1,30 @@
+package io.quarkus.kafka.streams.deployment.devui;
+
+import io.quarkus.deployment.IsDevelopment;
+import io.quarkus.deployment.annotations.BuildProducer;
+import io.quarkus.deployment.annotations.BuildStep;
+import io.quarkus.devui.spi.JsonRPCProvidersBuildItem;
+import io.quarkus.devui.spi.page.CardPageBuildItem;
+import io.quarkus.devui.spi.page.Page;
+import io.quarkus.kafka.streams.runtime.devui.KafkaStreamsJsonRPCService;
+
+public class KafkaStreamsDevUIProcessor {
+
+    @BuildStep(onlyIf = IsDevelopment.class)
+    public void createPages(BuildProducer<CardPageBuildItem> cardPageProducer) {
+
+        CardPageBuildItem cardPageBuildItem = new CardPageBuildItem();
+
+        cardPageBuildItem.addPage(Page.webComponentPageBuilder()
+                .componentLink("qwc-kafka-streams-topology.js")
+                .title("Topology")
+                .icon("font-awesome-solid:diagram-project"));
+
+        cardPageProducer.produce(cardPageBuildItem);
+    }
+
+    @BuildStep(onlyIf = IsDevelopment.class)
+    public void createJsonRPCService(BuildProducer<JsonRPCProvidersBuildItem> jsonRPCServiceProducer) {
+        jsonRPCServiceProducer.produce(new JsonRPCProvidersBuildItem(KafkaStreamsJsonRPCService.class));
+    }
+}

extensions/kafka-streams/deployment/src/main/resources/dev-ui/qwc-kafka-streams-topology.js

@@ -0,0 +1,124 @@
+import { QwcHotReloadElement, html, css } from 'qwc-hot-reload-element';
+import { unsafeHTML } from 'lit/directives/unsafe-html.js';
+import { JsonRpc } from 'jsonrpc';
+
+import { Graphviz } from "https://cdn.jsdelivr.net/npm/@hpcc-js/wasm/dist/graphviz.js";
+
+import '@vaadin/details';
+import '@vaadin/tabsheet';
+import '@vaadin/tabs';
+import '@vaadin/vertical-layout';
+import 'qui-badge';
+import 'qui-code-block';
+
+/**
+ * This component shows the Kafka Streams Topology
+ */
+export class QwcKafkaStreamsTopology extends QwcHotReloadElement {
+
+    jsonRpc = new JsonRpc(this);
+
+    static styles = css`
+        .codeBlock {
+            width: 100%;
+            height: auto;
+        }
+    `;
+
+    static properties = {
+        _topology: {state: true},
+        _graphviz: {state: true},
+        _tabContent: {state: true}
+    };
+
+    constructor() {
+        super();
+        this._topology = null;
+        this._graphviz = null;
+        this._tabContent = '';
+    }
+
+    connectedCallback() {
+        super.connectedCallback();
+        Graphviz.load().then(r => this._graphviz = r);
+        this.hotReload()
+    }
+
+    render() {
+        if (this._topology) {
+            return html`<vaadin-tabs @selected-changed="${(e) => this._tabSelectedChanged(e.detail.value)}">
+                          <vaadin-tab id="graphTab">Graph</vaadin-tab>
+                          <vaadin-tab id="detailsTab">Details</vaadin-tab>
+                          <vaadin-tab id="describeTab">Describe</vaadin-tab>
+                          <vaadin-tab id="graphvizTab">Graphviz</vaadin-tab>
+                          <vaadin-tab id="mermaidTab">Mermaid</vaadin-tab>
+                        </vaadin-tabs>
+                        <vaadin-vertical-layout theme="padding"><p>${this._tabContent}</p></vaadin-vertical-layout>`;
+        }
+
+        return html`<qwc-no-data message="You do not have any Topology."
+                                    link="https://quarkus.io/guides/kafka-streams"
+                                linkText="Learn how to write Kafka Streams">
+                    </qwc-no-data>`;
+    }
+
+    hotReload() {
+        this._topology = null;
+        this.jsonRpc.getTopology().then(jsonRpcResponse => {
+            this._topology = jsonRpcResponse.result;
+        });
+    }
+
+    _tabSelectedChanged(n) {
+      switch(n) {
+        case 1 : this._selectDetailsTab(); break;
+        case 2 : this._selectDescribeTab(); break;
+        case 3 : this._selectGraphvizTab(); break;
+        case 4 : this._selectMermaidTab(); break;
+        default : this._selectGraphTab();
+      }
+    }
+
+    _selectGraphTab() {
+      if (this._graphviz) {
+        let g = this._graphviz.dot(this._topology.graphviz);
+        this._tabContent = html`${unsafeHTML(g)}`;
+      } else {
+        this._tabContent = html`Graph engine not started.`;
+      }
+    }
+
+    _selectDetailsTab() {
+      this._tabContent = html`<table>
+                                <tr>
+                                  <td>Sub-topologies</td><td><qui-badge>${this._topology.subTopologies.length}</qui-badge></td>
+                                  <td>${this._topology.subTopologies.map((subTopology) => html`<qui-badge level="contrast" icon="font-awesome-solid:diagram-project" style="margin-right:5px">${subTopology}</qui-badge>`)}</td>
+                                </tr>
+                                <tr>
+                                  <td>Sources</td><td><qui-badge>${this._topology.sources.length}</qui-badge></td>
+                                  <td>${this._topology.sources.map((source) => html`<qui-badge level="contrast" icon="font-awesome-solid:right-to-bracket" style="margin-right:5px">${source}</qui-badge>`)}</td>
+                                </tr>
+                                <tr>
+                                  <td>Sinks</td><td><qui-badge>${this._topology.sinks.length}</qui-badge></td>
+                                  <td>${this._topology.sinks.map((sink) => html`<qui-badge level="contrast" icon="font-awesome-solid:right-from-bracket" style="margin-right:5px">${sink}</qui-badge>`)}</td>
+                                </tr>
+                                <tr>
+                                  <td>Stores</td><td><qui-badge>${this._topology.stores.length}</qui-badge></td>
+                                  <td>${this._topology.stores.map((store) => html`<qui-badge level="contrast" icon="font-awesome-solid:database" style="margin-right:5px">${store}</qui-badge>`)}</td>
+                                </tr>
+                              </table>`;
+    }
+
+    _selectDescribeTab() {
+      this._tabContent = html`<qui-code-block mode='text' content='${this._topology.describe}' class="codeBlock"></qui-code-block>`;
+    }
+
+    _selectGraphvizTab() {
+      this._tabContent = html`<qui-code-block mode='gv' content='${this._topology.graphviz}' class="codeBlock"></qui-code-block>`;
+    }
+
+    _selectMermaidTab() {
+      this._tabContent = html`<qui-code-block mode='mermaid' content='${this._topology.mermaid}' class="codeBlock"></qui-code-block>`;
+    }
+}
+customElements.define('qwc-kafka-streams-topology', QwcKafkaStreamsTopology);

extensions/kafka-streams/runtime/src/main/java/io/quarkus/kafka/streams/runtime/devui/KafkaStreamsJsonRPCService.java

@@ -0,0 +1,134 @@
+package io.quarkus.kafka.streams.runtime.devui;
+
+import java.util.Arrays;
+import java.util.function.Consumer;
+import java.util.function.Predicate;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+import java.util.stream.Stream;
+
+import jakarta.inject.Inject;
+
+import org.apache.kafka.streams.Topology;
+
+import io.smallrye.common.annotation.NonBlocking;
+import io.vertx.core.json.JsonObject;
+
+public class KafkaStreamsJsonRPCService {
+    @Inject
+    Topology topology;
+
+    @NonBlocking
+    public JsonObject getTopology() {
+        return parseTopologyDescription(topology.describe() != null ? topology.describe().toString() : "");
+    }
+
+    JsonObject parseTopologyDescription(String topologyDescription) {
+        final var res = new JsonObject();
+
+        final var context = new TopologyParserContext();
+        Arrays.stream(topologyDescription.split("\n"))
+                .map(String::trim)
+                .forEachOrdered(line -> Stream.of(SUB_TOPOLOGY, SOURCE, PROCESSOR, SINK, RIGHT_ARROW)
+                        .filter(itemParser -> itemParser.test(line))
+                        .forEachOrdered(itemParser -> itemParser.accept(context)));
+
+        res
+                .put("describe", topologyDescription)
+                .put("subTopologies", context.subTopologies)
+                .put("sources", context.sources)
+                .put("sinks", context.sinks)
+                .put("stores", context.stores)
+                .put("graphviz", context.graphviz.toGraph())
+                .put("mermaid", context.mermaid.toGraph());
+
+        return res;
+    }
+
+    private interface RawTopologyItemParser extends Predicate<String>, Consumer<TopologyParserContext> {
+    }
+
+    private static final RawTopologyItemParser SUB_TOPOLOGY = new RawTopologyItemParser() {
+        private final Pattern subTopologyPattern = Pattern.compile("Sub-topology: (?<subTopology>[0-9]*).*");
+        private Matcher matcher;
+
+        @Override
+        public boolean test(String line) {
+            matcher = subTopologyPattern.matcher(line);
+            return matcher.matches();
+        }
+
+        @Override
+        public void accept(TopologyParserContext context) {
+            context.addSubTopology(matcher.group("subTopology"));
+        }
+    };
+
+    private static final RawTopologyItemParser SOURCE = new RawTopologyItemParser() {
+        private final Pattern sourcePattern = Pattern
+                .compile("Source:\\s+(?<source>\\S+)\\s+\\(topics:\\s+\\[(?<topics>.*)\\]\\).*");
+        private Matcher matcher;
+
+        @Override
+        public boolean test(String line) {
+            matcher = sourcePattern.matcher(line);
+            return matcher.matches();
+        }
+
+        @Override
+        public void accept(TopologyParserContext context) {
+            context.addSources(matcher.group("source"), matcher.group("topics").split(","));
+        }
+    };
+
+    private static final RawTopologyItemParser PROCESSOR = new RawTopologyItemParser() {
+        private final Pattern processorPattern = Pattern
+                .compile("Processor:\\s+(?<processor>\\S+)\\s+\\(stores:\\s+\\[(?<stores>.*)\\]\\).*");
+        private Matcher matcher;
+        private String line;
+
+        @Override
+        public boolean test(String line) {
+            this.line = line;
+            matcher = processorPattern.matcher(line);
+            return matcher.matches();
+        }
+
+        @Override
+        public void accept(TopologyParserContext context) {
+            context.addStores(matcher.group("stores").split(","), matcher.group("processor"), line.contains("JOIN"));
+        }
+    };
+
+    private static final RawTopologyItemParser SINK = new RawTopologyItemParser() {
+        private final Pattern sinkPattern = Pattern.compile("Sink:\\s+(?<sink>\\S+)\\s+\\(topic:\\s+(?<topic>.*)\\).*");
+        private Matcher matcher;
+
+        @Override
+        public boolean test(String line) {
+            matcher = sinkPattern.matcher(line);
+            return matcher.matches();
+        }
+
+        @Override
+        public void accept(TopologyParserContext context) {
+            context.addSink(matcher.group("sink"), matcher.group("topic"));
+        }
+    };
+
+    private static final RawTopologyItemParser RIGHT_ARROW = new RawTopologyItemParser() {
+        private final Pattern rightArrowPattern = Pattern.compile("\\s*-->\\s+(?<targets>.*)");
+        private Matcher matcher;
+
+        @Override
+        public boolean test(String line) {
+            matcher = rightArrowPattern.matcher(line);
+            return matcher.matches();
+        }
+
+        @Override
+        public void accept(TopologyParserContext context) {
+            context.addTargets(matcher.group("targets").split(","));
+        }
+    };
+}