fix(#2944): Fix connection error in PLC S7 connector

streampipes-extensions/streampipes-connectors-plc/src/main/java/org/apache/streampipes/extensions/connectors/plc/PlcConnectorsModuleExport.java

@@ -28,18 +28,23 @@
 import org.apache.streampipes.extensions.connectors.plc.adapter.modbus.Plc4xModbusAdapter;
 import org.apache.streampipes.extensions.connectors.plc.adapter.s7.Plc4xS7Adapter;
 
+import org.apache.plc4x.java.api.PlcDriverManager;
+
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
 
 public class PlcConnectorsModuleExport implements IExtensionModuleExport {
+
   @Override
   public List<StreamPipesAdapter> adapters() {
+    var driverManager = PlcDriverManager.getDefault();
+    var connectionManager = driverManager.getConnectionManager();
     var adapters = new ArrayList<StreamPipesAdapter>(List.of(
         new Plc4xModbusAdapter(),
-        new Plc4xS7Adapter()
+        new Plc4xS7Adapter(driverManager, connectionManager)
     ));
-    adapters.addAll(new GenericAdapterGenerator().makeAvailableAdapters());
+    adapters.addAll(new GenericAdapterGenerator().makeAvailableAdapters(driverManager, connectionManager));
     return adapters;
   }
 

streampipes-extensions/streampipes-connectors-plc/src/main/java/org/apache/streampipes/extensions/connectors/plc/adapter/GenericAdapterGenerator.java

@@ -21,6 +21,7 @@
 import org.apache.streampipes.extensions.api.connect.StreamPipesAdapter;
 import org.apache.streampipes.extensions.connectors.plc.adapter.generic.GenericPlc4xAdapter;
 
+import org.apache.plc4x.java.api.PlcConnectionManager;
 import org.apache.plc4x.java.api.PlcDriverManager;
 import org.apache.plc4x.java.api.exceptions.PlcConnectionException;
 import org.slf4j.Logger;
@@ -41,11 +42,10 @@ public class GenericAdapterGenerator {
   );
 
 
-  public List<StreamPipesAdapter> makeAvailableAdapters() {
+  public List<StreamPipesAdapter> makeAvailableAdapters(PlcDriverManager driverManager,
+                                                        PlcConnectionManager connectionManager) {
     var adapters = new ArrayList<StreamPipesAdapter>();
-    var protocolCodes = getDrivers();
-    var driverManager = PlcDriverManager.getDefault();
-    var connectionManager = driverManager.getConnectionManager();
+    var protocolCodes = getDrivers(driverManager);
     protocolCodes
         .stream()
         .filter(pc -> !excludedDrivers.contains(pc))
@@ -60,9 +60,8 @@ public List<StreamPipesAdapter> makeAvailableAdapters() {
     return adapters;
   }
 
-  private Set<String> getDrivers() {
-    return PlcDriverManager
-        .getDefault()
+  private Set<String> getDrivers(PlcDriverManager driverManager) {
+    return driverManager
         .getProtocolCodes();
   }
 }

streampipes-extensions/streampipes-connectors-plc/src/main/java/org/apache/streampipes/extensions/connectors/plc/adapter/generic/connection/ContinuousPlcRequestReader.java

@@ -22,21 +22,20 @@
 import org.apache.streampipes.extensions.api.connect.IPollingSettings;
 import org.apache.streampipes.extensions.api.connect.IPullAdapter;
 import org.apache.streampipes.extensions.connectors.plc.adapter.generic.model.Plc4xConnectionSettings;
-import org.apache.streampipes.extensions.connectors.plc.adapter.s7.PlcReadResponseHandler;
 import org.apache.streampipes.extensions.management.connect.adapter.util.PollingSettings;
 
 import org.apache.plc4x.java.api.PlcConnection;
 import org.apache.plc4x.java.api.PlcConnectionManager;
 import org.apache.plc4x.java.api.exceptions.PlcConnectionException;
-import org.apache.plc4x.java.api.messages.PlcReadResponse;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
 
 public class ContinuousPlcRequestReader
-    extends OneTimePlcRequestReader implements IPullAdapter, PlcReadResponseHandler {
+    extends OneTimePlcRequestReader implements IPullAdapter {
 
   private static final Logger LOG = LoggerFactory.getLogger(ContinuousPlcRequestReader.class);
 
@@ -55,7 +54,7 @@ public ContinuousPlcRequestReader(PlcConnectionManager connectionManager,
   public void pullData() throws RuntimeException {
     try {
       var connection = getConnection();
-      readPlcData(connection, this);
+      readPlcData(connection);
     } catch (Exception e) {
       LOG.error("Error while reading from PLC with connection string {} ", settings.connectionString(), e);
     }
@@ -68,27 +67,19 @@ private PlcConnection getConnection() throws PlcConnectionException {
     return this.plcConnection;
   }
 
-  private void readPlcData(PlcConnection plcConnection, PlcReadResponseHandler handler) {
+  private void readPlcData(PlcConnection plcConnection)
+      throws ExecutionException, InterruptedException, TimeoutException {
     var readRequest = requestProvider.makeReadRequest(plcConnection, settings.nodes());
-    CompletableFuture<? extends PlcReadResponse> asyncResponse = readRequest.execute();
-    asyncResponse.whenComplete(handler::onReadResult);
+    var response = readRequest.execute().get(5000, TimeUnit.MILLISECONDS);
+    var event = eventGenerator.makeEvent(response);
+    collector.collect(event);
   }
 
   @Override
   public IPollingSettings getPollingInterval() {
     return PollingSettings.from(TimeUnit.MILLISECONDS, settings.pollingInterval());
   }
 
-  @Override
-  public void onReadResult(PlcReadResponse response, Throwable throwable) {
-    if (throwable != null) {
-      LOG.error(throwable.getMessage());
-    } else {
-      var event = eventGenerator.makeEvent(response);
-      collector.collect(event);
-    }
-  }
-
   public void closeConnection() throws Exception {
     if (this.plcConnection != null && this.plcConnection.isConnected()) {
       this.plcConnection.close();

streampipes-extensions/streampipes-connectors-plc/src/main/java/org/apache/streampipes/extensions/connectors/plc/adapter/generic/connection/OneTimePlcRequestReader.java

@@ -18,6 +18,7 @@
 
 package org.apache.streampipes.extensions.connectors.plc.adapter.generic.connection;
 
+import org.apache.streampipes.commons.exceptions.connect.AdapterException;
 import org.apache.streampipes.extensions.connectors.plc.adapter.generic.model.Plc4xConnectionSettings;
 
 import org.apache.plc4x.java.api.PlcConnection;
@@ -46,6 +47,9 @@ public OneTimePlcRequestReader(PlcConnectionManager connectionManager,
   public Map<String, Object> readPlcDataSynchronized() throws Exception {
     var connectionString = settings.connectionString();
     try (PlcConnection plcConnection = connectionManager.getConnection(connectionString)) {
+      if (!plcConnection.getMetadata().isReadSupported()) {
+        throw new AdapterException("This PLC does not support reading data");
+      }
       var readRequest = requestProvider.makeReadRequest(plcConnection, settings.nodes());
       var readResponse = readRequest.execute().get(5000, TimeUnit.MILLISECONDS);
       return eventGenerator.makeEvent(readResponse);