Repository: apache/streampipes
Pull request #3326 (2024-11-11 13:32:57)

Add topics information to pipeline editor in view mode
<!--
  ~ 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.
  ~
  -->

  <!--
Thanks for contributing! Here are some tips you can follow to help us incorporate your contribution quickly and easily:
1. If this is your first time, please read our contributor guidelines:
    - https://streampipes.apache.org/community/get-involved/
    - https://cwiki.apache.org/confluence/display/STREAMPIPES/Getting+Started
2. Make sure the PR title is formatted like: `[#<GitHub issue id>] PR title ...`
3. If the PR is unfinished, add '[WIP]' in your PR title, e.g., `[WIP][#<GitHub issue id>] PR title ...`.
4. Please write your PR title to summarize what this PR proposes/fixes.
5. Link the PR to the corresponding GitHub issue (if present) in the `Development` section in the right menu bar.
6. Be sure to keep the PR description updated to reflect all changes.
7. If possible, provide a concise example to reproduce the issue for a faster review.
8. Make sure tests pass via `mvn clean install`.
9. (Optional) If the contribution is large, please file an Apache ICLA
    - http://apache.org/licenses/icla.pdf
-->

### Purpose
<!--
Please clarify what changes you are proposing and describe how those changes will address the issue.
Furthermore, describe potential consequences the changes might have.
--> This change introduces the option to view the input and output topics of elements in the pipeline editor’s view mode. For each element, a "View Topics" button is added, which opens a dialog displaying all topics associated with that element. The dialog window also includes a "Code" tab, which will allow code display functionality in the future. Additionally, the topics table in the dialog includes a copy button for copying the content to the user's clipboard. This button is also added to the "View Pipeline as Code" dialog window as option to copy the code to the user's clipboard.

### Remarks
<!--
Is there anything left we need to pay attention on?
Are there some references that might be important? E.g. links to Confluence, or discussions
on the mailing list or GitHub.
-->
PR introduces (a) breaking change(s): No

PR introduces (a) deprecation(s): No



