Add async API for data change subscriptions

Subscriptions and monitored items are now handled entirely in the backend,
no objects are passed to the user.

Monitoring of attributes is activated using enableMonitoring() on the node.
Different attributes of the node can be monitored with different intervals.
Subscriptions are shared by default. Parameters allow requesting the creation
of an exclusive subscription or to assign monitored items for attributes to
an existing subscription. disableMonitoring() removes a monitored item.

The enableMonitoringFinished and disableMonitoringFinished signals inform
the user about the success of adding and removing monitored items.

Data changes are reported by the attributeUpdated signal, this also updates
the attribute cache.

Monitoring parameters can be modified using modifyMonitoring(). Changed
settings are reported by the monitoringStatusChanged signal, the value can be
checked by calling monitoringStatus() for an attribute.

Modifying monitoring parameters is currently not supported in freeopcua.
Open62541 allows modifying subscription settings, monitored items can't be
modified.

Change-Id: I3ba5b3f03bd9f0413c98feefa9ad9571f82e8f0d
Reviewed-by: Frank Meerkoetter <frank.meerkoetter@basyskom.com>

1 files changed, 140 insertions, 107 deletions

diff --git a/src/plugins/opcua/freeopcua/qfreeopcuasubscription.cpp b/src/plugins/opcua/freeopcua/qfreeopcuasubscription.cpp
index 487b3ed..f89ae57 100644
--- a/src/plugins/opcua/freeopcua/qfreeopcuasubscription.cpp
+++ b/src/plugins/opcua/freeopcua/qfreeopcuasubscription.cpp
@@ -35,13 +35,9 @@
 ****************************************************************************/
 
 #include "qfreeopcuaclient.h"
-#include "qfreeopcuanode.h"
 #include "qfreeopcuasubscription.h"
 #include "qfreeopcuavalueconverter.h"
-#include <QtOpcUa/qopcuamonitoredvalue.h>
-#include <QtOpcUa/qopcuasubscription.h>
-#include <private/qopcuamonitoredevent_p.h>
-#include <private/qopcuamonitoredvalue_p.h>
+#include "qfreeopcuaworker.h"
 #include <private/qopcuanode_p.h>
 
 #include <QtCore/qloggingcategory.h>
@@ -50,157 +46,194 @@ QT_BEGIN_NAMESPACE
 
 Q_DECLARE_LOGGING_CATEGORY(QT_OPCUA_PLUGINS_FREEOPCUA)
 
-QFreeOpcUaSubscription::QFreeOpcUaSubscription(OpcUa::UaClient *client, quint32 interval)
-    : m_client(client)
+QFreeOpcUaSubscription::QFreeOpcUaSubscription(QFreeOpcUaWorker *backend, const QOpcUaMonitoringParameters &settings)
+    : m_interval(settings.publishingInterval())
+    , m_shared(settings.shared())
+    , m_backend(backend)
 {
-    Q_ASSERT(m_client);
+    Q_ASSERT(m_backend);
+}
+
+QFreeOpcUaSubscription::~QFreeOpcUaSubscription()
+{
+    removeOnServer();
+}
+
+quint32 QFreeOpcUaSubscription::createOnServer()
+{
+    if (m_subscription)
+        return 0;
 
     try {
-        m_subscription = m_client->CreateSubscription(interval, *this);
+        m_subscription = m_backend->CreateSubscription(m_interval, *this);
+        m_interval = m_subscription->GetData().RevisedPublishingInterval;
     } catch (const std::exception &ex) {
-        qCWarning(QT_OPCUA_PLUGINS_FREEOPCUA, "Caught: %s", ex.what());
+        qCWarning(QT_OPCUA_PLUGINS_FREEOPCUA, "CreateOnServer caught: %s", ex.what());
+        return 0;
     }
+    return m_subscription->GetId();
 }
 
