Fix subscription sharing for revised publishing intervalsv5.11.0-rc2v5.11.0

The current implementation uses an existing subscription for
sharing if a subscription with the requested publishing interval
exists. If the publishing interval of a subscription is revised
by the server, the lookup for a second monitoring with the same
requested interval fails and a new subscription is created.

This patch adds a member variable which stores the last revised
publishing interval to each backend. Requests with a publishing
interval < last revised publishing interval are "revised" by
the backend to ensure subscription sharing.

A test case which checks for correct subscription sharing is
added to tst_client.cpp.

Change-Id: I5e433501f23bf99e841c5c41e750ed8631ea7162
Reviewed-by: Rainer Keller <Rainer.Keller@qt.io>

1 files changed, 20 insertions, 8 deletions

diff --git a/src/plugins/opcua/open62541/qopen62541backend.cpp b/src/plugins/opcua/open62541/qopen62541backend.cpp
index 15d1166..93a568a 100644
--- a/src/plugins/opcua/open62541/qopen62541backend.cpp
+++ b/src/plugins/opcua/open62541/qopen62541backend.cpp
@@ -66,6 +66,7 @@ Open62541AsyncBackend::Open62541AsyncBackend(QOpen62541Client *parent)
     , m_useStateCallback(false)
     , m_subscriptionTimer(this)
     , m_sendPublishRequests(false)
+    , m_minPublishingInterval(0)
 {
     m_subscriptionTimer.setSingleShot(true);
     QObject::connect(&m_subscriptionTimer, &QTimer::timeout,
@@ -74,7 +75,7 @@ Open62541AsyncBackend::Open62541AsyncBackend(QOpen62541Client *parent)
 
 Open62541AsyncBackend::~Open62541AsyncBackend()
 {
-    qDeleteAll(m_subscriptions);
+    cleanupSubscriptions();
     if (m_uaclient)
         UA_Client_delete(m_uaclient);
 }
@@ -280,8 +281,10 @@ void Open62541AsyncBackend::modifyMonitoring(uintptr_t handle, QOpcUa::NodeAttri
 QOpen62541Subscription *Open62541AsyncBackend::getSubscription(const QOpcUaMonitoringParameters &settings)
 {
     if (settings.shared() == QOpcUaMonitoringParameters::SubscriptionType::Shared) {
+        // Requesting multiple subscriptions with publishing interval < minimum publishing interval breaks subscription sharing
+        double interval = revisePublishingInterval(settings.publishingInterval(), m_minPublishingInterval);
         for (auto entry : qAsConst(m_subscriptions)) {
-            if (qFuzzyCompare(entry->interval(), settings.publishingInterval()) && entry->shared() == QOpcUaMonitoringParameters::SubscriptionType::Shared)
+            if (qFuzzyCompare(entry->interval(), interval) && entry->shared() == QOpcUaMonitoringParameters::SubscriptionType::Shared)
                 return entry;
         }
     }
@@ -293,6 +296,8 @@ QOpen62541Subscription *Open62541AsyncBackend::getSubscription(const QOpcUaMonit
         return nullptr;
     }
     m_subscriptions[id] = sub;
+    if (sub->interval() > settings.samplingInterval()) // The publishing interval has been revised by the server.
+        m_minPublishingInterval = sub->interval();
     // This must be a queued connection to prevent the slot from being called while the client is inside UA_Client_runAsync().
     QObject::connect(sub, &QOpen62541Subscription::timeout, this, &Open62541AsyncBackend::handleSubscriptionTimeout, Qt::QueuedConnection);
     return sub;
@@ -429,11 +434,14 @@ static void clientStateCallback(UA_Client *client, UA_ClientState state)
     if (state == UA_CLIENTSTATE_DISCONNECTED) {
         emit backend->stateAndOrErrorChanged(QOpcUaClient::Disconnected, QOpcUaClient::ConnectionError);
         backend->m_useStateCallback = false;
+        backend->cleanupSubscriptions();
     }
 }
 
 void Open62541AsyncBackend::connectToEndpoint(const QUrl &url)
 {
+    cleanupSubscriptions();
+
     if (m_uaclient)
         UA_Client_delete(m_uaclient);
 
@@ -468,9 +476,7 @@ void Open62541AsyncBackend::connectToEndpoint(const QUrl &url)
 void Open62541AsyncBackend::disconnectFromEndpoint()
 {
     m_subscriptionTimer.stop();
-    qDeleteAll(m_subscriptions);
-    m_subscriptions.clear();
-    m_attributeMapping.clear();
+    cleanupSubscriptions();
 
     m_useStateCallback = false;
 
@@ -500,9 +506,7 @@ void Open62541AsyncBackend::sendPublishRequest()
     if (UA_Client_runAsync(m_uaclient, 1) == UA_STATUSCODE_BADSERVERNOTCONNECTED) {
         qCWarning(QT_OPCUA_PLUGINS_OPEN62541) << "Unable to send publish request";
         m_sendPublishRequests = false;
-        qDeleteAll(m_subscriptions);
-        m_subscriptions.clear();
-        m_attributeMapping.clear();
+        cleanupSubscriptions();
         return;
     }
 
@@ -549,4 +553,12 @@ QOpen62541Subscription *Open62541AsyncBackend::getSubscriptionForItem(uintptr_t
     return subscription.value();
 }
 
+void Open62541AsyncBackend::cleanupSubscriptions()
+{
+    qDeleteAll(m_subscriptions);
+    m_subscriptions.clear();
+    m_attributeMapping.clear();
+    m_minPublishingInterval = 0;
+}
+
 QT_END_NAMESPACE