This is an automated email from the ASF dual-hosted git repository. dominikriemer pushed a commit to branch improve-opcua-timeout-handling in repository https://gitbox.apache.org/repos/asf/streampipes.git
commit dbc871cf1aadbc22b8961a217d106a820dcaf6ca Author: Dominik Riemer <[email protected]> AuthorDate: Wed Jul 15 23:16:00 2026 +0200 chore: Improve timeout handling of OPC UA adapter --- .../connectors/opcua/adapter/OpcUaAdapter.java | 13 +- .../opcua/alarms/OpcUaAlarmEventSubscriber.java | 32 +++- .../opcua/client/ConnectedOpcUaClient.java | 37 +++++ .../alarms/OpcUaAlarmEventSubscriberTest.java | 68 ++++++++ .../opcua/client/ConnectedOpcUaClientTest.java | 66 ++++++++ .../OpcUaAdapterReadTimeoutReproductionTest.java | 177 +++++++++++++++++++++ .../adapters/opcua/OpcUaAdapterTestHarness.java | 34 +++- .../containers/OpcUaDemoServerContainer.java | 7 +- 8 files changed, 418 insertions(+), 16 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 20799195a6..81a922af1b 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 @@ -50,7 +50,6 @@ import org.apache.streampipes.sdk.helpers.Locales; import org.eclipse.milo.opcua.sdk.client.subscriptions.OpcUaMonitoredItem; import org.eclipse.milo.opcua.stack.core.types.builtin.DataValue; import org.eclipse.milo.opcua.stack.core.types.builtin.StatusCode; -import org.eclipse.milo.opcua.stack.core.types.enumerated.TimestampsToReturn; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -176,14 +175,16 @@ public class OpcUaAdapter implements StreamPipesAdapter, IPullAdapter, SupportsR LOG.debug("Reading {} OPC UA nodes: {}", nodeIds.size(), nodeIds); var response = - this.connectedClient.getClient().readValuesAsync( - 0, - TimestampsToReturn.Both, - nodeIds); + this.connectedClient.readValuesAsync( + nodeIds, + this.getPollingInterval().timeUnit().toMillis(this.getPollingInterval().value())); boolean badStatusCodeReceived = false; boolean emptyValueReceived = false; List<DataValue> returnValues = - response.get(this.getPollingInterval().value(), this.getPollingInterval().timeUnit()); + response.get( + 2L * this.getPollingInterval().value(), + this.getPollingInterval().timeUnit() + ); if (returnValues == null) { emptyValueReceived = true; LOG.debug("Null value object returned for OPC UA nodes {} - event will not be sent", nodeIds); diff --git a/streampipes-extensions/streampipes-connectors-opcua/src/main/java/org/apache/streampipes/extensions/connectors/opcua/alarms/OpcUaAlarmEventSubscriber.java b/streampipes-extensions/streampipes-connectors-opcua/src/main/java/org/apache/streampipes/extensions/connectors/opcua/alarms/OpcUaAlarmEventSubscriber.java index 866bae83a1..ea06e69302 100644 --- a/streampipes-extensions/streampipes-connectors-opcua/src/main/java/org/apache/streampipes/extensions/connectors/opcua/alarms/OpcUaAlarmEventSubscriber.java +++ b/streampipes-extensions/streampipes-connectors-opcua/src/main/java/org/apache/streampipes/extensions/connectors/opcua/alarms/OpcUaAlarmEventSubscriber.java @@ -70,11 +70,25 @@ public class OpcUaAlarmEventSubscriber implements AutoCloseable { OpcUaAlarmEventSubscriber(ConnectedOpcUaClient connectedClient, OpcUaAlarmAdapterConfig config, Consumer<Map<String, Object>> eventConsumer) { + this( + connectedClient, + config, + eventConsumer, + OpcUaAlarmEventMapper.create(connectedClient.getClient(), config), + new OpcUaAlarmEventFilter(config) + ); + } + + OpcUaAlarmEventSubscriber(ConnectedOpcUaClient connectedClient, + OpcUaAlarmAdapterConfig config, + Consumer<Map<String, Object>> eventConsumer, + OpcUaAlarmEventMapper eventMapper, + OpcUaAlarmEventFilter eventFilter) { this.connectedClient = connectedClient; this.config = config; this.eventConsumer = eventConsumer; - this.eventMapper = OpcUaAlarmEventMapper.create(connectedClient.getClient(), config); - this.eventFilter = new OpcUaAlarmEventFilter(config); + this.eventMapper = eventMapper; + this.eventFilter = eventFilter; this.sessionActivityListener = new SessionActivityListener() { @Override public void onSessionActive(org.eclipse.milo.opcua.sdk.client.UaSession session) { @@ -119,6 +133,17 @@ public class OpcUaAlarmEventSubscriber implements AutoCloseable { private void createSubscription() throws UaException { setLastSubscriptionOperation("create-subscription"); OpcUaSubscription newSubscription = createManagedSubscription(); + this.subscription = newSubscription; + + try { + initializeSubscription(newSubscription); + } catch (UaException | RuntimeException e) { + deleteSubscriptionQuietly(); + throw e; + } + } + + private void initializeSubscription(OpcUaSubscription newSubscription) throws UaException { newSubscription.setSubscriptionListener(new OpcUaSubscription.SubscriptionListener() { @Override public void onKeepAliveReceived(OpcUaSubscription subscription) { @@ -196,14 +221,13 @@ public class OpcUaAlarmEventSubscriber implements AutoCloseable { results.get(0).operationResult().orElse(results.get(0).serviceResult()) ); - this.subscription = newSubscription; requestConditionRefresh(newSubscription); setLastSubscriptionOperation( "subscription-ready(subscriptionId=%s)".formatted(newSubscription.getSubscriptionId().orElse(null)) ); } - private OpcUaSubscription createManagedSubscription() throws UaException { + OpcUaSubscription createManagedSubscription() throws UaException { var subscription = new OpcUaSubscription(connectedClient.getClient(), PUBLISHING_INTERVAL_MS); subscription.create(); return subscription; 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 1441695cbe..a8f0c7dd06 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 @@ -25,15 +25,24 @@ import org.eclipse.milo.opcua.sdk.client.OpcUaClient; import org.eclipse.milo.opcua.sdk.client.subscriptions.MonitoredItemServiceOperationResult; import org.eclipse.milo.opcua.sdk.client.subscriptions.OpcUaMonitoredItem; import org.eclipse.milo.opcua.sdk.client.subscriptions.OpcUaSubscription; +import org.eclipse.milo.opcua.stack.core.AttributeId; 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.QualifiedName; import org.eclipse.milo.opcua.stack.core.types.builtin.StatusCode; +import org.eclipse.milo.opcua.stack.core.types.enumerated.TimestampsToReturn; +import org.eclipse.milo.opcua.stack.core.types.structured.ReadRequest; +import org.eclipse.milo.opcua.stack.core.types.structured.ReadResponse; +import org.eclipse.milo.opcua.stack.core.types.structured.ReadValueId; import org.jspecify.annotations.NonNull; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.util.ArrayList; +import java.util.Arrays; import java.util.List; +import java.util.concurrent.CompletableFuture; import static org.eclipse.milo.opcua.stack.core.types.builtin.unsigned.Unsigned.uint; @@ -122,6 +131,34 @@ public class ConnectedOpcUaClient { return this.client; } + public CompletableFuture<List<DataValue>> readValuesAsync(List<NodeId> nodeIds, + long requestTimeoutMillis) { + var readValueIds = nodeIds.stream() + .map(nodeId -> new ReadValueId( + nodeId, + AttributeId.Value.uid(), + null, + QualifiedName.NULL_VALUE + )) + .toArray(ReadValueId[]::new); + + return client.getSessionAsync().thenCompose(session -> { + var request = new ReadRequest( + client.newRequestHeader(session.getAuthenticationToken(), uint(requestTimeoutMillis)), + 0.0, + TimestampsToReturn.Both, + readValueIds + ); + + return client.sendRequestAsync(request) + .thenApply(ReadResponse.class::cast) + .thenApply(response -> { + var results = response.getResults(); + return results == null ? List.of() : Arrays.asList(results); + }); + }); + } + public void disconnect() { try { client.disconnect(); diff --git a/streampipes-extensions/streampipes-connectors-opcua/src/test/java/org/apache/streampipes/extensions/connectors/opcua/alarms/OpcUaAlarmEventSubscriberTest.java b/streampipes-extensions/streampipes-connectors-opcua/src/test/java/org/apache/streampipes/extensions/connectors/opcua/alarms/OpcUaAlarmEventSubscriberTest.java new file mode 100644 index 0000000000..16cd04f89a --- /dev/null +++ b/streampipes-extensions/streampipes-connectors-opcua/src/test/java/org/apache/streampipes/extensions/connectors/opcua/alarms/OpcUaAlarmEventSubscriberTest.java @@ -0,0 +1,68 @@ +/* + * 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.alarms; + +import org.apache.streampipes.extensions.connectors.opcua.client.ConnectedOpcUaClient; + +import org.eclipse.milo.opcua.sdk.client.OpcUaClient; +import org.eclipse.milo.opcua.sdk.client.subscriptions.OpcUaSubscription; +import org.eclipse.milo.opcua.stack.core.types.structured.EventFilter; +import org.junit.jupiter.api.Test; + +import java.util.List; + +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.mockito.Mockito.doReturn; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.spy; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +class OpcUaAlarmEventSubscriberTest { + + @Test + void deletesSubscriptionWhenMonitoredItemCreationFails() throws Exception { + var client = mock(OpcUaClient.class); + var connectedClient = mock(ConnectedOpcUaClient.class); + var config = new OpcUaAlarmAdapterConfig(); + var eventMapper = mock(OpcUaAlarmEventMapper.class); + var eventFilter = mock(OpcUaAlarmEventFilter.class); + var subscription = mock(OpcUaSubscription.class); + + when(connectedClient.getClient()).thenReturn(client); + when(eventMapper.makeEventFilter(client.getStaticEncodingContext())) + .thenReturn(mock(EventFilter.class)); + when(subscription.createMonitoredItems()).thenReturn(List.of()); + + var subscriber = spy(new OpcUaAlarmEventSubscriber( + connectedClient, + config, + event -> { }, + eventMapper, + eventFilter + )); + doReturn(subscription).when(subscriber).createManagedSubscription(); + + assertThrows(org.eclipse.milo.opcua.stack.core.UaException.class, subscriber::start); + subscriber.close(); + + verify(subscription, times(1)).delete(); + } +} diff --git a/streampipes-extensions/streampipes-connectors-opcua/src/test/java/org/apache/streampipes/extensions/connectors/opcua/client/ConnectedOpcUaClientTest.java b/streampipes-extensions/streampipes-connectors-opcua/src/test/java/org/apache/streampipes/extensions/connectors/opcua/client/ConnectedOpcUaClientTest.java new file mode 100644 index 0000000000..554c14a348 --- /dev/null +++ b/streampipes-extensions/streampipes-connectors-opcua/src/test/java/org/apache/streampipes/extensions/connectors/opcua/client/ConnectedOpcUaClientTest.java @@ -0,0 +1,66 @@ +/* + * 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.client; + +import org.eclipse.milo.opcua.sdk.client.OpcUaClient; +import org.eclipse.milo.opcua.sdk.client.OpcUaSession; +import org.eclipse.milo.opcua.stack.core.types.UaResponseMessageType; +import org.eclipse.milo.opcua.stack.core.types.builtin.NodeId; +import org.eclipse.milo.opcua.stack.core.types.structured.ReadRequest; +import org.eclipse.milo.opcua.stack.core.types.structured.RequestHeader; +import org.junit.jupiter.api.Test; +import org.mockito.ArgumentCaptor; + +import java.util.List; +import java.util.concurrent.CompletableFuture; + +import static org.eclipse.milo.opcua.stack.core.types.builtin.unsigned.Unsigned.uint; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertSame; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +class ConnectedOpcUaClientTest { + + @Test + void usesPullTimeoutAsRequestTimeoutHint() { + var client = mock(OpcUaClient.class); + var session = mock(OpcUaSession.class); + var authenticationToken = new NodeId(1, "authentication-token"); + var requestHeader = mock(RequestHeader.class); + var responseFuture = new CompletableFuture<UaResponseMessageType>(); + + when(session.getAuthenticationToken()).thenReturn(authenticationToken); + when(client.getSessionAsync()).thenReturn(CompletableFuture.completedFuture(session)); + when(client.newRequestHeader(authenticationToken, uint(250))).thenReturn(requestHeader); + when(client.sendRequestAsync(any())).thenReturn(responseFuture); + + var connectedClient = new ConnectedOpcUaClient(client); + connectedClient.readValuesAsync(List.of(new NodeId(2, "value")), 250); + + var requestCaptor = ArgumentCaptor.forClass(org.eclipse.milo.opcua.stack.core.types.UaRequestMessageType.class); + verify(client).sendRequestAsync(requestCaptor.capture()); + + var request = (ReadRequest) requestCaptor.getValue(); + assertSame(requestHeader, request.getRequestHeader()); + assertEquals(1, request.getNodesToRead().length); + } +} diff --git a/streampipes-integration-tests/src/test/java/org/apache/streampipes/integration/adapters/OpcUaAdapterReadTimeoutReproductionTest.java b/streampipes-integration-tests/src/test/java/org/apache/streampipes/integration/adapters/OpcUaAdapterReadTimeoutReproductionTest.java new file mode 100644 index 0000000000..beb54341fa --- /dev/null +++ b/streampipes-integration-tests/src/test/java/org/apache/streampipes/integration/adapters/OpcUaAdapterReadTimeoutReproductionTest.java @@ -0,0 +1,177 @@ +/* + * 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.integration.adapters; + +import org.apache.streampipes.extensions.api.monitoring.SpMonitoringManager; +import org.apache.streampipes.extensions.connectors.opcua.client.OpcUaClientProvider; +import org.apache.streampipes.integration.adapters.opcua.OpcUaAdapterTestHarness; +import org.apache.streampipes.integration.adapters.opcua.OpcUaAdapterTestHarness.RunningOpcUaAdapter; +import org.apache.streampipes.integration.containers.OpcUaDemoServerContainer; + +import org.eclipse.milo.opcua.sdk.client.OpcUaClient; +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.NodeId; +import org.eclipse.milo.opcua.stack.core.types.builtin.Variant; +import org.eclipse.milo.opcua.stack.core.types.enumerated.MessageSecurityMode; +import org.eclipse.milo.opcua.stack.core.types.structured.EndpointDescription; +import org.junit.jupiter.api.AfterAll; +import org.junit.jupiter.api.Assumptions; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Test; + +import java.lang.management.ManagementFactory; +import java.time.Duration; +import java.time.Instant; +import java.util.ArrayList; +import java.util.List; + +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * Opt-in load test for OPC UA pull reads whose server-side processing exceeds the request timeout. + */ +public class OpcUaAdapterReadTimeoutReproductionTest { + + private static final String ENABLE_PROPERTY = "streampipes.opcua.heap-reproduction"; + private static final String IMAGE_PROPERTY = "streampipes.opcua.demo.image"; + private static final String DELAYED_VALUE_NODE = + "ns=2;s=Demo.StreamPipesTestCases.ReadFailures.DelayedValue"; + private static final NodeId DELAY_ENABLED_NODE = + NodeId.parse("ns=2;s=Demo.StreamPipesTestCases.ReadFailures.DelayEnabled"); + + private static OpcUaDemoServerContainer opcUaContainer; + + @BeforeAll + public static void startContainer() { + Assumptions.assumeTrue( + Boolean.getBoolean(ENABLE_PROPERTY), + "Enable explicitly with -D" + ENABLE_PROPERTY + "=true" + ); + String image = System.getProperty(IMAGE_PROPERTY, "opc-ua-demo-server:heap-repro"); + opcUaContainer = new OpcUaDemoServerContainer(image); + opcUaContainer.start(); + } + + @AfterAll + public static void stopContainer() { + if (opcUaContainer != null) { + opcUaContainer.stop(); + } + } + + @Test + public void expiresDelayedReadsAtTransportLevel() throws Exception { + int adapterCount = Integer.getInteger("streampipes.opcua.reproduction.adapters", 25); + int pollingIntervalMillis = + Integer.getInteger("streampipes.opcua.reproduction.poll-interval-ms", 100); + int durationSeconds = Integer.getInteger("streampipes.opcua.reproduction.duration-seconds", 120); + + var harness = new OpcUaAdapterTestHarness(); + var clientProvider = new OpcUaClientProvider(); + var adapters = new ArrayList<RunningOpcUaAdapter>(); + + try { + for (int i = 0; i < adapterCount; i++) { + String adapterId = "opcua-timeout-reproduction-" + i; + adapters.add( + harness.startPullAdapter( + clientProvider, + opcUaContainer.getEndpointUrl(), + List.of(DELAYED_VALUE_NODE), + pollingIntervalMillis, + adapterId + ) + ); + } + + enableDelayedReads(); + sampleHeapAndWait(Duration.ofSeconds(durationSeconds)); + + var monitoring = SpMonitoringManager.INSTANCE.getMonitoringInfo().getLogInfos(); + long adaptersWithTimeout = monitoring.entrySet().stream() + .filter(entry -> entry.getKey().startsWith("opcua-timeout-reproduction-")) + .filter(entry -> entry.getValue().stream().anyMatch(logEntry -> + logEntry.getErrorMessage().getFullStackTrace().contains("Bad_Timeout"))) + .count(); + + assertTrue( + adaptersWithTimeout == adapterCount, + () -> "Expected timeout errors for all adapters, but found " + + adaptersWithTimeout + " of " + adapterCount + ); + } finally { + for (int i = adapters.size() - 1; i >= 0; i--) { + adapters.get(i).close(); + } + } + } + + private void enableDelayedReads() throws Exception { + OpcUaClient controlClient = OpcUaClient.create( + opcUaContainer.getEndpointUrl(), + endpoints -> endpoints.stream() + .filter(endpoint -> endpoint.getSecurityMode() == MessageSecurityMode.None) + .filter(endpoint -> SecurityPolicy.None.getUri().equals(endpoint.getSecurityPolicyUri())) + .findFirst() + .map(this::useMappedEndpoint), + transportConfig -> { }, + clientConfig -> { } + ); + try { + controlClient.connect(); + var statuses = controlClient.writeValues( + List.of(DELAY_ENABLED_NODE), + List.of(DataValue.valueOnly(Variant.ofBoolean(true))) + ); + assertTrue(statuses.get(0).isGood(), "Could not enable delayed reads in demo server"); + } finally { + controlClient.disconnect(); + } + } + + private EndpointDescription useMappedEndpoint(EndpointDescription endpoint) { + return new EndpointDescription( + opcUaContainer.getEndpointUrl(), + endpoint.getServer(), + endpoint.getServerCertificate(), + endpoint.getSecurityMode(), + endpoint.getSecurityPolicyUri(), + endpoint.getUserIdentityTokens(), + endpoint.getTransportProfileUri(), + endpoint.getSecurityLevel() + ); + } + + private void sampleHeapAndWait(Duration duration) throws InterruptedException { + var memoryBean = ManagementFactory.getMemoryMXBean(); + Instant deadline = Instant.now().plus(duration); + + while (Instant.now().isBefore(deadline)) { + var heap = memoryBean.getHeapMemoryUsage(); + System.out.printf( + "OPC-UA heap regression: used=%d MiB, committed=%d MiB, max=%d MiB%n", + heap.getUsed() / 1024 / 1024, + heap.getCommitted() / 1024 / 1024, + heap.getMax() / 1024 / 1024 + ); + Thread.sleep(1000); + } + } +} diff --git a/streampipes-integration-tests/src/test/java/org/apache/streampipes/integration/adapters/opcua/OpcUaAdapterTestHarness.java b/streampipes-integration-tests/src/test/java/org/apache/streampipes/integration/adapters/opcua/OpcUaAdapterTestHarness.java index 4522f2af5d..d332ddf310 100644 --- a/streampipes-integration-tests/src/test/java/org/apache/streampipes/integration/adapters/opcua/OpcUaAdapterTestHarness.java +++ b/streampipes-integration-tests/src/test/java/org/apache/streampipes/integration/adapters/opcua/OpcUaAdapterTestHarness.java @@ -53,7 +53,7 @@ public class OpcUaAdapterTestHarness { public Map<String, Object> readSingleEvent(String endpointUrl, List<String> selectedNodeIds) throws Exception { var collectorQueue = new LinkedBlockingQueue<Map<String, Object>>(); - var extractor = makeExtractor(endpointUrl, selectedNodeIds); + var extractor = makeExtractor(endpointUrl, selectedNodeIds, 1000, "opcua-adapter-it"); var runtimeContext = makeRuntimeContext(); var adapter = new OpcUaAdapter(new OpcUaClientProvider()); var started = false; @@ -72,7 +72,22 @@ public class OpcUaAdapterTestHarness { } } - private IAdapterParameterExtractor makeExtractor(String endpointUrl, List<String> selectedNodeIds) { + public RunningOpcUaAdapter startPullAdapter(OpcUaClientProvider clientProvider, + String endpointUrl, + List<String> selectedNodeIds, + int pollingIntervalMillis, + String adapterId) throws Exception { + var extractor = makeExtractor(endpointUrl, selectedNodeIds, pollingIntervalMillis, adapterId); + var runtimeContext = makeRuntimeContext(); + var adapter = new OpcUaAdapter(clientProvider); + adapter.onAdapterStarted(extractor, event -> { }, runtimeContext); + return new RunningOpcUaAdapter(adapter, extractor, runtimeContext); + } + + private IAdapterParameterExtractor makeExtractor(String endpointUrl, + List<String> selectedNodeIds, + int pollingIntervalMillis, + String adapterId) { IStaticPropertyExtractor staticExtractor = mock(IStaticPropertyExtractor.class); when(staticExtractor.selectedAlternativeInternalId(ADAPTER_TYPE.name())) @@ -90,7 +105,7 @@ public class OpcUaAdapterTestHarness { when(staticExtractor.singleValueParameter(OPC_SERVER_URL.name(), String.class)) .thenReturn(endpointUrl); when(staticExtractor.singleValueParameter(PULLING_INTERVAL.name(), Integer.class)) - .thenReturn(1000); + .thenReturn(pollingIntervalMillis); when(staticExtractor.selectedSingleValueInternalName( SharedUserConfiguration.INCOMPLETE_EVENT_HANDLING_KEY, String.class @@ -99,7 +114,7 @@ public class OpcUaAdapterTestHarness { .thenReturn(OpcUaNamingStrategy.DISPLAY_NAME.name()); AdapterDescription adapterDescription = new AdapterDescription(); - adapterDescription.setElementId("opcua-adapter-it"); + adapterDescription.setElementId(adapterId); IAdapterParameterExtractor extractor = mock(IAdapterParameterExtractor.class); when(extractor.getStaticPropertyExtractor()).thenReturn(staticExtractor); @@ -113,5 +128,14 @@ public class OpcUaAdapterTestHarness { when(runtimeContext.getStreamPipesClient()).thenReturn(streamPipesClient); return runtimeContext; } -} + public record RunningOpcUaAdapter(OpcUaAdapter adapter, + IAdapterParameterExtractor extractor, + IAdapterRuntimeContext runtimeContext) implements AutoCloseable { + + @Override + public void close() throws Exception { + adapter.onAdapterStopped(extractor, runtimeContext); + } + } +} diff --git a/streampipes-integration-tests/src/test/java/org/apache/streampipes/integration/containers/OpcUaDemoServerContainer.java b/streampipes-integration-tests/src/test/java/org/apache/streampipes/integration/containers/OpcUaDemoServerContainer.java index 50ab2dd932..e3c9e7fcec 100644 --- a/streampipes-integration-tests/src/test/java/org/apache/streampipes/integration/containers/OpcUaDemoServerContainer.java +++ b/streampipes-integration-tests/src/test/java/org/apache/streampipes/integration/containers/OpcUaDemoServerContainer.java @@ -27,9 +27,14 @@ import java.time.Duration; public class OpcUaDemoServerContainer extends GenericContainer<OpcUaDemoServerContainer> { public static final int OPC_UA_PORT = 4840; + public static final String DEFAULT_IMAGE = "digitalpetri/opc-ua-demo-server:latest"; public OpcUaDemoServerContainer() { - super(DockerImageName.parse("digitalpetri/opc-ua-demo-server:latest")); + this(DEFAULT_IMAGE); + } + + public OpcUaDemoServerContainer(String imageName) { + super(DockerImageName.parse(imageName)); } @Override
