davsclaus commented on code in PR #27910:
URL: https://github.com/apache/camel/pull/27910#discussion_r4291166014
##########
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:
The await above returns as soon as `create()` installs the new
SubscriptionId, which happens before `putSubscriptions` has added and created
the monitored items. In that window `getMonitoredItems()` reports 0 (the items
were removed in `recreate()`) or 3 (all added, the failed one not yet removed),
so this assertion can fail intermittently. Waiting for the count avoids it:
```suggestion
await().atMost(10, TimeUnit.SECONDS).untilAsserted(() ->
assertEquals(2, subscription.getMonitoredItems().size()));
```
--
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]
