etzel-mes commented on code in PR #27910:
URL: https://github.com/apache/camel/pull/27910#discussion_r4296942625
##########
components/camel-milo/src/test/java/org/apache/camel/component/milo/MonitorItemLifecycleTest.java:
##########
@@ -241,6 +248,128 @@ void
failedCreateMonitoredItemsCallRebuildsTheConnection() throws Exception {
.getBody(DataValue.class)));
}
+ @Test
+ @Timeout(value = 60, unit = TimeUnit.SECONDS)
+ void subscriptionThatStopsPublishingIsRecreatedInTheSameSession() throws
Exception {
+ item1Endpoint.expectedMinimumMessageCount(1);
+ testBody(item1Endpoint.message(0), assertGoodValue("Foo"));
+ producer1.sendBody("Foo");
+ item1Endpoint.assertIsSatisfied();
+
+ final MiloClientEndpoint endpoint =
context.getEndpoint(resolve(MILO_CLIENT_ITEM_1), MiloClientEndpoint.class);
+ final MiloClientConnection connection = endpoint.createConnection();
+ try {
+ final OpcUaClient client = connection.getOpcUaClient();
+ final OpcUaSubscription subscription =
client.getSubscriptions().get(0);
+ // shortens the watchdog of the subscription to a few publishing
intervals
+ subscription.setMaxKeepAliveCount(uint(1));
+ subscription.setLifetimeCount(uint(3));
+ subscription.modify();
+
+ // the server deletes the subscription and answers the pending
publish requests without one
+ final UInteger subscriptionId =
subscription.getSubscriptionId().orElseThrow();
+
assertTrue(client.deleteSubscriptions(List.of(subscriptionId)).getResults()[0].isGood());
+
+ producer1.sendBody("Bar");
+ await().atMost(30, TimeUnit.SECONDS)
+ .untilAsserted(() ->
assertGoodValue("Bar").accept(lastExchange(item1Endpoint).getMessage()
+ .getBody(DataValue.class)));
+
+ assertSame(client, connection.getOpcUaClient());
+ assertNotEquals(subscriptionId,
subscription.getSubscriptionId().orElseThrow());
+ } finally {
+ endpoint.releaseConnection(connection);
+ }
+ }
+
+ @Test
+ @Timeout(value = 60, unit = TimeUnit.SECONDS)
+ void subscriptionThatTimesOutOnTheServerIsRecreatedInTheSameSession()
throws Exception {
+ item1Endpoint.expectedMinimumMessageCount(1);
+ testBody(item1Endpoint.message(0), assertGoodValue("Foo"));
+ producer1.sendBody("Foo");
+ item1Endpoint.assertIsSatisfied();
+
+ final MiloClientEndpoint endpoint =
context.getEndpoint(resolve(MILO_CLIENT_ITEM_1), MiloClientEndpoint.class);
+ final MiloClientConnection connection = endpoint.createConnection();
+ try {
+ final OpcUaClient client = connection.getOpcUaClient();
+ final UInteger subscriptionId =
client.getSubscriptions().get(0).getSubscriptionId().orElseThrow();
+
+ timeOutSubscription(subscriptionId);
+
+ producer1.sendBody("Bar");
+ await().atMost(30, TimeUnit.SECONDS)
+ .untilAsserted(() ->
assertGoodValue("Bar").accept(lastExchange(item1Endpoint).getMessage()
+ .getBody(DataValue.class)));
+
+ assertSame(client, connection.getOpcUaClient());
+ assertNotEquals(subscriptionId,
client.getSubscriptions().get(0).getSubscriptionId().orElseThrow());
+ } finally {
+ endpoint.releaseConnection(connection);
+ }
+ }
+
+ @Test
+ @Timeout(value = 60, unit = TimeUnit.SECONDS)
+ void
nodeThatCannotBeMonitoredAgainIsReportedWithoutRebuildingTheConnection() throws
Exception {
+ producer1.sendBody("Foo");
+ producer2.sendBody("Bar");
+
+ items12Endpoint.expectedMessageCount(2);
+ context.getRouteController().startRoute("items12");
+ items12Endpoint.assertIsSatisfied();
+
+ final MiloClientEndpoint endpoint =
context.getEndpoint(resolve(MILO_CLIENT_ITEM_1), MiloClientEndpoint.class);
+ final MiloClientConnection connection = endpoint.createConnection();
+ try {
+ final OpcUaClient client = connection.getOpcUaClient();
+ final OpcUaSubscription subscription =
client.getSubscriptions().get(0);
+ final UInteger subscriptionId =
subscription.getSubscriptionId().orElseThrow();
+
+ // the second node is removed on the server before the
subscription times out there
+
getServer().getAddressSpaceManager().getManagedNode(ExpandedNodeId.parse(NODE_ITEM_2)).orElseThrow().delete();
+ timeOutSubscription(subscriptionId);
+
+ // the subscription is recreated in the same session, with the
monitored items of the first node only
+ await().atMost(30, TimeUnit.SECONDS)
+ .until(() -> subscription.getSubscriptionId().filter(id ->
!id.equals(subscriptionId)).isPresent());
+ assertSame(client, connection.getOpcUaClient());
+ assertEquals(2, subscription.getMonitoredItems().size());
Review Comment:
Applied in 70aca32 as the condition of the await, with one addition: it also
waits for `isMonitoredItemsSynchronized()`. `putSubscriptions` adds the items
one by one before it creates them, so the count alone passes 2 on its way to 3.
_Claude Code on behalf of etzel-mes_
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
