[Server] Durable Subscriptions

Applications/ConsoleReferenceServer/Quickstarts.ReferenceServer.Config.xml

@@ -215,6 +215,10 @@
     <MaxPublishRequestCount>20</MaxPublishRequestCount>
     <MaxSubscriptionCount>100</MaxSubscriptionCount>
     <MaxEventQueueSize>10000</MaxEventQueueSize>
+    <DurableSubscriptionsEnabled>true</DurableSubscriptionsEnabled>
+    <MaxDurableNotificationQueueSize>10000</MaxDurableNotificationQueueSize>
+    <MaxDurableEventQueueSize>10000</MaxDurableEventQueueSize>
+    <MaxDurableSubscriptionLifetimeInHours>10</MaxDurableSubscriptionLifetimeInHours>
 
     <!-- see https://opcfoundation-onlineapplications.org/profilereporting/ for list of available profiles -->
     <ServerProfileArray>

Applications/Quickstarts.Servers/DurableSubscription/DurableMonitoredItemQueue.cs

@@ -0,0 +1,112 @@
+/* ========================================================================
+ * Copyright (c) 2005-2024 The OPC Foundation, Inc. All rights reserved.
+ *
+ * OPC Foundation MIT License 1.00
+ * 
+ * Permission is hereby granted, free of charge, to any person
+ * obtaining a copy of this software and associated documentation
+ * files (the "Software"), to deal in the Software without
+ * restriction, including without limitation the rights to use,
+ * copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the
+ * Software is furnished to do so, subject to the following
+ * conditions:
+ * 
+ * The above copyright notice and this permission notice shall be
+ * included in all copies or substantial portions of the Software.
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
+ * EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES
+ * OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
+ * NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT
+ * HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY,
+ * WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
+ * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR
+ * OTHER DEALINGS IN THE SOFTWARE.
+ *
+ * The complete license agreement can be found here:
+ * http://opcfoundation.org/License/MIT/1.00/
+ * ======================================================================*/
+
+
+using Opc.Ua.Server;
+
+namespace Quickstarts.Servers
+{
+    /// <summary>
+    /// A factory for <see cref="IDataChangeMonitoredItemQueue"> and </see> <see cref="IEventMonitoredItemQueue"/>
+    /// </summary>
+    public class DurableMonitoredItemQueueFactory : IMonitoredItemQueueFactory
+    {
+        /// <inheritdoc/>
+        public bool SupportsDurableQueues => true;
+        /// <inheritdoc/>
+        public IDataChangeMonitoredItemQueue CreateDataChangeQueue(bool createDurable)
+        {
+            //use durable queue only if MI is durable
+            if (createDurable)
+            {
+
+                return new DurableDataChangeMonitoredItemQueue(createDurable);
+            }
+            else
+            {
+                return new DataChangeMonitoredItemQueue(createDurable);
+            }
+
+        }
+
+        /// <inheritdoc/>
+        public IEventMonitoredItemQueue CreateEventQueue(bool createDurable)
+        {
+            //use durable queue only if MI is durable
+            if (createDurable)
+            {
+
+                return new DurableEventMonitoredItemQueue(createDurable);
+            }
+            else
+            {
+                return new EventMonitoredItemQueue(createDurable);
+            }
+        }
+
+        /// <inheritdoc/>
+        public void Dispose()
+        {
+            //only needed for managed resources
+        }
+    }
+
+    public class DurableEventMonitoredItemQueue : EventMonitoredItemQueue
+    {
+        /// <summary>
+        /// Creates an empty queue.
+        /// </summary>
+        public DurableEventMonitoredItemQueue(bool createDurable) : base(false)
+        {
+            IsDurable = createDurable;
+        }
+
+        #region Public Methods
+        /// <inheritdoc/>
+        public override bool IsDurable { get; }
+        #endregion
+    }
+
+    public class DurableDataChangeMonitoredItemQueue : DataChangeMonitoredItemQueue
+    {
+        /// <summary>
+        /// Creates an empty queue.
+        /// </summary>
+        public DurableDataChangeMonitoredItemQueue(bool createDurable) : base(false)
+        {
+            IsDurable = createDurable;
+        }
+
+        #region Public Methods
+        /// <inheritdoc/>
+        public override bool IsDurable { get; }
+        #endregion
+    }
+
+}

Applications/Quickstarts.Servers/MemoryBuffer/MemoryBufferMonitoredItem.cs

@@ -61,7 +61,8 @@ public MemoryBufferMonitoredItem(
             double samplingInterval,
             uint queueSize,
             bool discardOldest,
-            double minimumSamplingInterval)
+            double minimumSamplingInterval,
+            bool createDurable)
         :
             base(
                 server,
@@ -80,7 +81,8 @@ public MemoryBufferMonitoredItem(
                 samplingInterval,
                 queueSize,
                 discardOldest,
-                minimumSamplingInterval)
+                minimumSamplingInterval,
+                createDurable)
         {
             m_offset = offset;
         }

