This is an automated email from the ASF dual-hosted git repository. dominikriemer pushed a commit to branch improve-opc-ua-subscription-mode-configuration in repository https://gitbox.apache.org/repos/asf/streampipes.git
commit 6f326438b59f3a2dff295cb799a783566bfff999 Author: Dominik Riemer <[email protected]> AuthorDate: Mon Jun 29 22:07:45 2026 +0200 feat: Extend subscription mode configuration options of OPC UA adapter --- .../opcua/OpcUaConnectorsModuleExport.java | 2 + .../connectors/opcua/adapter/OpcUaAdapter.java | 88 ++++++--- .../opcua/client/ConnectedOpcUaClient.java | 31 ++-- .../opcua/config/OpcUaAdapterConfig.java | 41 +++++ .../opcua/config/SharedUserConfiguration.java | 61 +++++- .../opcua/config/SpOpcUaConfigExtractor.java | 54 ++++++ .../opcua/migration/OpcUaAdapterMigrationV7.java | 59 ++++++ .../connectors/opcua/utils/OpcUaLabels.java | 4 + .../strings.en | 15 ++ .../adapter/OpcUaAdapterSubscriptionTest.java | 204 +++++++++++++++++++++ .../opcua/config/SpOpcUaConfigExtractorTest.java | 48 +++++ 11 files changed, 564 insertions(+), 43 deletions(-) diff --git a/streampipes-extensions/streampipes-connectors-opcua/src/main/java/org/apache/streampipes/extensions/connectors/opcua/OpcUaConnectorsModuleExport.java b/streampipes-extensions/streampipes-connectors-opcua/src/main/java/org/apache/streampipes/extensions/connectors/opcua/OpcUaConnectorsModuleExport.java index 0c7bf2088f..9ea53e469b 100644 --- a/streampipes-extensions/streampipes-connectors-opcua/src/main/java/org/apache/streampipes/extensions/connectors/opcua/OpcUaConnectorsModuleExport.java +++ b/streampipes-extensions/streampipes-connectors-opcua/src/main/java/org/apache/streampipes/extensions/connectors/opcua/OpcUaConnectorsModuleExport.java @@ -31,6 +31,7 @@ import org.apache.streampipes.extensions.connectors.opcua.migration.OpcUaAdapter import org.apache.streampipes.extensions.connectors.opcua.migration.OpcUaAdapterMigrationV4; import org.apache.streampipes.extensions.connectors.opcua.migration.OpcUaAdapterMigrationV5; import org.apache.streampipes.extensions.connectors.opcua.migration.OpcUaAdapterMigrationV6; +import org.apache.streampipes.extensions.connectors.opcua.migration.OpcUaAdapterMigrationV7; import org.apache.streampipes.extensions.connectors.opcua.migration.OpcUaSinkMigrationV1; import org.apache.streampipes.extensions.connectors.opcua.migration.OpcUaSinkMigrationV2; import org.apache.streampipes.extensions.connectors.opcua.sink.OpcUaSink; @@ -69,6 +70,7 @@ public class OpcUaConnectorsModuleExport implements IExtensionModuleExport { new OpcUaAdapterMigrationV4(), new OpcUaAdapterMigrationV5(), new OpcUaAdapterMigrationV6(), + new OpcUaAdapterMigrationV7(), new OpcUaSinkMigrationV1(), new OpcUaSinkMigrationV2() ); diff --git a/streampipes-extensions/streampipes-connectors-opcua/src/main/java/org/apache/streampipes/extensions/connectors/opcua/adapter/OpcUaAdapter.java b/streampipes-extensions/streampipes-connectors-opcua/src/main/java/org/apache/streampipes/extensions/connectors/opcua/adapter/OpcUaAdapter.java index 33571ec394..20799195a6 100644 --- a/streampipes-extensions/streampipes-connectors-opcua/src/main/java/org/apache/streampipes/extensions/connectors/opcua/adapter/OpcUaAdapter.java +++ b/streampipes-extensions/streampipes-connectors-opcua/src/main/java/org/apache/streampipes/extensions/connectors/opcua/adapter/OpcUaAdapter.java @@ -84,14 +84,14 @@ public class OpcUaAdapter implements StreamPipesAdapter, IPullAdapter, SupportsR private int numberOfEventProperties = 0; /** - * This variable is used to map the node ids during the subscription to the labels of the nodes + * This variable is used to map the node ids during the subscription to the selected nodes. */ - private final Map<String, String> nodeIdToLabelMapping; + private final Map<String, OpcUaNode> nodeIdToNodeMapping; public OpcUaAdapter(OpcUaClientProvider clientProvider) { this.clientProvider = clientProvider; this.event = new HashMap<>(); - this.nodeIdToLabelMapping = new HashMap<>(); + this.nodeIdToNodeMapping = new HashMap<>(); } private void prepareAdapter() throws AdapterException { @@ -130,14 +130,10 @@ public class OpcUaAdapter implements StreamPipesAdapter, IPullAdapter, SupportsR } this.pullingIntervalMilliSeconds = effectiveIntervalMs; - } else { - var allNodeIds = this.allNodes.stream() - .map(node -> node.nodeInfo().getNodeId()).toList(); - this.connectedClient.createListSubscription(allNodeIds, this); } - this.allNodes.forEach(node -> this.nodeIdToLabelMapping - .put(node.nodeInfo().getNodeId().toString(), node.nodeInfo().getDisplayName())); + this.allNodes.forEach(node -> this.nodeIdToNodeMapping + .put(node.nodeInfo().getNodeId().toString(), node)); } catch (Exception e) { var errorMessage = buildStartupErrorMessage(e); @@ -274,29 +270,61 @@ public class OpcUaAdapter implements StreamPipesAdapter, IPullAdapter, SupportsR public void onSubscriptionValue(OpcUaMonitoredItem item, DataValue value) { - String key = this.nodeIdToLabelMapping.get(item.getReadValueId().getNodeId().toString()); - - var currNode = this.allNodes.stream() - .filter(node -> key.equals(node.nodeInfo().getDisplayName())) - .findFirst() - .orElse(null); + var nodeId = item.getReadValueId().getNodeId(); + var currNode = this.nodeIdToNodeMapping.get(nodeId.toString()); if (currNode != null) { - currNode.addToEvent(connectedClient.getClient(), event, value.getValue()); - // ensure that event is complete and all opc ua subscriptions transmitted at least one value - if (event.size() >= numberOfEventProperties) { - Map<String, Object> newEvent = new HashMap<>(); - // deep copy of event to prevent preprocessor error - for (String k : event.keySet()) { - newEvent.put(k, event.get(k)); + if (value == null) { + LOG.warn("Received null DataValue for OPC UA subscription node label: {}, node id: {}", + currNode.nodeInfo().getDisplayName(), + nodeId); + return; + } + + var status = value.getStatusCode(); + if (status != null && status.isGood()) { + synchronized (event) { + currNode.addToEvent(connectedClient.getClient(), event, value.getValue()); + publishSubscriptionEventIfComplete(); + } + } else if (shouldSkipEvent(true)) { + LOG.warn("Received status code {} for OPC UA subscription node label: {}, node id: {}", + status, + currNode.nodeInfo().getDisplayName(), + nodeId); + } else { + synchronized (event) { + event.remove(currNode.nodeInfo().getDesiredName("")); + publishSubscriptionEvent(); } - collector.collect(newEvent); } } else { LOG.error("No event is produced, because subscription item {} could not be found within all nodes", item); } } + private void publishSubscriptionEvent() { + if (!event.isEmpty()) { + collector.collect(copySubscriptionEvent()); + } + } + + private void publishSubscriptionEventIfComplete() { + // ensure that event is complete and all opc ua subscriptions transmitted at least one value + if (event.size() >= numberOfEventProperties) { + collector.collect(copySubscriptionEvent()); + } + } + + private Map<String, Object> copySubscriptionEvent() { + Map<String, Object> newEvent = new HashMap<>(); + // deep copy of event to prevent preprocessor error + for (String k : event.keySet()) { + newEvent.put(k, event.get(k)); + } + return newEvent; + } + @Override public PollingSettings getPollingInterval() { return PollingSettings.from(TimeUnit.MILLISECONDS, this.pullingIntervalMilliSeconds); @@ -320,6 +348,15 @@ public class OpcUaAdapter implements StreamPipesAdapter, IPullAdapter, SupportsR if (this.opcUaAdapterConfig.inPullMode()) { this.pullAdapterScheduler = new PullAdapterScheduler(); this.pullAdapterScheduler.schedule(this, extractor.getAdapterDescription().getElementId()); + } else { + var allNodeIds = this.allNodes.stream() + .map(node -> node.nodeInfo().getNodeId()).toList(); + try { + this.connectedClient.createListSubscription(allNodeIds, this.opcUaAdapterConfig, this); + } catch (Exception e) { + var errorMessage = buildStartupErrorMessage(e); + throw new AdapterException(errorMessage, e); + } } } @@ -341,14 +378,15 @@ public class OpcUaAdapter implements StreamPipesAdapter, IPullAdapter, SupportsR @Override public IAdapterConfiguration declareConfig() { - var builder = AdapterConfigurationBuilder.create(ID, 6, () -> new OpcUaAdapter(clientProvider)) + var builder = AdapterConfigurationBuilder.create(ID, 7, () -> new OpcUaAdapter(clientProvider)) .withAssets(ExtensionAssetType.DOCUMENTATION, ExtensionAssetType.ICON) .withLocales(Locales.EN) .requiredAlternatives(Labels.withId(ADAPTER_TYPE), Alternatives.from(Labels.withId(PULL_MODE), SharedUserConfiguration.getPullModeGroup() ), - Alternatives.from(Labels.withId(SUBSCRIPTION_MODE))); + Alternatives.from(Labels.withId(SUBSCRIPTION_MODE), + SharedUserConfiguration.getSubscriptionModeGroup())); SharedUserConfiguration.appendSharedOpcUaConfig(builder, true); builder.requiredStaticProperty(SharedUserConfiguration.makeNamingStrategyOption()); return builder.buildConfiguration(); diff --git a/streampipes-extensions/streampipes-connectors-opcua/src/main/java/org/apache/streampipes/extensions/connectors/opcua/client/ConnectedOpcUaClient.java b/streampipes-extensions/streampipes-connectors-opcua/src/main/java/org/apache/streampipes/extensions/connectors/opcua/client/ConnectedOpcUaClient.java index ddb27e5bd3..1441695cbe 100644 --- a/streampipes-extensions/streampipes-connectors-opcua/src/main/java/org/apache/streampipes/extensions/connectors/opcua/client/ConnectedOpcUaClient.java +++ b/streampipes-extensions/streampipes-connectors-opcua/src/main/java/org/apache/streampipes/extensions/connectors/opcua/client/ConnectedOpcUaClient.java @@ -19,6 +19,7 @@ package org.apache.streampipes.extensions.connectors.opcua.client; import org.apache.streampipes.extensions.connectors.opcua.adapter.OpcUaAdapter; +import org.apache.streampipes.extensions.connectors.opcua.config.OpcUaAdapterConfig; import org.eclipse.milo.opcua.sdk.client.OpcUaClient; import org.eclipse.milo.opcua.sdk.client.subscriptions.MonitoredItemServiceOperationResult; @@ -27,7 +28,6 @@ import org.eclipse.milo.opcua.sdk.client.subscriptions.OpcUaSubscription; import org.eclipse.milo.opcua.stack.core.UaException; import org.eclipse.milo.opcua.stack.core.types.builtin.NodeId; import org.eclipse.milo.opcua.stack.core.types.builtin.StatusCode; -import org.eclipse.milo.opcua.stack.core.types.enumerated.TimestampsToReturn; import org.jspecify.annotations.NonNull; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -49,32 +49,28 @@ public class ConnectedOpcUaClient { /*** * Register subscriptions for given OPC UA nodes * @param nodes List of {@link org.eclipse.milo.opcua.stack.core.types.builtin.NodeId} + * @param config subscription configuration * @param opcUaAdapter current instance of {@link OpcUaAdapter} * @throws Exception */ public void createListSubscription(List<NodeId> nodes, + OpcUaAdapterConfig config, OpcUaAdapter opcUaAdapter) throws Exception { - initSubscription(nodes, opcUaAdapter); + initSubscription(nodes, config, opcUaAdapter); } public void initSubscription(List<NodeId> nodes, + OpcUaAdapterConfig config, OpcUaAdapter opcUaAdapter) throws Exception { - var subscription = getOpcUaSubscription(nodes, opcUaAdapter); - - for (NodeId node : nodes) { - var value = this.client.readValue(0, TimestampsToReturn.Both, node); - if (value == null || value.getValue().getValue() == null) { - LOG.error("Node has no value"); - } - } + var subscription = getOpcUaSubscription(nodes, config, opcUaAdapter); List<OpcUaMonitoredItem> items = new ArrayList<>(); for (NodeId node : nodes) { var item = OpcUaMonitoredItem.newDataItem(node); - item.setSamplingInterval(1000.0); - item.setQueueSize(uint(10)); - item.setDiscardOldest(true); + item.setSamplingInterval((double) config.getSubscriptionSamplingIntervalMs()); + item.setQueueSize(uint(config.getSubscriptionQueueSize())); + item.setDiscardOldest(config.isSubscriptionDiscardOldest()); item.setDataValueListener(opcUaAdapter::onSubscriptionValue); items.add(item); } @@ -95,14 +91,15 @@ public class ConnectedOpcUaClient { } private @NonNull OpcUaSubscription getOpcUaSubscription(List<NodeId> nodes, + OpcUaAdapterConfig config, OpcUaAdapter opcUaAdapter) throws UaException { - OpcUaSubscription subscription = createManagedSubscription(); + OpcUaSubscription subscription = createManagedSubscription(config); subscription.setSubscriptionListener(new OpcUaSubscription.SubscriptionListener() { @Override public void onTransferFailed(OpcUaSubscription subscription, StatusCode statusCode) { LOG.warn("Transfer for subscriptionId={} failed: {}", subscription.getSubscriptionId(), statusCode); try { - initSubscription(nodes, opcUaAdapter); + initSubscription(nodes, config, opcUaAdapter); } catch (Exception e) { LOG.error("Re-creating the subscription failed", e); } @@ -111,8 +108,8 @@ public class ConnectedOpcUaClient { return subscription; } - private OpcUaSubscription createManagedSubscription() throws UaException { - var subscription = new OpcUaSubscription(this.client, 1000.0); + private OpcUaSubscription createManagedSubscription(OpcUaAdapterConfig config) throws UaException { + var subscription = new OpcUaSubscription(this.client, (double) config.getSubscriptionPublishingIntervalMs()); subscription.create(); return subscription; } diff --git a/streampipes-extensions/streampipes-connectors-opcua/src/main/java/org/apache/streampipes/extensions/connectors/opcua/config/OpcUaAdapterConfig.java b/streampipes-extensions/streampipes-connectors-opcua/src/main/java/org/apache/streampipes/extensions/connectors/opcua/config/OpcUaAdapterConfig.java index a0bac48efd..14b5683f33 100644 --- a/streampipes-extensions/streampipes-connectors-opcua/src/main/java/org/apache/streampipes/extensions/connectors/opcua/config/OpcUaAdapterConfig.java +++ b/streampipes-extensions/streampipes-connectors-opcua/src/main/java/org/apache/streampipes/extensions/connectors/opcua/config/OpcUaAdapterConfig.java @@ -22,9 +22,18 @@ import org.apache.streampipes.extensions.connectors.opcua.utils.OpcUaNamingStrat public class OpcUaAdapterConfig extends OpcUaConfig { + public static final int DEFAULT_SUBSCRIPTION_PUBLISHING_INTERVAL_MS = 1000; + public static final int DEFAULT_SUBSCRIPTION_SAMPLING_INTERVAL_MS = 1000; + public static final int DEFAULT_SUBSCRIPTION_QUEUE_SIZE = 10; + public static final boolean DEFAULT_SUBSCRIPTION_DISCARD_OLDEST = true; + private Integer pullIntervalMilliSeconds; private String incompleteEventStrategy; private OpcUaNamingStrategy namingStrategy; + private int subscriptionPublishingIntervalMs = DEFAULT_SUBSCRIPTION_PUBLISHING_INTERVAL_MS; + private int subscriptionSamplingIntervalMs = DEFAULT_SUBSCRIPTION_SAMPLING_INTERVAL_MS; + private int subscriptionQueueSize = DEFAULT_SUBSCRIPTION_QUEUE_SIZE; + private boolean subscriptionDiscardOldest = DEFAULT_SUBSCRIPTION_DISCARD_OLDEST; public Integer getPullIntervalMilliSeconds() { return pullIntervalMilliSeconds; @@ -54,4 +63,36 @@ public class OpcUaAdapterConfig extends OpcUaConfig { this.namingStrategy = namingStrategy; } + public int getSubscriptionPublishingIntervalMs() { + return subscriptionPublishingIntervalMs; + } + + public void setSubscriptionPublishingIntervalMs(int subscriptionPublishingIntervalMs) { + this.subscriptionPublishingIntervalMs = subscriptionPublishingIntervalMs; + } + + public int getSubscriptionSamplingIntervalMs() { + return subscriptionSamplingIntervalMs; + } + + public void setSubscriptionSamplingIntervalMs(int subscriptionSamplingIntervalMs) { + this.subscriptionSamplingIntervalMs = subscriptionSamplingIntervalMs; + } + + public int getSubscriptionQueueSize() { + return subscriptionQueueSize; + } + + public void setSubscriptionQueueSize(int subscriptionQueueSize) { + this.subscriptionQueueSize = subscriptionQueueSize; + } + + public boolean isSubscriptionDiscardOldest() { + return subscriptionDiscardOldest; + } + + public void setSubscriptionDiscardOldest(boolean subscriptionDiscardOldest) { + this.subscriptionDiscardOldest = subscriptionDiscardOldest; + } + } diff --git a/streampipes-extensions/streampipes-connectors-opcua/src/main/java/org/apache/streampipes/extensions/connectors/opcua/config/SharedUserConfiguration.java b/streampipes-extensions/streampipes-connectors-opcua/src/main/java/org/apache/streampipes/extensions/connectors/opcua/config/SharedUserConfiguration.java index dbf25d83c6..e939b641ff 100644 --- a/streampipes-extensions/streampipes-connectors-opcua/src/main/java/org/apache/streampipes/extensions/connectors/opcua/config/SharedUserConfiguration.java +++ b/streampipes-extensions/streampipes-connectors-opcua/src/main/java/org/apache/streampipes/extensions/connectors/opcua/config/SharedUserConfiguration.java @@ -21,8 +21,10 @@ package org.apache.streampipes.extensions.connectors.opcua.config; import org.apache.streampipes.extensions.connectors.opcua.utils.OpcUaLabels; import org.apache.streampipes.extensions.connectors.opcua.utils.OpcUaNamingStrategy; import org.apache.streampipes.extensions.connectors.opcua.utils.SecurityUtils; +import org.apache.streampipes.model.staticproperty.FreeTextStaticProperty; import org.apache.streampipes.model.staticproperty.OneOfStaticProperty; import org.apache.streampipes.model.staticproperty.Option; +import org.apache.streampipes.model.staticproperty.PropertyValueSpecification; import org.apache.streampipes.model.staticproperty.StaticPropertyGroup; import org.apache.streampipes.sdk.StaticProperties; import org.apache.streampipes.sdk.builder.AbstractConfigurablePipelineElementBuilder; @@ -43,6 +45,10 @@ import static org.apache.streampipes.extensions.connectors.opcua.utils.OpcUaLabe import static org.apache.streampipes.extensions.connectors.opcua.utils.OpcUaLabels.OPC_URL; import static org.apache.streampipes.extensions.connectors.opcua.utils.OpcUaLabels.PASSWORD; import static org.apache.streampipes.extensions.connectors.opcua.utils.OpcUaLabels.PULLING_INTERVAL; +import static org.apache.streampipes.extensions.connectors.opcua.utils.OpcUaLabels.SUBSCRIPTION_DISCARD_OLDEST; +import static org.apache.streampipes.extensions.connectors.opcua.utils.OpcUaLabels.SUBSCRIPTION_PUBLISHING_INTERVAL; +import static org.apache.streampipes.extensions.connectors.opcua.utils.OpcUaLabels.SUBSCRIPTION_QUEUE_SIZE; +import static org.apache.streampipes.extensions.connectors.opcua.utils.OpcUaLabels.SUBSCRIPTION_SAMPLING_INTERVAL; import static org.apache.streampipes.extensions.connectors.opcua.utils.OpcUaLabels.USERNAME; import static org.apache.streampipes.extensions.connectors.opcua.utils.OpcUaLabels.USERNAME_GROUP; @@ -51,6 +57,9 @@ public class SharedUserConfiguration { public static final String INCOMPLETE_EVENT_HANDLING_KEY = "incomplete-event-handling"; public static final String INCOMPLETE_OPTION_IGNORE = "ignore-event"; public static final String INCOMPLETE_OPTION_SEND = "send-event"; + public static final String SUBSCRIPTION_GROUP = "subscription-mode-group"; + public static final String SUBSCRIPTION_DISCARD_OLDEST_TRUE = "true"; + public static final String SUBSCRIPTION_DISCARD_OLDEST_FALSE = "false"; public static final String SECURITY_MODE = "securityMode"; public static final String SECURITY_POLICY = "securityPolicy"; @@ -146,14 +155,64 @@ public class SharedUserConfiguration { return group; } + public static StaticPropertyGroup getSubscriptionModeGroup() { + var group = StaticProperties.group( + Labels.withId(SUBSCRIPTION_GROUP), + false, + positiveIntegerProperty( + SUBSCRIPTION_PUBLISHING_INTERVAL, + OpcUaAdapterConfig.DEFAULT_SUBSCRIPTION_PUBLISHING_INTERVAL_MS, + 100, + 3600000), + positiveIntegerProperty( + SUBSCRIPTION_SAMPLING_INTERVAL, + OpcUaAdapterConfig.DEFAULT_SUBSCRIPTION_SAMPLING_INTERVAL_MS, + 100, + 3600000), + positiveIntegerProperty( + SUBSCRIPTION_QUEUE_SIZE, + OpcUaAdapterConfig.DEFAULT_SUBSCRIPTION_QUEUE_SIZE, + 1, + 100), + getSubscriptionDiscardOldestConfig(), + getIncompleteEventConfig() + ); + group.setHorizontalRendering(false); + return group; + } + + private static FreeTextStaticProperty positiveIntegerProperty(OpcUaLabels label, + int defaultValue, + int min, + int max) { + var property = StaticProperties.integerFreeTextProperty(Labels.withId(label), defaultValue); + property.setValueSpecification(new PropertyValueSpecification(min, max, 1)); + return property; + } + + public static OneOfStaticProperty getSubscriptionDiscardOldestConfig() { + var property = StaticProperties.singleValueSelection( + Labels.withId(SUBSCRIPTION_DISCARD_OLDEST), + List.of( + new Option("Discard oldest", SUBSCRIPTION_DISCARD_OLDEST_TRUE), + new Option("Discard newest", SUBSCRIPTION_DISCARD_OLDEST_FALSE) + ) + ); + property.getOptions().get(0).setSelected(OpcUaAdapterConfig.DEFAULT_SUBSCRIPTION_DISCARD_OLDEST); + property.getOptions().get(1).setSelected(!OpcUaAdapterConfig.DEFAULT_SUBSCRIPTION_DISCARD_OLDEST); + return property; + } + public static OneOfStaticProperty getIncompleteEventConfig() { - return StaticProperties.singleValueSelection( + var property = StaticProperties.singleValueSelection( Labels.withId(INCOMPLETE_EVENT_HANDLING_KEY), List.of( new Option("Ignore (only complete messages are sent)", INCOMPLETE_OPTION_IGNORE), new Option("Send (incomplete messages are sent)", INCOMPLETE_OPTION_SEND) ) ); + property.getOptions().get(0).setSelected(true); + return property; } public static List<String> getDependsOn(boolean adapterConfig) { diff --git a/streampipes-extensions/streampipes-connectors-opcua/src/main/java/org/apache/streampipes/extensions/connectors/opcua/config/SpOpcUaConfigExtractor.java b/streampipes-extensions/streampipes-connectors-opcua/src/main/java/org/apache/streampipes/extensions/connectors/opcua/config/SpOpcUaConfigExtractor.java index fae0dfcde8..6c7c8b3491 100644 --- a/streampipes-extensions/streampipes-connectors-opcua/src/main/java/org/apache/streampipes/extensions/connectors/opcua/config/SpOpcUaConfigExtractor.java +++ b/streampipes-extensions/streampipes-connectors-opcua/src/main/java/org/apache/streampipes/extensions/connectors/opcua/config/SpOpcUaConfigExtractor.java @@ -47,6 +47,10 @@ import static org.apache.streampipes.extensions.connectors.opcua.utils.OpcUaLabe import static org.apache.streampipes.extensions.connectors.opcua.utils.OpcUaLabels.PASSWORD; import static org.apache.streampipes.extensions.connectors.opcua.utils.OpcUaLabels.PULLING_INTERVAL; import static org.apache.streampipes.extensions.connectors.opcua.utils.OpcUaLabels.PULL_MODE; +import static org.apache.streampipes.extensions.connectors.opcua.utils.OpcUaLabels.SUBSCRIPTION_DISCARD_OLDEST; +import static org.apache.streampipes.extensions.connectors.opcua.utils.OpcUaLabels.SUBSCRIPTION_PUBLISHING_INTERVAL; +import static org.apache.streampipes.extensions.connectors.opcua.utils.OpcUaLabels.SUBSCRIPTION_QUEUE_SIZE; +import static org.apache.streampipes.extensions.connectors.opcua.utils.OpcUaLabels.SUBSCRIPTION_SAMPLING_INTERVAL; import static org.apache.streampipes.extensions.connectors.opcua.utils.OpcUaLabels.USERNAME; import static org.apache.streampipes.extensions.connectors.opcua.utils.OpcUaLabels.USERNAME_GROUP; @@ -83,6 +87,34 @@ public class SpOpcUaConfigExtractor { config.setPullIntervalMilliSeconds(pullIntervalSeconds); config.setIncompleteEventStrategy(incompleteEventStrategy); + } else { + config.setSubscriptionPublishingIntervalMs(extractOptionalInteger( + extractor, + SUBSCRIPTION_PUBLISHING_INTERVAL.name(), + OpcUaAdapterConfig.DEFAULT_SUBSCRIPTION_PUBLISHING_INTERVAL_MS + )); + config.setSubscriptionSamplingIntervalMs(extractOptionalInteger( + extractor, + SUBSCRIPTION_SAMPLING_INTERVAL.name(), + OpcUaAdapterConfig.DEFAULT_SUBSCRIPTION_SAMPLING_INTERVAL_MS + )); + config.setSubscriptionQueueSize(extractOptionalInteger( + extractor, + SUBSCRIPTION_QUEUE_SIZE.name(), + OpcUaAdapterConfig.DEFAULT_SUBSCRIPTION_QUEUE_SIZE + )); + var discardOldest = extractOptionalSelectedInternalName( + extractor, + SUBSCRIPTION_DISCARD_OLDEST.name(), + String.valueOf(OpcUaAdapterConfig.DEFAULT_SUBSCRIPTION_DISCARD_OLDEST) + ); + config.setSubscriptionDiscardOldest(Boolean.parseBoolean(discardOldest)); + var incompleteEventStrategy = extractOptionalSelectedInternalName( + extractor, + SharedUserConfiguration.INCOMPLETE_EVENT_HANDLING_KEY, + SharedUserConfiguration.INCOMPLETE_OPTION_IGNORE + ); + config.setIncompleteEventStrategy(incompleteEventStrategy); } var namingStrategySelection = extractor.selectedSingleValueInternalName( @@ -94,6 +126,28 @@ public class SpOpcUaConfigExtractor { return config; } + private static Integer extractOptionalInteger(IStaticPropertyExtractor extractor, + String internalName, + Integer defaultValue) { + try { + var value = extractor.singleValueParameter(internalName, Integer.class); + return value != null ? value : defaultValue; + } catch (RuntimeException e) { + return defaultValue; + } + } + + private static String extractOptionalSelectedInternalName(IStaticPropertyExtractor extractor, + String internalName, + String defaultValue) { + try { + var value = extractor.selectedSingleValueInternalName(internalName, String.class); + return value != null ? value : defaultValue; + } catch (RuntimeException e) { + return defaultValue; + } + } + public static OpcUaConfig extractSinkConfig(IParameterExtractor extractor, IStreamPipesClient streamPipesClient) { return extractSharedConfig(extractor, new OpcUaConfig(), streamPipesClient); diff --git a/streampipes-extensions/streampipes-connectors-opcua/src/main/java/org/apache/streampipes/extensions/connectors/opcua/migration/OpcUaAdapterMigrationV7.java b/streampipes-extensions/streampipes-connectors-opcua/src/main/java/org/apache/streampipes/extensions/connectors/opcua/migration/OpcUaAdapterMigrationV7.java new file mode 100644 index 0000000000..91f5acbe0f --- /dev/null +++ b/streampipes-extensions/streampipes-connectors-opcua/src/main/java/org/apache/streampipes/extensions/connectors/opcua/migration/OpcUaAdapterMigrationV7.java @@ -0,0 +1,59 @@ +/* + * 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.migration; + +import org.apache.streampipes.extensions.api.extractor.IStaticPropertyExtractor; +import org.apache.streampipes.extensions.api.migration.IAdapterMigrator; +import org.apache.streampipes.extensions.connectors.opcua.adapter.OpcUaAdapter; +import org.apache.streampipes.extensions.connectors.opcua.config.SharedUserConfiguration; +import org.apache.streampipes.extensions.connectors.opcua.utils.OpcUaLabels; +import org.apache.streampipes.model.connect.adapter.AdapterDescription; +import org.apache.streampipes.model.extensions.svcdiscovery.SpServiceTagPrefix; +import org.apache.streampipes.model.migration.MigrationResult; +import org.apache.streampipes.model.migration.ModelMigratorConfig; +import org.apache.streampipes.model.staticproperty.StaticPropertyAlternatives; + +public class OpcUaAdapterMigrationV7 implements IAdapterMigrator { + + @Override + public ModelMigratorConfig config() { + return new ModelMigratorConfig( + OpcUaAdapter.ID, + SpServiceTagPrefix.ADAPTER, + 6, + 7 + ); + } + + @Override + public MigrationResult<AdapterDescription> migrate(AdapterDescription element, + IStaticPropertyExtractor extractor) throws RuntimeException { + element.getConfig().forEach(sp -> { + if (sp.getInternalName().equalsIgnoreCase(OpcUaLabels.ADAPTER_TYPE.name())) { + var alternatives = (StaticPropertyAlternatives) sp; + alternatives.getAlternatives().stream() + .filter(alternative -> alternative.getInternalName().equals(OpcUaLabels.SUBSCRIPTION_MODE.name())) + .findFirst() + .ifPresent(alternative -> + alternative.setStaticProperty(SharedUserConfiguration.getSubscriptionModeGroup())); + } + }); + return MigrationResult.success(element); + } +} diff --git a/streampipes-extensions/streampipes-connectors-opcua/src/main/java/org/apache/streampipes/extensions/connectors/opcua/utils/OpcUaLabels.java b/streampipes-extensions/streampipes-connectors-opcua/src/main/java/org/apache/streampipes/extensions/connectors/opcua/utils/OpcUaLabels.java index 98ce898dff..725ea32d59 100644 --- a/streampipes-extensions/streampipes-connectors-opcua/src/main/java/org/apache/streampipes/extensions/connectors/opcua/utils/OpcUaLabels.java +++ b/streampipes-extensions/streampipes-connectors-opcua/src/main/java/org/apache/streampipes/extensions/connectors/opcua/utils/OpcUaLabels.java @@ -41,6 +41,10 @@ public enum OpcUaLabels { ADAPTER_TYPE, PULL_MODE, SUBSCRIPTION_MODE, + SUBSCRIPTION_PUBLISHING_INTERVAL, + SUBSCRIPTION_SAMPLING_INTERVAL, + SUBSCRIPTION_QUEUE_SIZE, + SUBSCRIPTION_DISCARD_OLDEST, NAMING_STRATEGY, MAPPING_PROPERY; } diff --git a/streampipes-extensions/streampipes-connectors-opcua/src/main/resources/org.apache.streampipes.connect.iiot.adapters.opcua/strings.en b/streampipes-extensions/streampipes-connectors-opcua/src/main/resources/org.apache.streampipes.connect.iiot.adapters.opcua/strings.en index 853b9f11ea..026dc4e889 100644 --- a/streampipes-extensions/streampipes-connectors-opcua/src/main/resources/org.apache.streampipes.connect.iiot.adapters.opcua/strings.en +++ b/streampipes-extensions/streampipes-connectors-opcua/src/main/resources/org.apache.streampipes.connect.iiot.adapters.opcua/strings.en @@ -80,6 +80,21 @@ PULL_MODE.description= SUBSCRIPTION_MODE.title=Subscription mode SUBSCRIPTION_MODE.description= +subscription-mode-group.title=Subscription Settings +subscription-mode-group.description= + +SUBSCRIPTION_PUBLISHING_INTERVAL.title=Publishing Interval +SUBSCRIPTION_PUBLISHING_INTERVAL.description=Requested OPC UA subscription publishing interval, in milliseconds + +SUBSCRIPTION_SAMPLING_INTERVAL.title=Sampling Interval +SUBSCRIPTION_SAMPLING_INTERVAL.description=Requested monitored item sampling interval, in milliseconds + +SUBSCRIPTION_QUEUE_SIZE.title=Queue Size +SUBSCRIPTION_QUEUE_SIZE.description=Requested queue size for each monitored item + +SUBSCRIPTION_DISCARD_OLDEST.title=Queue Overflow Handling +SUBSCRIPTION_DISCARD_OLDEST.description=Select which queued value is discarded when the monitored item queue is full + incomplete-event-handling.title=Incomplete Events incomplete-event-handling.description=Select how events with missing values (e.g., due to bad status codes) are handled. diff --git a/streampipes-extensions/streampipes-connectors-opcua/src/test/java/org/apache/streampipes/extensions/connectors/opcua/adapter/OpcUaAdapterSubscriptionTest.java b/streampipes-extensions/streampipes-connectors-opcua/src/test/java/org/apache/streampipes/extensions/connectors/opcua/adapter/OpcUaAdapterSubscriptionTest.java new file mode 100644 index 0000000000..449c1f4d5e --- /dev/null +++ b/streampipes-extensions/streampipes-connectors-opcua/src/test/java/org/apache/streampipes/extensions/connectors/opcua/adapter/OpcUaAdapterSubscriptionTest.java @@ -0,0 +1,204 @@ +/* + * 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.adapter; + +import org.apache.streampipes.extensions.api.connect.IEventCollector; +import org.apache.streampipes.extensions.connectors.opcua.client.ConnectedOpcUaClient; +import org.apache.streampipes.extensions.connectors.opcua.client.OpcUaClientProvider; +import org.apache.streampipes.extensions.connectors.opcua.config.OpcUaAdapterConfig; +import org.apache.streampipes.extensions.connectors.opcua.config.SharedUserConfiguration; +import org.apache.streampipes.extensions.connectors.opcua.model.node.BasicVariableNodeInfo; +import org.apache.streampipes.extensions.connectors.opcua.model.node.OpcUaNode; +import org.apache.streampipes.model.connect.guess.FieldStatusInfo; + +import org.eclipse.milo.opcua.sdk.client.OpcUaClient; +import org.eclipse.milo.opcua.sdk.client.subscriptions.OpcUaMonitoredItem; +import org.eclipse.milo.opcua.stack.core.StatusCodes; +import org.eclipse.milo.opcua.stack.core.types.builtin.DataValue; +import org.eclipse.milo.opcua.stack.core.types.builtin.NodeId; +import org.eclipse.milo.opcua.stack.core.types.builtin.StatusCode; +import org.eclipse.milo.opcua.stack.core.types.builtin.Variant; +import org.junit.jupiter.api.Test; +import org.mockito.ArgumentCaptor; + +import java.lang.reflect.Field; +import java.util.Map; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +class OpcUaAdapterSubscriptionTest { + + @Test + @SuppressWarnings("unchecked") + void shouldRouteSubscriptionValuesByNodeIdForDuplicateDisplayNames() throws Exception { + var firstNodeId = new NodeId(2, "LineA.Temperature"); + var secondNodeId = new NodeId(2, "LineB.Temperature"); + var firstNode = new TestNode(firstNodeId, "Temperature", "lineA"); + var secondNode = new TestNode(secondNodeId, "Temperature", "lineB"); + var collector = mock(IEventCollector.class); + var adapter = makeAdapter(collector, firstNode, secondNode); + + adapter.onSubscriptionValue( + OpcUaMonitoredItem.newDataItem(firstNodeId), + DataValue.valueOnly(new Variant(101)) + ); + adapter.onSubscriptionValue( + OpcUaMonitoredItem.newDataItem(secondNodeId), + DataValue.valueOnly(new Variant(202)) + ); + + ArgumentCaptor<Map<String, Object>> eventCaptor = ArgumentCaptor.forClass(Map.class); + verify(collector).collect(eventCaptor.capture()); + assertEquals(101, eventCaptor.getValue().get("lineA")); + assertEquals(202, eventCaptor.getValue().get("lineB")); + } + + @Test + void shouldNotPublishBadStatusSubscriptionValuesWhenIncompleteEventsAreIgnored() throws Exception { + var nodeId = new NodeId(2, "IntermittentValue"); + var node = new TestNode(nodeId, "IntermittentValue", "intermittent"); + var collector = mock(IEventCollector.class); + var adapter = makeAdapter(collector, node); + + adapter.onSubscriptionValue( + OpcUaMonitoredItem.newDataItem(nodeId), + new DataValue(new Variant(1), new StatusCode(StatusCodes.Bad_NotReadable)) + ); + + verify(collector, never()).collect(org.mockito.ArgumentMatchers.any()); + } + + @Test + @SuppressWarnings("unchecked") + void shouldRemoveStaleSubscriptionValueBeforeSendingIncompleteEvent() throws Exception { + var firstNodeId = new NodeId(2, "StableCounter"); + var secondNodeId = new NodeId(2, "IntermittentValue"); + var firstNode = new TestNode(firstNodeId, "StableCounter", "stable"); + var secondNode = new TestNode(secondNodeId, "IntermittentValue", "intermittent"); + var collector = mock(IEventCollector.class); + var adapter = makeAdapter(collector, SharedUserConfiguration.INCOMPLETE_OPTION_SEND, firstNode, secondNode); + + adapter.onSubscriptionValue( + OpcUaMonitoredItem.newDataItem(firstNodeId), + DataValue.valueOnly(new Variant(1)) + ); + adapter.onSubscriptionValue( + OpcUaMonitoredItem.newDataItem(secondNodeId), + DataValue.valueOnly(new Variant(2)) + ); + adapter.onSubscriptionValue( + OpcUaMonitoredItem.newDataItem(secondNodeId), + new DataValue(new Variant(3), new StatusCode(StatusCodes.Bad_NotReadable)) + ); + + ArgumentCaptor<Map<String, Object>> eventCaptor = ArgumentCaptor.forClass(Map.class); + verify(collector, org.mockito.Mockito.times(2)).collect(eventCaptor.capture()); + + var incompleteEvent = eventCaptor.getAllValues().get(1); + assertEquals(1, incompleteEvent.get("stable")); + assertFalse(incompleteEvent.containsKey("intermittent")); + } + + private OpcUaAdapter makeAdapter(IEventCollector collector, + TestNode... nodes) throws Exception { + return makeAdapter(collector, SharedUserConfiguration.INCOMPLETE_OPTION_IGNORE, nodes); + } + + private OpcUaAdapter makeAdapter(IEventCollector collector, + String incompleteEventStrategy, + TestNode... nodes) throws Exception { + var adapter = new OpcUaAdapter(mock(OpcUaClientProvider.class)); + var nodeIdToNodeMapping = getNodeIdToNodeMapping(adapter); + for (var node : nodes) { + nodeIdToNodeMapping.put(node.nodeInfo().getNodeId().toString(), node); + } + + var config = new OpcUaAdapterConfig(); + config.setIncompleteEventStrategy(incompleteEventStrategy); + + setField(adapter, "collector", collector); + setField(adapter, "connectedClient", new ConnectedOpcUaClient(mock(OpcUaClient.class))); + setField(adapter, "numberOfEventProperties", nodes.length); + setField(adapter, "opcUaAdapterConfig", config); + return adapter; + } + + @SuppressWarnings("unchecked") + private Map<String, OpcUaNode> getNodeIdToNodeMapping(OpcUaAdapter adapter) throws Exception { + Field field = adapter.getClass().getDeclaredField("nodeIdToNodeMapping"); + field.setAccessible(true); + return (Map<String, OpcUaNode>) field.get(adapter); + } + + private void setField(Object target, + String fieldName, + Object value) throws Exception { + Field field = target.getClass().getDeclaredField(fieldName); + field.setAccessible(true); + field.set(target, value); + } + + private static class TestNode implements OpcUaNode { + + private final BasicVariableNodeInfo nodeInfo; + private final String eventPropertyName; + + TestNode(NodeId nodeId, + String displayName, + String eventPropertyName) { + this.nodeInfo = mock(BasicVariableNodeInfo.class); + this.eventPropertyName = eventPropertyName; + + when(this.nodeInfo.getNodeId()).thenReturn(nodeId); + when(this.nodeInfo.getDisplayName()).thenReturn(displayName); + when(this.nodeInfo.getDesiredName("")).thenReturn(eventPropertyName); + } + + @Override + public BasicVariableNodeInfo nodeInfo() { + return nodeInfo; + } + + @Override + public int getNumberOfEventProperties(OpcUaClient client) { + return 1; + } + + @Override + public void addToEvent(OpcUaClient client, + Map<String, Object> event, + Variant variant) { + event.put(eventPropertyName, variant.getValue()); + } + + @Override + public void addToEventPreview(OpcUaClient client, + Map<String, Object> eventPreview, + Map<String, FieldStatusInfo> fieldStatusInfos, + Variant variant, + FieldStatusInfo fieldStatusInfo) { + eventPreview.put(eventPropertyName, variant.getValue()); + } + } +} diff --git a/streampipes-extensions/streampipes-connectors-opcua/src/test/java/org/apache/streampipes/extensions/connectors/opcua/config/SpOpcUaConfigExtractorTest.java b/streampipes-extensions/streampipes-connectors-opcua/src/test/java/org/apache/streampipes/extensions/connectors/opcua/config/SpOpcUaConfigExtractorTest.java index 33b80be2cf..ee49eabbc5 100644 --- a/streampipes-extensions/streampipes-connectors-opcua/src/test/java/org/apache/streampipes/extensions/connectors/opcua/config/SpOpcUaConfigExtractorTest.java +++ b/streampipes-extensions/streampipes-connectors-opcua/src/test/java/org/apache/streampipes/extensions/connectors/opcua/config/SpOpcUaConfigExtractorTest.java @@ -36,7 +36,13 @@ import static org.apache.streampipes.extensions.connectors.opcua.utils.OpcUaLabe import static org.apache.streampipes.extensions.connectors.opcua.utils.OpcUaLabels.OPC_URL; import static org.apache.streampipes.extensions.connectors.opcua.utils.OpcUaLabels.PULLING_INTERVAL; import static org.apache.streampipes.extensions.connectors.opcua.utils.OpcUaLabels.PULL_MODE; +import static org.apache.streampipes.extensions.connectors.opcua.utils.OpcUaLabels.SUBSCRIPTION_DISCARD_OLDEST; +import static org.apache.streampipes.extensions.connectors.opcua.utils.OpcUaLabels.SUBSCRIPTION_MODE; +import static org.apache.streampipes.extensions.connectors.opcua.utils.OpcUaLabels.SUBSCRIPTION_PUBLISHING_INTERVAL; +import static org.apache.streampipes.extensions.connectors.opcua.utils.OpcUaLabels.SUBSCRIPTION_QUEUE_SIZE; +import static org.apache.streampipes.extensions.connectors.opcua.utils.OpcUaLabels.SUBSCRIPTION_SAMPLING_INTERVAL; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.when; @@ -73,4 +79,46 @@ class SpOpcUaConfigExtractorTest { assertEquals(List.of("ns=2;s=Demo.DataTypeTest.ExtensionObject"), config.getSelectedNodeNames()); } + + @Test + void shouldExtractSubscriptionSettings() { + IStaticPropertyExtractor extractor = mock(IStaticPropertyExtractor.class); + + when(extractor.selectedAlternativeInternalId(ADAPTER_TYPE.name())) + .thenReturn(SUBSCRIPTION_MODE.name()); + when(extractor.selectedAlternativeInternalId(OPC_HOST_OR_URL.name())) + .thenReturn(OPC_URL.name()); + when(extractor.selectedAlternativeInternalId(SharedUserConfiguration.USER_AUTHENTICATION)) + .thenReturn(SharedUserConfiguration.USER_AUTHENTICATION_ANONYMOUS); + when(extractor.selectedTreeNodesInternalNames(AVAILABLE_NODES.name(), String.class)) + .thenReturn(List.of("ns=2;s=Demo.Subscription.HighFrequencyCounter")); + when(extractor.selectedSingleValueInternalName(SharedUserConfiguration.SECURITY_MODE, String.class)) + .thenReturn(MessageSecurityMode.None.name()); + when(extractor.selectedSingleValue(SharedUserConfiguration.SECURITY_POLICY, String.class)) + .thenReturn(SecurityPolicy.None.name()); + when(extractor.singleValueParameter(OPC_SERVER_URL.name(), String.class)) + .thenReturn("opc.tcp://localhost:4840/milo"); + when(extractor.singleValueParameter(SUBSCRIPTION_PUBLISHING_INTERVAL.name(), Integer.class)) + .thenReturn(250); + when(extractor.singleValueParameter(SUBSCRIPTION_SAMPLING_INTERVAL.name(), Integer.class)) + .thenReturn(100); + when(extractor.singleValueParameter(SUBSCRIPTION_QUEUE_SIZE.name(), Integer.class)) + .thenReturn(25); + when(extractor.selectedSingleValueInternalName(SUBSCRIPTION_DISCARD_OLDEST.name(), String.class)) + .thenReturn(SharedUserConfiguration.SUBSCRIPTION_DISCARD_OLDEST_FALSE); + when(extractor.selectedSingleValueInternalName( + SharedUserConfiguration.INCOMPLETE_EVENT_HANDLING_KEY, + String.class + )).thenReturn(SharedUserConfiguration.INCOMPLETE_OPTION_SEND); + when(extractor.selectedSingleValueInternalName(OpcUaLabels.NAMING_STRATEGY.name(), String.class)) + .thenReturn(OpcUaNamingStrategy.DISPLAY_NAME.name()); + + var config = SpOpcUaConfigExtractor.extractAdapterConfig(extractor, mock(IStreamPipesClient.class)); + + assertEquals(250, config.getSubscriptionPublishingIntervalMs()); + assertEquals(100, config.getSubscriptionSamplingIntervalMs()); + assertEquals(25, config.getSubscriptionQueueSize()); + assertFalse(config.isSubscriptionDiscardOldest()); + assertEquals(SharedUserConfiguration.INCOMPLETE_OPTION_SEND, config.getIncompleteEventStrategy()); + } }
