diff --git a/Applications/Quickstarts.Servers/SampleNodeManager/DataChangeMonitoredItem.cs b/Applications/Quickstarts.Servers/SampleNodeManager/DataChangeMonitoredItem.cs index f24d3ed025..e17a44260c 100644 --- a/Applications/Quickstarts.Servers/SampleNodeManager/DataChangeMonitoredItem.cs +++ b/Applications/Quickstarts.Servers/SampleNodeManager/DataChangeMonitoredItem.cs @@ -465,6 +465,11 @@ namespace Opc.Ua.Sample } } + public void SetupResendDataTrigger() + { + // Does nothing since this type of Monitored Item does not support Resend data functionality + } + /// /// Returns the results for the create request. /// diff --git a/Applications/Quickstarts.Servers/SampleNodeManager/SampleNodeManager.cs b/Applications/Quickstarts.Servers/SampleNodeManager/SampleNodeManager.cs index 3cf7976040..93c4b49068 100644 --- a/Applications/Quickstarts.Servers/SampleNodeManager/SampleNodeManager.cs +++ b/Applications/Quickstarts.Servers/SampleNodeManager/SampleNodeManager.cs @@ -2964,20 +2964,14 @@ namespace Opc.Ua.Sample // owned by this node manager. processedItems[ii] = true; - var monitoredItem = monitoredItems[ii]; - transferredItems.Add(monitoredItem); + transferredItems.Add(monitoredItems[ii]); - if (sendInitialValues && !monitoredItem.IsReadyToPublish) + if (sendInitialValues) { - if (monitoredItem is DataChangeMonitoredItem dataChangeMonitoredItem) - { - errors[ii] = ReadInitialValue(systemContext, monitoredNode, dataChangeMonitoredItem, true); - } - } - else - { - errors[ii] = StatusCodes.Good; + monitoredItems[ii].SetupResendDataTrigger(); } + + errors[ii] = StatusCodes.Good; } } @@ -2998,6 +2992,7 @@ namespace Opc.Ua.Sample // does nothing. } + /// /// Changes the monitoring mode for a set of monitored items. /// diff --git a/Libraries/Opc.Ua.Server/Diagnostics/CustomNodeManager.cs b/Libraries/Opc.Ua.Server/Diagnostics/CustomNodeManager.cs index dd4317b175..297c0c2b5d 100644 --- a/Libraries/Opc.Ua.Server/Diagnostics/CustomNodeManager.cs +++ b/Libraries/Opc.Ua.Server/Diagnostics/CustomNodeManager.cs @@ -4234,20 +4234,14 @@ namespace Opc.Ua.Server // owned by this node manager. processedItems[ii] = true; - var monitoredItem = monitoredItems[ii]; - transferredItems.Add(monitoredItem); + transferredItems.Add(monitoredItems[ii]); - if (sendInitialValues && !monitoredItem.IsReadyToPublish) + if (sendInitialValues) { - if (monitoredItem is IDataChangeMonitoredItem2 dataChangeMonitoredItem) - { - errors[ii] = ReadInitialValue(systemContext, handle, dataChangeMonitoredItem); - } - } - else - { - errors[ii] = StatusCodes.Good; + monitoredItems[ii].SetupResendDataTrigger(); } + + errors[ii] = StatusCodes.Good; } } diff --git a/Libraries/Opc.Ua.Server/Diagnostics/DiagnosticsNodeManager.cs b/Libraries/Opc.Ua.Server/Diagnostics/DiagnosticsNodeManager.cs index 1da77d5c39..92b7de79b5 100644 --- a/Libraries/Opc.Ua.Server/Diagnostics/DiagnosticsNodeManager.cs +++ b/Libraries/Opc.Ua.Server/Diagnostics/DiagnosticsNodeManager.cs @@ -199,6 +199,17 @@ namespace Opc.Ua.Server } } #endif + + // hookup server ResendData method. + + ResendDataMethodState resendData = (ResendDataMethodState)FindPredefinedNode( + MethodIds.Server_ResendData, + typeof(ResendDataMethodState)); + + if (resendData != null) + { + resendData.OnCallMethod = OnResendData; + } } } @@ -282,6 +293,46 @@ namespace Opc.Ua.Server return StatusCodes.BadSubscriptionIdInvalid; } + /// + /// Called when a client initiates resending of all data monitored items in a Subscription. + /// + public ServiceResult OnResendData( + ISystemContext context, + MethodState method, + IList inputArguments, + IList outputArguments) + { + if (inputArguments == null || inputArguments.Count != 1) + { + return StatusCodes.BadInvalidArgument; + } + + uint? subscriptionId = inputArguments[0] as uint?; + + if (subscriptionId == null) + { + return StatusCodes.BadInvalidArgument; + } + + foreach (Subscription subscription in Server.SubscriptionManager.GetSubscriptions()) + { + if (subscription.Id == subscriptionId) + { + if (subscription.SessionId != context.SessionId) + { + // user tries to access subscription of different session + return StatusCodes.BadUserAccessDenied; + } + + subscription.ResendData((OperationContext)((SystemContext)context)?.OperationContext); + + return ServiceResult.Good; + } + } + + return StatusCodes.BadSubscriptionIdInvalid; + } + /// /// Called when a client locks the server. /// diff --git a/Libraries/Opc.Ua.Server/NodeManager/CoreNodeManager.cs b/Libraries/Opc.Ua.Server/NodeManager/CoreNodeManager.cs index 425c949939..cc331258cb 100644 --- a/Libraries/Opc.Ua.Server/NodeManager/CoreNodeManager.cs +++ b/Libraries/Opc.Ua.Server/NodeManager/CoreNodeManager.cs @@ -28,14 +28,7 @@ * ======================================================================*/ using System; -using System.Collections; using System.Collections.Generic; -using System.Text; -using System.Xml; -using System.Threading; -using System.Reflection; -using System.IO; -using System.Diagnostics; using System.Threading.Tasks; #pragma warning disable 0618 @@ -69,8 +62,8 @@ namespace Opc.Ua.Server m_monitoredItems = new Dictionary(); m_defaultMinimumSamplingInterval = 1000; m_namespaceUris = new List(); - m_dynamicNamespaceIndex = dynamicNamespaceIndex; - + m_dynamicNamespaceIndex = dynamicNamespaceIndex; + // use namespace 1 if out of range. if (m_dynamicNamespaceIndex == 0 || m_dynamicNamespaceIndex >= server.NamespaceUris.Count) { @@ -1512,7 +1505,7 @@ namespace Opc.Ua.Server IList processedItems, IList errors) { - if (context == null) throw new ArgumentNullException(nameof(context)); + if (context == null) throw new ArgumentNullException(nameof(context)); if (monitoredItems == null) throw new ArgumentNullException(nameof(monitoredItems)); if (processedItems == null) throw new ArgumentNullException(nameof(processedItems)); @@ -1525,7 +1518,7 @@ namespace Opc.Ua.Server { continue; } - + // check if the node manager created the item. if (!Object.ReferenceEquals(this, monitoredItems[ii].NodeManager)) { @@ -1533,7 +1526,7 @@ namespace Opc.Ua.Server } // owned by this node manager. - processedItems[ii] = true; + processedItems[ii] = true; // validate monitored item. IMonitoredItem monitoredItem = monitoredItems[ii]; @@ -1545,17 +1538,12 @@ namespace Opc.Ua.Server continue; } - if (sendInitialValues && !monitoredItem.IsReadyToPublish) + if (sendInitialValues) { - if (monitoredItem is IDataChangeMonitoredItem2 dataChangeMonitoredItem) - { - errors[ii] = ReadInitialValue(context, node, dataChangeMonitoredItem); - } - } - else - { - errors[ii] = StatusCodes.Good; + monitoredItem.SetupResendDataTrigger(); } + + errors[ii] = StatusCodes.Good; } } } diff --git a/Libraries/Opc.Ua.Server/Subscription/IMonitoredItem.cs b/Libraries/Opc.Ua.Server/Subscription/IMonitoredItem.cs index 12bd6dfa42..e0b7218017 100644 --- a/Libraries/Opc.Ua.Server/Subscription/IMonitoredItem.cs +++ b/Libraries/Opc.Ua.Server/Subscription/IMonitoredItem.cs @@ -35,6 +35,8 @@ using System.Threading; namespace Opc.Ua.Server { + + /// /// Manages a monitored item created by a client. /// @@ -94,6 +96,11 @@ namespace Opc.Ua.Server /// bool IsReadyToTrigger { get; set; } + /// + /// Setup the resend data trigger by setting the monitor item in ResendData state + /// + void SetupResendDataTrigger(); + /// /// Returns the result after creating the monitor item. /// diff --git a/Libraries/Opc.Ua.Server/Subscription/MonitoredItem.cs b/Libraries/Opc.Ua.Server/Subscription/MonitoredItem.cs index 3396624a90..92700a89e0 100644 --- a/Libraries/Opc.Ua.Server/Subscription/MonitoredItem.cs +++ b/Libraries/Opc.Ua.Server/Subscription/MonitoredItem.cs @@ -30,10 +30,23 @@ using System; using System.Collections.Generic; using System.Globalization; +using System.Threading; using System.Xml; namespace Opc.Ua.Server { + /// + /// Specifies the values of the ResendData state for a monitored item + /// The state is set on the ResendData method call + /// + internal enum ResendDataState : int + { + ///The Monitored item does not participate in the ResendData + NonResendData = 0, + ///The Monitored item participates in ResendData + ResendData = 1 + } + /// /// A handle that describes how to access a node/attribute via an i/o manager. /// @@ -194,6 +207,7 @@ namespace Opc.Ua.Server m_readyToTrigger = false; m_sourceSamplingInterval = 0; m_samplingError = ServiceResult.Good; + m_resendDataState = (int)ResendDataState.NonResendData; } #endregion @@ -238,8 +252,12 @@ namespace Opc.Ua.Server } } - // check if not ready to publish. - if (!m_readyToPublish) + bool isResendData = (int)ResendDataState.ResendData == Interlocked.CompareExchange(ref m_resendDataState, + (int)ResendDataState.ResendData, + (int)ResendDataState.ResendData); + + // check if not ready to publish in case it doesn't ResendData + if (!m_readyToPublish && (!isResendData)) { ServerUtils.EventLog.MonitoredItemReady(m_id, "FALSE"); return false; @@ -261,10 +279,10 @@ namespace Opc.Ua.Server if (m_sourceSamplingInterval == 0) { - // re-queue if too little time has passed since the last publish. + // re-queue if too little time has passed since the last publish, in case it doesn't ResendData long now = HiResClock.TickCount64; - if (m_nextSamplingTime > now) + if ((m_nextSamplingTime > now) && !isResendData) { ServerUtils.EventLog.MonitoredItemReady(m_id, Utils.Format("FALSE {0}ms", m_nextSamplingTime - now)); return false; @@ -303,6 +321,14 @@ namespace Opc.Ua.Server } } + /// + /// Setup the resend data trigger by setting the monitor item in ResendData state + /// + public void SetupResendDataTrigger() + { + Interlocked.Exchange(ref m_resendDataState, (int)ResendDataState.ResendData); + } + /// /// Sets a flag indicating that the item has been triggered and should publish. /// @@ -881,7 +907,6 @@ namespace Opc.Ua.Server m_lastValue = value; m_lastError = error; m_readyToPublish = true; - ServerUtils.EventLog.QueueValue(m_id, m_lastValue.WrappedValue, m_lastValue.StatusCode); } @@ -1238,8 +1263,11 @@ namespace Opc.Ua.Server return false; } - // only publish if reporting. - if (!IsReadyToPublish) + // only publish if reporting or resending data + ResendDataState selectedToResendData = (ResendDataState)Interlocked.CompareExchange(ref m_resendDataState, (int)ResendDataState.NonResendData, (int)ResendDataState.ResendData); + bool isResendData = selectedToResendData == ResendDataState.ResendData; + + if (!IsReadyToPublish && !isResendData) { return false; } @@ -1264,23 +1292,55 @@ namespace Opc.Ua.Server // go to the next sampling interval. IncrementSampleTime(); - // check if queueing enabled. - if (m_queue != null) + if (isResendData) { - DataValue value = null; - ServiceResult error = null; - - while (m_queue.Publish(out value, out error)) + // check if queueing enabled. + if (m_queue != null && m_queue.ItemsInQueue != 0) { - Publish(context, notifications, diagnostics, value, error); + DataValue value = null; + ServiceResult error = null; + + if (m_queue.ItemsInQueue > 1) + { + // pop the first value + m_queue.Publish(out value, out error); + // publish the next + m_queue.Publish(out value, out error); + Publish(context, notifications, diagnostics, value, error); + } + else // m_queue.ItemsInQueue == 1 + { + m_queue.Publish(out value, out error); + Publish(context, notifications, diagnostics, value, error); + } + } + // publish last value if no queuing or no items are queued + else + { + ServerUtils.EventLog.DequeueValue(m_lastValue.WrappedValue, m_lastValue.StatusCode); + Publish(context, notifications, diagnostics, m_lastValue, m_lastError); } } - - // publish last value if no queuing. else { - ServerUtils.EventLog.DequeueValue(m_lastValue.WrappedValue, m_lastValue.StatusCode); - Publish(context, notifications, diagnostics, m_lastValue, m_lastError); + // check if queueing enabled. + if (m_queue != null) + { + DataValue value = null; + ServiceResult error = null; + + while (m_queue.Publish(out value, out error)) + { + Publish(context, notifications, diagnostics, value, error); + } + } + + // publish last value if no queuing or no items are queued + else + { + ServerUtils.EventLog.DequeueValue(m_lastValue.WrappedValue, m_lastValue.StatusCode); + Publish(context, notifications, diagnostics, m_lastValue, m_lastError); + } } // reset state variables. @@ -1862,6 +1922,7 @@ namespace Opc.Ua.Server private ServiceResult m_samplingError; private IAggregateCalculator m_calculator; private bool m_triggered; + private int m_resendDataState; #endregion } } diff --git a/Libraries/Opc.Ua.Server/Subscription/Subscription.cs b/Libraries/Opc.Ua.Server/Subscription/Subscription.cs index c0f079dc5d..20108fa620 100644 --- a/Libraries/Opc.Ua.Server/Subscription/Subscription.cs +++ b/Libraries/Opc.Ua.Server/Subscription/Subscription.cs @@ -494,7 +494,7 @@ namespace Opc.Ua.Server m_itemsToCheck.Remove(current); m_itemsToPublish.AddLast(current); } - + current = next; } } @@ -535,11 +535,9 @@ namespace Opc.Ua.Server /// Whether the first Publish response shall contain current values. public void TransferSession(OperationContext context, bool sendInitialValues) { - lock (m_lock) - { - m_session = context.Session; - } - + // locked by caller + m_session = context.Session; + var monitoredItems = m_monitoredItems.Select(v => v.Value.Value).ToList(); var errors = new List(monitoredItems.Count); for (int ii = 0; ii < monitoredItems.Count; ii++) @@ -569,6 +567,30 @@ namespace Opc.Ua.Server } } + /// + /// Initiates resending of all data monitored items in a Subscription + /// + /// + public void ResendData(OperationContext context) + { + // check session. + VerifySession(context); + lock (m_lock) + { + var monitoredItems = m_monitoredItems.Select(v => v.Value.Value).ToList(); + // process MI when MonitoringMode is set to Reporting + foreach (IMonitoredItem monitoredItem in monitoredItems) + { + if ((monitoredItem.MonitoringMode == MonitoringMode.Reporting) && + ((monitoredItem.MonitoredItemType & MonitoredItemTypeMask.DataChange) != 0)) + { + IDataChangeMonitoredItem2 dataChangeMonitoredItem = (IDataChangeMonitoredItem2)monitoredItem; + dataChangeMonitoredItem.SetupResendDataTrigger(); + } + } + } + } + /// /// Tells the subscription that the owning session is being closed. /// @@ -782,6 +804,7 @@ namespace Opc.Ua.Server return message; } + /// /// Returns all available notifications. /// @@ -796,8 +819,8 @@ namespace Opc.Ua.Server // TraceState(LogLevel.Trace, TraceStateId.Items, "PUBLISH"); // check if a keep alive should be sent if there is no data. - bool keepAliveIfNoData = (m_keepAliveCounter >= m_maxKeepAliveCount); - + bool keepAliveIfNoData = m_keepAliveCounter >= m_maxKeepAliveCount; + availableSequenceNumbers = new UInt32Collection(); moreNotifications = false; diff --git a/Tests/Opc.Ua.Server.Tests/ReferenceServerTest.cs b/Tests/Opc.Ua.Server.Tests/ReferenceServerTest.cs index f55036c431..0360b06c1b 100644 --- a/Tests/Opc.Ua.Server.Tests/ReferenceServerTest.cs +++ b/Tests/Opc.Ua.Server.Tests/ReferenceServerTest.cs @@ -410,6 +410,171 @@ namespace Opc.Ua.Server.Tests SecureChannelContext.Current = securityContext; } } + + + /// + /// Create a subscription with a monitored item. + /// Call ResendData. + /// + [Test] + public void ResendData() + { + var serverTestServices = new ServerTestServices(m_server); + // save old security context, test fixture can only work with one session + var securityContext = SecureChannelContext.Current; + try + { + var namespaceUris = m_server.CurrentInstance.NamespaceUris; + NodeId[] testSet = CommonTestWorkers.NodeIdTestSetStatic.Select(n => ExpandedNodeId.ToNodeId(n, namespaceUris)).ToArray(); + //Re-use method CreateSubscriptionForTransfer to create a subscription + CommonTestWorkers.CreateSubscriptionForTransfer(serverTestServices, m_requestHeader, + testSet, out var subscriptionIds); + + RequestHeader resendDataRequestHeader = m_server.CreateAndActivateSession("ResendData"); + var resendDataSecurityContext = SecureChannelContext.Current; + + // After the ResendData call there will be data to publish again + MethodState methodStateInstance = (MethodState)m_server.CurrentInstance. + DiagnosticsNodeManager.FindPredefinedNode(MethodIds.Server_ResendData, typeof(MethodState)); + var nodesToCall = new CallMethodRequestCollection(); + nodesToCall.Add(new CallMethodRequest() { + ObjectId = ObjectIds.Server, + MethodId = MethodIds.Server_ResendData, + InputArguments = new VariantCollection() { new Variant(subscriptionIds.Last()) } + }); + + //call ResendData method from the same session context + m_requestHeader.Timestamp = DateTime.UtcNow; + var response = m_server.Call(m_requestHeader, + nodesToCall, + out var results, + out var diagnosticInfos); + + Assert.IsTrue(StatusCode.IsGood(results[0].StatusCode)); + ServerFixtureUtils.ValidateResponse(response); + + Thread.Sleep(1000); + + // Make sure publish queue becomes empty by consuming it + Assert.AreEqual(1, subscriptionIds.Count); + // Issue a Publish request + m_requestHeader.Timestamp = DateTime.UtcNow; + var acknoledgements = new SubscriptionAcknowledgementCollection(); + response = serverTestServices.Publish(m_requestHeader, acknoledgements, + out uint publishedId, out UInt32Collection availableSequenceNumbers, + out bool moreNotifications, out NotificationMessage notificationMessage, + out StatusCodeCollection _, out diagnosticInfos); + + Assert.AreEqual(StatusCodes.Good, response.ServiceResult.Code); + ServerFixtureUtils.ValidateResponse(response); + ServerFixtureUtils.ValidateDiagnosticInfos(diagnosticInfos, acknoledgements); + Assert.AreEqual(subscriptionIds[0], publishedId); + Assert.AreEqual(1, notificationMessage.NotificationData.Count); + + // Validate nothing to publish a few times + const int timesToCallPublish = 3; + for (int i = 0; i < timesToCallPublish; i++) + { + m_requestHeader.Timestamp = DateTime.UtcNow; + response = serverTestServices.Publish(m_requestHeader, acknoledgements, + out publishedId, out availableSequenceNumbers, + out moreNotifications, out notificationMessage, + out StatusCodeCollection _, out diagnosticInfos); + + Assert.AreEqual(StatusCodes.Good, response.ServiceResult.Code); + ServerFixtureUtils.ValidateResponse(response); + ServerFixtureUtils.ValidateDiagnosticInfos(diagnosticInfos, acknoledgements); + Assert.AreEqual(subscriptionIds[0], publishedId); + Assert.AreEqual(0, notificationMessage.NotificationData.Count); + } + + // Validate ResendData method call from same and different session contexts + + // call ResendData method from different session context + resendDataRequestHeader.Timestamp = DateTime.UtcNow; + response = m_server.Call(resendDataRequestHeader, + nodesToCall, + out results, + out diagnosticInfos); + + Assert.AreEqual(StatusCodes.BadUserAccessDenied, results[0].StatusCode.Code); + ServerFixtureUtils.ValidateResponse(response); + + // Still nothing to publish since previous ResendData call did not execute + m_requestHeader.Timestamp = DateTime.UtcNow; + response = serverTestServices.Publish(m_requestHeader, acknoledgements, + out publishedId, out availableSequenceNumbers, + out moreNotifications, out notificationMessage, + out StatusCodeCollection _, out diagnosticInfos); + + Assert.AreEqual(StatusCodes.Good, response.ServiceResult.Code); + ServerFixtureUtils.ValidateResponse(response); + ServerFixtureUtils.ValidateDiagnosticInfos(diagnosticInfos, acknoledgements); + Assert.AreEqual(subscriptionIds[0], publishedId); + Assert.AreEqual(0, notificationMessage.NotificationData.Count); + + //call ResendData method from the same session context + m_requestHeader.Timestamp = DateTime.UtcNow; + response = m_server.Call(m_requestHeader, + nodesToCall, + out results, + out diagnosticInfos); + + Assert.IsTrue(StatusCode.IsGood(results[0].StatusCode)); + ServerFixtureUtils.ValidateResponse(response); + + // Data should be available for publishing now + m_requestHeader.Timestamp = DateTime.UtcNow; + response = serverTestServices.Publish(m_requestHeader, acknoledgements, + out publishedId, out availableSequenceNumbers, + out moreNotifications, out notificationMessage, + out StatusCodeCollection _, out diagnosticInfos); + + Assert.AreEqual(StatusCodes.Good, response.ServiceResult.Code); + ServerFixtureUtils.ValidateResponse(response); + ServerFixtureUtils.ValidateDiagnosticInfos(diagnosticInfos, acknoledgements); + Assert.AreEqual(subscriptionIds[0], publishedId); + Assert.AreEqual(1, notificationMessage.NotificationData.Count); + + // Call ResendData method with invalid subscription Id + nodesToCall = new CallMethodRequestCollection(); + nodesToCall.Add(new CallMethodRequest() { + ObjectId = ObjectIds.Server, + MethodId = MethodIds.Server_ResendData, + InputArguments = new VariantCollection() { new Variant(subscriptionIds.Last() + 20) } + }); + m_requestHeader.Timestamp = DateTime.UtcNow; + response = m_server.Call(m_requestHeader, + nodesToCall, + out results, + out diagnosticInfos); + + Assert.AreEqual(StatusCodes.BadSubscriptionIdInvalid, results[0].StatusCode.Code); + ServerFixtureUtils.ValidateResponse(response); + + // Nothing to publish since previous ResendData call did not execute + m_requestHeader.Timestamp = DateTime.UtcNow; + response = serverTestServices.Publish(m_requestHeader, acknoledgements, + out publishedId, out availableSequenceNumbers, + out moreNotifications, out notificationMessage, + out StatusCodeCollection _, out diagnosticInfos); + + Assert.AreEqual(StatusCodes.Good, response.ServiceResult.Code); + ServerFixtureUtils.ValidateResponse(response); + ServerFixtureUtils.ValidateDiagnosticInfos(diagnosticInfos, acknoledgements); + Assert.AreEqual(subscriptionIds[0], publishedId); + Assert.AreEqual(0, notificationMessage.NotificationData.Count); + + resendDataRequestHeader.Timestamp = DateTime.UtcNow; + SecureChannelContext.Current = resendDataSecurityContext; + m_server.CloseSession(resendDataRequestHeader); + } + finally + { + //restore security context, that close connection can work + SecureChannelContext.Current = securityContext; + } + } #endregion } }