This is an automated email from the ASF dual-hosted git repository.
riemer pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/streampipes.git
The following commit(s) were added to refs/heads/dev by this push:
new 27839afdf Harmonize OPC-UA adapter and sink, add timestamp to metadata
(#899) (#1682)
27839afdf is described below
commit 27839afdf1fdaca78086a9482d5a78c5a9dee8c8
Author: Dominik Riemer <[email protected]>
AuthorDate: Mon Jun 26 20:53:53 2023 +0200
Harmonize OPC-UA adapter and sink, add timestamp to metadata (#899) (#1682)
* Add source and server time to metadata (#899)
* Refactor and harmonize OPC adapter and sink
---
.../connectors/opcua/adapter/OpcUaAdapter.java | 82 +++---------
.../connectors/opcua/adapter/OpcUaNodeBrowser.java | 21 ++-
.../opcua/adapter/configuration/SpOpcUaConfig.java | 148 ---------------------
.../opcua/{adapter => client}/SpOpcUaClient.java | 30 ++---
.../MiloOpcUaConfigurationProvider.java | 7 +-
.../OpcUaAdapterConfig.java} | 36 ++---
.../connectors/opcua/config/OpcUaConfig.java | 70 ++++++++++
.../opcua/config/SharedUserConfiguration.java | 89 +++++++++++++
.../SpOpcUaConfigExtractor.java} | 101 ++++++--------
.../opcua/{adapter => model}/OpcNode.java | 2 +-
.../extensions/connectors/opcua/sink/OpcUa.java | 83 +++---------
.../connectors/opcua/sink/OpcUaParameters.java | 45 +++----
.../connectors/opcua/sink/OpcUaSink.java | 62 +++++----
.../utils/ExceptionMessageExtractor.java | 2 +-
.../opcua/{adapter => }/utils/OpcUaTypes.java | 2 +-
.../opcua/{adapter => }/utils/OpcUaUtil.java | 18 ++-
.../strings.en | 42 ++++--
.../RuntimeResolvableTreeInputStaticProperty.java | 10 ++
.../apache/streampipes/sdk/StaticProperties.java | 5 +-
...AbstractConfigurablePipelineElementBuilder.java | 12 +-
.../src/lib/model/gen/streampipes-model.ts | 4 +-
21 files changed, 405 insertions(+), 466 deletions(-)
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 e4d471869..bfa035c0c 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
@@ -29,8 +29,12 @@ import
org.apache.streampipes.extensions.api.connect.context.IAdapterRuntimeCont
import
org.apache.streampipes.extensions.api.extractor.IAdapterParameterExtractor;
import
org.apache.streampipes.extensions.api.extractor.IStaticPropertyExtractor;
import org.apache.streampipes.extensions.api.runtime.SupportsRuntimeConfig;
-import
org.apache.streampipes.extensions.connectors.opcua.adapter.configuration.SpOpcUaConfigBuilder;
-import
org.apache.streampipes.extensions.connectors.opcua.adapter.utils.OpcUaUtil;
+import org.apache.streampipes.extensions.connectors.opcua.client.SpOpcUaClient;
+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.config.SpOpcUaConfigExtractor;
+import org.apache.streampipes.extensions.connectors.opcua.model.OpcNode;
+import org.apache.streampipes.extensions.connectors.opcua.utils.OpcUaUtil;
import
org.apache.streampipes.extensions.management.connect.PullAdapterScheduler;
import
org.apache.streampipes.extensions.management.connect.adapter.util.PollingSettings;
import org.apache.streampipes.model.AdapterType;
@@ -53,7 +57,6 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.ArrayList;
-import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
@@ -62,26 +65,11 @@ import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.stream.Collectors;
-import static
org.apache.streampipes.extensions.connectors.opcua.adapter.utils.OpcUaUtil.getSchema;
-import static
org.apache.streampipes.extensions.connectors.opcua.utils.OpcUaLabels.ACCESS_MODE;
import static
org.apache.streampipes.extensions.connectors.opcua.utils.OpcUaLabels.ADAPTER_TYPE;
-import static
org.apache.streampipes.extensions.connectors.opcua.utils.OpcUaLabels.AVAILABLE_NODES;
-import static
org.apache.streampipes.extensions.connectors.opcua.utils.OpcUaLabels.HOST_PORT;
-import static
org.apache.streampipes.extensions.connectors.opcua.utils.OpcUaLabels.NAMESPACE_INDEX;
-import static
org.apache.streampipes.extensions.connectors.opcua.utils.OpcUaLabels.NODE_ID;
-import static
org.apache.streampipes.extensions.connectors.opcua.utils.OpcUaLabels.OPC_HOST;
-import static
org.apache.streampipes.extensions.connectors.opcua.utils.OpcUaLabels.OPC_HOST_OR_URL;
-import static
org.apache.streampipes.extensions.connectors.opcua.utils.OpcUaLabels.OPC_SERVER_HOST;
-import static
org.apache.streampipes.extensions.connectors.opcua.utils.OpcUaLabels.OPC_SERVER_PORT;
-import static
org.apache.streampipes.extensions.connectors.opcua.utils.OpcUaLabels.OPC_SERVER_URL;
-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.PULL_MODE;
import static
org.apache.streampipes.extensions.connectors.opcua.utils.OpcUaLabels.SUBSCRIPTION_MODE;
-import static
org.apache.streampipes.extensions.connectors.opcua.utils.OpcUaLabels.UNAUTHENTICATED;
-import static
org.apache.streampipes.extensions.connectors.opcua.utils.OpcUaLabels.USERNAME;
-import static
org.apache.streampipes.extensions.connectors.opcua.utils.OpcUaLabels.USERNAME_GROUP;
+import static
org.apache.streampipes.extensions.connectors.opcua.utils.OpcUaUtil.getSchema;
public class OpcUaAdapter implements StreamPipesAdapter, IPullAdapter,
SupportsRuntimeConfig {
@@ -89,7 +77,7 @@ public class OpcUaAdapter implements StreamPipesAdapter,
IPullAdapter, SupportsR
private static final Logger LOG =
LoggerFactory.getLogger(OpcUaAdapter.class);
private int pullingIntervalMilliSeconds;
- private SpOpcUaClient spOpcUaClient;
+ private SpOpcUaClient<OpcUaAdapterConfig> spOpcUaClient;
private List<OpcNode> allNodes;
private List<NodeId> allNodeIds;
private int numberProperties;
@@ -129,11 +117,11 @@ public class OpcUaAdapter implements StreamPipesAdapter,
IPullAdapter, SupportsR
for (OpcNode node : this.allNodes) {
- this.allNodeIds.add(node.nodeId);
+ this.allNodeIds.add(node.getNodeId());
}
- if (spOpcUaClient.inPullMode()) {
- this.pullingIntervalMilliSeconds =
spOpcUaClient.getPullIntervalMilliSeconds();
+ if (spOpcUaClient.getSpOpcConfig().inPullMode()) {
+ this.pullingIntervalMilliSeconds =
spOpcUaClient.getSpOpcConfig().getPullIntervalMilliSeconds();
} else {
this.numberProperties = this.allNodeIds.size();
this.spOpcUaClient.createListSubscription(this.allNodeIds, this);
@@ -213,11 +201,13 @@ public class OpcUaAdapter implements StreamPipesAdapter,
IPullAdapter, SupportsR
public void onAdapterStarted(IAdapterParameterExtractor extractor,
IEventCollector collector,
IAdapterRuntimeContext adapterRuntimeContext)
throws AdapterException {
- this.spOpcUaClient = new
SpOpcUaClient(SpOpcUaConfigBuilder.from(extractor.getStaticPropertyExtractor()));
+ this.spOpcUaClient = new SpOpcUaClient<>(
+
SpOpcUaConfigExtractor.extractAdapterConfig(extractor.getStaticPropertyExtractor())
+ );
this.collector = collector;
this.prepareAdapter(extractor);
- if (this.spOpcUaClient.inPullMode()) {
+ if (this.spOpcUaClient.getSpOpcConfig().inPullMode()) {
this.pullAdapterScheduler = new PullAdapterScheduler();
this.pullAdapterScheduler.schedule(this,
extractor.getAdapterDescription().getElementId());
}
@@ -228,7 +218,7 @@ public class OpcUaAdapter implements StreamPipesAdapter,
IPullAdapter, SupportsR
IAdapterRuntimeContext adapterRuntimeContext)
throws AdapterException {
this.spOpcUaClient.disconnect();
- if (this.spOpcUaClient.inPullMode()) {
+ if (this.spOpcUaClient.getSpOpcConfig().inPullMode()) {
this.pullAdapterScheduler.shutdown();
}
}
@@ -241,7 +231,7 @@ public class OpcUaAdapter implements StreamPipesAdapter,
IPullAdapter, SupportsR
@Override
public IAdapterConfiguration declareConfig() {
- return AdapterConfigurationBuilder.create(ID, OpcUaAdapter::new)
+ var builder = AdapterConfigurationBuilder.create(ID, OpcUaAdapter::new)
.withAssets(Assets.DOCUMENTATION, Assets.ICON)
.withLocales(Locales.EN)
.withCategory(AdapterType.Generic, AdapterType.Manufacturing)
@@ -249,41 +239,9 @@ public class OpcUaAdapter implements StreamPipesAdapter,
IPullAdapter, SupportsR
Alternatives.from(Labels.withId(PULL_MODE),
StaticProperties.integerFreeTextProperty(
Labels.withId(PULLING_INTERVAL))),
- Alternatives.from(Labels.withId(SUBSCRIPTION_MODE)))
- .requiredAlternatives(Labels.withId(ACCESS_MODE),
- Alternatives.from(Labels.withId(UNAUTHENTICATED)),
- Alternatives.from(Labels.withId(USERNAME_GROUP),
- StaticProperties.group(
- Labels.withId(USERNAME_GROUP),
- StaticProperties.stringFreeTextProperty(
- Labels.withId(USERNAME)),
- StaticProperties.secretValue(Labels.withId(PASSWORD))
- ))
- )
- .requiredAlternatives(Labels.withId(OPC_HOST_OR_URL),
- Alternatives.from(
- Labels.withId(OPC_URL),
- StaticProperties.stringFreeTextProperty(
- Labels.withId(OPC_SERVER_URL), "opc.tcp://localhost:4840"))
- ,
- Alternatives.from(Labels.withId(OPC_HOST),
- StaticProperties.group(
- Labels.withId(HOST_PORT),
- StaticProperties.stringFreeTextProperty(
- Labels.withId(OPC_SERVER_HOST)),
- StaticProperties.stringFreeTextProperty(
- Labels.withId(OPC_SERVER_PORT))
- ))
- )
- .requiredRuntimeResolvableTreeInput(
- Labels.withId(AVAILABLE_NODES.name()),
- List.of(
- ADAPTER_TYPE.name(),
- ACCESS_MODE.name(),
- OPC_HOST_OR_URL.name()),
- true
- )
- .buildConfiguration();
+ Alternatives.from(Labels.withId(SUBSCRIPTION_MODE)));
+ SharedUserConfiguration.appendSharedOpcUaConfig(builder, true);
+ return builder.buildConfiguration();
}
diff --git
a/streampipes-extensions/streampipes-connectors-opcua/src/main/java/org/apache/streampipes/extensions/connectors/opcua/adapter/OpcUaNodeBrowser.java
b/streampipes-extensions/streampipes-connectors-opcua/src/main/java/org/apache/streampipes/extensions/connectors/opcua/adapter/OpcUaNodeBrowser.java
index 10456a8bb..b9ef833c5 100644
---
a/streampipes-extensions/streampipes-connectors-opcua/src/main/java/org/apache/streampipes/extensions/connectors/opcua/adapter/OpcUaNodeBrowser.java
+++
b/streampipes-extensions/streampipes-connectors-opcua/src/main/java/org/apache/streampipes/extensions/connectors/opcua/adapter/OpcUaNodeBrowser.java
@@ -18,8 +18,9 @@
package org.apache.streampipes.extensions.connectors.opcua.adapter;
-import
org.apache.streampipes.extensions.connectors.opcua.adapter.configuration.SpOpcUaConfig;
-import
org.apache.streampipes.extensions.connectors.opcua.adapter.utils.OpcUaTypes;
+import org.apache.streampipes.extensions.connectors.opcua.config.OpcUaConfig;
+import org.apache.streampipes.extensions.connectors.opcua.model.OpcNode;
+import org.apache.streampipes.extensions.connectors.opcua.utils.OpcUaTypes;
import org.apache.streampipes.model.staticproperty.TreeInputNode;
import org.eclipse.milo.opcua.sdk.client.AddressSpace;
@@ -27,8 +28,8 @@ import org.eclipse.milo.opcua.sdk.client.OpcUaClient;
import org.eclipse.milo.opcua.sdk.client.nodes.UaNode;
import org.eclipse.milo.opcua.sdk.client.nodes.UaVariableNode;
import org.eclipse.milo.opcua.stack.core.Identifiers;
+import org.eclipse.milo.opcua.stack.core.StatusCodes;
import org.eclipse.milo.opcua.stack.core.UaException;
-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.unsigned.UInteger;
@@ -48,12 +49,12 @@ import java.util.stream.Collectors;
public class OpcUaNodeBrowser {
private final OpcUaClient client;
- private final SpOpcUaConfig spOpcConfig;
+ private final OpcUaConfig spOpcConfig;
private static final Logger LOG =
LoggerFactory.getLogger(OpcUaNodeBrowser.class);
public OpcUaNodeBrowser(OpcUaClient client,
- SpOpcUaConfig spOpcUaClientConfig) {
+ OpcUaConfig spOpcUaClientConfig) {
this.client = client;
this.spOpcConfig = spOpcUaClientConfig;
}
@@ -134,8 +135,16 @@ public class OpcUaNodeBrowser {
metadata.put("Value", String.valueOf(((UaVariableNode)
node).getValue().getValue().getValue()));
try {
var dataTypeNode = client.getAddressSpace().getNode(dataTypeNodeId);
+ var value = client.readValue(0, TimestampsToReturn.Both,
node.getNodeId()).get();
+ metadata.put("SourceTime",
value.getSourceTime().getJavaDate().toString());
+ metadata.put("ServerTime",
value.getServerTime().getJavaDate().toString());
+ var statusCode = value.getStatusCode().getValue();
+ var statusCodeName = StatusCodes.lookup(statusCode).get();
+ if (statusCodeName.length > 0) {
+ metadata.put("Status", statusCodeName[0]);
+ }
metadata.put("DataType", dataTypeNode.getDisplayName().getText());
- } catch (UaException e) {
+ } catch (UaException | ExecutionException | InterruptedException e) {
throw new RuntimeException(e);
}
}
diff --git
a/streampipes-extensions/streampipes-connectors-opcua/src/main/java/org/apache/streampipes/extensions/connectors/opcua/adapter/configuration/SpOpcUaConfig.java
b/streampipes-extensions/streampipes-connectors-opcua/src/main/java/org/apache/streampipes/extensions/connectors/opcua/adapter/configuration/SpOpcUaConfig.java
deleted file mode 100644
index 2cfecea09..000000000
---
a/streampipes-extensions/streampipes-connectors-opcua/src/main/java/org/apache/streampipes/extensions/connectors/opcua/adapter/configuration/SpOpcUaConfig.java
+++ /dev/null
@@ -1,148 +0,0 @@
-/*
- * 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.configuration;
-
-import java.util.List;
-
-public class SpOpcUaConfig {
-
- private final String opcServerURL;
-
- private boolean unauthenticated;
-
- private String username;
- private String password;
-
- private final List<String> selectedNodeNames;
- private final Integer pullIntervalMilliSeconds;
-
- /**
- * Constructor for security level {@code None}, OPC server given by url and
subscription-based
- *
- * @param opcServerURL complete OPC UA server url
- * @param pullIntervalMilliSeconds duration of pull interval in
milliseconds, {@code null} if in subscription
- * mode
- * @param selectedNodeNames list of node names provided
- */
- public SpOpcUaConfig(String opcServerURL,
- Integer pullIntervalMilliSeconds,
- List<String> selectedNodeNames) {
-
- this.opcServerURL = opcServerURL;
- this.unauthenticated = true;
- this.pullIntervalMilliSeconds = pullIntervalMilliSeconds;
- this.selectedNodeNames = selectedNodeNames;
- }
-
- /**
- * Constructor for security level {@code None} and OPC server given by
hostname and port number
- *
- * @param opcServer OPC UA hostname
- * @param opcServerPort OPC UA port number
- * @param pullIntervalMilliSeconds duration of pull interval in
milliseconds, {@code null} if in
- * subscription mode
- * @param selectedNodeNames list of node names provided
- */
- public SpOpcUaConfig(String opcServer,
- int opcServerPort,
- Integer pullIntervalMilliSeconds,
- List<String> selectedNodeNames) {
- this(opcServer + ":" + opcServerPort, pullIntervalMilliSeconds,
selectedNodeNames);
- }
-
- /**
- * Constructor for security level {@code Sign} and OPC server given by url
- *
- * @param opcServerURL complete OPC UA server url
- * @param username username to authenticate at the OPC UA
server
- * @param password corresponding password to given user name
- * @param pullIntervalMilliSeconds duration of pull interval in milliseconds,
- * {@code null} if in subscription mode
- * @param selectedNodeNames list of node names provided
- */
- public SpOpcUaConfig(String opcServerURL,
- String username,
- String password,
- Integer pullIntervalMilliSeconds,
- List<String> selectedNodeNames) {
- this(opcServerURL, pullIntervalMilliSeconds, selectedNodeNames);
- this.unauthenticated = false;
- this.username = username;
- this.password = password;
- }
-
- /**
- * Constructor for OPC UA security level {@code Sign} and OPC server given
by hostname and port number
- *
- * @param opcServer OPC UA hostname
- * @param opcServerPort OPC UA port number
- * @param username username to authenticate at the OPC UA
server
- * @param password corresponding password to given user name
- * @param pullIntervalMilliSeconds duration of pull interval in milliseconds,
- * {@code null} if in subscription mode
- * @param selectedNodeNames list of node names provided
- */
- public SpOpcUaConfig(String opcServer,
- int opcServerPort,
- String username,
- String password,
- int pullIntervalMilliSeconds,
- List<String> selectedNodeNames) {
- this(opcServer, opcServerPort, pullIntervalMilliSeconds,
selectedNodeNames);
- this.unauthenticated = false;
- this.username = username;
- this.password = password;
- }
-
- public String getOpcServerURL() {
- return opcServerURL;
- }
-
- public boolean isUnauthenticated() {
- return unauthenticated;
- }
-
- public void setUnauthenticated(boolean unauthenticated) {
- this.unauthenticated = unauthenticated;
- }
-
- public String getUsername() {
- return username;
- }
-
- public void setUsername(String username) {
- this.username = username;
- }
-
- public String getPassword() {
- return password;
- }
-
- public void setPassword(String password) {
- this.password = password;
- }
-
- public List<String> getSelectedNodeNames() {
- return selectedNodeNames;
- }
-
- public Integer getPullIntervalMilliSeconds() {
- return pullIntervalMilliSeconds;
- }
-}
diff --git
a/streampipes-extensions/streampipes-connectors-opcua/src/main/java/org/apache/streampipes/extensions/connectors/opcua/adapter/SpOpcUaClient.java
b/streampipes-extensions/streampipes-connectors-opcua/src/main/java/org/apache/streampipes/extensions/connectors/opcua/client/SpOpcUaClient.java
similarity index 89%
rename from
streampipes-extensions/streampipes-connectors-opcua/src/main/java/org/apache/streampipes/extensions/connectors/opcua/adapter/SpOpcUaClient.java
rename to
streampipes-extensions/streampipes-connectors-opcua/src/main/java/org/apache/streampipes/extensions/connectors/opcua/client/SpOpcUaClient.java
index 755710e2e..95fda81db 100644
---
a/streampipes-extensions/streampipes-connectors-opcua/src/main/java/org/apache/streampipes/extensions/connectors/opcua/adapter/SpOpcUaClient.java
+++
b/streampipes-extensions/streampipes-connectors-opcua/src/main/java/org/apache/streampipes/extensions/connectors/opcua/client/SpOpcUaClient.java
@@ -16,11 +16,13 @@
*
*/
-package org.apache.streampipes.extensions.connectors.opcua.adapter;
+package org.apache.streampipes.extensions.connectors.opcua.client;
import org.apache.streampipes.commons.exceptions.SpConfigurationException;
-import
org.apache.streampipes.extensions.connectors.opcua.adapter.configuration.SpOpcUaConfig;
+import org.apache.streampipes.extensions.connectors.opcua.adapter.OpcUaAdapter;
+import
org.apache.streampipes.extensions.connectors.opcua.config.MiloOpcUaConfigurationProvider;
+import org.apache.streampipes.extensions.connectors.opcua.config.OpcUaConfig;
import org.eclipse.milo.opcua.sdk.client.OpcUaClient;
import org.eclipse.milo.opcua.sdk.client.api.config.OpcUaClientConfig;
@@ -54,16 +56,16 @@ import static
org.eclipse.milo.opcua.stack.core.types.builtin.unsigned.Unsigned.
/***
* Wrapper class for all OPC UA specific stuff.
*/
-public class SpOpcUaClient {
+public class SpOpcUaClient<T extends OpcUaConfig> {
private static final Logger LOG =
LoggerFactory.getLogger(SpOpcUaClient.class);
private OpcUaClient client;
- private final SpOpcUaConfig spOpcConfig;
+ private final T spOpcConfig;
private static final AtomicLong clientHandles = new AtomicLong(1L);
- public SpOpcUaClient(SpOpcUaConfig config) {
+ public SpOpcUaClient(T config) {
this.spOpcConfig = config;
}
@@ -158,12 +160,8 @@ public class SpOpcUaClient {
requests.add(new MonitoredItemCreateRequest(readValue,
MonitoringMode.Reporting, parameters));
}
- UaSubscription.ItemCreationCallback onItemCreated = new
UaSubscription.ItemCreationCallback() {
- @Override
- public void onItemCreated(UaMonitoredItem item, int i) {
- item.setValueConsumer(opcUaAdapter::onSubscriptionValue);
- }
- };
+ UaSubscription.ItemCreationCallback onItemCreated =
+ (item, i) -> item.setValueConsumer(opcUaAdapter::onSubscriptionValue);
List<UaMonitoredItem> items = subscription.createMonitoredItems(
TimestampsToReturn.Both,
requests,
@@ -180,15 +178,7 @@ public class SpOpcUaClient {
}
}
- public boolean inPullMode() {
- return !(spOpcConfig.getPullIntervalMilliSeconds() == null);
- }
-
- public int getPullIntervalMilliSeconds() {
- return spOpcConfig.getPullIntervalMilliSeconds();
- }
-
- public SpOpcUaConfig getSpOpcConfig() {
+ public T getSpOpcConfig() {
return spOpcConfig;
}
}
diff --git
a/streampipes-extensions/streampipes-connectors-opcua/src/main/java/org/apache/streampipes/extensions/connectors/opcua/adapter/MiloOpcUaConfigurationProvider.java
b/streampipes-extensions/streampipes-connectors-opcua/src/main/java/org/apache/streampipes/extensions/connectors/opcua/config/MiloOpcUaConfigurationProvider.java
similarity index 92%
rename from
streampipes-extensions/streampipes-connectors-opcua/src/main/java/org/apache/streampipes/extensions/connectors/opcua/adapter/MiloOpcUaConfigurationProvider.java
rename to
streampipes-extensions/streampipes-connectors-opcua/src/main/java/org/apache/streampipes/extensions/connectors/opcua/config/MiloOpcUaConfigurationProvider.java
index 00287d85a..9813d52fb 100644
---
a/streampipes-extensions/streampipes-connectors-opcua/src/main/java/org/apache/streampipes/extensions/connectors/opcua/adapter/MiloOpcUaConfigurationProvider.java
+++
b/streampipes-extensions/streampipes-connectors-opcua/src/main/java/org/apache/streampipes/extensions/connectors/opcua/config/MiloOpcUaConfigurationProvider.java
@@ -16,10 +16,9 @@
*
*/
-package org.apache.streampipes.extensions.connectors.opcua.adapter;
+package org.apache.streampipes.extensions.connectors.opcua.config;
import org.apache.streampipes.commons.exceptions.SpConfigurationException;
-import
org.apache.streampipes.extensions.connectors.opcua.adapter.configuration.SpOpcUaConfig;
import org.eclipse.milo.opcua.sdk.client.api.config.OpcUaClientConfig;
import org.eclipse.milo.opcua.sdk.client.api.config.OpcUaClientConfigBuilder;
@@ -37,7 +36,7 @@ import java.util.concurrent.ExecutionException;
public class MiloOpcUaConfigurationProvider {
- public OpcUaClientConfig makeClientConfig(SpOpcUaConfig spOpcConfig)
+ public OpcUaClientConfig makeClientConfig(OpcUaConfig spOpcConfig)
throws ExecutionException, InterruptedException,
SpConfigurationException, URISyntaxException {
String opcServerUrl = spOpcConfig.getOpcServerURL();
List<EndpointDescription> endpoints =
DiscoveryClient.getEndpoints(opcServerUrl).get();
@@ -61,7 +60,7 @@ public class MiloOpcUaConfigurationProvider {
}
private OpcUaClientConfig buildConfig(EndpointDescription endpoint,
- SpOpcUaConfig spOpcConfig) {
+ OpcUaConfig spOpcConfig) {
OpcUaClientConfigBuilder builder = defaultBuilder(endpoint);
if (!spOpcConfig.isUnauthenticated()) {
diff --git
a/streampipes-extensions/streampipes-connectors-opcua/src/main/java/org/apache/streampipes/extensions/connectors/opcua/adapter/utils/OpcUaNodeVariants.java
b/streampipes-extensions/streampipes-connectors-opcua/src/main/java/org/apache/streampipes/extensions/connectors/opcua/config/OpcUaAdapterConfig.java
similarity index 55%
rename from
streampipes-extensions/streampipes-connectors-opcua/src/main/java/org/apache/streampipes/extensions/connectors/opcua/adapter/utils/OpcUaNodeVariants.java
rename to
streampipes-extensions/streampipes-connectors-opcua/src/main/java/org/apache/streampipes/extensions/connectors/opcua/config/OpcUaAdapterConfig.java
index 414465cad..c6736ff8c 100644
---
a/streampipes-extensions/streampipes-connectors-opcua/src/main/java/org/apache/streampipes/extensions/connectors/opcua/adapter/utils/OpcUaNodeVariants.java
+++
b/streampipes-extensions/streampipes-connectors-opcua/src/main/java/org/apache/streampipes/extensions/connectors/opcua/config/OpcUaAdapterConfig.java
@@ -16,39 +16,21 @@
*
*/
-package org.apache.streampipes.extensions.connectors.opcua.adapter.utils;
+package org.apache.streampipes.extensions.connectors.opcua.config;
+public class OpcUaAdapterConfig extends OpcUaConfig {
-import jakarta.annotation.Nullable;
+ private Integer pullIntervalMilliSeconds;
-/**
- * Enum that maintains different variants of OPC UA nodes. <br>
- * Not yet completed. <br>
- */
-public enum OpcUaNodeVariants {
- Property(68),
- EUInformation(887);
-
- // ID as specified in OPC UA standard
- private final int id;
-
- private OpcUaNodeVariants(int id) {
- this.id = id;
+ public Integer getPullIntervalMilliSeconds() {
+ return pullIntervalMilliSeconds;
}
- public int getId() {
- return this.id;
+ public void setPullIntervalMilliSeconds(Integer pullIntervalMilliSeconds) {
+ this.pullIntervalMilliSeconds = pullIntervalMilliSeconds;
}
- @Nullable
- public static OpcUaNodeVariants from(int id) {
- switch (id) {
- case 68:
- return Property;
- case 887:
- return EUInformation;
- default:
- return null;
- }
+ public boolean inPullMode() {
+ return pullIntervalMilliSeconds != null;
}
}
diff --git
a/streampipes-extensions/streampipes-connectors-opcua/src/main/java/org/apache/streampipes/extensions/connectors/opcua/config/OpcUaConfig.java
b/streampipes-extensions/streampipes-connectors-opcua/src/main/java/org/apache/streampipes/extensions/connectors/opcua/config/OpcUaConfig.java
new file mode 100644
index 000000000..6db027d86
--- /dev/null
+++
b/streampipes-extensions/streampipes-connectors-opcua/src/main/java/org/apache/streampipes/extensions/connectors/opcua/config/OpcUaConfig.java
@@ -0,0 +1,70 @@
+/*
+ * 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.config;
+
+import java.util.List;
+
+public class OpcUaConfig {
+
+ private String opcServerURL;
+ private boolean unauthenticated;
+ private String username;
+ private String password;
+ private List<String> selectedNodeNames;
+
+ public String getOpcServerURL() {
+ return opcServerURL;
+ }
+
+ public void setOpcServerURL(String opcServerURL) {
+ this.opcServerURL = opcServerURL;
+ }
+
+ public boolean isUnauthenticated() {
+ return unauthenticated;
+ }
+
+ public void setUnauthenticated(boolean unauthenticated) {
+ this.unauthenticated = unauthenticated;
+ }
+
+ public String getUsername() {
+ return username;
+ }
+
+ public void setUsername(String username) {
+ this.username = username;
+ }
+
+ public String getPassword() {
+ return password;
+ }
+
+ public void setPassword(String password) {
+ this.password = password;
+ }
+
+ public List<String> getSelectedNodeNames() {
+ return selectedNodeNames;
+ }
+
+ public void setSelectedNodeNames(List<String> selectedNodeNames) {
+ this.selectedNodeNames = selectedNodeNames;
+ }
+}
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
new file mode 100644
index 000000000..647e66fdb
--- /dev/null
+++
b/streampipes-extensions/streampipes-connectors-opcua/src/main/java/org/apache/streampipes/extensions/connectors/opcua/config/SharedUserConfiguration.java
@@ -0,0 +1,89 @@
+/*
+ * 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.config;
+
+import org.apache.streampipes.sdk.StaticProperties;
+import
org.apache.streampipes.sdk.builder.AbstractConfigurablePipelineElementBuilder;
+import org.apache.streampipes.sdk.helpers.Alternatives;
+import org.apache.streampipes.sdk.helpers.Labels;
+
+import java.util.List;
+
+import static
org.apache.streampipes.extensions.connectors.opcua.utils.OpcUaLabels.ACCESS_MODE;
+import static
org.apache.streampipes.extensions.connectors.opcua.utils.OpcUaLabels.ADAPTER_TYPE;
+import static
org.apache.streampipes.extensions.connectors.opcua.utils.OpcUaLabels.AVAILABLE_NODES;
+import static
org.apache.streampipes.extensions.connectors.opcua.utils.OpcUaLabels.HOST_PORT;
+import static
org.apache.streampipes.extensions.connectors.opcua.utils.OpcUaLabels.OPC_HOST;
+import static
org.apache.streampipes.extensions.connectors.opcua.utils.OpcUaLabels.OPC_HOST_OR_URL;
+import static
org.apache.streampipes.extensions.connectors.opcua.utils.OpcUaLabels.OPC_SERVER_HOST;
+import static
org.apache.streampipes.extensions.connectors.opcua.utils.OpcUaLabels.OPC_SERVER_PORT;
+import static
org.apache.streampipes.extensions.connectors.opcua.utils.OpcUaLabels.OPC_SERVER_URL;
+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.UNAUTHENTICATED;
+import static
org.apache.streampipes.extensions.connectors.opcua.utils.OpcUaLabels.USERNAME;
+import static
org.apache.streampipes.extensions.connectors.opcua.utils.OpcUaLabels.USERNAME_GROUP;
+
+public class SharedUserConfiguration {
+
+ public static void
appendSharedOpcUaConfig(AbstractConfigurablePipelineElementBuilder<?, ?>
builder,
+ boolean adapterConfig) {
+
+ var dependsOn = adapterConfig ? List.of(
+ ADAPTER_TYPE.name(),
+ ACCESS_MODE.name(),
+ OPC_HOST_OR_URL.name()
+ ) : List.of(
+ ACCESS_MODE.name(),
+ OPC_HOST_OR_URL.name());
+
+ builder
+ .requiredAlternatives(Labels.withId(ACCESS_MODE),
+ Alternatives.from(Labels.withId(UNAUTHENTICATED)),
+ Alternatives.from(Labels.withId(USERNAME_GROUP),
+ StaticProperties.group(
+ Labels.withId(USERNAME_GROUP),
+ StaticProperties.stringFreeTextProperty(
+ Labels.withId(USERNAME)),
+ StaticProperties.secretValue(Labels.withId(PASSWORD))
+ ))
+ )
+ .requiredAlternatives(Labels.withId(OPC_HOST_OR_URL),
+ Alternatives.from(
+ Labels.withId(OPC_URL),
+ StaticProperties.stringFreeTextProperty(
+ Labels.withId(OPC_SERVER_URL), "opc.tcp://localhost:4840"))
+ ,
+ Alternatives.from(Labels.withId(OPC_HOST),
+ StaticProperties.group(
+ Labels.withId(HOST_PORT),
+ StaticProperties.stringFreeTextProperty(
+ Labels.withId(OPC_SERVER_HOST)),
+ StaticProperties.stringFreeTextProperty(
+ Labels.withId(OPC_SERVER_PORT))
+ ))
+ )
+ .requiredRuntimeResolvableTreeInput(
+ Labels.withId(AVAILABLE_NODES.name()),
+ dependsOn,
+ true,
+ adapterConfig
+ );
+ }
+}
diff --git
a/streampipes-extensions/streampipes-connectors-opcua/src/main/java/org/apache/streampipes/extensions/connectors/opcua/adapter/configuration/SpOpcUaConfigBuilder.java
b/streampipes-extensions/streampipes-connectors-opcua/src/main/java/org/apache/streampipes/extensions/connectors/opcua/config/SpOpcUaConfigExtractor.java
similarity index 63%
rename from
streampipes-extensions/streampipes-connectors-opcua/src/main/java/org/apache/streampipes/extensions/connectors/opcua/adapter/configuration/SpOpcUaConfigBuilder.java
rename to
streampipes-extensions/streampipes-connectors-opcua/src/main/java/org/apache/streampipes/extensions/connectors/opcua/config/SpOpcUaConfigExtractor.java
index 7d8cbd97e..4f461c977 100644
---
a/streampipes-extensions/streampipes-connectors-opcua/src/main/java/org/apache/streampipes/extensions/connectors/opcua/adapter/configuration/SpOpcUaConfigBuilder.java
+++
b/streampipes-extensions/streampipes-connectors-opcua/src/main/java/org/apache/streampipes/extensions/connectors/opcua/config/SpOpcUaConfigExtractor.java
@@ -16,10 +16,11 @@
*
*/
-package
org.apache.streampipes.extensions.connectors.opcua.adapter.configuration;
+package org.apache.streampipes.extensions.connectors.opcua.config;
+import org.apache.streampipes.extensions.api.extractor.IParameterExtractor;
import
org.apache.streampipes.extensions.api.extractor.IStaticPropertyExtractor;
-import
org.apache.streampipes.extensions.connectors.opcua.adapter.utils.OpcUaUtil;
+import org.apache.streampipes.extensions.connectors.opcua.utils.OpcUaUtil;
import java.util.List;
@@ -27,8 +28,6 @@ import static
org.apache.kafka.common.config.ConfigDef.Type.PASSWORD;
import static
org.apache.streampipes.extensions.connectors.opcua.utils.OpcUaLabels.ACCESS_MODE;
import static
org.apache.streampipes.extensions.connectors.opcua.utils.OpcUaLabels.ADAPTER_TYPE;
import static
org.apache.streampipes.extensions.connectors.opcua.utils.OpcUaLabels.AVAILABLE_NODES;
-import static
org.apache.streampipes.extensions.connectors.opcua.utils.OpcUaLabels.NAMESPACE_INDEX;
-import static
org.apache.streampipes.extensions.connectors.opcua.utils.OpcUaLabels.NODE_ID;
import static
org.apache.streampipes.extensions.connectors.opcua.utils.OpcUaLabels.OPC_HOST_OR_URL;
import static
org.apache.streampipes.extensions.connectors.opcua.utils.OpcUaLabels.OPC_SERVER_HOST;
import static
org.apache.streampipes.extensions.connectors.opcua.utils.OpcUaLabels.OPC_SERVER_PORT;
@@ -39,84 +38,72 @@ import static
org.apache.streampipes.extensions.connectors.opcua.utils.OpcUaLabe
import static
org.apache.streampipes.extensions.connectors.opcua.utils.OpcUaLabels.UNAUTHENTICATED;
import static
org.apache.streampipes.extensions.connectors.opcua.utils.OpcUaLabels.USERNAME;
-public class SpOpcUaConfigBuilder {
+public class SpOpcUaConfigExtractor {
/**
- * Creates {@link SpOpcUaConfig} instance in accordance with the given
+ * Creates {@link OpcUaAdapterConfig} instance in accordance with the given
* {@link org.apache.streampipes.sdk.extractor.StaticPropertyExtractor}.
*
* @param extractor extractor for user inputs
- * @return {@link SpOpcUaConfig} instance based on information from {@code
extractor}
+ * @return {@link OpcUaAdapterConfig} instance based on information from
{@code extractor}
*/
- public static SpOpcUaConfig from(IStaticPropertyExtractor extractor) {
-
- String selectedAlternativeConnection =
- extractor.selectedAlternativeInternalId(OPC_HOST_OR_URL.name());
- String selectedAlternativeAuthentication =
- extractor.selectedAlternativeInternalId(ACCESS_MODE.name());
-
+ public static OpcUaAdapterConfig
extractAdapterConfig(IStaticPropertyExtractor extractor) {
+ var config = extractSharedConfig(extractor, new OpcUaAdapterConfig());
boolean usePullMode =
extractor.selectedAlternativeInternalId(ADAPTER_TYPE.name())
.equals(PULL_MODE.name());
- boolean useURL = selectedAlternativeConnection.equals(OPC_URL.name());
- boolean unauthenticated =
selectedAlternativeAuthentication.equals(UNAUTHENTICATED.name());
- Integer pullIntervalSeconds = null;
if (usePullMode) {
- pullIntervalSeconds =
+ Integer pullIntervalSeconds =
extractor.singleValueParameter(PULLING_INTERVAL.name(),
Integer.class);
+
+ config.setPullIntervalMilliSeconds(pullIntervalSeconds);
}
+ return config;
+ }
+
+ public static OpcUaConfig extractSinkConfig(IParameterExtractor<?>
extractor) {
+ return extractSharedConfig(extractor, new OpcUaConfig());
+ }
+
+ public static <T extends OpcUaConfig> T
extractSharedConfig(IParameterExtractor<?> extractor,
+ T config) {
+
+ String selectedAlternativeConnection =
+ extractor.selectedAlternativeInternalId(OPC_HOST_OR_URL.name());
+ String selectedAlternativeAuthentication =
+ extractor.selectedAlternativeInternalId(ACCESS_MODE.name());
List<String> selectedNodeNames =
extractor.selectedTreeNodesInternalNames(AVAILABLE_NODES.name(),
String.class, true);
- if (useURL && unauthenticated) {
-
- String serverAddress =
- extractor.singleValueParameter(OPC_SERVER_URL.name(), String.class);
- serverAddress = OpcUaUtil.formatServerAddress(serverAddress);
+ config.setSelectedNodeNames(selectedNodeNames);
- return new SpOpcUaConfig(serverAddress,
- pullIntervalSeconds,
- selectedNodeNames);
+ boolean useURL = selectedAlternativeConnection.equals(OPC_URL.name());
+ boolean unauthenticated =
selectedAlternativeAuthentication.equals(UNAUTHENTICATED.name());
- } else if (!useURL && unauthenticated) {
+ if (useURL) {
String serverAddress =
- extractor.singleValueParameter(OPC_SERVER_HOST.name(), String.class);
- serverAddress = OpcUaUtil.formatServerAddress(serverAddress);
+ extractor.singleValueParameter(OPC_SERVER_URL.name(), String.class);
+ config.setOpcServerURL(OpcUaUtil.formatServerAddress(serverAddress));
+ } else {
+ String serverAddress = OpcUaUtil.formatServerAddress(
+ extractor.singleValueParameter(OPC_SERVER_HOST.name(), String.class)
+ );
int port = extractor.singleValueParameter(OPC_SERVER_PORT.name(),
int.class);
+ config.setOpcServerURL(serverAddress + ":" + port);
+ }
- return new SpOpcUaConfig(serverAddress,
- port,
- pullIntervalSeconds,
- selectedNodeNames);
+ if (unauthenticated) {
+ config.setUnauthenticated(true);
} else {
-
String username = extractor.singleValueParameter(USERNAME.name(),
String.class);
String password = extractor.secretValue(PASSWORD.name());
- if (useURL) {
- String serverAddress =
- extractor.singleValueParameter(OPC_SERVER_URL.name(),
String.class);
- serverAddress = OpcUaUtil.formatServerAddress(serverAddress);
-
- return new SpOpcUaConfig(serverAddress,
- username,
- password,
- pullIntervalSeconds,
- selectedNodeNames);
- } else {
- String serverAddress =
- extractor.singleValueParameter(OPC_SERVER_HOST.name(),
String.class);
- serverAddress = OpcUaUtil.formatServerAddress(serverAddress);
- int port = extractor.singleValueParameter(OPC_SERVER_PORT.name(),
int.class);
-
- return new SpOpcUaConfig(serverAddress,
- port,
- username,
- password,
- pullIntervalSeconds,
- selectedNodeNames);
- }
+ config.setUsername(username);
+ config.setPassword(password);
+ config.setUnauthenticated(false);
}
+
+ return config;
}
}
diff --git
a/streampipes-extensions/streampipes-connectors-opcua/src/main/java/org/apache/streampipes/extensions/connectors/opcua/adapter/OpcNode.java
b/streampipes-extensions/streampipes-connectors-opcua/src/main/java/org/apache/streampipes/extensions/connectors/opcua/model/OpcNode.java
similarity index 98%
rename from
streampipes-extensions/streampipes-connectors-opcua/src/main/java/org/apache/streampipes/extensions/connectors/opcua/adapter/OpcNode.java
rename to
streampipes-extensions/streampipes-connectors-opcua/src/main/java/org/apache/streampipes/extensions/connectors/opcua/model/OpcNode.java
index 36b400e51..f27b3321f 100644
---
a/streampipes-extensions/streampipes-connectors-opcua/src/main/java/org/apache/streampipes/extensions/connectors/opcua/adapter/OpcNode.java
+++
b/streampipes-extensions/streampipes-connectors-opcua/src/main/java/org/apache/streampipes/extensions/connectors/opcua/model/OpcNode.java
@@ -16,7 +16,7 @@
*
*/
-package org.apache.streampipes.extensions.connectors.opcua.adapter;
+package org.apache.streampipes.extensions.connectors.opcua.model;
import org.apache.streampipes.sdk.utils.Datatypes;
diff --git
a/streampipes-extensions/streampipes-connectors-opcua/src/main/java/org/apache/streampipes/extensions/connectors/opcua/sink/OpcUa.java
b/streampipes-extensions/streampipes-connectors-opcua/src/main/java/org/apache/streampipes/extensions/connectors/opcua/sink/OpcUa.java
index 870ae02b8..78babe002 100644
---
a/streampipes-extensions/streampipes-connectors-opcua/src/main/java/org/apache/streampipes/extensions/connectors/opcua/sink/OpcUa.java
+++
b/streampipes-extensions/streampipes-connectors-opcua/src/main/java/org/apache/streampipes/extensions/connectors/opcua/sink/OpcUa.java
@@ -19,27 +19,22 @@
package org.apache.streampipes.extensions.connectors.opcua.sink;
import org.apache.streampipes.commons.exceptions.SpRuntimeException;
+import org.apache.streampipes.extensions.connectors.opcua.client.SpOpcUaClient;
import org.apache.streampipes.model.runtime.Event;
import org.apache.streampipes.model.runtime.field.PrimitiveField;
import org.apache.streampipes.vocabulary.XSD;
import org.eclipse.milo.opcua.sdk.client.OpcUaClient;
-import org.eclipse.milo.opcua.sdk.client.api.config.OpcUaClientConfig;
-import org.eclipse.milo.opcua.stack.client.DiscoveryClient;
import org.eclipse.milo.opcua.stack.core.UaException;
-import org.eclipse.milo.opcua.stack.core.security.SecurityPolicy;
import org.eclipse.milo.opcua.stack.core.types.builtin.DataValue;
-import org.eclipse.milo.opcua.stack.core.types.builtin.LocalizedText;
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.eclipse.milo.opcua.stack.core.types.structured.EndpointDescription;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.Arrays;
import java.util.HashMap;
-import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
@@ -49,13 +44,14 @@ public class OpcUa {
private OpcUaClient opcUaClient;
private OpcUaParameters params;
- private String serverUrl;
+
private NodeId node;
- private Class targetDataType;
- private Class sourceDataType;
+
+ private Class<?> targetDataType;
+ private Class<?> sourceDataType;
// define a mapping of StreamPipes data types to Java classes
- private static final HashMap<String, Class> XSDMatchings = new HashMap<>();
+ private static final HashMap<String, Class<?>> XSDMatchings = new
HashMap<>();
static {
XSDMatchings.put(XSD.DOUBLE.toString(), Double.class);
@@ -67,7 +63,7 @@ public class OpcUa {
}
// define potential mappings, left can be mapped to right
- private static HashMap<Class, Class[]> compatibleDataTypes = new HashMap<>();
+ private static final HashMap<Class<?>, Class<?>[]> compatibleDataTypes = new
HashMap<>();
static {
compatibleDataTypes.put(Double.class, new Class[]{Float.class,
String.class});
@@ -77,55 +73,26 @@ public class OpcUa {
compatibleDataTypes.put(String.class, new Class[]{String.class});
}
- public void onInvocation(OpcUaParameters parameters) throws
+ public void onInvocation(OpcUaParameters params) throws
SpRuntimeException {
- if (!parameters.hostname().startsWith("opc.tcp://")) {
- serverUrl = "opc.tcp://" + parameters.hostname() + ":" +
parameters.port();
- } else {
- serverUrl = parameters.hostname() + ":" + parameters.port();
- }
- if (isInteger(parameters.nodeId())) {
- int integerNodeId = Integer.parseInt(parameters.nodeId());
- node = new NodeId(parameters.nameSpaceIndex(), integerNodeId);
- } else {
- node = new NodeId(parameters.nameSpaceIndex(), parameters.nodeId());
- }
-
-
- this.params = parameters;
-
- List<EndpointDescription> endpoints;
-
try {
-
- endpoints = DiscoveryClient.getEndpoints(serverUrl).get();
-
- EndpointDescription endpoint = endpoints
- .stream()
- .filter(e ->
e.getSecurityPolicyUri().equals(SecurityPolicy.None.getUri()))
- .findFirst().orElseThrow(() -> new Exception("no desired endpoints
returned"));
-
- OpcUaClientConfig config = OpcUaClientConfig.builder()
- .setApplicationName(LocalizedText.english("eclipse milo opc-ua
client"))
- .setApplicationUri("urn:eclipse:milo:examples:client")
- .setEndpoint(endpoint)
- .build();
-
- opcUaClient = OpcUaClient.create(config);
+ this.params = params;
+ this.node = NodeId.parse(params.getSelectedNode());
+ opcUaClient = new SpOpcUaClient<>(params.getConfig()).getClient();
opcUaClient.connect().get();
} catch (Exception e) {
- throw new SpRuntimeException("Could not connect to OPC-UA server: " +
serverUrl);
+ throw new SpRuntimeException("Could not connect to OPC-UA server: " +
params.getConfig().getOpcServerURL());
}
// check whether input data type and target data type are compatible
try {
Variant value =
opcUaClient.getAddressSpace().getVariableNode(node).readValue().getValue();
targetDataType = value.getValue().getClass();
- sourceDataType = XSDMatchings.get(params.mappingPropertyType());
+ sourceDataType = XSDMatchings.get(params.getMappingPropertyType());
if (!sourceDataType.equals(targetDataType)) {
- if
(!Arrays.stream(compatibleDataTypes.get(sourceDataType)).anyMatch(dt ->
dt.equals(targetDataType))) {
+ if
(Arrays.stream(compatibleDataTypes.get(sourceDataType)).noneMatch(dt ->
dt.equals(targetDataType))) {
throw new SpRuntimeException("Data Type of event of target node are
not compatible");
}
}
@@ -140,7 +107,7 @@ public class OpcUa {
Variant v = getValue(inputEvent);
if (v == null) {
- LOG.error("Mapping property type: " + this.params.mappingPropertyType()
+ " is not supported");
+ LOG.error("Mapping property type: " +
this.params.getMappingPropertyType() + " is not supported");
} else {
DataValue value = new DataValue(v);
@@ -150,18 +117,18 @@ public class OpcUa {
StatusCode status = f.get();
if (status.isBad()) {
if (status.getValue() == 0x80740000L) {
- LOG.error("Type missmatch! Tried to write value of type: " +
this.params.mappingPropertyType()
+ LOG.error("Type missmatch! Tried to write value of type: " +
this.params.getMappingPropertyType()
+ " but server did not accept this");
} else if (status.getValue() == 0x803B0000L) {
LOG.error("Wrong access level. Not allowed to write to nodes");
}
LOG.error(
"Value: " + value.getValue().toString() + " could not be written
to node Id: "
- + this.params.nodeId() + " on " + "OPC-UA server: " +
this.serverUrl);
+ + node.getIdentifier() + " on " + "OPC-UA server: " +
params.getConfig().getOpcServerURL());
}
} catch (InterruptedException | ExecutionException e) {
LOG.error("Exception: Value: " + value.getValue().toString() + " could
not be written to node Id: "
- + this.params.nodeId() + " on " + "OPC-UA server: " +
this.serverUrl);
+ + node.getIdentifier() + " on " + "OPC-UA server: " +
params.getConfig().getOpcServerURL());
}
}
}
@@ -173,7 +140,7 @@ public class OpcUa {
private Variant getValue(Event inputEvent) {
Variant result = null;
PrimitiveField propertyPrimitive =
-
inputEvent.getFieldBySelector(this.params.mappingPropertySelector()).getAsPrimitive();
+
inputEvent.getFieldBySelector(this.params.getMappingPropertySelector()).getAsPrimitive();
if (targetDataType.equals(Integer.class)) {
result = new Variant(propertyPrimitive.getAsInt());
@@ -189,16 +156,4 @@ public class OpcUa {
return result;
}
-
- public static boolean isInteger(String s) {
- try {
- Integer.parseInt(s);
- } catch (NumberFormatException e) {
- return false;
- } catch (NullPointerException e) {
- return false;
- }
- // only got here if we didn't return false
- return true;
- }
}
diff --git
a/streampipes-extensions/streampipes-connectors-opcua/src/main/java/org/apache/streampipes/extensions/connectors/opcua/sink/OpcUaParameters.java
b/streampipes-extensions/streampipes-connectors-opcua/src/main/java/org/apache/streampipes/extensions/connectors/opcua/sink/OpcUaParameters.java
index 0a0f45ea2..15906af4a 100644
---
a/streampipes-extensions/streampipes-connectors-opcua/src/main/java/org/apache/streampipes/extensions/connectors/opcua/sink/OpcUaParameters.java
+++
b/streampipes-extensions/streampipes-connectors-opcua/src/main/java/org/apache/streampipes/extensions/connectors/opcua/sink/OpcUaParameters.java
@@ -18,45 +18,38 @@
package org.apache.streampipes.extensions.connectors.opcua.sink;
+import org.apache.streampipes.extensions.connectors.opcua.config.OpcUaConfig;
+
public final class OpcUaParameters {
- private final String hostname;
- private final Integer port;
- private final String nodeId;
- private final Integer nameSpaceIndex;
+ private final String selectedNode;
private final String mappingPropertySelector;
private final String mappingPropertyType;
- public OpcUaParameters(String hostname, Integer port, String nodeId, Integer
nameSpaceIndex,
- String mappingPropertySelector, String
mappingPropertyType) {
- this.hostname = hostname;
- this.port = port;
- this.nodeId = nodeId;
- this.nameSpaceIndex = nameSpaceIndex;
+ private final OpcUaConfig config;
+
+ public OpcUaParameters(OpcUaConfig config,
+ String mappingPropertySelector,
+ String mappingPropertyType,
+ String selectedNode) {
+ this.config = config;
this.mappingPropertySelector = mappingPropertySelector;
this.mappingPropertyType = mappingPropertyType;
+ this.selectedNode = selectedNode;
}
- public String hostname() {
- return hostname;
- }
-
- public Integer port() {
- return port;
- }
-
- public String nodeId() {
- return nodeId;
- }
-
- public Integer nameSpaceIndex() {
- return nameSpaceIndex;
+ public String getSelectedNode() {
+ return selectedNode;
}
- public String mappingPropertySelector() {
+ public String getMappingPropertySelector() {
return mappingPropertySelector;
}
- public String mappingPropertyType() {
+ public String getMappingPropertyType() {
return mappingPropertyType;
}
+
+ public OpcUaConfig getConfig() {
+ return config;
+ }
}
diff --git
a/streampipes-extensions/streampipes-connectors-opcua/src/main/java/org/apache/streampipes/extensions/connectors/opcua/sink/OpcUaSink.java
b/streampipes-extensions/streampipes-connectors-opcua/src/main/java/org/apache/streampipes/extensions/connectors/opcua/sink/OpcUaSink.java
index 8532172b9..c58c8cab0 100644
---
a/streampipes-extensions/streampipes-connectors-opcua/src/main/java/org/apache/streampipes/extensions/connectors/opcua/sink/OpcUaSink.java
+++
b/streampipes-extensions/streampipes-connectors-opcua/src/main/java/org/apache/streampipes/extensions/connectors/opcua/sink/OpcUaSink.java
@@ -18,14 +18,21 @@
package org.apache.streampipes.extensions.connectors.opcua.sink;
+import org.apache.streampipes.commons.exceptions.SpConfigurationException;
import org.apache.streampipes.commons.exceptions.SpRuntimeException;
+import
org.apache.streampipes.extensions.api.extractor.IStaticPropertyExtractor;
import org.apache.streampipes.extensions.api.pe.IStreamPipesDataSink;
import org.apache.streampipes.extensions.api.pe.config.IDataSinkConfiguration;
import
org.apache.streampipes.extensions.api.pe.context.EventSinkRuntimeContext;
import org.apache.streampipes.extensions.api.pe.param.IDataSinkParameters;
+import org.apache.streampipes.extensions.api.runtime.SupportsRuntimeConfig;
+import
org.apache.streampipes.extensions.connectors.opcua.config.SharedUserConfiguration;
+import
org.apache.streampipes.extensions.connectors.opcua.config.SpOpcUaConfigExtractor;
+import org.apache.streampipes.extensions.connectors.opcua.utils.OpcUaUtil;
import org.apache.streampipes.model.DataSinkType;
import org.apache.streampipes.model.runtime.Event;
import org.apache.streampipes.model.schema.PropertyScope;
+import org.apache.streampipes.model.staticproperty.StaticProperty;
import org.apache.streampipes.sdk.builder.DataSinkBuilder;
import org.apache.streampipes.sdk.builder.StreamRequirementsBuilder;
import org.apache.streampipes.sdk.builder.sink.DataSinkConfiguration;
@@ -35,33 +42,28 @@ import org.apache.streampipes.sdk.helpers.Locales;
import org.apache.streampipes.sdk.utils.Assets;
import static
org.apache.streampipes.extensions.connectors.opcua.utils.OpcUaLabels.MAPPING_PROPERY;
-import static
org.apache.streampipes.extensions.connectors.opcua.utils.OpcUaLabels.NAMESPACE_INDEX;
-import static
org.apache.streampipes.extensions.connectors.opcua.utils.OpcUaLabels.NODE_ID;
-import static
org.apache.streampipes.extensions.connectors.opcua.utils.OpcUaLabels.OPC_SERVER_HOST;
-import static
org.apache.streampipes.extensions.connectors.opcua.utils.OpcUaLabels.OPC_SERVER_PORT;
-public class OpcUaSink implements IStreamPipesDataSink {
+public class OpcUaSink implements IStreamPipesDataSink, SupportsRuntimeConfig {
private OpcUa opcUa;
@Override
public IDataSinkConfiguration declareConfig() {
+ var builder =
DataSinkBuilder.create("org.apache.streampipes.sinks.databases.jvm.opcua")
+ .withLocales(Locales.EN)
+ .withAssets(Assets.DOCUMENTATION, Assets.ICON)
+ .category(DataSinkType.FORWARD)
+ .requiredStream(StreamRequirementsBuilder
+ .create()
+ .requiredPropertyWithUnaryMapping(EpRequirements.anyProperty(),
+ Labels.withId(MAPPING_PROPERY),
+ PropertyScope.NONE).build());
+
+ SharedUserConfiguration.appendSharedOpcUaConfig(builder, false);
+
return DataSinkConfiguration.create(
OpcUaSink::new,
-
DataSinkBuilder.create("org.apache.streampipes.sinks.databases.jvm.opcua")
- .withLocales(Locales.EN)
- .withAssets(Assets.DOCUMENTATION, Assets.ICON)
- .category(DataSinkType.FORWARD)
- .requiredStream(StreamRequirementsBuilder
- .create()
- .requiredPropertyWithUnaryMapping(EpRequirements.anyProperty(),
- Labels.withId(MAPPING_PROPERY),
- PropertyScope.NONE).build())
- .requiredTextParameter(Labels.withId(OPC_SERVER_HOST))
- .requiredIntegerParameter(Labels.withId(OPC_SERVER_PORT))
- .requiredIntegerParameter(Labels.withId(NAMESPACE_INDEX))
- .requiredTextParameter(Labels.withId(NODE_ID))
- .build()
+ builder.build()
);
}
@@ -69,13 +71,9 @@ public class OpcUaSink implements IStreamPipesDataSink {
public void onPipelineStarted(IDataSinkParameters parameters,
EventSinkRuntimeContext runtimeContext) {
var extractor = parameters.extractor();
- String hostname = extractor.singleValueParameter(OPC_SERVER_HOST.name(),
String.class);
- Integer port = extractor.singleValueParameter(OPC_SERVER_PORT.name(),
Integer.class);
+ var config = SpOpcUaConfigExtractor.extractSinkConfig(extractor);
- String nodeId = extractor.singleValueParameter(NODE_ID.name(),
String.class);
- Integer nameSpaceIndex =
extractor.singleValueParameter(NAMESPACE_INDEX.name(), Integer.class);
-
- String mappingPropertySelector =
extractor.mappingPropertyValue(NODE_ID.name());
+ String mappingPropertySelector =
extractor.mappingPropertyValue(MAPPING_PROPERY.name());
String mappingPropertyType = "";
try {
@@ -84,8 +82,12 @@ public class OpcUaSink implements IStreamPipesDataSink {
e.printStackTrace();
}
- OpcUaParameters params = new OpcUaParameters(hostname, port, nodeId,
nameSpaceIndex,
- mappingPropertySelector, mappingPropertyType);
+ OpcUaParameters params = new OpcUaParameters(
+ config,
+ mappingPropertySelector,
+ mappingPropertyType,
+ config.getSelectedNodeNames().get(0)
+ );
this.opcUa = new OpcUa();
this.opcUa.onInvocation(params);
@@ -100,4 +102,10 @@ public class OpcUaSink implements IStreamPipesDataSink {
public void onPipelineStopped() {
this.opcUa.onDetach();
}
+
+ @Override
+ public StaticProperty resolveConfiguration(String staticPropertyInternalName,
+ IStaticPropertyExtractor
extractor) throws SpConfigurationException {
+ return OpcUaUtil.resolveConfiguration(staticPropertyInternalName,
extractor);
+ }
}
diff --git
a/streampipes-extensions/streampipes-connectors-opcua/src/main/java/org/apache/streampipes/extensions/connectors/opcua/adapter/utils/ExceptionMessageExtractor.java
b/streampipes-extensions/streampipes-connectors-opcua/src/main/java/org/apache/streampipes/extensions/connectors/opcua/utils/ExceptionMessageExtractor.java
similarity index 94%
rename from
streampipes-extensions/streampipes-connectors-opcua/src/main/java/org/apache/streampipes/extensions/connectors/opcua/adapter/utils/ExceptionMessageExtractor.java
rename to
streampipes-extensions/streampipes-connectors-opcua/src/main/java/org/apache/streampipes/extensions/connectors/opcua/utils/ExceptionMessageExtractor.java
index df0dfae8c..9c37b0901 100644
---
a/streampipes-extensions/streampipes-connectors-opcua/src/main/java/org/apache/streampipes/extensions/connectors/opcua/adapter/utils/ExceptionMessageExtractor.java
+++
b/streampipes-extensions/streampipes-connectors-opcua/src/main/java/org/apache/streampipes/extensions/connectors/opcua/utils/ExceptionMessageExtractor.java
@@ -16,7 +16,7 @@
*
*/
-package org.apache.streampipes.extensions.connectors.opcua.adapter.utils;
+package org.apache.streampipes.extensions.connectors.opcua.utils;
import org.eclipse.milo.opcua.stack.core.UaException;
diff --git
a/streampipes-extensions/streampipes-connectors-opcua/src/main/java/org/apache/streampipes/extensions/connectors/opcua/adapter/utils/OpcUaTypes.java
b/streampipes-extensions/streampipes-connectors-opcua/src/main/java/org/apache/streampipes/extensions/connectors/opcua/utils/OpcUaTypes.java
similarity index 96%
rename from
streampipes-extensions/streampipes-connectors-opcua/src/main/java/org/apache/streampipes/extensions/connectors/opcua/adapter/utils/OpcUaTypes.java
rename to
streampipes-extensions/streampipes-connectors-opcua/src/main/java/org/apache/streampipes/extensions/connectors/opcua/utils/OpcUaTypes.java
index 956191c67..46692e96e 100644
---
a/streampipes-extensions/streampipes-connectors-opcua/src/main/java/org/apache/streampipes/extensions/connectors/opcua/adapter/utils/OpcUaTypes.java
+++
b/streampipes-extensions/streampipes-connectors-opcua/src/main/java/org/apache/streampipes/extensions/connectors/opcua/utils/OpcUaTypes.java
@@ -16,7 +16,7 @@
*
*/
-package org.apache.streampipes.extensions.connectors.opcua.adapter.utils;
+package org.apache.streampipes.extensions.connectors.opcua.utils;
import org.apache.streampipes.sdk.utils.Datatypes;
diff --git
a/streampipes-extensions/streampipes-connectors-opcua/src/main/java/org/apache/streampipes/extensions/connectors/opcua/adapter/utils/OpcUaUtil.java
b/streampipes-extensions/streampipes-connectors-opcua/src/main/java/org/apache/streampipes/extensions/connectors/opcua/utils/OpcUaUtil.java
similarity index 92%
rename from
streampipes-extensions/streampipes-connectors-opcua/src/main/java/org/apache/streampipes/extensions/connectors/opcua/adapter/utils/OpcUaUtil.java
rename to
streampipes-extensions/streampipes-connectors-opcua/src/main/java/org/apache/streampipes/extensions/connectors/opcua/utils/OpcUaUtil.java
index 32c31ce6a..9acd31190 100644
---
a/streampipes-extensions/streampipes-connectors-opcua/src/main/java/org/apache/streampipes/extensions/connectors/opcua/adapter/utils/OpcUaUtil.java
+++
b/streampipes-extensions/streampipes-connectors-opcua/src/main/java/org/apache/streampipes/extensions/connectors/opcua/utils/OpcUaUtil.java
@@ -16,7 +16,7 @@
*
*/
-package org.apache.streampipes.extensions.connectors.opcua.adapter.utils;
+package org.apache.streampipes.extensions.connectors.opcua.utils;
import org.apache.streampipes.commons.exceptions.SpConfigurationException;
import org.apache.streampipes.commons.exceptions.connect.AdapterException;
@@ -24,11 +24,11 @@ import
org.apache.streampipes.commons.exceptions.connect.ParseException;
import
org.apache.streampipes.extensions.api.extractor.IAdapterParameterExtractor;
import
org.apache.streampipes.extensions.api.extractor.IStaticPropertyExtractor;
import
org.apache.streampipes.extensions.api.runtime.ResolvesContainerProvidedOptions;
-import org.apache.streampipes.extensions.connectors.opcua.adapter.OpcNode;
import
org.apache.streampipes.extensions.connectors.opcua.adapter.OpcUaNodeBrowser;
-import
org.apache.streampipes.extensions.connectors.opcua.adapter.SpOpcUaClient;
-import
org.apache.streampipes.extensions.connectors.opcua.adapter.configuration.SpOpcUaConfigBuilder;
-import org.apache.streampipes.extensions.connectors.opcua.utils.OpcUaLabels;
+import org.apache.streampipes.extensions.connectors.opcua.client.SpOpcUaClient;
+import org.apache.streampipes.extensions.connectors.opcua.config.OpcUaConfig;
+import
org.apache.streampipes.extensions.connectors.opcua.config.SpOpcUaConfigExtractor;
+import org.apache.streampipes.extensions.connectors.opcua.model.OpcNode;
import org.apache.streampipes.model.connect.guess.FieldStatusInfo;
import org.apache.streampipes.model.connect.guess.GuessSchema;
import org.apache.streampipes.model.connect.guess.GuessTypeInfo;
@@ -89,7 +89,9 @@ public class OpcUaUtil {
Map<String, FieldStatusInfo> fieldStatusInfos = new HashMap<>();
List<EventProperty> allProperties = new ArrayList<>();
- SpOpcUaClient spOpcUaClient = new
SpOpcUaClient(SpOpcUaConfigBuilder.from(extractor.getStaticPropertyExtractor()));
+ SpOpcUaClient<OpcUaConfig> spOpcUaClient = new SpOpcUaClient<>(
+
SpOpcUaConfigExtractor.extractSharedConfig(extractor.getStaticPropertyExtractor(),
new OpcUaConfig())
+ );
try {
spOpcUaClient.connect();
@@ -181,7 +183,9 @@ public class OpcUaUtil {
return config;
}
- SpOpcUaClient spOpcUaClient = new
SpOpcUaClient(SpOpcUaConfigBuilder.from(parameterExtractor));
+ SpOpcUaClient spOpcUaClient = new SpOpcUaClient(
+ SpOpcUaConfigExtractor.extractSharedConfig(parameterExtractor, new
OpcUaConfig())
+ );
try {
spOpcUaClient.connect();
diff --git
a/streampipes-extensions/streampipes-connectors-opcua/src/main/resources/org.apache.streampipes.sinks.databases.jvm.opcua/strings.en
b/streampipes-extensions/streampipes-connectors-opcua/src/main/resources/org.apache.streampipes.sinks.databases.jvm.opcua/strings.en
index 4df4e7598..7e78d16a1 100644
---
a/streampipes-extensions/streampipes-connectors-opcua/src/main/resources/org.apache.streampipes.sinks.databases.jvm.opcua/strings.en
+++
b/streampipes-extensions/streampipes-connectors-opcua/src/main/resources/org.apache.streampipes.sinks.databases.jvm.opcua/strings.en
@@ -18,17 +18,41 @@
org.apache.streampipes.sinks.databases.jvm.opcua.title=OPC-UA
org.apache.streampipes.sinks.databases.jvm.opcua.description=Writes values in
an OPC-UA server
-OPC_SERVER_HOST.title=Host Address
-OPC_SERVER_HOST.description=The host address of the OPC-UA server (Example:
opc-ua-server.com, opc.tcp://opc-ua-server.com)
+OPC_HOST_OR_URL.title=OPC Server
+OPC_HOST_OR_URL.description=
+
+OPC_URL.title=URL
+OPC_URL.description=
+
+OPC_HOST.title=Host/Port
+OPC_HOST.description=
+
+OPC_SERVER_URL.title=URL
+OPC_SERVER_URL.description=Example: opc.tcp://test-server.com:4840,
+
+OPC_SERVER_HOST.title=Host
+OPC_SERVER_HOST.description=Example: test-server.com,
opc.tcp://test-server.com)
OPC_SERVER_PORT.title=Port
-OPC_SERVER_PORT.description=The port of the OPC-UA server (Example: 4840)
+OPC_SERVER_PORT.description=Example: 4840
+
+ACCESS_MODE.title=Security Level
+ACCESS_MODE.description=Select the OPC UA security level for the connection
+
+USERNAME_GROUP.title=Sign (username & password)
+USERNAME_GROUP.description=
+
+UNAUTHENTICATED.title=None
+UNAUTHENTICATED.description=
+
+USERNAME.title=Username
+USERNAME.description=
-NAMESPACE_INDEX.title=Namespace Index
-NAMESPACE_INDEX.description=The number of the namespace
+PASSWORD.title=Password
+PASSWORD.description=
-NODE_ID.title=Node Id
-NODE_ID.description=Id of the node the result should be written to
+AVAILABLE_NODES.title=Available Nodes
+AVAILABLE_NODES.description=Select the nodes that are relevant for you. Please
ensure to select one option each for access mode and OPC UA server and to
specify the polling interval
-MAPPING_PROPERY.title=Event Property
-MAPPING_PROPERY.description=The event property that should be written to the
OPC-UA server
+MAPPING_PROPERY.title=Field to write
+MAPPING_PROPERY.description=The field that should be written to the OPC-UA
server
diff --git
a/streampipes-model/src/main/java/org/apache/streampipes/model/staticproperty/RuntimeResolvableTreeInputStaticProperty.java
b/streampipes-model/src/main/java/org/apache/streampipes/model/staticproperty/RuntimeResolvableTreeInputStaticProperty.java
index 8e3cd9099..f6d48c744 100644
---
a/streampipes-model/src/main/java/org/apache/streampipes/model/staticproperty/RuntimeResolvableTreeInputStaticProperty.java
+++
b/streampipes-model/src/main/java/org/apache/streampipes/model/staticproperty/RuntimeResolvableTreeInputStaticProperty.java
@@ -28,6 +28,7 @@ public class RuntimeResolvableTreeInputStaticProperty extends
StaticProperty {
private List<TreeInputNode> nodes;
private boolean resolveDynamically;
+ private boolean multiSelection;
private String nextBaseNodeToResolve;
private List<TreeInputNode> latestFetchedNodes;
@@ -58,6 +59,7 @@ public class RuntimeResolvableTreeInputStaticProperty extends
StaticProperty {
this.latestFetchedNodes = other.getLatestFetchedNodes();
this.selectedNodesInternalNames = other.getSelectedNodesInternalNames();
this.nextBaseNodeToResolve = other.getNextBaseNodeToResolve();
+ this.multiSelection = other.isMultiSelection();
}
@Override
@@ -112,4 +114,12 @@ public class RuntimeResolvableTreeInputStaticProperty
extends StaticProperty {
public void setSelectedNodesInternalNames(List<String>
selectedNodesInternalNames) {
this.selectedNodesInternalNames = selectedNodesInternalNames;
}
+
+ public boolean isMultiSelection() {
+ return multiSelection;
+ }
+
+ public void setMultiSelection(boolean multiSelection) {
+ this.multiSelection = multiSelection;
+ }
}
diff --git
a/streampipes-sdk/src/main/java/org/apache/streampipes/sdk/StaticProperties.java
b/streampipes-sdk/src/main/java/org/apache/streampipes/sdk/StaticProperties.java
index ecce2e722..6707617e8 100644
---
a/streampipes-sdk/src/main/java/org/apache/streampipes/sdk/StaticProperties.java
+++
b/streampipes-sdk/src/main/java/org/apache/streampipes/sdk/StaticProperties.java
@@ -137,13 +137,16 @@ public class StaticProperties {
public static RuntimeResolvableTreeInputStaticProperty
runtimeResolvableTreeInput(Label label,
List<String> dependsOn,
-
boolean resolveDynamically) {
+
boolean resolveDynamically,
+
boolean multiSelection) {
RuntimeResolvableTreeInputStaticProperty treeInput = new
RuntimeResolvableTreeInputStaticProperty(
label.getInternalId(),
label.getLabel(),
label.getDescription());
treeInput.setDependsOn(dependsOn);
+ treeInput.setResolveDynamically(resolveDynamically);
+ treeInput.setMultiSelection(multiSelection);
return treeInput;
}
diff --git
a/streampipes-sdk/src/main/java/org/apache/streampipes/sdk/builder/AbstractConfigurablePipelineElementBuilder.java
b/streampipes-sdk/src/main/java/org/apache/streampipes/sdk/builder/AbstractConfigurablePipelineElementBuilder.java
index e7d2878c6..64d7a05ae 100644
---
a/streampipes-sdk/src/main/java/org/apache/streampipes/sdk/builder/AbstractConfigurablePipelineElementBuilder.java
+++
b/streampipes-sdk/src/main/java/org/apache/streampipes/sdk/builder/AbstractConfigurablePipelineElementBuilder.java
@@ -1012,14 +1012,18 @@ public abstract class
AbstractConfigurablePipelineElementBuilder<K extends
}
public K requiredRuntimeResolvableTreeInput(Label label,
- List<String> dependsOn) {
- return requiredRuntimeResolvableTreeInput(label, dependsOn, false);
+ List<String> dependsOn,
+ boolean multiSelection) {
+ return requiredRuntimeResolvableTreeInput(label, dependsOn, false,
multiSelection);
}
public K requiredRuntimeResolvableTreeInput(Label label,
List<String> dependsOn,
- boolean resolveDynamically) {
-
this.staticProperties.add(StaticProperties.runtimeResolvableTreeInput(label,
dependsOn, resolveDynamically));
+ boolean resolveDynamically,
+ boolean multiSelection) {
+ this.staticProperties.add(
+ StaticProperties.runtimeResolvableTreeInput(label, dependsOn,
resolveDynamically, multiSelection)
+ );
return me();
}
diff --git
a/ui/projects/streampipes/platform-services/src/lib/model/gen/streampipes-model.ts
b/ui/projects/streampipes/platform-services/src/lib/model/gen/streampipes-model.ts
index ae13fdf96..c8029745c 100644
---
a/ui/projects/streampipes/platform-services/src/lib/model/gen/streampipes-model.ts
+++
b/ui/projects/streampipes/platform-services/src/lib/model/gen/streampipes-model.ts
@@ -19,7 +19,7 @@
/* tslint:disable */
/* eslint-disable */
// @ts-nocheck
-// Generated using typescript-generator version 3.1.1185 on 2023-06-12
22:17:35.
+// Generated using typescript-generator version 3.1.1185 on 2023-06-17
23:53:33.
export class NamedStreamPipesEntity {
'@class':
@@ -3140,6 +3140,7 @@ export class RuntimeResolvableTreeInputStaticProperty
extends StaticProperty {
'@class':
'org.apache.streampipes.model.staticproperty.RuntimeResolvableTreeInputStaticProperty';
'dependsOn': string[];
'latestFetchedNodes': TreeInputNode[];
+ 'multiSelection': boolean;
'nextBaseNodeToResolve': string;
'nodes': TreeInputNode[];
'resolveDynamically': boolean;
@@ -3161,6 +3162,7 @@ export class RuntimeResolvableTreeInputStaticProperty
extends StaticProperty {
instance.latestFetchedNodes = __getCopyArrayFn(TreeInputNode.fromData)(
data.latestFetchedNodes,
);
+ instance.multiSelection = data.multiSelection;
instance.nextBaseNodeToResolve = data.nextBaseNodeToResolve;
instance.nodes = __getCopyArrayFn(TreeInputNode.fromData)(data.nodes);
instance.resolveDynamically = data.resolveDynamically;
