This is an automated email from the ASF dual-hosted git repository.
riemer pushed a commit to branch
2944-creating-new-plc4x-s7-basic-adapter-is-working-only-once
in repository https://gitbox.apache.org/repos/asf/streampipes.git
The following commit(s) were added to
refs/heads/2944-creating-new-plc4x-s7-basic-adapter-is-working-only-once by
this push:
new 6ab094b471 fix(#2944): Fix connection error in PLC S7 connector
6ab094b471 is described below
commit 6ab094b471d09d234c26356c74ad7fb794f11614
Author: Dominik Riemer <[email protected]>
AuthorDate: Tue Jun 18 23:48:19 2024 +0200
fix(#2944): Fix connection error in PLC S7 connector
---
.../connectors/plc/PlcConnectorsModuleExport.java | 9 +-
.../plc/adapter/GenericAdapterGenerator.java | 13 +-
.../connection/ContinuousPlcRequestReader.java | 27 +--
.../connection/OneTimePlcRequestReader.java | 4 +
.../connectors/plc/adapter/s7/Plc4xS7Adapter.java | 188 +++++++--------------
5 files changed, 84 insertions(+), 157 deletions(-)
diff --git
a/streampipes-extensions/streampipes-connectors-plc/src/main/java/org/apache/streampipes/extensions/connectors/plc/PlcConnectorsModuleExport.java
b/streampipes-extensions/streampipes-connectors-plc/src/main/java/org/apache/streampipes/extensions/connectors/plc/PlcConnectorsModuleExport.java
index e5ecfe14b4..c6468f3fca 100644
---
a/streampipes-extensions/streampipes-connectors-plc/src/main/java/org/apache/streampipes/extensions/connectors/plc/PlcConnectorsModuleExport.java
+++
b/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.migration.Plc4xS
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;
}
diff --git
a/streampipes-extensions/streampipes-connectors-plc/src/main/java/org/apache/streampipes/extensions/connectors/plc/adapter/GenericAdapterGenerator.java
b/streampipes-extensions/streampipes-connectors-plc/src/main/java/org/apache/streampipes/extensions/connectors/plc/adapter/GenericAdapterGenerator.java
index 56f3d83095..e8a0ad2150 100644
---
a/streampipes-extensions/streampipes-connectors-plc/src/main/java/org/apache/streampipes/extensions/connectors/plc/adapter/GenericAdapterGenerator.java
+++
b/streampipes-extensions/streampipes-connectors-plc/src/main/java/org/apache/streampipes/extensions/connectors/plc/adapter/GenericAdapterGenerator.java
@@ -21,6 +21,7 @@ package
org.apache.streampipes.extensions.connectors.plc.adapter;
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 class GenericAdapterGenerator {
return adapters;
}
- private Set<String> getDrivers() {
- return PlcDriverManager
- .getDefault()
+ private Set<String> getDrivers(PlcDriverManager driverManager) {
+ return driverManager
.getProtocolCodes();
}
}
diff --git
a/streampipes-extensions/streampipes-connectors-plc/src/main/java/org/apache/streampipes/extensions/connectors/plc/adapter/generic/connection/ContinuousPlcRequestReader.java
b/streampipes-extensions/streampipes-connectors-plc/src/main/java/org/apache/streampipes/extensions/connectors/plc/adapter/generic/connection/ContinuousPlcRequestReader.java
index bb23816a16..f10a93bc12 100644
---
a/streampipes-extensions/streampipes-connectors-plc/src/main/java/org/apache/streampipes/extensions/connectors/plc/adapter/generic/connection/ContinuousPlcRequestReader.java
+++
b/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.IEventCollector;
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 class ContinuousPlcRequestReader
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,10 +67,12 @@ public class ContinuousPlcRequestReader
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
@@ -79,16 +80,6 @@ public class ContinuousPlcRequestReader
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();
diff --git
a/streampipes-extensions/streampipes-connectors-plc/src/main/java/org/apache/streampipes/extensions/connectors/plc/adapter/generic/connection/OneTimePlcRequestReader.java
b/streampipes-extensions/streampipes-connectors-plc/src/main/java/org/apache/streampipes/extensions/connectors/plc/adapter/generic/connection/OneTimePlcRequestReader.java
index e03566d99f..cab3a53ad6 100644
---
a/streampipes-extensions/streampipes-connectors-plc/src/main/java/org/apache/streampipes/extensions/connectors/plc/adapter/generic/connection/OneTimePlcRequestReader.java
+++
b/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 class OneTimePlcRequestReader {
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);
diff --git
a/streampipes-extensions/streampipes-connectors-plc/src/main/java/org/apache/streampipes/extensions/connectors/plc/adapter/s7/Plc4xS7Adapter.java
b/streampipes-extensions/streampipes-connectors-plc/src/main/java/org/apache/streampipes/extensions/connectors/plc/adapter/s7/Plc4xS7Adapter.java
index 1e98d4e224..5349a19563 100644
---
a/streampipes-extensions/streampipes-connectors-plc/src/main/java/org/apache/streampipes/extensions/connectors/plc/adapter/s7/Plc4xS7Adapter.java
+++
b/streampipes-extensions/streampipes-connectors-plc/src/main/java/org/apache/streampipes/extensions/connectors/plc/adapter/s7/Plc4xS7Adapter.java
@@ -22,20 +22,21 @@ package
org.apache.streampipes.extensions.connectors.plc.adapter.s7;
import org.apache.streampipes.commons.exceptions.connect.AdapterException;
import org.apache.streampipes.extensions.api.connect.IAdapterConfiguration;
import org.apache.streampipes.extensions.api.connect.IEventCollector;
-import org.apache.streampipes.extensions.api.connect.IPullAdapter;
import org.apache.streampipes.extensions.api.connect.StreamPipesAdapter;
import
org.apache.streampipes.extensions.api.connect.context.IAdapterGuessSchemaContext;
import
org.apache.streampipes.extensions.api.connect.context.IAdapterRuntimeContext;
import
org.apache.streampipes.extensions.api.extractor.IAdapterParameterExtractor;
import
org.apache.streampipes.extensions.api.extractor.IStaticPropertyExtractor;
import
org.apache.streampipes.extensions.connectors.plc.adapter.generic.config.EventSchemaProvider;
-import
org.apache.streampipes.extensions.connectors.plc.adapter.generic.connection.PlcEventGenerator;
+import
org.apache.streampipes.extensions.connectors.plc.adapter.generic.connection.ContinuousPlcRequestReader;
+import
org.apache.streampipes.extensions.connectors.plc.adapter.generic.connection.OneTimePlcRequestReader;
import
org.apache.streampipes.extensions.connectors.plc.adapter.generic.connection.PlcRequestProvider;
+import
org.apache.streampipes.extensions.connectors.plc.adapter.generic.model.Plc4xConnectionSettings;
import
org.apache.streampipes.extensions.connectors.plc.adapter.s7.config.ConfigurationParser;
import
org.apache.streampipes.extensions.management.connect.PullAdapterScheduler;
-import
org.apache.streampipes.extensions.management.connect.adapter.util.PollingSettings;
import org.apache.streampipes.model.AdapterType;
import org.apache.streampipes.model.connect.guess.GuessSchema;
+import org.apache.streampipes.model.extensions.ExtensionAssetType;
import org.apache.streampipes.model.schema.EventProperty;
import org.apache.streampipes.model.staticproperty.CollectionStaticProperty;
import org.apache.streampipes.model.staticproperty.StaticProperty;
@@ -49,12 +50,9 @@ import org.apache.streampipes.sdk.helpers.CodeLanguage;
import org.apache.streampipes.sdk.helpers.Labels;
import org.apache.streampipes.sdk.helpers.Locales;
import org.apache.streampipes.sdk.helpers.Options;
-import org.apache.streampipes.sdk.utils.Assets;
-import org.apache.plc4x.java.api.PlcConnection;
+import org.apache.plc4x.java.api.PlcConnectionManager;
import org.apache.plc4x.java.api.PlcDriverManager;
-import org.apache.plc4x.java.api.exceptions.PlcConnectionException;
-import org.apache.plc4x.java.api.messages.PlcReadResponse;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -62,10 +60,8 @@ import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
-import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.TimeUnit;
-public class Plc4xS7Adapter implements StreamPipesAdapter, IPullAdapter,
PlcReadResponseHandler {
+public class Plc4xS7Adapter implements StreamPipesAdapter {
/**
* A unique id to identify the Plc4xS7Adapter
@@ -105,121 +101,21 @@ public class Plc4xS7Adapter implements
StreamPipesAdapter, IPullAdapter, PlcRead
/**
* Values of user configuration parameters
*/
- private String ip;
- private int pollingInterval;
private Map<String, String> nodes;
- private PlcDriverManager driverManager;
+ private final PlcDriverManager driverManager;
+ private final PlcConnectionManager connectionManager;
private PullAdapterScheduler pullAdapterScheduler;
-
- private IEventCollector collector;
-
- private PlcEventGenerator eventGenerator;
+ private ContinuousPlcRequestReader plcRequestReader;
private final PlcRequestProvider requestProvider;
- public Plc4xS7Adapter() {
+ public Plc4xS7Adapter(PlcDriverManager driverManager,
+ PlcConnectionManager connectionManager) {
this.requestProvider = new PlcRequestProvider();
- }
-
- /**
- * This method is executed when the adapter is started. A connection to the
PLC is initialized
- */
- private void before(IStaticPropertyExtractor extractor) {
- // Extract user input
- getConfigurations(extractor);
-
- this.driverManager = PlcDriverManager.getDefault();
- try (PlcConnection plcConnection =
this.driverManager.getConnectionManager().getConnection(S7_URL + this.ip)) {
- if (!plcConnection.getMetadata().isReadSupported()) {
- LOG.error("The S7 on IP: " + this.ip + " does not support reading
data");
- }
- } catch (PlcConnectionException e) {
- LOG.error("Could not establish connection to S7 with ip " + this.ip, e);
- } catch (Exception e) {
- LOG.error("Could not close connection to S7 with ip " + this.ip, e);
- }
- }
-
-
- /**
- * pullData is called iteratively according to the polling interval defined
in getPollInterval.
- */
- @Override
- public void pullData() {
- // Create PLC read request
- try (PlcConnection plcConnection =
this.driverManager.getConnectionManager().getConnection(S7_URL + this.ip)) {
- readPlcData(plcConnection, this);
- } catch (Exception e) {
- LOG.error("Error while reading from PLC with IP {} ", this.ip, e);
- }
- }
-
- private void readPlcData(PlcConnection plcConnection, PlcReadResponseHandler
handler) {
- var readRequest = requestProvider.makeReadRequest(plcConnection,
this.nodes);
- // Execute the request
- CompletableFuture<? extends PlcReadResponse> asyncResponse =
readRequest.execute();
- asyncResponse.whenComplete(handler::onReadResult);
- }
-
- private Map<String, Object> readPlcDataSynchronized() throws Exception {
- try (PlcConnection plcConnection =
this.driverManager.getConnectionManager().getConnection(S7_URL + this.ip)) {
- var readRequest = requestProvider.makeReadRequest(plcConnection,
this.nodes);
- // Execute the request
- var readResponse = readRequest.execute().get(5000,
TimeUnit.MILLISECONDS);
- return eventGenerator.makeEvent(readResponse);
- }
- }
-
- /**
- * Define the polling interval of this adapter. Default is to poll every
second
- *
- * @return PollingSettings
- */
- @Override
- public PollingSettings getPollingInterval() {
- return PollingSettings.from(TimeUnit.MILLISECONDS, this.pollingInterval);
- }
-
- /**
- * Extracts the user configuration from the SpecificAdapterStreamDescription
and sets the local variales
- *
- * @param extractor StaticPropertyExtractor
- */
- private void getConfigurations(IStaticPropertyExtractor extractor) {
-
- this.ip = extractor.singleValueParameter(PLC_IP, String.class);
- this.pollingInterval =
extractor.singleValueParameter(PLC_POLLING_INTERVAL, Integer.class);
-
- this.nodes = new HashMap<>();
-
- var selectedAlternative =
extractor.selectedAlternativeInternalId(PLC_NODE_INPUT_ALTERNATIVES);
-
- if (selectedAlternative.equals(PLC_NODE_INPUT_COLLECTION_ALTERNATIVE)) {
- // Alternative Simple
- var csp = (CollectionStaticProperty)
extractor.getStaticPropertyByName(PLC_NODES);
- this.nodes = getNodeInformationFromCollectionStaticProperty(csp);
-
- } else {
- // Alternative Advanced
- var codePropertyInput = extractor.codeblockValue(PLC_CODE_BLOCK);
- this.nodes = new
ConfigurationParser().getNodeInformationFromCodeProperty(codePropertyInput);
- }
- this.eventGenerator = new PlcEventGenerator(this.nodes);
- }
-
-
- @Override
- public void onReadResult(PlcReadResponse response, Throwable throwable) {
- if (throwable != null) {
- throwable.printStackTrace();
- LOG.error(throwable.getMessage());
- } else {
- var event = eventGenerator.makeEvent(response);
- // publish the final event
- collector.collect(event);
- }
+ this.driverManager = driverManager;
+ this.connectionManager = connectionManager;
}
/**
@@ -230,9 +126,9 @@ public class Plc4xS7Adapter implements StreamPipesAdapter,
IPullAdapter, PlcRead
*/
@Override
public IAdapterConfiguration declareConfig() {
- return AdapterConfigurationBuilder.create(ID, 1, Plc4xS7Adapter::new)
+ return AdapterConfigurationBuilder.create(ID, 1, () -> new
Plc4xS7Adapter(driverManager, connectionManager))
.withLocales(Locales.EN)
- .withAssets(Assets.DOCUMENTATION, Assets.ICON)
+ .withAssets(ExtensionAssetType.DOCUMENTATION, ExtensionAssetType.ICON)
.withCategory(AdapterType.Manufacturing)
.requiredTextParameter(Labels.withId(PLC_IP))
.requiredIntegerParameter(Labels.withId(PLC_POLLING_INTERVAL), 1000)
@@ -255,34 +151,38 @@ public class Plc4xS7Adapter implements
StreamPipesAdapter, IPullAdapter, PlcRead
public void onAdapterStarted(IAdapterParameterExtractor extractor,
IEventCollector collector,
IAdapterRuntimeContext adapterRuntimeContext) {
- this.before(extractor.getStaticPropertyExtractor());
- this.collector = collector;
+ var settings = getConfigurations(extractor.getStaticPropertyExtractor());
+ this.plcRequestReader = new ContinuousPlcRequestReader(connectionManager,
settings, requestProvider, collector);
this.pullAdapterScheduler = new PullAdapterScheduler();
- this.pullAdapterScheduler.schedule(this,
extractor.getAdapterDescription().getElementId());
+ this.pullAdapterScheduler.schedule(plcRequestReader,
extractor.getAdapterDescription().getElementId());
}
@Override
public void onAdapterStopped(IAdapterParameterExtractor extractor,
IAdapterRuntimeContext adapterRuntimeContext) {
+ try {
+ this.plcRequestReader.closeConnection();
+ } catch (Exception e) {
+ LOG.error("Error when closing connection", e);
+ }
this.pullAdapterScheduler.shutdown();
}
@Override
public GuessSchema onSchemaRequested(IAdapterParameterExtractor extractor,
IAdapterGuessSchemaContext
adapterGuessSchemaContext) throws AdapterException {
- // Extract user input
try {
- getConfigurations(extractor.getStaticPropertyExtractor());
+ var settings = getConfigurations(extractor.getStaticPropertyExtractor());
- if (this.pollingInterval < 10) {
- throw new AdapterException("Polling interval must be higher than 10.
Current value: " + this.pollingInterval);
+ if (settings.pollingInterval() < 10) {
+ throw new AdapterException(
+ String.format("Polling interval must be higher than 10. Current
value: %s", settings.pollingInterval())
+ );
}
GuessSchemaBuilder builder = GuessSchemaBuilder.create();
- List<EventProperty> allProperties = new
EventSchemaProvider().makeSchema(this.nodes);
-
- this.before(extractor.getStaticPropertyExtractor());
- var event = readPlcDataSynchronized();
+ List<EventProperty> allProperties = new
EventSchemaProvider().makeSchema(settings.nodes());
+ var event = new OneTimePlcRequestReader(connectionManager, settings,
requestProvider).readPlcDataSynchronized();
builder.properties(allProperties);
builder.preview(event);
@@ -293,6 +193,34 @@ public class Plc4xS7Adapter implements StreamPipesAdapter,
IPullAdapter, PlcRead
}
}
+ /**
+ * Extracts the user configuration from the SpecificAdapterStreamDescription
and sets the local variales
+ *
+ * @param extractor StaticPropertyExtractor
+ */
+ private Plc4xConnectionSettings getConfigurations(IStaticPropertyExtractor
extractor) {
+
+ var ip = extractor.singleValueParameter(PLC_IP, String.class);
+ var pollingInterval = extractor.singleValueParameter(PLC_POLLING_INTERVAL,
Integer.class);
+
+ Map<String, String> nodes = Map.of();
+
+ var selectedAlternative =
extractor.selectedAlternativeInternalId(PLC_NODE_INPUT_ALTERNATIVES);
+
+ if (selectedAlternative.equals(PLC_NODE_INPUT_COLLECTION_ALTERNATIVE)) {
+ // Alternative Simple
+ var csp = (CollectionStaticProperty)
extractor.getStaticPropertyByName(PLC_NODES);
+ nodes = getNodeInformationFromCollectionStaticProperty(csp);
+
+ } else {
+ // Alternative Advanced
+ var codePropertyInput = extractor.codeblockValue(PLC_CODE_BLOCK);
+ this.nodes = new
ConfigurationParser().getNodeInformationFromCodeProperty(codePropertyInput);
+ }
+
+ return new Plc4xConnectionSettings(S7_URL + ip, pollingInterval, nodes);
+ }
+
private Map<String, String>
getNodeInformationFromCollectionStaticProperty(CollectionStaticProperty csp) {
var result = new HashMap<String, String>();