Patch:
diff --git a/streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/DataLakeMeasurementCounterInflux.java b/streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/DataLakeMeasurementCounterInflux.java
index db11222324..0a58b8b29a 100644
--- a/streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/DataLakeMeasurementCounterInflux.java
+++ b/streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/DataLakeMeasurementCounterInflux.java
@@ -22,16 +22,23 @@ import org.apache.streampipes.dataexplorer.query.DataLakeMeasurementCounter;
 import org.apache.streampipes.model.datalake.AggregationFunction;
 import org.apache.streampipes.model.datalake.DataLakeMeasure;
 
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
 import java.util.List;
 import java.util.Optional;
 import java.util.concurrent.CompletableFuture;
 
 public class DataLakeMeasurementCounterInflux extends DataLakeMeasurementCounter {
 
+  private static final Logger LOG = LoggerFactory.getLogger(DataLakeMeasurementCounterInflux.class);
+
   private static final String COUNT_FIELD = "count";
 
-  public DataLakeMeasurementCounterInflux(List<DataLakeMeasure> allMeasurements,
-                                          List<String> measurementNames) {
+  public DataLakeMeasurementCounterInflux(
+      List<DataLakeMeasure> allMeasurements,
+      List<String> measurementNames
+  ) {
     super(allMeasurements, measurementNames);
   }
 
@@ -39,15 +46,21 @@ public class DataLakeMeasurementCounterInflux extends DataLakeMeasurementCounter
   protected CompletableFuture<Integer> createQueryAsAsyncFuture(DataLakeMeasure measure) {
     return CompletableFuture.supplyAsync(() -> {
       var firstColumn = getFirstMeasurementProperty(measure);
+      if (firstColumn == null) {
+        LOG.error(
+            "Could not count events in measurement: {}, because no measurement property was found in event schema",
+            measure.getMeasureName()
+        );
+        return 0;
+      }
+
       var builder = DataLakeInfluxQueryBuilder
-          .create(measure.getMeasureName()).withEndTime(System.currentTimeMillis())
+          .create(measure.getMeasureName())
+          .withEndTime(System.currentTimeMillis())
           .withAggregatedColumn(firstColumn, AggregationFunction.COUNT);
       var queryResult = new DataExplorerInfluxQueryExecutor().executeQuery(builder.build(), Optional.empty(), true);
-      if (queryResult.getTotal() > 0) {
-        return extractResult(queryResult, COUNT_FIELD);
-      } else {
-        return 0;
-      }
+
+      return queryResult.getTotal() > 0 ? extractResult(queryResult, COUNT_FIELD) : 0;
     });
   }
 }
diff --git a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/query/DataLakeMeasurementCounter.java b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/query/DataLakeMeasurementCounter.java
index 844ce5b573..bc3a551ed7 100644
--- a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/query/DataLakeMeasurementCounter.java
+++ b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/query/DataLakeMeasurementCounter.java
@@ -42,8 +42,10 @@ public abstract class DataLakeMeasurementCounter implements IDataLakeMeasurement
   protected final List<DataLakeMeasure> allMeasurements;
   protected final List<String> measurementNames;
 
-  public DataLakeMeasurementCounter(List<DataLakeMeasure> allMeasurements,
-                                         List<String> measurementNames) {
+  public DataLakeMeasurementCounter(
+      List<DataLakeMeasure> allMeasurements,
+      List<String> measurementNames
+  ) {
     this.allMeasurements = allMeasurements;
     this.measurementNames = measurementNames;
   }
@@ -52,12 +54,14 @@ public abstract class DataLakeMeasurementCounter implements IDataLakeMeasurement
   public Map<String, Integer> countMeasurementSizes() {
 
     // create async futures so that count queries can be executed parallel
-    Map<String, CompletableFuture<Integer>> countQueriesFutures = measurementNames.stream()
+    Map<String, CompletableFuture<Integer>> countQueriesFutures = measurementNames
+        .stream()
         .map(this::getMeasure)
         .filter(Objects::nonNull)
         .collect(Collectors.toMap(
-            DataLakeMeasure::getMeasureName,
-            this::createQueryAsAsyncFuture)
+                     DataLakeMeasure::getMeasureName,
+                     this::createQueryAsAsyncFuture
+                 )
         );
 
     return getQueryResults(countQueriesFutures);
@@ -72,7 +76,8 @@ public abstract class DataLakeMeasurementCounter implements IDataLakeMeasurement
   private DataLakeMeasure getMeasure(String measureName) {
     return allMeasurements
         .stream()
-        .filter(m -> m.getMeasureName().equals(measureName))
+        .filter(m -> m.getMeasureName()
+                      .equals(measureName))
         .findFirst()
         .orElse(null);
   }
@@ -83,7 +88,7 @@ public abstract class DataLakeMeasurementCounter implements IDataLakeMeasurement
    * @param queryFutures A Map containing the futures of
    *                     asynchronous count queries mapped by their respective keys.
    * @return A Map representing the results of the queries, where each key corresponds to
-   *         a measure name and the value is the count result.
+   * a measure name and the value is the count result.
    */
   private Map<String, Integer> getQueryResults(Map<String, CompletableFuture<Integer>> queryFutures) {
     Map<String, Integer> resultPerMeasure = new HashMap<>();
@@ -106,18 +111,34 @@ public abstract class DataLakeMeasurementCounter implements IDataLakeMeasurement
    * @return The runtime name of the first measurement property, or null if no such property is found.
    */
   protected String getFirstMeasurementProperty(DataLakeMeasure measure) {
-    return measure.getEventSchema().getEventProperties()
+    var propertyRuntimeName = measure
+        .getEventSchema()
+        .getEventProperties()
         .stream()
         .filter(ep -> ep.getPropertyScope() != null
-            && ep.getPropertyScope().equals(PropertyScope.MEASUREMENT_PROPERTY.name()))
+            && ep.getPropertyScope()
+                 .equals(PropertyScope.MEASUREMENT_PROPERTY.name()))
         .map(EventProperty::getRuntimeName)
         .findFirst()
         .orElse(null);
+
+    if (propertyRuntimeName == null) {
+      LOG.error("No measurement property was found in the event schema found for measure {}", measure.getMeasureName());
+    }
+
+    return propertyRuntimeName;
   }
 
   protected Integer extractResult(SpQueryResult queryResult, String fieldName) {
-    return ((Double) (
-        queryResult.getAllDataSeries().get(0).getRows().get(0).get(queryResult.getHeaders().indexOf(fieldName)))
+    return (
+        (Double) (
+            queryResult.getAllDataSeries()
+                       .get(0)
+                       .getRows()
+                       .get(0)
+                       .get(queryResult.getHeaders()
+                                       .indexOf(fieldName))
+        )
     ).intValue();
   }
 
diff --git a/streampipes-extensions/streampipes-processors-enricher-jvm/src/main/java/org/apache/streampipes/processors/enricher/jvm/processor/math/MathOpProcessor.java b/streampipes-extensions/streampipes-processors-enricher-jvm/src/main/java/org/apache/streampipes/processors/enricher/jvm/processor/math/MathOpProcessor.java
index 7a2409d1b2..0571496f6b 100644
--- a/streampipes-extensions/streampipes-processors-enricher-jvm/src/main/java/org/apache/streampipes/processors/enricher/jvm/processor/math/MathOpProcessor.java
+++ b/streampipes-extensions/streampipes-processors-enricher-jvm/src/main/java/org/apache/streampipes/processors/enricher/jvm/processor/math/MathOpProcessor.java
@@ -47,10 +47,10 @@ import org.apache.streampipes.wrapper.standalone.StreamPipesDataProcessor;
 
 public class MathOpProcessor extends StreamPipesDataProcessor {
 
-  private static final String RESULT_FIELD = "calculationResult";
-  private static final String LEFT_OPERAND = "leftOperand";
-  private static final String RIGHT_OPERAND = "rightOperand";
-  private static final String OPERATION = "operation";
+  protected static final String RESULT_FIELD = "calculationResult";
+  protected static final String LEFT_OPERAND = "leftOperand";
+  protected static final String RIGHT_OPERAND = "rightOperand";
+  protected static final String OPERATION = "operation";
 
   Operation arithmeticOperation = null;
   String leftOperand;
diff --git a/ui/cypress/fixtures/connect/compact/fileReplay.yml b/ui/cypress/fixtures/connect/compact/fileReplay.yml
new file mode 100644
index 0000000000..3c57795f10
--- /dev/null
+++ b/ui/cypress/fixtures/connect/compact/fileReplay.yml
@@ -0,0 +1,40 @@
+# 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.
+
+id: sp:adapterdescription:IfmzfQ
+name: File Stream Adapter Test
+description: ''
+appId: org.apache.streampipes.connect.iiot.protocol.stream.file
+configuration:
+    - filePath: compactTest.csv
+    - replaceTimestamp:
+          - ''
+    - replayOnce: 'no'
+    - speed: keepOriginalTime
+    - delimiter: ;
+      format: CSV
+      header:
+          - Header
+schema:
+    timestamp:
+        description: ''
+        propertyScope: HEADER_PROPERTY
+        semanticType: http://schema.org/DateTime
+transform:
+    rename: {}
+    measurementUnit: {}
+createOptions:
+    persist: true
+    start: true
diff --git a/ui/src/app/connect/components/adapter-code-panel/adapter-code-panel.component.html b/ui/src/app/connect/components/adapter-code-panel/adapter-code-panel.component.html
index 8a05afabd3..6f553c63e3 100644
--- a/ui/src/app/connect/components/adapter-code-panel/adapter-code-panel.component.html
+++ b/ui/src/app/connect/components/adapter-code-panel/adapter-code-panel.component.html
@@ -18,6 +18,7 @@
 
 <div fxFlex="100" *ngIf="compactAdapter" class="mt-10">
     <sp-configuration-code-panel
+        *ngIf="compactAdapter"
         [configuration]="compactAdapter"
         [maxHeight]="maxHeight"
     >
diff --git a/ui/src/app/core-ui/configuration-code-panel/configuration-code-panel.component.html b/ui/src/app/core-ui/configuration-code-panel/configuration-code-panel.component.html
index a6590ce049..7e232c3089 100644
--- a/ui/src/app/core-ui/configuration-code-panel/configuration-code-panel.component.html
+++ b/ui/src/app/core-ui/configuration-code-panel/configuration-code-panel.component.html
@@ -16,8 +16,23 @@
   ~
   -->
 
-<mat-tab-group color="accent" [mat-stretch-tabs]="false">
+<mat-tab-group
+    color="accent"
+    [mat-stretch-tabs]="false"
+    (selectedTabChange)="onTabChanged($event)"
+>
     <mat-tab label="YAML">
+        <div class="container-copy-button-container">
+            <div class="copy-button-container">
+                <button
+                    mat-icon-button
+                    [cdkCopyToClipboard]="currentConfiguration"
+                    matTooltip="Copy"
+                >
+                    <mat-icon>content_copy</mat-icon>
+                </button>
+            </div>
+        </div>
         <pre
             [innerHTML]="configuration | yamlpretty"
             class="preview-text"
@@ -26,6 +41,17 @@
         ></pre>
     </mat-tab>
     <mat-tab label="JSON">
+        <div class="container-copy-button-container">
+            <div class="copy-button-container">
+                <button
+                    mat-icon-button
+                    [cdkCopyToClipboard]="currentConfiguration"
+                    matTooltip="Copy"
+                >
+                    <mat-icon>content_copy</mat-icon>
+                </button>
+            </div>
+        </div>
         <pre
             [innerHTML]="configuration | jsonpretty"
             class="preview-text"
diff --git a/ui/src/app/core-ui/configuration-code-panel/configuration-code-panel.component.scss b/ui/src/app/core-ui/configuration-code-panel/configuration-code-panel.component.scss
index c02ba8acc6..4025845c82 100644
--- a/ui/src/app/core-ui/configuration-code-panel/configuration-code-panel.component.scss
+++ b/ui/src/app/core-ui/configuration-code-panel/configuration-code-panel.component.scss
@@ -28,3 +28,17 @@
     overflow-y: scroll;
     white-space: pre-wrap;
 }
+
+.mat-mdc-icon-button {
+    color: #fff;
+}
+
+.copy-button-container {
+    position: absolute;
+    right: 20px;
+    top: 20px;
+}
+
+.container-copy-button-container {
+    position: relative;
+}
diff --git a/ui/src/app/core-ui/configuration-code-panel/configuration-code-panel.component.ts b/ui/src/app/core-ui/configuration-code-panel/configuration-code-panel.component.ts
index bf8deb6418..9c3a88a970 100644
--- a/ui/src/app/core-ui/configuration-code-panel/configuration-code-panel.component.ts
+++ b/ui/src/app/core-ui/configuration-code-panel/configuration-code-panel.component.ts
@@ -17,16 +17,37 @@
  */
 
 import { Component, Input, OnInit } from '@angular/core';
+import { stringify } from 'yaml';
+import { MatTabChangeEvent } from '@angular/material/tabs';
 
 @Component({
     selector: 'sp-configuration-code-panel',
     templateUrl: './configuration-code-panel.component.html',
     styleUrls: ['./configuration-code-panel.component.scss'],
 })
-export class ConfigurationCodePanelComponent {
+export class ConfigurationCodePanelComponent implements OnInit {
     @Input()
     configuration: any;
 
     @Input()
     maxHeight = '300px';
+
+    configurationYaml: string;
+    configurationJson: string;
+
+    currentConfiguration: string;
+
+    ngOnInit() {
+        this.configurationYaml = stringify(this.configuration);
+        this.configurationJson = JSON.stringify(this.configuration);
+        this.currentConfiguration = this.configurationYaml;
+    }
+
+    onTabChanged(event: MatTabChangeEvent) {
+        if (event.index === 0) {
+            this.currentConfiguration = this.configurationYaml;
+        } else {
+            this.currentConfiguration = this.configurationJson;
+        }
+    }
 }
diff --git a/ui/src/app/core-ui/core-ui.module.ts b/ui/src/app/core-ui/core-ui.module.ts
index 5a8815dce0..bd0c996d6e 100644
--- a/ui/src/app/core-ui/core-ui.module.ts
+++ b/ui/src/app/core-ui/core-ui.module.ts
@@ -26,6 +26,7 @@ import { MatNativeDateModule } from '@angular/material/core';
 import { MatDatepickerModule } from '@angular/material/datepicker';
 import { MatProgressSpinnerModule } from '@angular/material/progress-spinner';
 import { MatSnackBarModule } from '@angular/material/snack-bar';
+import { ClipboardModule } from '@angular/cdk/clipboard';
 
 import { MatChipsModule } from '@angular/material/chips';
 import { MatSliderModule } from '@angular/material/slider';
@@ -119,6 +120,7 @@ import { PipelineElementTemplateConfigItemComponent } from './pipeline-element-t
 import { ConfigurationCodePanelComponent } from './configuration-code-panel/configuration-code-panel.component';
 import { JsonPrettyPrintPipe } from './pipes/json-pretty-print.pipe';
 import { YamlPrettyPrintPipe } from './pipes/yaml-pretty-print.pipe';
+import { TopicsComponent } from './topics/topics.component';
 
 @NgModule({
     imports: [
@@ -151,6 +153,7 @@ import { YamlPrettyPrintPipe } from './pipes/yaml-pretty-print.pipe';
         ReactiveFormsModule,
         FormsModule,
         CdkTableModule,
+        ClipboardModule,
         MatAutocompleteModule,
         MatSnackBarModule,
         MatProgressSpinnerModule,
@@ -182,6 +185,7 @@ import { YamlPrettyPrintPipe } from './pipes/yaml-pretty-print.pipe';
         PipelineElementRuntimeInfoComponent,
         PipelineElementDocumentationComponent,
         HelpComponent,
+        TopicsComponent,
         StaticAnyInputComponent,
         StaticPropertyComponent,
         StaticFreeInputComponent,
@@ -236,6 +240,7 @@ import { YamlPrettyPrintPipe } from './pipes/yaml-pretty-print.pipe';
         PipelineElementRuntimeInfoComponent,
         PipelineElementDocumentationComponent,
         HelpComponent,
+        TopicsComponent,
         StaticAnyInputComponent,
         StaticPropertyComponent,
         StaticFreeInputComponent,
diff --git a/ui/src/app/core-ui/pipes/yaml-pretty-print.pipe.ts b/ui/src/app/core-ui/pipes/yaml-pretty-print.pipe.ts
index 88f8b4b459..233bb6f11d 100644
--- a/ui/src/app/core-ui/pipes/yaml-pretty-print.pipe.ts
+++ b/ui/src/app/core-ui/pipes/yaml-pretty-print.pipe.ts
@@ -17,7 +17,7 @@
  */
 
 import { Injectable, Pipe, PipeTransform } from '@angular/core';
-import { parse, stringify } from 'yaml';
+import { stringify } from 'yaml';
 
 @Pipe({
     name: 'yamlpretty',
diff --git a/ui/src/app/core-ui/topics/topics.component.html b/ui/src/app/core-ui/topics/topics.component.html
new file mode 100644
index 0000000000..73c5be895c
--- /dev/null
+++ b/ui/src/app/core-ui/topics/topics.component.html
@@ -0,0 +1,213 @@
+<!--
+  ~ 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.
+  ~
+  -->
+
+<div class="sp-dialog-container">
+    <div class="sp-dialog-content p-15" fxLayout="column">
+        <div fxLayout="row" fxFlex="100">
+            <div fxLayout="column" fxFlex>
+                <h4>{{ pipelineElement?.name }}</h4>
+                <small>{{ pipelineElement?.description }}</small>
+            </div>
+            <div class="element-id" fxLayoutAlign="end start">
+                <span>ID</span>&nbsp;
+                <b>{{
+                    isDataStream
+                        ? pipelineElement?.elementId
+                        : pipelineElement?.appId
+                }}</b>
+            </div>
+        </div>
+
+        <mat-tab-group
+            color="accent"
+            [selectedIndex]="selectedTabIndex"
+            (selectedIndexChange)="selectedTabIndex = $event"
+        >
+            <mat-tab *ngFor="let tab of tabs" [label]="tab"></mat-tab>
+        </mat-tab-group>
+
+        <div *ngIf="selectedTabIndex === 0">
+            <ng-container *ngIf="isSpDataStream()">
+                <table
+                    mat-table
+                    [dataSource]="
+                        pipelineElement?.eventGrounding?.transportProtocols
+                    "
+                >
+                    <ng-container matColumnDef="outputTopics">
+                        <th mat-header-cell *matHeaderCellDef>
+                            <strong>Output Topics</strong>
+                        </th>
+                        <td mat-cell *matCellDef="let element">
+                            <div
+                                fxLayout="row"
+                                fxLayoutAlign="space-between center"
+                            >
+                                <span>{{
+                                    element?.topicDefinition?.actualTopicName
+                                }}</span>
+                                <button
+                                    mat-icon-button
+                                    [cdkCopyToClipboard]="
+                                        element.topicDefinition.actualTopicName
+                                    "
+                                    matTooltip="Copy"
+                                >
+                                    <mat-icon>content_copy</mat-icon>
+                                </button>
+                            </div>
+                        </td>
+                    </ng-container>
+                    <tr mat-header-row *matHeaderRowDef="['outputTopics']"></tr>
+                    <tr
+                        mat-row
+                        *matRowDef="let row; columns: ['outputTopics']"
+                    ></tr>
+                </table>
+            </ng-container>
+
+            <ng-container *ngIf="isDataProcessorInvocation()">
+                <table mat-table [dataSource]="pipelineElement?.inputStreams">
+                    <ng-container matColumnDef="inputTopics">
+                        <th mat-header-cell *matHeaderCellDef>
+                            <strong>Input Topics</strong>
+                        </th>
+                        <td mat-cell *matCellDef="let element">
+                            <div
+                                fxLayout="row"
+                                fxLayoutAlign="space-between center"
+                            >
+                                <span>{{
+                                    element?.eventGrounding
+                                        ?.transportProtocols?.[0]
+                                        ?.topicDefinition?.actualTopicName
+                                }}</span>
+                                <button
+                                    mat-icon-button
+                                    [cdkCopyToClipboard]="
+                                        element?.eventGrounding
+                                            ?.transportProtocols?.[0]
+                                            ?.topicDefinition?.actualTopicName
+                                    "
+                                    matTooltip="Copy"
+                                >
+                                    <mat-icon>content_copy</mat-icon>
+                                </button>
+                            </div>
+                        </td>
+                    </ng-container>
+                    <tr mat-header-row *matHeaderRowDef="['inputTopics']"></tr>
+                    <tr
+                        mat-row
+                        *matRowDef="let row; columns: ['inputTopics']"
+                    ></tr>
+                </table>
+
+                <table
+                    mat-table
+                    [dataSource]="
+                        pipelineElement?.outputStream?.eventGrounding
+                            ?.transportProtocols
+                    "
+                >
+                    <ng-container matColumnDef="outputTopics">
+                        <th mat-header-cell *matHeaderCellDef>
+                            <strong>Output Topics</strong>
+                        </th>
+                        <td mat-cell *matCellDef="let element">
+                            <div
+                                fxLayout="row"
+                                fxLayoutAlign="space-between center"
+                            >
+                                <span>{{
+                                    element?.topicDefinition?.actualTopicName
+                                }}</span>
+                                <button
+                                    mat-icon-button
+                                    [cdkCopyToClipboard]="
+                                        element?.topicDefinition
+                                            ?.actualTopicName
+                                    "
+                                    matTooltip="Copy"
+                                >
+                                    <mat-icon>content_copy</mat-icon>
+                                </button>
+                            </div>
+                        </td>
+                    </ng-container>
+                    <tr mat-header-row *matHeaderRowDef="['outputTopics']"></tr>
+                    <tr
+                        mat-row
+                        *matRowDef="let row; columns: ['outputTopics']"
+                    ></tr>
+                </table>
+            </ng-container>
+
+            <ng-container *ngIf="isDataSinkInvocation()">
+                <table mat-table [dataSource]="pipelineElement?.inputStreams">
+                    <ng-container matColumnDef="inputTopics">
+                        <th mat-header-cell *matHeaderCellDef>
+                            <strong>Input Topics</strong>
+                        </th>
+                        <td mat-cell *matCellDef="let element">
+                            <div
+                                fxLayout="row"
+                                fxLayoutAlign="space-between center"
+                            >
+                                <span>{{
+                                    element?.eventGrounding
+                                        ?.transportProtocols?.[0]
+                                        ?.topicDefinition?.actualTopicName
+                                }}</span>
+                                <button
+                                    mat-icon-button
+                                    [cdkCopyToClipboard]="
+                                        element?.eventGrounding
+                                            ?.transportProtocols?.[0]
+                                            ?.topicDefinition?.actualTopicName
+                                    "
+                                    matTooltip="Copy"
+                                >
+                                    <mat-icon>content_copy</mat-icon>
+                                </button>
+                            </div>
+                        </td>
+                    </ng-container>
+                    <tr mat-header-row *matHeaderRowDef="['inputTopics']"></tr>
+                    <tr
+                        mat-row
+                        *matRowDef="let row; columns: ['inputTopics']"
+                    ></tr>
+                </table>
+            </ng-container>
+        </div>
+    </div>
+
+    <mat-divider></mat-divider>
+
+    <div class="sp-dialog-actions actions-align-right">
+        <button
+            mat-button
+            mat-raised-button
+            class="mat-basic"
+            (click)="close()"
+        >
+            Close
+        </button>
+    </div>
+</div>
diff --git a/ui/src/app/core-ui/topics/topics.component.scss b/ui/src/app/core-ui/topics/topics.component.scss
new file mode 100644
index 0000000000..c077f68f35
--- /dev/null
+++ b/ui/src/app/core-ui/topics/topics.component.scss
@@ -0,0 +1,29 @@
+/*
+ * 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.
+ *
+ */
+
+@import '../../../scss/sp/sp-dialog.scss';
+
+.element-id {
+    border-radius: 5px;
+    margin-right: 10px;
+    margin-top: 5px;
+    margin-bottom: 5px;
+    font-size: small;
+    display: inline-block;
+    padding: 5px;
+}
diff --git a/ui/src/app/core-ui/topics/topics.component.ts b/ui/src/app/core-ui/topics/topics.component.ts
new file mode 100644
index 0000000000..2ce28d6dc7
--- /dev/null
+++ b/ui/src/app/core-ui/topics/topics.component.ts
@@ -0,0 +1,77 @@
+/*
+ * 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.
+ *
+ */
+
+import { Component, Input, OnInit } from '@angular/core';
+import {
+    DataProcessorInvocation,
+    DataSinkInvocation,
+    SpDataStream,
+} from '@streampipes/platform-services';
+import { DialogRef } from '@streampipes/shared-ui';
+import { PipelineElementUnion } from '../../editor/model/editor.model';
+
+@Component({
+    selector: 'sp-pipeline-element-topics',
+    templateUrl: './topics.component.html',
+    styleUrls: ['./topics.component.scss'],
+})
+export class TopicsComponent implements OnInit {
+    selectedTabIndex = 0;
+
+    availableTabs = ['Topics', 'Code'];
+    tabs: string[] = [];
+
+    @Input()
+    pipelineElement: PipelineElementUnion;
+    isDataStream: boolean;
+
+    constructor(private dialogRef: DialogRef<TopicsComponent>) {}
+
+    ngOnInit() {
+        if (
+            this.pipelineElement instanceof SpDataStream ||
+            this.pipelineElement instanceof DataProcessorInvocation ||
+            this.pipelineElement instanceof DataSinkInvocation
+        ) {
+            this.tabs = this.availableTabs;
+        } else {
+            this.tabs = [this.availableTabs[1]];
+            this.selectedTabIndex = 1;
+        }
+    }
+
+    isSpDataStream(): boolean {
+        return this.pipelineElement instanceof SpDataStream;
+    }
+
+    isDataProcessorInvocation(): boolean {
+        return this.pipelineElement instanceof DataProcessorInvocation;
+    }
+
+    isDataSinkInvocation(): boolean {
+        return this.pipelineElement instanceof DataSinkInvocation;
+    }
+
+    close() {
+        setTimeout(() => {
+            this.dialogRef.close();
+        });
+    }
+
+    protected readonly SpDataStream = SpDataStream;
+}
diff --git a/ui/src/app/editor/components/pipeline/pipeline.component.html b/ui/src/app/editor/components/pipeline/pipeline.component.html
index 20b984a570..8b77b5b017 100644
--- a/ui/src/app/editor/components/pipeline/pipeline.component.html
+++ b/ui/src/app/editor/components/pipeline/pipeline.component.html
@@ -26,6 +26,7 @@
     *ngFor="
         let pipelineElementConfig of rawPipelineModel | enabledPipelineElement
     "
+    style="position: relative"
 >
     <sp-dropped-pipeline-element
         [allElements]="allElements"
@@ -41,4 +42,15 @@
         (deleteEmitter)="handleDeleteOption($event)"
     >
     </sp-dropped-pipeline-element>
+    <div class="topics-button customize-button" *ngIf="readonly">
+        <button
+            class="topics-icon-button"
+            mat-icon-button
+            matTooltip="View Topics"
+            [matTooltipPosition]="'above'"
+            (click)="openTopicsDialog(pipelineElementConfig)"
+        >
+            <i class="material-icons topics-icon-size">code</i>
+        </button>
+    </div>
 </div>
diff --git a/ui/src/app/editor/components/pipeline/pipeline.component.scss b/ui/src/app/editor/components/pipeline/pipeline.component.scss
new file mode 100644
index 0000000000..4d3c139be2
--- /dev/null
+++ b/ui/src/app/editor/components/pipeline/pipeline.component.scss
@@ -0,0 +1,43 @@
+/*
+ * 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.
+ *
+ */
+
+.topics-button {
+    cursor: pointer;
+    width: 30px;
+    height: 30px;
+    border-radius: 20%;
+    box-shadow: 0 0 3px var(--color-accent);
+    background-color: var(--color-accent);
+    color: rgba(255, 255, 255, 0.75);
+    position: absolute;
+    left: 35%;
+    top: 75%;
+    z-index: 50;
+    margin-top: 5px;
+}
+
+.topics-icon-button {
+    width: 100%;
+    height: 100%;
+    padding: 0;
+}
+
+.topics-icon-size {
+    font-size: 16px;
+    color: white;
+}
diff --git a/ui/src/app/editor/components/pipeline/pipeline.component.ts b/ui/src/app/editor/components/pipeline/pipeline.component.ts
index 71843f0692..7ff4bc3007 100644
--- a/ui/src/app/editor/components/pipeline/pipeline.component.ts
+++ b/ui/src/app/editor/components/pipeline/pipeline.component.ts
@@ -73,6 +73,7 @@ import { IdGeneratorService } from '../../../core-services/id-generator/id-gener
 @Component({
     selector: 'sp-pipeline',
     templateUrl: './pipeline.component.html',
+    styleUrls: ['./pipeline.component.scss'],
 })
 export class PipelineComponent implements OnInit, OnDestroy {
     @Input()
@@ -81,6 +82,9 @@ export class PipelineComponent implements OnInit, OnDestroy {
     @Input()
     rawPipelineModel: PipelineElementConfig[];
 
+    @Input()
+    pipelineElement: PipelineElementConfig;
+
     @Input()
     allElements: PipelineElementUnion[];
 
@@ -586,4 +590,8 @@ export class PipelineComponent implements OnInit, OnDestroy {
                     );
             });
     }
+
+    openTopicsDialog(pipelineElementConfig: PipelineElementConfig) {
+        this.editorService.openTopicsDialog(pipelineElementConfig);
+    }
 }
diff --git a/ui/src/app/editor/dialog/save-pipeline/save-pipeline-settings/save-pipeline-settings.component.html b/ui/src/app/editor/dialog/save-pipeline/save-pipeline-settings/save-pipeline-settings.component.html
index 18f221cbfc..482d98e358 100644
--- a/ui/src/app/editor/dialog/save-pipeline/save-pipeline-settings/save-pipeline-settings.component.html
+++ b/ui/src/app/editor/dialog/save-pipeline/save-pipeline-settings/save-pipeline-settings.component.html
@@ -95,6 +95,7 @@
                 >Show pipeline configuration as code</mat-expansion-panel-header
             >
             <sp-configuration-code-panel
+                *ngIf="compactPipeline"
                 [configuration]="compactPipeline"
                 maxHeight="none"
             >
diff --git a/ui/src/app/editor/services/editor.service.ts b/ui/src/app/editor/services/editor.service.ts
index 45ba82218d..0e0f19954b 100644
--- a/ui/src/app/editor/services/editor.service.ts
+++ b/ui/src/app/editor/services/editor.service.ts
@@ -40,6 +40,7 @@ import { DialogService, PanelType } from '@streampipes/shared-ui';
 import { map } from 'rxjs/operators';
 import { NGX_LOADING_BAR_IGNORED } from '@ngx-loading-bar/http-client';
 import { HelpComponent } from '../../core-ui/help/help.component';
+import { TopicsComponent } from 'src/app/core-ui/topics/topics.component';
 
 @Injectable({ providedIn: 'root' })
 export class EditorService {
@@ -175,6 +176,17 @@ export class EditorService {
         });
     }
 
+    openTopicsDialog(pipelineElementConfig: PipelineElementConfig) {
+        this.dialogService.open(TopicsComponent, {
+            panelType: PanelType.STANDARD_PANEL,
+            title: 'View Topics of ' + pipelineElementConfig.payload.name,
+            width: '70vw',
+            data: {
+                pipelineElement: pipelineElementConfig.payload,
+            },
+        });
+    }
+
     initiatePipelinePreview(
         pipeline: Pipeline,
     ): Observable<PipelinePreviewModel> {
diff --git a/ui/src/app/pipeline-details/dialogs/pipeline-code/pipeline-code-dialog.component.html b/ui/src/app/pipeline-details/dialogs/pipeline-code/pipeline-code-dialog.component.html
index 2b89fa567e..9e17a966ea 100644
--- a/ui/src/app/pipeline-details/dialogs/pipeline-code/pipeline-code-dialog.component.html
+++ b/ui/src/app/pipeline-details/dialogs/pipeline-code/pipeline-code-dialog.component.html
@@ -20,6 +20,7 @@
     <div class="sp-dialog-content">
         <div fxFlex="100" fxLayout="column" class="p-10">
             <sp-configuration-code-panel
+                *ngIf="compactPipeline"
                 [configuration]="compactPipeline"
                 maxHeight="none"
             >
diff --git a/ui/src/app/pipeline-details/pipeline-details.module.ts b/ui/src/app/pipeline-details/pipeline-details.module.ts
index f05009a578..6eb5f6fb11 100644
--- a/ui/src/app/pipeline-details/pipeline-details.module.ts
+++ b/ui/src/app/pipeline-details/pipeline-details.module.ts
@@ -22,6 +22,7 @@ import { FormsModule, ReactiveFormsModule } from '@angular/forms';
 import { MatTabsModule } from '@angular/material/tabs';
 import { MatButtonModule } from '@angular/material/button';
 import { CommonModule } from '@angular/common';
+import { ClipboardModule } from '@angular/cdk/clipboard';
 import { MatProgressSpinnerModule } from '@angular/material/progress-spinner';
 import { PipelinePreviewComponent } from './components/preview/pipeline-preview.component';
 import { EditorModule } from '../editor/editor.module';
@@ -62,6 +63,7 @@ import { PipelineCodeDialogComponent } from './dialogs/pipeline-code/pipeline-co
         MatExpansionModule,
         MatSlideToggleModule,
         MatDivider,
+        ClipboardModule,
     ],
     declarations: [
         PipelineActionsComponent,


Test patch:
diff --git a/streampipes-extensions/streampipes-processors-enricher-jvm/src/test/java/org/apache/streampipes/processors/enricher/jvm/processor/math/MathOpProcessorTest.java b/streampipes-extensions/streampipes-processors-enricher-jvm/src/test/java/org/apache/streampipes/processors/enricher/jvm/processor/math/MathOpProcessorTest.java
new file mode 100644
index 0000000000..fb2a1d5666
--- /dev/null
+++ b/streampipes-extensions/streampipes-processors-enricher-jvm/src/test/java/org/apache/streampipes/processors/enricher/jvm/processor/math/MathOpProcessorTest.java
@@ -0,0 +1,121 @@
+/*
+ * 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.apache.streampipes.processors.enricher.jvm.processor.math;
+
+import org.apache.streampipes.test.executors.ProcessingElementTestExecutor;
+import org.apache.streampipes.test.executors.TestConfiguration;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.util.List;
+import java.util.Map;
+
+public class MathOpProcessorTest{
+  private MathOpProcessor processor;
+
+  @BeforeEach
+  public void setup(){
+    processor = new MathOpProcessor();
+  }
+
+  @Test
+  public void detectAdditionOperator(){
+    TestConfiguration leftOperandConfiguration = TestConfiguration.builder()
+            .configWithDefaultPrefix(MathOpProcessor.LEFT_OPERAND, "leftOperand")
+            .configWithDefaultPrefix(MathOpProcessor.RIGHT_OPERAND, "rightOperand")
+            .config(MathOpProcessor.OPERATION, "+").build();
+
+    List<Map<String, Object>> inputEvents = List.of(
+            Map.of("leftOperand", 5.0d, "rightOperand", 5.0d));
+    List<Map<String, Object>> outputEvents = List.of(
+            Map.of("leftOperand", 5.0d, "rightOperand", 5.0d, "calculationResult", 10.0d));
+    ProcessingElementTestExecutor testExecutor = new ProcessingElementTestExecutor(
+            processor, leftOperandConfiguration);
+
+    testExecutor.run(inputEvents, outputEvents);
+  }
+
+  @Test
+  public void detectSubtractionOperator(){
+    TestConfiguration leftOperandConfiguration = TestConfiguration.builder()
+            .configWithDefaultPrefix(MathOpProcessor.LEFT_OPERAND, "leftOperand")
+            .configWithDefaultPrefix(MathOpProcessor.RIGHT_OPERAND, "rightOperand")
+            .config(MathOpProcessor.OPERATION, "-").build();
+
+    List<Map<String, Object>> inputEvents = List.of(
+            Map.of("leftOperand", 5.0d, "rightOperand", 5.0d));
+    List<Map<String, Object>> outputEvents = List.of(
+            Map.of("leftOperand", 5.0d, "rightOperand", 5.0d, "calculationResult", 0.0d));
+    ProcessingElementTestExecutor testExecutor = new ProcessingElementTestExecutor(
+            processor, leftOperandConfiguration);
+
+    testExecutor.run(inputEvents, outputEvents);
+  }
+
+  @Test
+  public void detectMultiplicationOperator(){
+    TestConfiguration leftOperandConfiguration = TestConfiguration.builder()
+            .configWithDefaultPrefix(MathOpProcessor.LEFT_OPERAND, "leftOperand")
+            .configWithDefaultPrefix(MathOpProcessor.RIGHT_OPERAND, "rightOperand")
+            .config(MathOpProcessor.OPERATION, "*").build();
+
+    List<Map<String, Object>> inputEvents = List.of(
+            Map.of("leftOperand", 5.0d, "rightOperand", 5.0d));
+    List<Map<String, Object>> outputEvents = List.of(
+            Map.of("leftOperand", 5.0d, "rightOperand", 5.0d, "calculationResult", 25.0d));
+    ProcessingElementTestExecutor testExecutor = new ProcessingElementTestExecutor(
+            processor, leftOperandConfiguration);
+
+    testExecutor.run(inputEvents, outputEvents);
+  }
+
+  @Test
+  public void detectDivisionOperator(){
+    TestConfiguration leftOperandConfiguration = TestConfiguration.builder()
+            .configWithDefaultPrefix(MathOpProcessor.LEFT_OPERAND, "leftOperand")
+            .configWithDefaultPrefix(MathOpProcessor.RIGHT_OPERAND, "rightOperand")
+            .config(MathOpProcessor.OPERATION, "/").build();
+
+    List<Map<String, Object>> inputEvents = List.of(
+            Map.of("leftOperand", 5.0d, "rightOperand", 5.0d));
+    List<Map<String, Object>> outputEvents = List.of(
+            Map.of("leftOperand", 5.0d, "rightOperand", 5.0d, "calculationResult", 1.0d));
+    ProcessingElementTestExecutor testExecutor = new ProcessingElementTestExecutor(
+            processor, leftOperandConfiguration);
+
+    testExecutor.run(inputEvents, outputEvents);
+  }
+
+  @Test
+  public void detectModulusOperator(){
+    TestConfiguration leftOperandConfiguration = TestConfiguration.builder()
+            .configWithDefaultPrefix(MathOpProcessor.LEFT_OPERAND, "leftOperand")
+            .configWithDefaultPrefix(MathOpProcessor.RIGHT_OPERAND, "rightOperand")
+            .config(MathOpProcessor.OPERATION, "%").build();
+
+    List<Map<String, Object>> inputEvents = List.of(
+            Map.of("leftOperand", 5.0d, "rightOperand", 5.0d));
+    List<Map<String, Object>> outputEvents = List.of(
+            Map.of("leftOperand", 5.0d, "rightOperand", 5.0d, "calculationResult", 0.0d));
+    ProcessingElementTestExecutor testExecutor = new ProcessingElementTestExecutor(
+            processor, leftOperandConfiguration);
+
+    testExecutor.run(inputEvents, outputEvents);
+  }
+}
\ No newline at end of file
diff --git a/ui/cypress/fixtures/connect/compact/compactTest.csv b/ui/cypress/fixtures/connect/compact/compactTest.csv
new file mode 100644
index 0000000000..6cf117f7c1
--- /dev/null
+++ b/ui/cypress/fixtures/connect/compact/compactTest.csv
@@ -0,0 +1,2 @@
+timestamp;value;temperature
+1000;1.0;0.1
diff --git a/ui/cypress/tests/connect/compact/addCompactAdapter.spec.ts b/ui/cypress/tests/connect/compact/addCompactAdapter.spec.ts
index c19e345f8d..57d2466c15 100644
--- a/ui/cypress/tests/connect/compact/addCompactAdapter.spec.ts
+++ b/ui/cypress/tests/connect/compact/addCompactAdapter.spec.ts
@@ -19,6 +19,7 @@
 import { ConnectUtils } from '../../../support/utils/connect/ConnectUtils';
 import { CompactAdapterUtils } from '../../../support/utils/connect/CompactAdapterUtils';
 import { PipelineUtils } from '../../../support/utils/pipeline/PipelineUtils';
+import { FileManagementUtils } from '../../../support/utils/FileManagementUtils';
 
 describe('Add Compact Adapters', () => {
     beforeEach('Setup Test', () => {
@@ -96,4 +97,19 @@ describe('Add Compact Adapters', () => {
             },
         );
     });
+
+    it('Add file stream adapter via the compact API. Start Adapter with Pipeline', () => {
+        FileManagementUtils.addFile('connect/compact/compactTest.csv');
+
+        cy.readFile('cypress/fixtures/connect/compact/fileReplay.yml').then(
+            ymlDescription => {
+                CompactAdapterUtils.storeCompactYmlAdapter(ymlDescription).then(
+                    () => {
+                        ConnectUtils.validateAdapterIsRunning();
+                        PipelineUtils.checkAmountOfPipelinesPipeline(1);
+                    },
+                );
+            },
+        );
+    });
 });