refactor: Remove unused code and minor refactoring (#2495)

streampipes-extensions/streampipes-connectors-opcua/src/main/java/org/apache/streampipes/extensions/connectors/opcua/config/SpOpcUaConfigExtractor.java

@@ -84,9 +84,9 @@ public static <T extends OpcUaConfig> T extractSharedConfig(IParameterExtractor
     if (useURL) {
       String serverAddress =
           extractor.singleValueParameter(OPC_SERVER_URL.name(), String.class);
-      config.setOpcServerURL(OpcUaUtil.formatServerAddress(serverAddress));
+      config.setOpcServerURL(OpcUaUtil.addOpcPrefixIfNotExists(serverAddress));
     } else {
-      String serverAddress = OpcUaUtil.formatServerAddress(
+      String serverAddress = OpcUaUtil.addOpcPrefixIfNotExists(
           extractor.singleValueParameter(OPC_SERVER_HOST.name(), String.class)
       );
       int port = extractor.singleValueParameter(OPC_SERVER_PORT.name(), int.class);

streampipes-extensions/streampipes-connectors-opcua/src/main/java/org/apache/streampipes/extensions/connectors/opcua/utils/OpcUaUtil.java

@@ -60,18 +60,15 @@
  */
 public class OpcUaUtil {
 
+  private static final String OPC_TCP_PREFIX = "opc.tcp://";
+
   /***
    * Ensures server address starts with {@code opc.tcp://}
    * @param serverAddress server address as given by user
    * @return correctly formated server address
    */
-  public static String formatServerAddress(String serverAddress) {
-
-    if (!serverAddress.startsWith("opc.tcp://")) {
-      serverAddress = "opc.tcp://" + serverAddress;
-    }
-
-    return serverAddress;
+  public static String addOpcPrefixIfNotExists(String serverAddress) {
+    return serverAddress.startsWith(OPC_TCP_PREFIX) ? serverAddress : OPC_TCP_PREFIX + serverAddress;
   }
 
   /***
@@ -103,21 +100,24 @@ public static GuessSchema getSchema(IAdapterParameterExtractor extractor)
         for (OpcNode opcNode : selectedNodes) {
           if (opcNode.hasUnitId()) {
             allProperties.add(PrimitivePropertyBuilder
-                .create(opcNode.getType(), opcNode.getLabel())
-                .label(opcNode.getLabel())
-                .measurementUnit(new URI(opcNode.getQudtURI()))
-                .build());
+                                  .create(opcNode.getType(), opcNode.getLabel())
+                                  .label(opcNode.getLabel())
+                                  .measurementUnit(new URI(opcNode.getQudtURI()))
+                                  .build());
           } else {
             allProperties.add(PrimitivePropertyBuilder
-                .create(opcNode.getType(), opcNode.getLabel())
-                .label(opcNode.getLabel())
-                .build());
+                                  .create(opcNode.getType(), opcNode.getLabel())
+                                  .label(opcNode.getLabel())
+                                  .build());
           }
         }
       }
 
-      var nodeIds = selectedNodes.stream().map(OpcNode::getNodeId).collect(Collectors.toList());
-      var response = spOpcUaClient.getClient().readValues(0, TimestampsToReturn.Both, nodeIds);
+      var nodeIds = selectedNodes.stream()
+                                 .map(OpcNode::getNodeId)
+                                 .collect(Collectors.toList());
+      var response = spOpcUaClient.getClient()
+                                  .readValues(0, TimestampsToReturn.Both, nodeIds);
 
       var returnValues = response.get();
 
@@ -138,20 +138,25 @@ public static GuessSchema getSchema(IAdapterParameterExtractor extractor)
     return builder.build();
   }
 
-  private static void makeEventPreview(List<OpcNode> selectedNodes,
-                                       Map<String, Object> eventPreview,
-                                       Map<String, FieldStatusInfo> fieldStatusInfos,
-                                       List<DataValue> dataValues) {
+  private static void makeEventPreview(
+      List<OpcNode> selectedNodes,
+      Map<String, Object> eventPreview,
+      Map<String, FieldStatusInfo> fieldStatusInfos,
+      List<DataValue> dataValues
+  ) {
 
     for (int i = 0; i < dataValues.size(); i++) {
       var dv = dataValues.get(i);
-      String label = selectedNodes.get(i).getLabel();
+      String label = selectedNodes.get(i)
+                                  .getLabel();
       if (StatusCode.GOOD.equals(dv.getStatusCode())) {
-        var value = dv.getValue().getValue();
+        var value = dv.getValue()
+                      .getValue();
         eventPreview.put(label, value);
         fieldStatusInfos.put(label, FieldStatusInfo.good());
       } else {
-        String additionalInfo = dv.getStatusCode() != null ? dv.getStatusCode().toString() : "Status code is null";
+        String additionalInfo = dv.getStatusCode() != null ? dv.getStatusCode()
+                                                               .toString() : "Status code is null";
         fieldStatusInfos.put(label, FieldStatusInfo.bad(additionalInfo, false));
       }
     }
@@ -179,7 +184,7 @@ public static RuntimeResolvableTreeInputStaticProperty resolveConfig(String inte
       return config;
     }
 
-    var spOpcUaClient = new SpOpcUaClient(
+    SpOpcUaClient spOpcUaClient = new SpOpcUaClient(
         SpOpcUaConfigExtractor.extractSharedConfig(parameterExtractor, new OpcUaConfig())
     );
 

streampipes-extensions/streampipes-connectors-opcua/src/test/java/org/apache/streampipes/extensions/connectors/opcua/utils/OpcUaUtilTest.java

@@ -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.
+ *
+ */
+
+package org.apache.streampipes.extensions.connectors.opcua.utils;
+
+import org.junit.Test;
+
+import static org.junit.Assert.assertEquals;
+
+public class OpcUaUtilTest {
+
+  private static final String SERVER_ADDRESS_WITH_OPC_PREFIX = "opc.tcp://example.com";
+
+  @Test
+  public void testAddOpcPrefixIfNotExistsWithPrefix() {
+    var result = OpcUaUtil.addOpcPrefixIfNotExists(SERVER_ADDRESS_WITH_OPC_PREFIX);
+    assertEquals(SERVER_ADDRESS_WITH_OPC_PREFIX, result);
+  }
+
+  @Test
+  public void testAddOpcPrefixIfNotExistsNoPrefix() {
+    var result = OpcUaUtil.addOpcPrefixIfNotExists("example.com");
+    assertEquals(SERVER_ADDRESS_WITH_OPC_PREFIX, result);
+  }
+}