Applications/Quickstarts.Servers/MemoryBuffer/MemoryBufferNodeManager.cs

@@ -268,6 +268,7 @@ protected override ServiceResult CreateMonitoredItem(
             DiagnosticsMasks diagnosticsMasks,
             TimestampsToReturn timestampsToReturn,
             MonitoredItemCreateRequest itemToCreate,
+            bool createDurable,
             ref long globalIdCounter,
             out MonitoringFilterResult filterError,
             out IMonitoredItem monitoredItem)
@@ -288,6 +289,7 @@ protected override ServiceResult CreateMonitoredItem(
                     diagnosticsMasks,
                     timestampsToReturn,
                     itemToCreate,
+                    createDurable,
                     ref globalIdCounter,
                     out filterError,
                     out monitoredItem);
@@ -366,7 +368,8 @@ protected override ServiceResult CreateMonitoredItem(
                 timestampsToReturn,
                 itemToCreate.MonitoringMode,
                 itemToCreate.RequestedParameters.ClientHandle,
-                samplingInterval);
+                samplingInterval,
+                createDurable);
 
             // report the initial value.
             datachangeItem.QueueValue(initialValue, null);

Applications/Quickstarts.Servers/MemoryBuffer/MemoryBufferState.cs

@@ -446,7 +446,8 @@ public MemoryBufferMonitoredItem CreateDataChangeItem(
             TimestampsToReturn timestampsToReturn,
             MonitoringMode monitoringMode,
             uint clientHandle,
-            double samplingInterval)
+            double samplingInterval,
+            bool createDurable)
 
         /*
         ISystemContext context,
@@ -479,7 +480,8 @@ public MemoryBufferMonitoredItem CreateDataChangeItem(
                     samplingInterval,
                     0,
                     false,
-                    0);
+                    0,
+                    createDurable);
 
                 /*
                 MemoryBufferMonitoredItem monitoredItem = new MemoryBufferMonitoredItem(

Applications/Quickstarts.Servers/ReferenceServer/ReferenceServer.cs

@@ -80,6 +80,15 @@ protected override MasterNodeManager CreateMasterNodeManager(IServerInternal ser
             return new MasterNodeManager(server, configuration, null, nodeManagers.ToArray());
         }
 
+        protected override IMonitoredItemQueueFactory CreateMonitoredItemQueueFactory(IServerInternal server, ApplicationConfiguration configuration)
+        {
+            if (configuration?.ServerConfiguration?.DurableSubscriptionsEnabled == true)
+            {
+                return new Servers.DurableMonitoredItemQueueFactory();
+            }
+            return new MonitoredItemQueueFactory();
+        }
+
         /// <summary>
         /// Loads the non-configurable properties for the application.
         /// </summary>

Applications/Quickstarts.Servers/SampleNodeManager/DataChangeMonitoredItem.cs

@@ -77,6 +77,7 @@ public DataChangeMonitoredItem(
         /// Constructs a new instance.
         /// </summary>
         public DataChangeMonitoredItem(
+            IMonitoredItemQueueFactory monitoredItemQueueFactory,
             MonitoredNode source,
             uint id,
             uint attributeId,
@@ -94,6 +95,7 @@ public DataChangeMonitoredItem(
             bool alwaysReportUpdates)
         {
             m_source = source;
+            m_monitoredItemQueueFactory = monitoredItemQueueFactory;
             m_id = id;
             m_attributeId = attributeId;
             m_indexRange = indexRange;
@@ -108,6 +110,7 @@ public DataChangeMonitoredItem(
             m_readyToTrigger = false;
             m_resendData = false;
             m_queue = null;
+            m_queueSize = queueSize;
             m_filter = filter;
             m_range = 0;
             m_alwaysReportUpdates = alwaysReportUpdates;
@@ -119,7 +122,7 @@ public DataChangeMonitoredItem(
 
             if (queueSize > 1)
             {
-                m_queue = new MonitoredItemQueue(id);
+                m_queue = new DataChangeQueueHandler(id, false, m_monitoredItemQueueFactory);
                 m_queue.SetQueueSize(queueSize, discardOldest, diagnosticsMasks);
                 m_queue.SetSamplingInterval(samplingInterval);
             }
@@ -241,6 +244,7 @@ public ServiceResult Modify(
                 m_diagnosticsMasks = diagnosticsMasks;
                 m_timestampsToReturn = timestampsToReturn;
                 m_clientHandle = clientHandle;
+                m_queueSize = queueSize;
 
                 // subtract the previous sampling interval.
                 long oldSamplingInterval = (long)(m_samplingInterval * TimeSpan.TicksPerMillisecond);
@@ -278,7 +282,7 @@ public ServiceResult Modify(
                 {
                     if (m_queue == null)
                     {
-                        m_queue = new MonitoredItemQueue(m_id);
+                        m_queue = new DataChangeQueueHandler(m_id, false, m_monitoredItemQueueFactory);
                     }
 
                     m_queue.SetQueueSize(queueSize, discardOldest, diagnosticsMasks);
@@ -497,7 +501,7 @@ public ServiceResult GetCreateResult(out MonitoredItemCreateResult result)
 
                 if (m_queue != null)
                 {
-                    result.RevisedQueueSize = m_queue.QueueSize;
+                    result.RevisedQueueSize = m_queueSize;
                 }
 
                 return ServiceResult.Good;
@@ -520,7 +524,7 @@ public ServiceResult GetModifyResult(out MonitoredItemModifyResult result)
 
                 if (m_queue != null)
                 {
-                    result.RevisedQueueSize = m_queue.QueueSize;
+                    result.RevisedQueueSize = m_queueSize;
                 }
 
                 return ServiceResult.Good;
@@ -668,6 +672,8 @@ public DataChangeFilter DataChangeFilter
             get { return m_filter; }
         }
 
+        public bool IsDurable => false;
+
         /// <summary>
         /// Increments the sample time to the next interval.
         /// </summary>
@@ -697,7 +703,7 @@ private void IncrementSampleTime()
         /// <summary>
         /// Called by the subscription to publish any notification.
         /// </summary>
-        public bool Publish(OperationContext context, Queue<MonitoredItemNotification> notifications, Queue<DiagnosticInfo> diagnostics)
+        public bool Publish(OperationContext context, Queue<MonitoredItemNotification> notifications, Queue<DiagnosticInfo> diagnostics, uint maxNotificationsPerPublish)
         {
             lock (m_lock)
             {
@@ -715,23 +721,21 @@ public bool Publish(OperationContext context, Queue<MonitoredItemNotification> n
                     IncrementSampleTime();
                 }
 
-                // update publish flag.
-                m_readyToPublish = false;
-                m_readyToTrigger = false;
 
                 // check if queuing is enabled.
                 if (m_queue != null && (!m_resendData || m_queue.ItemsInQueue != 0))
                 {
                     DataValue value = null;
                     ServiceResult error = null;
 
-                    while (m_queue.Publish(out value, out error))
+                    uint notificationCount = 0;
+                    while (notificationCount < maxNotificationsPerPublish && m_queue.PublishSingleValue(out value, out error))
                     {
                         Publish(context, value, error, notifications, diagnostics);
+                        notificationCount++;
 
                         if (m_resendData)
                         {
-                            m_readyToPublish = m_queue.ItemsInQueue > 0;
                             break;
                         }
                     }
@@ -741,10 +745,14 @@ public bool Publish(OperationContext context, Queue<MonitoredItemNotification> n
                     Publish(context, m_lastValue, m_lastError, notifications, diagnostics);
                 }
 
+                bool moreValuesToPublish = m_queue?.ItemsInQueue > 0;
+
                 // update flags
+                m_readyToPublish = moreValuesToPublish;
+                m_readyToTrigger = moreValuesToPublish;
                 m_resendData = false;
 
-                return true;
+                return moreValuesToPublish;
             }
         }
 
@@ -834,10 +842,16 @@ private void Publish(
 
             diagnostics.Enqueue(diagnosticInfo);
         }
+
+        public void Dispose()
+        {
+            //only durable queues need to be disposed
+        }
         #endregion
 
         #region Private Fields
         private readonly object m_lock = new object();
+        private IMonitoredItemQueueFactory m_monitoredItemQueueFactory;
         private MonitoredNode m_source;
         private ISubscription m_subscription;
         private uint m_id;
@@ -850,7 +864,8 @@ private void Publish(
         private DiagnosticsMasks m_diagnosticsMasks;
         private uint m_clientHandle;
         private double m_samplingInterval;
-        private MonitoredItemQueue m_queue;
+        private DataChangeQueueHandler m_queue;
+        private uint m_queueSize;
         private DataChangeFilter m_filter;
         private double m_range;
         private MonitoringMode m_monitoringMode;

Applications/Quickstarts.Servers/SampleNodeManager/MonitoredNode.cs

@@ -138,6 +138,7 @@ public DataChangeMonitoredItem CreateDataChangeItem(
             bool alwaysReportUpdates)
         {
             DataChangeMonitoredItem monitoredItem = new DataChangeMonitoredItem(
+                Server.MonitoredItemQueueFactory,
                 this,
                 monitoredItemId,
                 attributeId,