-QFreeOpcUaSubscription::~QFreeOpcUaSubscription()
+bool QFreeOpcUaSubscription::removeOnServer()
 {
+    bool success = false;
     try {
         if (m_subscription) {
-            for (int32_t handle : m_dataChangeHandles.keys())
-                m_subscription->UnSubscribe(handle);
-
-            for (int32_t handle : m_eventHandles.keys())
-                m_subscription->UnSubscribe(handle);
-
             m_subscription->Delete();
+            m_subscription.reset();
+            success = true;
         }
     } catch (const std::exception &ex) {
-        qCWarning(QT_OPCUA_PLUGINS_FREEOPCUA, "Caught: %s", ex.what());
+        qCWarning(QT_OPCUA_PLUGINS_FREEOPCUA, "RemoveOnServer caught: %s", ex.what());
     }
+
+    qDeleteAll(m_itemIdToItemMapping);
+
+    m_itemIdToItemMapping.clear();
+    m_handleToItemMapping.clear();
+
+    return success;
 }
 
-void QFreeOpcUaSubscription::DataChange(quint32 handle, const OpcUa::Node &node,
-                                        const OpcUa::Variant &val,
-                                        OpcUa::AttributeId attr)
+void QFreeOpcUaSubscription::modifyMonitoring(uintptr_t handle, QOpcUaNode::NodeAttribute attr, QOpcUaMonitoringParameters::Parameter item, QVariant value)
 {
-    OPCUA_UNUSED(node);
-    OPCUA_UNUSED(attr);
-
-    auto it = m_dataChangeHandles.find(handle);
-    if (it == m_dataChangeHandles.end()) {
-        qCWarning(QT_OPCUA_PLUGINS_FREEOPCUA, "Received event for unknown handle: %ul", handle);
+    Q_UNUSED(item);
+    Q_UNUSED(value);
+
+    if (!getItemForAttribute(handle, attr)) {
+        qCWarning(QT_OPCUA_PLUGINS_FREEOPCUA, "Could not modify parameter for %lu, there are no monitored items", handle);
+        QOpcUaMonitoringParameters p;
+        p.setStatusCode(QOpcUa::UaStatusCode::BadAttributeIdInvalid);
+        emit m_backend->monitoringStatusChanged(handle, attr, item, p);
         return;
     }
 
-    try {
-        (*it)->d_func()->triggerValueChanged(QFreeOpcUaValueConverter::toQVariant(val));
-    } catch (const std::exception &ex) {
-        qCWarning(QT_OPCUA_PLUGINS_FREEOPCUA, "Caught: %s", ex.what());
-    }
+    QOpcUaMonitoringParameters s;
+    s.setStatusCode(QOpcUa::UaStatusCode::BadNotImplemented);
+    emit m_backend->monitoringEnableDisable(handle, attr, true, s);
 }
 
-QOpcUaMonitoredValue *QFreeOpcUaSubscription::addValue(QOpcUaNode *node)
+bool QFreeOpcUaSubscription::addAttributeMonitoredItem(uintptr_t handle, QOpcUaNode::NodeAttribute attr, const OpcUa::Node &node, QOpcUaMonitoringParameters settings)
 {
-    if (!m_subscription)
-        return nullptr;
+    Q_UNUSED(settings); // Setting these options is not yet supported
 
-    try {
-        // Only add a monitored item if the node has a value attribute
-        QFreeOpcUaNode *nnode = static_cast<QFreeOpcUaNode *>(node->d_func()->m_impl.data());
-        if (nnode->m_node.GetAttribute(OpcUa::AttributeId::Value).Status != OpcUa::StatusCode::Good)
-            return nullptr;
+    quint32 monitoredItemId;
 
+    try {
         if (m_subscription) {
-            uint32_t handle = m_subscription->SubscribeDataChange(m_client->GetNode(node->nodeId().toStdString()));
-            QOpcUaMonitoredValue *monitoredValue = new QOpcUaMonitoredValue(node, m_qsubscription);
-            m_dataChangeHandles[handle] = monitoredValue;
-            return monitoredValue;
+            monitoredItemId = m_subscription->SubscribeDataChange(node, QFreeOpcUaValueConverter::toUaAttributeId(attr));
+        }
+        else {
+            QOpcUaMonitoringParameters s;
+            s.setStatusCode(QOpcUa::UaStatusCode::BadSubscriptionIdInvalid);
+            emit m_backend->monitoringEnableDisable(handle, attr, true, s);
+            return false;
         }
     } catch (const std::exception &ex) {
-        qCWarning(QT_OPCUA_PLUGINS_FREEOPCUA, "Caught: %s", ex.what());
+        qCWarning(QT_OPCUA_PLUGINS_FREEOPCUA, "addAttributeMonitoredItem caught: %s", ex.what());
+        QOpcUaMonitoringParameters s;
+        s.setStatusCode(QFreeOpcUaValueConverter::exceptionToStatusCode(ex));
+        emit m_backend->monitoringEnableDisable(handle, attr, true, s);
+        return false;
     }
-    return nullptr;
+
+    MonitoredItem *temp = new MonitoredItem(handle, attr, monitoredItemId);
+    m_handleToItemMapping[handle][attr] = temp;
+    m_itemIdToItemMapping[monitoredItemId] = temp;
+
+    QOpcUaMonitoringParameters s;
+    s.setSubscriptionId(m_subscription->GetId());
+    s.setPublishingInterval(m_interval);
+    s.setMaxKeepAliveCount(m_subscription->GetData().RevisedMaxKeepAliveCount);
+    s.setLifetimeCount(m_subscription->GetData().RevisedLifetimeCount);
+    s.setStatusCode(QOpcUa::UaStatusCode::Good);
+    s.setSamplingInterval(m_interval);
+    emit m_backend->monitoringEnableDisable(handle, attr, true, s);
+
+    return true;
 }
 
-void QFreeOpcUaSubscription::Event(quint32 handle, const OpcUa::Event &event)
+bool QFreeOpcUaSubscription::removeAttributeMonitoredItem(uintptr_t handle, QOpcUaNode::NodeAttribute attr)
 {
-    auto it = m_eventHandles.find(handle);
-    if (it == m_eventHandles.end()) {
-        qCWarning(QT_OPCUA_PLUGINS_FREEOPCUA) << "Received event for unknown handle:" << handle;
-        return;
+    QScopedPointer<MonitoredItem> item(getItemForAttribute(handle, attr));
+
+    if (!item) {
+        qCWarning(QT_OPCUA_PLUGINS_FREEOPCUA, "There is no monitored item for this attribute");
+        QOpcUaMonitoringParameters s;
+        s.setStatusCode(QOpcUa::UaStatusCode::BadMonitoredItemIdInvalid);
+        emit m_backend->monitoringEnableDisable(handle, attr, false, s);
+        return false;
     }
 
-    try {
-        QVector<QVariant> val;
-        val.reserve(3);
-
-        val.push_back(QVariant(QString::fromStdString(event.Message.Text)));
-        val.push_back(QVariant(QString::fromStdString(event.SourceName)));
-        val.push_back(QVariant(event.Severity));
+    QOpcUaMonitoringParameters s;
 
-        QOpcUaMonitoredEvent *me = *it;
-        me->d_func()->triggerNewEvent(val);
+    try {
+        m_subscription->UnSubscribe(item->monitoredItemId);
+        s.setStatusCode(QOpcUa::UaStatusCode::Good);
     } catch (const std::exception &ex) {
-        qCWarning(QT_OPCUA_PLUGINS_FREEOPCUA, "Caught: %s", ex.what());
+        qCWarning(QT_OPCUA_PLUGINS_FREEOPCUA, "removeAttributeMonitoredItem caught: %s", ex.what());
+        s.setStatusCode(QFreeOpcUaValueConverter::exceptionToStatusCode(ex));
     }
+
+    m_itemIdToItemMapping.remove(item->monitoredItemId);
+    auto it = m_handleToItemMapping.find(handle);
+    it->remove(attr);
+    if (it->empty())
+        m_handleToItemMapping.erase(it);
+
+    emit m_backend->monitoringEnableDisable(handle, attr, false, s);
+
+    return true;
 }
 
-QOpcUaMonitoredEvent *QFreeOpcUaSubscription::addEvent(QOpcUaNode *node)
+double QFreeOpcUaSubscription::interval() const
 {
-    // Note: Callback is not called due to some error in the event implementation in freeopcua
-    if (!m_subscription)
-        return nullptr;
+    return m_interval;
+}
 
-    try {
-        QFreeOpcUaNode *nnode = static_cast<QFreeOpcUaNode *>(node->d_func()->m_impl.data());
-        if (nnode->m_node.GetAttribute(OpcUa::AttributeId::EventNotifier).Status != OpcUa::StatusCode::Good)
-            return nullptr;
+QOpcUaMonitoringParameters::SubscriptionType QFreeOpcUaSubscription::shared() const
+{
+    return m_shared;
+}
 
-        OpcUa::Node serverNode = m_client->GetNode(OpcUa::ObjectId::Server);
-        OpcUa::Node typeNode = m_client->GetNode(node->nodeId().toStdString());
+quint32 QFreeOpcUaSubscription::subscriptionId() const
+{
+    if (m_subscription)
+        return m_subscription->GetId();
+    else
+        return 0;
+}
 
-        uint32_t handle = m_subscription->SubscribeEvents(serverNode, typeNode);
-        QOpcUaMonitoredEvent *monitoredEvent = new QOpcUaMonitoredEvent(node, m_qsubscription);
-        m_eventHandles[handle] = monitoredEvent;
-        return monitoredEvent;
-    } catch (const std::exception &ex) {
-        qCWarning(QT_OPCUA_PLUGINS_FREEOPCUA, "Caught: %s", ex.what());
-    }
-    return nullptr;
+int QFreeOpcUaSubscription::monitoredItemsCount() const
+{
+    return m_itemIdToItemMapping.size();
 }
 
-void QFreeOpcUaSubscription::removeEvent(QOpcUaMonitoredEvent *event)
+QFreeOpcUaSubscription::MonitoredItem *QFreeOpcUaSubscription::getItemForAttribute(uintptr_t handle, QOpcUaNode::NodeAttribute attr)
 {
-    try {
-        auto it = m_eventHandles.begin();
-        while (it != m_eventHandles.end()) {
-            if (it.value() == event) {
-                m_subscription->UnSubscribe(it.key());
-                m_eventHandles.erase(it);
-                break;
-            }
-            ++it;
-        }
-    } catch (const std::exception &ex) {
-        qCWarning(QT_OPCUA_PLUGINS_FREEOPCUA, "Caught: %s", ex.what());
-    }
+    auto nodeEntry = m_handleToItemMapping.find(handle);
+
+    if (nodeEntry == m_handleToItemMapping.end())
+        return nullptr;
+
+    auto item = nodeEntry->find(attr);
+    if (item == nodeEntry->end())
+        return nullptr;
+
+    return item.value();
 }
 
-void QFreeOpcUaSubscription::removeValue(QOpcUaMonitoredValue *value)
+void QFreeOpcUaSubscription::DataChange(quint32 handle, const OpcUa::Node &node,
+                                        const OpcUa::Variant &val,
+                                        OpcUa::AttributeId attr)
 {
-    try {
-        auto it = m_dataChangeHandles.begin();
-        while (it != m_dataChangeHandles.end()) {
-            if (it.value() == value) {
-                m_subscription->UnSubscribe(it.key());
-                m_dataChangeHandles.erase(it);
-                break;
-            }
-            ++it;
-        }
-    } catch (const std::exception &ex) {
-        qCWarning(QT_OPCUA_PLUGINS_FREEOPCUA, "Caught: %s", ex.what());
-    }
+    OPCUA_UNUSED(node);
+    OPCUA_UNUSED(attr);
+
+    auto item = m_itemIdToItemMapping.find(handle);
+    if (item == m_itemIdToItemMapping.end())
+        return;
+    emit m_backend->attributeUpdated(item.value()->handle, item.value()->attr, QFreeOpcUaValueConverter::toQVariant(val));
 }
 
 QT_END_NAMESPACE