GithubStar/UA-.NETStandard
1
0
Fork 0
mirror of https://github.com/OPCFoundation/UA-.NETStandard.git synced 2026-10-29 07:07:24 +08:00
Code Issues Packages Projects Releases Wiki Activity

Added server method ResendData functionality

Browse Source 
* Added ResendData functionality
This commit is contained in:
Suciu Mircea Adrian
2022-06-06 11:11:39 +03:00
committed by GitHub
parent e3ac2bf4db
commit d789d8bf51
9 changed files with 358 additions and 69 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Applications/Quickstarts.Servers/SampleNodeManager/DataChangeMonitoredItem.cs
+5
View File
@@ -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
}
/// <summary>
/// Returns the results for the create request.
/// </summary>
Applications/Quickstarts.Servers/SampleNodeManager/SampleNodeManager.cs
+6 -11
View File
@@ -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.
}
/// <summary>
/// Changes the monitoring mode for a set of monitored items.
/// </summary>
Libraries/Opc.Ua.Server/Diagnostics/CustomNodeManager.cs
+5 -11
View File
@@ -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;
}
}
Libraries/Opc.Ua.Server/Diagnostics/DiagnosticsNodeManager.cs
+51
View File
@@ -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;
}
/// <summary>
/// Called when a client initiates resending of all data monitored items in a Subscription.
/// </summary>
public ServiceResult OnResendData(
ISystemContext context,
MethodState method,
IList<object> inputArguments,
IList<object> 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;
}
/// <summary>
/// Called when a client locks the server.
/// </summary>
Libraries/Opc.Ua.Server/NodeManager/CoreNodeManager.cs
+9 -21
View File
@@ -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<uint,MonitoredItem>();
m_defaultMinimumSamplingInterval = 1000;
m_namespaceUris = new List<string>();
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<bool> processedItems,
IList<ServiceResult> 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;
}
}
}
Libraries/Opc.Ua.Server/Subscription/IMonitoredItem.cs
+7
View File
@@ -35,6 +35,8 @@ using System.Threading;
namespace Opc.Ua.Server
{
/// <summary>
/// Manages a monitored item created by a client.
/// </summary>
@@ -94,6 +96,11 @@ namespace Opc.Ua.Server
/// </summary>
bool IsReadyToTrigger { get; set; }
/// <summary>
/// Setup the resend data trigger by setting the monitor item in ResendData state
/// </summary>
void SetupResendDataTrigger();
/// <summary>
/// Returns the result after creating the monitor item.
/// </summary>
Libraries/Opc.Ua.Server/Subscription/MonitoredItem.cs
+79 -18
View File
@@ -30,10 +30,23 @@
using System;
using System.Collections.Generic;
using System.Globalization;
using System.Threading;
using System.Xml;
namespace Opc.Ua.Server
{
/// <summary>
/// Specifies the values of the ResendData state for a monitored item
/// The state is set on the ResendData method call
/// </summary>
internal enum ResendDataState : int
{
///<summary>The Monitored item does not participate in the ResendData</summary>
NonResendData = 0,
///<summary>The Monitored item participates in ResendData</summary>
ResendData = 1
}
/// <summary>
/// A handle that describes how to access a node/attribute via an i/o manager.
/// </summary>
@@ -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
}
}
/// <summary>
/// Setup the resend data trigger by setting the monitor item in ResendData state
/// </summary>
public void SetupResendDataTrigger()
{
Interlocked.Exchange(ref m_resendDataState, (int)ResendDataState.ResendData);
}
/// <summary>
/// Sets a flag indicating that the item has been triggered and should publish.
/// </summary>
@@ -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
}
}
Libraries/Opc.Ua.Server/Subscription/Subscription.cs
+31 -8
View File
@@ -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
/// <param name="sendInitialValues">Whether the first Publish response shall contain current values.</param>
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<ServiceResult>(monitoredItems.Count);
for (int ii = 0; ii < monitoredItems.Count; ii++)
@@ -569,6 +567,30 @@ namespace Opc.Ua.Server
}
}
/// <summary>
/// Initiates resending of all data monitored items in a Subscription
/// </summary>
/// <param name="context"></param>
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();
}
}
}
}
/// <summary>
/// Tells the subscription that the owning session is being closed.
/// </summary>
@@ -782,6 +804,7 @@ namespace Opc.Ua.Server
return message;
}
/// <summary>
/// Returns all available notifications.
/// </summary>
@@ -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;
Tests/Opc.Ua.Server.Tests/ReferenceServerTest.cs
+165
View File
@@ -410,6 +410,171 @@ namespace Opc.Ua.Server.Tests
SecureChannelContext.Current = securityContext;
}
}
/// <summary>
/// Create a subscription with a monitored item.
/// Call ResendData.
/// </summary>
[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
}
}