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

Merge pull request #2163 from OPCFoundation/master371

Browse Source 
Merge release 1.4.371.91 in master
This commit is contained in:
Martin Regen
2023-05-27 05:52:47 +02:00
committed by GitHub
parent cea95d213a a81a6961bd
commit d8f8a4e7ad
13 changed files with 767 additions and 106 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Applications/ConsoleReferenceClient/ClientSamples.cs
+176 -17
View File
@@ -224,6 +224,7 @@ namespace Quickstarts
browser.BrowseDirection = BrowseDirection.Forward;
browser.NodeClassMask = (int)NodeClass.Object | (int)NodeClass.Variable;
browser.ReferenceTypeId = ReferenceTypeIds.HierarchicalReferences;
browser.IncludeSubtypes = true;
NodeId nodeToBrowse = ObjectIds.Server;
@@ -380,7 +381,8 @@ namespace Quickstarts
NodeId startingNode,
bool fetchTree = false,
bool addRootNode = false,
bool filterUATypes = true)
bool filterUATypes = true,
bool clearNodeCache = true)
{
var stopwatch = new Stopwatch();
var nodeDictionary = new Dictionary<ExpandedNodeId, INode>();
@@ -389,19 +391,15 @@ namespace Quickstarts
startingNode
};
// clear NodeCache to fetch all nodes from server
uaClient.Session.NodeCache.Clear();
// start
stopwatch.Start();
// fetch the reference types first, otherwise browse for e.g. hierarchical references with subtypes won't work
var bindingFlags = BindingFlags.Instance | BindingFlags.Static | BindingFlags.Public;
var namespaceUris = uaClient.Session.NamespaceUris;
var referenceTypes = typeof(ReferenceTypeIds)
.GetFields(bindingFlags)
.Select(field => NodeId.ToExpandedNodeId((NodeId)field.GetValue(null), namespaceUris));
uaClient.Session.FetchTypeTree(new ExpandedNodeIdCollection(referenceTypes));
if (clearNodeCache)
{
// clear NodeCache to fetch all nodes from server
uaClient.Session.NodeCache.Clear();
FetchReferenceIdTypes(uaClient.Session);
}
// add root node
if (addRootNode)
@@ -421,7 +419,7 @@ namespace Quickstarts
searchDepth++;
Utils.LogInfo("{0}: Find {1} references after {2}ms", searchDepth, nodesToBrowse.Count, stopwatch.ElapsedMilliseconds);
var response = uaClient.Session.NodeCache.FindReferences(
IList<INode> response = uaClient.Session.NodeCache.FindReferences(
nodesToBrowse,
references,
false,
@@ -429,13 +427,33 @@ namespace Quickstarts
var nextNodesToBrowse = new ExpandedNodeIdCollection();
int duplicates = 0;
foreach (var node in response)
int leafNodes = 0;
foreach (INode node in response)
{
if (!nodeDictionary.ContainsKey(node.NodeId))
{
if (fetchTree)
{
nextNodesToBrowse.Add(node.NodeId);
bool leafNode = false;
// no need to browse property types
if (node is VariableNode variableNode)
{
var hasTypeDefinition = variableNode.ReferenceTable.FirstOrDefault(r => r.ReferenceTypeId.Equals(ReferenceTypeIds.HasTypeDefinition));
if (hasTypeDefinition != null)
{
leafNode = (hasTypeDefinition.TargetId == VariableTypeIds.PropertyType);
}
}
if (!leafNode)
{
nextNodesToBrowse.Add(node.NodeId);
}
else
{
leafNodes++;
}
}
if (filterUATypes)
@@ -460,6 +478,10 @@ namespace Quickstarts
{
Utils.LogInfo("Find References {0} duplicate nodes were ignored", duplicates);
}
if (leafNodes > 0)
{
Utils.LogInfo("Find References {0} leaf nodes were ignored", leafNodes);
}
nodesToBrowse = nextNodesToBrowse;
}
@@ -515,7 +537,7 @@ namespace Quickstarts
var referenceDescriptions = new Dictionary<ExpandedNodeId, ReferenceDescription>();
int searchDepth = 0;
uint maxNodesPerBrowse = 0;
uint maxNodesPerBrowse = uaClient.Session.OperationLimits.MaxNodesPerBrowse;
while (browseDescriptionCollection.Any() && searchDepth < kMaxSearchDepth)
{
searchDepth++;
@@ -625,7 +647,10 @@ namespace Quickstarts
if (!referenceDescriptions.ContainsKey(reference.NodeId))
{
referenceDescriptions[reference.NodeId] = reference;
browseTable.Add(ExpandedNodeId.ToNodeId(reference.NodeId, uaClient.Session.NamespaceUris));
if (reference.ReferenceTypeId != ReferenceTypeIds.HasProperty)
{
browseTable.Add(ExpandedNodeId.ToNodeId(reference.NodeId, uaClient.Session.NamespaceUris));
}
}
else
{
@@ -707,6 +732,28 @@ namespace Quickstarts
}
#endregion
#region Fetch ReferenceId Types
/// <summary>
/// Read all ReferenceTypeIds from the server that are not known by the client.
/// To reduce the number of calls due to traversal call pyramid, start with all
/// known reference types to reduce the number of FetchReferences/FetchNodes calls.
/// </summary>
/// <remarks>
/// The NodeCache needs this information to function properly with subtypes of hierarchical calls.
/// </remarks>
/// <param name="session">The session to use</param>
void FetchReferenceIdTypes(ISession session)
{
// fetch the reference types first, otherwise browse for e.g. hierarchical references with subtypes won't work
var bindingFlags = BindingFlags.Instance | BindingFlags.Static | BindingFlags.Public;
var namespaceUris = session.NamespaceUris;
var referenceTypes = typeof(ReferenceTypeIds)
.GetFields(bindingFlags)
.Select(field => NodeId.ToExpandedNodeId((NodeId)field.GetValue(null), namespaceUris));
session.FetchTypeTree(new ExpandedNodeIdCollection(referenceTypes));
}
#endregion
#region Read Values and output as JSON sample
/// <summary>
/// Output all values as JSON.
@@ -790,6 +837,84 @@ namespace Quickstarts
}
#endregion
#region Subscribe Values
/// <summary>
/// Output all values as JSON.
/// </summary>
/// <param name="uaClient">The UAClient with a session to use.</param>
/// <param name="variableIds">The variables to subscribe.</param>
public async Task SubscribeAllValuesAsync(
IUAClient uaClient,
NodeCollection variableIds,
int samplingInterval,
int publishingInterval,
uint queueSize,
uint lifetimeCount,
uint keepAliveCount)
{
if (uaClient.Session == null || !uaClient.Session.Connected)
{
m_output.WriteLine("Session not connected!");
return;
}
try
{
// Create a subscription for receiving data change notifications
var session = uaClient.Session;
// test for deferred ack of sequence numbers
session.PublishSequenceNumbersToAcknowledge += DeferSubscriptionAcknowledge;
// set a minimum amount of three publish requests per session
session.MinPublishRequestCount = 3;
// Define Subscription parameters
Subscription subscription = new Subscription(session.DefaultSubscription) {
DisplayName = "Console ReferenceClient Subscription",
PublishingEnabled = true,
PublishingInterval = publishingInterval,
LifetimeCount = lifetimeCount,
KeepAliveCount = keepAliveCount,
SequentialPublishing = true,
RepublishAfterTransfer = true,
MaxNotificationsPerPublish = 1000,
MinLifetimeInterval = (uint)session.SessionTimeout,
FastDataChangeCallback = FastDataChangeNotification,
};
session.AddSubscription(subscription);
// Create the subscription on Server side
await subscription.CreateAsync().ConfigureAwait(false);
m_output.WriteLine("New Subscription created with SubscriptionId = {0}.", subscription.Id);
// Create MonitoredItems for data changes
foreach (Node item in variableIds)
{
MonitoredItem monitoredItem = new MonitoredItem(subscription.DefaultItem) {
StartNodeId = item.NodeId,
AttributeId = Attributes.Value,
SamplingInterval = samplingInterval,
DisplayName = item.DisplayName.Text ?? item.BrowseName.Name,
QueueSize = queueSize,
DiscardOldest = true,
MonitoringMode = MonitoringMode.Reporting,
};
subscription.AddItem(monitoredItem);
if (subscription.CurrentKeepAliveCount > 1000) break;
}
// Create the monitored items on Server side
subscription.ApplyChanges();
m_output.WriteLine("MonitoredItems {0} created for SubscriptionId = {1}.", subscription.MonitoredItemCount, subscription.Id);
}
catch (Exception ex)
{
m_output.WriteLine("Subscribe error: {0}", ex.Message);
}
}
#endregion
#region Helper Methods
/// <summary>
/// Create a prettified JSON string of a DataValue.
@@ -834,6 +959,21 @@ namespace Quickstarts
#endregion
#region Private Methods
/// <summary>
/// The fast data change notification callback.
/// </summary>
private void FastDataChangeNotification(Subscription subscription, DataChangeNotification notification, IList<string> stringTable)
{
try
{
m_output.WriteLine("Notification: Id={0} Items={1}.", subscription.Id, notification.MonitoredItems.Count);
}
catch (Exception ex)
{
m_output.WriteLine("FastDataChangeNotification error: {0}", ex.Message);
}
}
/// <summary>
/// Handle DataChange notifications from Server
/// </summary>
@@ -843,7 +983,7 @@ namespace Quickstarts
{
// Log MonitoredItem Notification event
MonitoredItemNotification notification = e.NotificationValue as MonitoredItemNotification;
m_output.WriteLine("Notification: {0} \"{1}\" and Value = {2}.", notification.Message.SequenceNumber, monitoredItem.DisplayName, notification.Value);
m_output.WriteLine("Notification: {0} \"{1}\" and Value = {2}.", notification.Message.SequenceNumber, monitoredItem.ResolvedNodeId, notification.Value);
}
catch (Exception ex)
{
@@ -851,6 +991,25 @@ namespace Quickstarts
}
}
/// <summary>
/// Event handler to defer publish response sequence number acknowledge.
/// </summary>
private void DeferSubscriptionAcknowledge(ISession session, PublishSequenceNumbersToAcknowledgeEventArgs e)
{
// for testing keep the latest sequence numbers for a while
const int AckDelay = 5;
if (e.AcknowledgementsToSend.Count > 0)
{
// defer latest sequence numbers
var deferredItems = e.AcknowledgementsToSend.OrderByDescending(s => s.SequenceNumber).Take(AckDelay).ToList();
e.DeferredAcknowledgementsToSend.AddRange(deferredItems);
foreach (var deferredItem in deferredItems)
{
e.AcknowledgementsToSend.Remove(deferredItem);
}
}
}
/// <summary>
/// Create a browse description from a node id collection.
/// </summary>
Applications/ConsoleReferenceClient/Program.cs
+49 -6
View File
@@ -76,6 +76,7 @@ namespace Quickstarts.ConsoleReferenceClient
bool fetchall = false;
bool jsonvalues = false;
bool verbose = false;
bool subscribe = false;
string password = null;
int timeout = Timeout.Infinite;
string logFile = null;
@@ -97,6 +98,7 @@ namespace Quickstarts.ConsoleReferenceClient
{ "f|fetchall", "Fetch all nodes", f => { if (f != null) fetchall = true; } },
{ "j|json", "Output all Values as JSON", j => { if (j != null) jsonvalues = true; } },
{ "v|verbose", "Verbose output", v => { if (v != null) verbose = true; } },
{ "s|subscribe", "Subscribe", s => { if (s != null) subscribe = true; } },
};
try
@@ -128,7 +130,7 @@ namespace Quickstarts.ConsoleReferenceClient
};
// load the application configuration.
var config = await application.LoadApplicationConfiguration(silent: false);
var config = await application.LoadApplicationConfiguration(silent: false).ConfigureAwait(false);
// override logfile
if (logFile != null)
@@ -177,7 +179,8 @@ namespace Quickstarts.ConsoleReferenceClient
// create the UA Client object and connect to configured server.
using (UAClient uaClient = new UAClient(
application.ApplicationConfiguration, output, ClientBase.ValidateResponse) {
AutoAccept = autoAccept
AutoAccept = autoAccept,
SessionLifeTime = 60000,
})
{
// set user identity
@@ -186,12 +189,13 @@ namespace Quickstarts.ConsoleReferenceClient
uaClient.UserIdentity = new UserIdentity(username, userpassword ?? string.Empty);
}
bool connected = await uaClient.ConnectAsync(serverUrl.ToString(), false);
bool connected = await uaClient.ConnectAsync(serverUrl.ToString(), false).ConfigureAwait(false);
if (connected)
{
output.WriteLine("Connected! Ctrl-C to quit.");
// enable subscription transfer
uaClient.ReconnectPeriod = 1000;
uaClient.Session.TransferSubscriptionsOnReconnect = true;
var samples = new ClientSamples(output, ClientBase.ValidateResponse, quitEvent, verbose);
@@ -219,16 +223,55 @@ namespace Quickstarts.ConsoleReferenceClient
allNodes = samples.FetchAllNodesNodeCache(
uaClient, Objects.RootFolder, true, true, false);
variableIds = new NodeIdCollection(allNodes
.Where(r => r.NodeClass == NodeClass.Variable && ((VariableNode)r).DataType.NamespaceIndex != 0)
.Where(r => r.NodeClass == NodeClass.Variable && r is VariableNode && ((VariableNode)r).DataType.NamespaceIndex != 0)
.Select(r => ExpandedNodeId.ToNodeId(r.NodeId, uaClient.Session.NamespaceUris)));
}
if (jsonvalues && variableIds != null)
{
await samples.ReadAllValuesAsync(uaClient, variableIds);
await samples.ReadAllValuesAsync(uaClient, variableIds).ConfigureAwait(false);
}
if (subscribe)
{
const int MaxVariables = 100;
NodeCollection variables = new NodeCollection();
Random random = new Random(62541);
if (fetchall)
{
variables.AddRange(allNodes
.Where(r => r.NodeClass == NodeClass.Variable && r.NodeId.NamespaceIndex > 1)
.Select(r => ((VariableNode)r))
.OrderBy(o => random.Next())
.Take(MaxVariables));
}
else if (browseall)
{
var variableReferences = referenceDescriptions
.Where(r => r.NodeClass == NodeClass.Variable && r.NodeId.NamespaceIndex > 1)
.Select(r => r.NodeId)
.OrderBy(o => random.Next())
.Take(MaxVariables)
.ToList();
variables.AddRange(uaClient.Session.NodeCache.Find(variableReferences).Cast<Node>());
}
await samples.SubscribeAllValuesAsync(uaClient,
variableIds: new NodeCollection(variables.Take(100)),
samplingInterval: 1000,
publishingInterval: 5000,
queueSize: 10,
lifetimeCount: 12,
keepAliveCount: 2).ConfigureAwait(false);
// Wait for DataChange notifications from MonitoredItems
quit = quitEvent.WaitOne(timeout > 0 ? waitTime : Timeout.Infinite);
}
else
{
quit = true;
}
quit = true;
}
else
{
Applications/ConsoleReferenceClient/UAClient.cs
+36 -20
View File
@@ -62,6 +62,7 @@ namespace Quickstarts
{
Utils.SilentDispose(m_session);
m_configuration.CertificateValidator.CertificateValidation -= CertificateValidation;
GC.SuppressFinalize(this);
}
#endregion
@@ -89,7 +90,7 @@ namespace Quickstarts
/// <summary>
/// The session lifetime.
/// </summary>
public uint SessionLifeTime { get; set; } = 30 * 1000;
public uint SessionLifeTime { get; set; } = 60 * 1000;
/// <summary>
/// The user identity to use to connect to the server.
@@ -135,7 +136,7 @@ namespace Quickstarts
var session = await Opc.Ua.Client.Session.Create(
m_configuration,
endpoint,
false,
true,
false,
m_configuration.ApplicationName,
SessionLifeTime,
@@ -151,8 +152,15 @@ namespace Quickstarts
// override keep alive interval
m_session.KeepAliveInterval = KeepAliveInterval;
// support transfer
m_session.DeleteSubscriptionsOnClose = false;
m_session.TransferSubscriptionsOnReconnect = true;
// set up keep alive callback.
m_session.KeepAlive += Session_KeepAlive;
// prepare a reconnect handler
m_reconnectHandler = new SessionReconnectHandler(true, 30_000);
}
// Session created successfully.
@@ -227,19 +235,14 @@ namespace Quickstarts
return;
}
lock (m_lock)
var state = m_reconnectHandler.BeginReconnect(m_session, ReconnectPeriod, Client_ReconnectComplete);
if (state == SessionReconnectHandler.ReconnectState.Triggered)
{
if (m_reconnectHandler == null)
{
Utils.LogInfo("KeepAlive status {0}, reconnecting in {1}ms.", e.Status, ReconnectPeriod);
m_output.WriteLine("--- RECONNECTING {0} ---", e.Status);
m_reconnectHandler = new SessionReconnectHandler(true);
m_reconnectHandler.BeginReconnect(m_session, ReconnectPeriod, Client_ReconnectComplete);
}
else
{
Utils.LogInfo("KeepAlive status {0}, reconnect in progress.", e.Status);
}
Utils.LogInfo("KeepAlive status {0}, reconnect status {1}, reconnect period {2}ms.", e.Status, state, ReconnectPeriod);
}
else
{
Utils.LogInfo("KeepAlive status {0}, reconnect status {1}.", e.Status, state);
}
return;
@@ -267,14 +270,27 @@ namespace Quickstarts
// if session recovered, Session property is null
if (m_reconnectHandler.Session != null)
{
m_session = m_reconnectHandler.Session as Session;
// ensure only a new instance is disposed
// after reactivate, the same session instance may be returned
if (!Object.ReferenceEquals(m_session, m_reconnectHandler.Session))
{
m_output.WriteLine("--- RECONNECTED TO NEW SESSION --- {0}", m_reconnectHandler.Session.SessionId);
var session = m_session;
session.KeepAlive -= Session_KeepAlive;
m_session = m_reconnectHandler.Session as Session;
m_session.KeepAlive += Session_KeepAlive;
Utils.SilentDispose(session);
}
else
{
m_output.WriteLine("--- REACTIVATED SESSION --- {0}", m_reconnectHandler.Session.SessionId);
}
}
else
{
m_output.WriteLine("--- RECONNECT KeepAlive recovered ---");
}
m_reconnectHandler.Dispose();
m_reconnectHandler = null;
}
m_output.WriteLine("--- RECONNECTED ---");
}
#endregion
Libraries/Opc.Ua.Client/ISession.cs
+19
View File
@@ -55,6 +55,11 @@ namespace Opc.Ua.Client
/// </summary>
public delegate void PublishErrorEventHandler(ISession session, PublishErrorEventArgs e);
/// <summary>
/// The delegate used to modify publish response sequence numbers to acknowledge.
/// </summary>
public delegate void PublishSequenceNumbersToAcknowledgeEventHandler(ISession session, PublishSequenceNumbersToAcknowledgeEventArgs e);
/// <summary>
/// Manages a session with a server.
/// </summary>
@@ -93,6 +98,15 @@ namespace Opc.Ua.Client
/// </remarks>
event PublishErrorEventHandler PublishError;
/// <summary>
/// Raised when a publish request is about to acknolegde sequence numbers.
/// </summary>
/// <remarks>
/// If the client chose to defer acknowledge of sequenece numbers, it is responsible
/// to transfer these <see cref="SubscriptionAcknowledgement"/> to the deferred list.
/// </remarks>
event PublishSequenceNumbersToAcknowledgeEventHandler PublishSequenceNumbersToAcknowledge;
/// <summary>
/// Raised when a subscription is added or removed
/// </summary>
@@ -243,6 +257,11 @@ namespace Opc.Ua.Client
/// </summary>
int GoodPublishRequestCount { get; }
/// <summary>
/// Gets and sets the minimum number of publish requests to be used in the session.
/// </summary>
int MinPublishRequestCount { get; set; }
/// <summary>
/// Stores the operation limits of a OPC UA Server.
/// </summary>
Libraries/Opc.Ua.Client/ReverseConnectManager.cs
+6
View File
@@ -216,6 +216,11 @@ namespace Opc.Ua.Client
Utils.SilentDispose(m_configurationWatcher);
m_configurationWatcher = null;
}
if (m_cts != null)
{
Utils.SilentDispose(m_cts);
m_cts = null;
}
DisposeHosts();
}
#endregion
@@ -725,6 +730,7 @@ namespace Opc.Ua.Client
CancellationTokenSource cts = m_cts;
m_cts = new CancellationTokenSource();
cts.Cancel();
cts.Dispose();
}
#endregion
Libraries/Opc.Ua.Client/Session.cs
+170 -15
View File
@@ -114,6 +114,7 @@ namespace Opc.Ua.Client
m_transferSubscriptionsOnReconnect = template.m_transferSubscriptionsOnReconnect;
m_sessionTimeout = template.m_sessionTimeout;
m_maxRequestMessageSize = template.m_maxRequestMessageSize;
m_minPublishRequestCount = template.m_minPublishRequestCount;
m_preferredLocales = template.PreferredLocales;
m_sessionName = template.SessionName;
m_handle = template.Handle;
@@ -130,6 +131,7 @@ namespace Opc.Ua.Client
m_KeepAlive = template.m_KeepAlive;
m_Publish = template.m_Publish;
m_PublishError = template.m_PublishError;
m_PublishSequenceNumbersToAcknowledge = template.m_PublishSequenceNumbersToAcknowledge;
m_SubscriptionsChanged = template.m_SubscriptionsChanged;
m_SessionClosing = template.m_SessionClosing;
}
@@ -267,6 +269,7 @@ namespace Opc.Ua.Client
m_outstandingRequests = new LinkedList<AsyncRequestState>();
m_keepAliveInterval = 5000;
m_tooManyPublishRequests = 0;
m_minPublishRequestCount = kDefaultPublishRequestCount;
m_sessionName = "";
m_deleteSubscriptionsOnClose = true;
m_transferSubscriptionsOnReconnect = false;
@@ -474,6 +477,27 @@ namespace Opc.Ua.Client
}
}
/// <inheritdoc/>
public event PublishSequenceNumbersToAcknowledgeEventHandler PublishSequenceNumbersToAcknowledge
{
add
{
lock (m_eventLock)
{
m_PublishSequenceNumbersToAcknowledge += value;
}
}
remove
{
lock (m_eventLock)
{
m_PublishSequenceNumbersToAcknowledge -= value;
}
}
}
/// <summary>
/// Raised when a subscription is added or removed
/// </summary>
@@ -767,6 +791,29 @@ namespace Opc.Ua.Client
}
}
}
/// <summary>
/// Gets and sets the minimum number of publish requests to be used in the session.
/// </summary>
public int MinPublishRequestCount
{
get => m_minPublishRequestCount;
set
{
lock (SyncRoot)
{
if (value >= kDefaultPublishRequestCount && value <= kMinPublishRequestCountMax)
{
m_minPublishRequestCount = value;
}
else
{
throw new ArgumentOutOfRangeException(nameof(MinPublishRequestCount),
$"Minimum publish request count must be between {kDefaultPublishRequestCount} and {kMinPublishRequestCountMax}.");
}
}
}
}
#endregion
#region Public Static Methods
@@ -1356,7 +1403,7 @@ namespace Opc.Ua.Client
m_previousServerNonce = m_serverNonce;
m_serverNonce = serverNonce;
m_reconnecting = false;
publishCount = m_subscriptions.Count;
publishCount = GetMinPublishRequestCount();
}
// refill pipeline.
@@ -1369,7 +1416,10 @@ namespace Opc.Ua.Client
}
finally
{
m_reconnecting = false;
lock (SyncRoot)
{
m_reconnecting = false;
}
}
}
@@ -4223,11 +4273,7 @@ namespace Opc.Ua.Client
}
}
lock (SyncRoot)
{
count = m_subscriptions.Count;
}
count = GetMinPublishRequestCount();
while (count-- > 0)
{
BeginPublish(OperationTimeout);
@@ -5044,13 +5090,39 @@ namespace Opc.Ua.Client
return null;
}
SubscriptionAcknowledgementCollection acknowledgementsToSend = null;
// get event handler to modify ack list
PublishSequenceNumbersToAcknowledgeEventHandler callback = null;
lock (m_eventLock)
{
callback = m_PublishSequenceNumbersToAcknowledge;
}
// collect the current set if acknowledgements.
SubscriptionAcknowledgementCollection acknowledgementsToSend = null;
lock (SyncRoot)
{
acknowledgementsToSend = m_acknowledgementsToSend;
m_acknowledgementsToSend = new SubscriptionAcknowledgementCollection();
if (callback != null)
{
try
{
var deferredAcknowledgementsToSend = new SubscriptionAcknowledgementCollection();
callback(this, new PublishSequenceNumbersToAcknowledgeEventArgs(m_acknowledgementsToSend, deferredAcknowledgementsToSend));
acknowledgementsToSend = m_acknowledgementsToSend;
m_acknowledgementsToSend = deferredAcknowledgementsToSend;
}
catch (Exception e2)
{
Utils.LogError(e2, "Session: Unexpected error invoking PublishSequenceNumbersToAcknowledgeEventArgs.");
}
}
if (acknowledgementsToSend == null)
{
// send all ack values, clear list
acknowledgementsToSend = m_acknowledgementsToSend;
m_acknowledgementsToSend = new SubscriptionAcknowledgementCollection();
}
foreach (var toSend in acknowledgementsToSend)
{
m_latestAcknowledgementsSent[toSend.SubscriptionId] = toSend.SequenceNumber;
@@ -5245,14 +5317,14 @@ namespace Opc.Ua.Client
}
int requestCount = GoodPublishRequestCount;
var subscriptionsCount = m_subscriptions.Count;
if (requestCount < subscriptionsCount)
var minPublishRequestCount = GetMinPublishRequestCount();
if (requestCount < minPublishRequestCount)
{
BeginPublish(OperationTimeout);
}
else
{
Utils.LogInfo("PUBLISH - Did not send another publish request. GoodPublishRequestCount={0}, Subscriptions={1}", requestCount, subscriptionsCount);
Utils.LogInfo("PUBLISH - Did not send another publish request. GoodPublishRequestCount={0}, MinPublishRequestCount={1}", requestCount, minPublishRequestCount);
}
}
@@ -5304,7 +5376,7 @@ namespace Opc.Ua.Client
Utils.LogWarning("Message {0}-{1} no longer available.", subscriptionId, sequenceNumber);
break;
// if encoding limits are exceeded, the issue is logged and
// the published data is acknoledged to prevent the endless republish loop.
// the published data is acknowledged to prevent the endless republish loop.
case StatusCodes.BadEncodingLimitsExceeded:
Utils.LogError(e, "Message {0}-{1} exceeded size limits, ignored.", subscriptionId, sequenceNumber);
var ack = new SubscriptionAcknowledgement {
@@ -5537,7 +5609,15 @@ namespace Opc.Ua.Client
}
catch (ServiceResultException sre)
{
Utils.LogError(sre, "Transfer subscriptions failed.");
if (sre.StatusCode == StatusCodes.BadServiceUnsupported)
{
TransferSubscriptionsOnReconnect = false;
Utils.LogWarning("Transfer subscription unsupported, TransferSubscriptionsOnReconnect set to false.");
}
else
{
Utils.LogError(sre, "Transfer subscriptions failed.");
}
}
catch (Exception ex)
{
@@ -5682,6 +5762,24 @@ namespace Opc.Ua.Client
return (m_tooManyPublishRequests == 0) ||
(requestCount < m_tooManyPublishRequests);
}
/// <summary>
/// Returns the minimum number of active publish request that should be used.
/// </summary>
/// <remarks>
/// Returns 0 if there are no subscriptions.
/// </remarks>
private int GetMinPublishRequestCount()
{
lock (SyncRoot)
{
if (m_subscriptions.Count == 0)
{
return 0;
}
return Math.Max(m_subscriptions.Count, m_minPublishRequestCount);
}
}
#endregion
#region Protected Fields
@@ -5760,6 +5858,9 @@ namespace Opc.Ua.Client
private long m_keepAliveCounter;
private bool m_reconnecting;
private const int kReconnectTimeout = 15000;
private const int kMinPublishRequestCountMax = 100;
private const int kDefaultPublishRequestCount = 1;
private int m_minPublishRequestCount;
private LinkedList<AsyncRequestState> m_outstandingRequests;
private readonly EndpointDescriptionCollection m_discoveryServerEndpoints;
private readonly StringCollection m_discoveryProfileUris;
@@ -5777,6 +5878,7 @@ namespace Opc.Ua.Client
private event KeepAliveEventHandler m_KeepAlive;
private event NotificationEventHandler m_Publish;
private event PublishErrorEventHandler m_PublishError;
private event PublishSequenceNumbersToAcknowledgeEventHandler m_PublishSequenceNumbersToAcknowledge;
private event EventHandler m_SubscriptionsChanged;
private event EventHandler m_SessionClosing;
#endregion
@@ -5934,4 +6036,57 @@ namespace Opc.Ua.Client
#endregion
}
#endregion
#region PublishSequenceNumbersToAcknowledgeEventArgs Class
/// <summary>
/// Represents the event arguments provided when publish response
/// sequence numbers are about to be achknoledged with a publish request.
/// </summary>
/// <remarks>
/// A callee can defer an acknowledge to the next publish request by
/// moving the <see cref="SubscriptionAcknowledgement"/> to the deferred list.
/// The callee can modify the list of acknowledgements to send, it is the
/// responsibility of the caller to protect the lists for modifications.
/// </remarks>
public class PublishSequenceNumbersToAcknowledgeEventArgs : EventArgs
{
#region Constructors
/// <summary>
/// Creates a new instance.
/// </summary>
internal PublishSequenceNumbersToAcknowledgeEventArgs(
SubscriptionAcknowledgementCollection acknowledgementsToSend,
SubscriptionAcknowledgementCollection deferredAcknowledgementsToSend)
{
m_acknowledgementsToSend = acknowledgementsToSend;
m_deferredAcknowledgementsToSend = deferredAcknowledgementsToSend;
}
#endregion
#region Public Properties
/// <summary>
/// The acknowledgements which are sent with the next publish request.
/// </summary>
/// <remarks>
/// A client may also chose to remove an acknowledgement from this list to add it back
/// to the list in a subsequent callback when the request is fully processed.
/// </remarks>
public SubscriptionAcknowledgementCollection AcknowledgementsToSend => m_acknowledgementsToSend;
/// <summary>
/// The deferred list of acknowledgements.
/// </summary>
/// <remarks>
/// The callee can transfer an outstanding <see cref="SubscriptionAcknowledgement"/>
/// to this list to defer the acknowledge of a sequence number to the next publish request.
/// </remarks>
public SubscriptionAcknowledgementCollection DeferredAcknowledgementsToSend => m_deferredAcknowledgementsToSend;
#endregion
#region Private Fields
private readonly SubscriptionAcknowledgementCollection m_acknowledgementsToSend;
private readonly SubscriptionAcknowledgementCollection m_deferredAcknowledgementsToSend;
#endregion
}
#endregion
}
Libraries/Opc.Ua.Client/SessionAsync.cs
+4
View File
@@ -493,6 +493,10 @@ namespace Opc.Ua.Client
Utils.LogError("Session close error: " + result);
}
finally
{
this.OperationTimeout = existingTimeout;
}
}
// clean up.
Libraries/Opc.Ua.Client/SessionReconnectHandler.cs
+276 -31
View File
@@ -1,5 +1,5 @@
/* ========================================================================
* Copyright (c) 2005-2020 The OPC Foundation, Inc. All rights reserved.
* Copyright (c) 2005-2023 The OPC Foundation, Inc. All rights reserved.
*
* OPC Foundation MIT License 1.00
*
@@ -38,13 +38,66 @@ namespace Opc.Ua.Client
/// </summary>
public class SessionReconnectHandler : IDisposable
{
/// <summary>
/// The minimum reconnect period in ms.
/// </summary>
public const int MinReconnectPeriod = 500;
/// <summary>
/// The maximum reconnect period in ms.
/// </summary>
public const int MaxReconnectPeriod = 30000;
/// <summary>
/// The default reconnect period in ms.
/// </summary>
public const int DefaultReconnectPeriod = 1000;
/// <summary>
/// The internal state of the reconnect handler.
/// </summary>
public enum ReconnectState
{
/// <summary>
/// The reconnect handler is ready to start the reconnect timer.
/// </summary>
Ready = 0,
/// <summary>
/// The reconnect timer is triggered and waiting to reconnect.
/// </summary>
Triggered = 1,
/// <summary>
/// The reconnection is in progress.
/// </summary>
Reconnecting = 2,
/// <summary>
/// The reconnect handler is disposed and can not be used for further reconnect attempts.
/// </summary>
Disposed = 4
};
/// <summary>
/// Create a reconnect handler.
/// </summary>
/// <param name="reconnectAbort">Set to <c>true</c> to allow reconnect abort if keep alive recovered.</param>
public SessionReconnectHandler(bool reconnectAbort = false)
/// <param name="maxReconnectPeriod">
/// The upper limit for the reconnect period after exponential backoff.
/// -1 (default) indicates that no exponential backoff should be used.
/// </param>
public SessionReconnectHandler(bool reconnectAbort = false, int maxReconnectPeriod = -1)
{
m_reconnectAbort = reconnectAbort;
m_reconnectTimer = new Timer(OnReconnect, this, Timeout.Infinite, Timeout.Infinite);
m_state = ReconnectState.Ready;
m_cancelReconnect = false;
m_updateFromServer = false;
m_baseReconnectPeriod = DefaultReconnectPeriod;
m_maxReconnectPeriod = maxReconnectPeriod < 0 ? -1 :
Math.Max(MinReconnectPeriod, Math.Min(maxReconnectPeriod, MaxReconnectPeriod));
m_random = new Random();
}
#region IDisposable Members
@@ -71,6 +124,7 @@ namespace Opc.Ua.Client
m_reconnectTimer.Dispose();
m_reconnectTimer = null;
}
m_state = ReconnectState.Disposed;
}
}
}
@@ -83,32 +137,139 @@ namespace Opc.Ua.Client
/// <value>The session.</value>
public ISession Session => m_session;
/// <summary>
/// The internal state of the reconnect handler.
/// </summary>
public ReconnectState State
{
get
{
lock (m_lock)
{
if (m_reconnectTimer == null)
{
return ReconnectState.Disposed;
}
return m_state;
}
}
}
/// <summary>
/// Cancel a reconnect in progress.
/// </summary>
public void CancelReconnect()
{
lock (m_lock)
{
if (m_reconnectTimer == null)
{
return;
}
if (m_state == ReconnectState.Triggered)
{
m_session = null;
EnterReadyState();
return;
}
m_cancelReconnect = true;
}
}
/// <summary>
/// Begins the reconnect process.
/// </summary>
public void BeginReconnect(ISession session, int reconnectPeriod, EventHandler callback)
public ReconnectState BeginReconnect(ISession session, int reconnectPeriod, EventHandler callback)
{
BeginReconnect(session, null, reconnectPeriod, callback);
return BeginReconnect(session, null, reconnectPeriod, callback);
}
/// <summary>
/// Begins the reconnect process using a reverse connection.
/// </summary>
public void BeginReconnect(ISession session, ReverseConnectManager reverseConnectManager, int reconnectPeriod, EventHandler callback)
public ReconnectState BeginReconnect(ISession session, ReverseConnectManager reverseConnectManager, int reconnectPeriod, EventHandler callback)
{
lock (m_lock)
{
if (m_reconnectTimer != null)
if (m_reconnectTimer == null)
{
throw new ServiceResultException(StatusCodes.BadInvalidState);
}
m_session = session;
m_reconnectFailed = false;
m_reconnectPeriod = reconnectPeriod;
m_callback = callback;
m_reverseConnectManager = reverseConnectManager;
m_reconnectTimer = new System.Threading.Timer(OnReconnect, null, reconnectPeriod, Timeout.Infinite);
// cancel reconnect requested, if possible
if (session == null)
{
if (m_state == ReconnectState.Triggered)
{
m_session = null;
EnterReadyState();
return m_state;
}
// reconnect already in progress, schedule cancel
m_cancelReconnect = true;
return m_state;
}
// set reconnect period within boundaries
m_baseReconnectPeriod = reconnectPeriod = CheckedReconnectPeriod(reconnectPeriod);
// ignore subsequent trigger requests
if (m_state == ReconnectState.Ready)
{
m_session = session;
m_reconnectFailed = false;
m_cancelReconnect = false;
m_callback = callback;
m_reverseConnectManager = reverseConnectManager;
m_reconnectTimer.Change(JitteredReconnectPeriod(reconnectPeriod), Timeout.Infinite);
m_reconnectPeriod = CheckedReconnectPeriod(reconnectPeriod, true);
m_state = ReconnectState.Triggered;
return m_state;
}
// if triggered, reset timer if requested reconnect period is smaller
if (m_state == ReconnectState.Triggered && reconnectPeriod < m_reconnectPeriod)
{
m_reconnectTimer.Change(JitteredReconnectPeriod(reconnectPeriod), Timeout.Infinite);
m_reconnectPeriod = CheckedReconnectPeriod(reconnectPeriod, true);
}
return m_state;
}
}
/// <summary>
/// Returns the reconnect period with a random jitter.
/// </summary>
public virtual int JitteredReconnectPeriod(int reconnectPeriod)
{
// The factors result in a jitter of 10%.
const int JitterResolution = 1000;
const int JitterFactor = 10;
int jitter = (reconnectPeriod * m_random.Next(-JitterResolution, JitterResolution)) /
(JitterResolution * JitterFactor);
return reconnectPeriod + jitter;
}
/// <summary>
/// Returns the reconnect period within the min and max boundaries.
/// </summary>
public virtual int CheckedReconnectPeriod(int reconnectPeriod, bool exponentialBackoff = false)
{
// exponential backoff is controlled by m_maxReconnectPeriod
if (m_maxReconnectPeriod > MinReconnectPeriod)
{
if (exponentialBackoff)
{
reconnectPeriod *= 2;
}
return Math.Min(Math.Max(reconnectPeriod, MinReconnectPeriod), m_maxReconnectPeriod);
}
else
{
return Math.Max(reconnectPeriod, MinReconnectPeriod);
}
}
#endregion
@@ -123,9 +284,18 @@ namespace Opc.Ua.Client
try
{
// check for exit.
if (m_reconnectTimer == null)
lock (m_lock)
{
return;
if (m_reconnectTimer == null || m_session == null)
{
return;
}
if (m_state != ReconnectState.Triggered)
{
return;
}
// enter reconnecting state
m_state = ReconnectState.Reconnecting;
}
bool keepaliveRecovered = false;
@@ -147,11 +317,7 @@ namespace Opc.Ua.Client
{
lock (m_lock)
{
if (m_reconnectTimer != null)
{
m_reconnectTimer.Dispose();
m_reconnectTimer = null;
}
EnterReadyState();
}
// notify the caller.
@@ -168,12 +334,21 @@ namespace Opc.Ua.Client
// schedule the next reconnect.
lock (m_lock)
{
int adjustedReconnectPeriod = m_reconnectPeriod - (int)DateTime.UtcNow.Subtract(reconnectStart).TotalMilliseconds;
if (adjustedReconnectPeriod <= 0)
if (m_state != ReconnectState.Disposed)
{
adjustedReconnectPeriod = 100;
if (m_cancelReconnect)
{
EnterReadyState();
}
else
{
int adjustedReconnectPeriod = JitteredReconnectPeriod(m_reconnectPeriod) -
(int)DateTime.UtcNow.Subtract(reconnectStart).TotalMilliseconds;
m_reconnectTimer.Change(CheckedReconnectPeriod(adjustedReconnectPeriod), Timeout.Infinite);
m_reconnectPeriod = CheckedReconnectPeriod(m_reconnectPeriod, true);
m_state = ReconnectState.Triggered;
}
}
m_reconnectTimer = new Timer(OnReconnect, null, adjustedReconnectPeriod, Timeout.Infinite);
}
}
@@ -182,15 +357,17 @@ namespace Opc.Ua.Client
/// </summary>
private async Task<bool> DoReconnect()
{
// override operation timeout
var operationTimeout = m_session.OperationTimeout;
// helper to override operation timeout
int operationTimeout = m_session.OperationTimeout;
int reconnectOperationTimeout = m_reconnectPeriod >= DefaultReconnectPeriod ?
m_reconnectPeriod : DefaultReconnectPeriod;
// try a reconnect.
if (!m_reconnectFailed)
{
try
{
m_session.OperationTimeout = m_reconnectPeriod;
m_session.OperationTimeout = reconnectOperationTimeout;
if (m_reverseConnectManager != null)
{
var connection = await m_reverseConnectManager.WaitForConnection(
@@ -229,6 +406,13 @@ namespace Opc.Ua.Client
return false;
}
}
// check if the security configuration may have changed
if (sre.StatusCode == StatusCodes.BadSecurityChecksFailed)
{
m_updateFromServer = true;
Utils.LogInfo("Reconnect failed due to security check. Request endpoint update from server. {0}", sre.Message);
}
}
else
{
@@ -247,24 +431,68 @@ namespace Opc.Ua.Client
try
{
ISession session;
m_session.OperationTimeout = m_reconnectPeriod;
m_session.OperationTimeout = reconnectOperationTimeout;
if (m_reverseConnectManager != null)
{
var connection = await m_reverseConnectManager.WaitForConnection(
new Uri(m_session.Endpoint.EndpointUrl),
m_session.Endpoint.Server.ApplicationUri
).ConfigureAwait(false);
ITransportWaitingConnection connection;
do
{
connection = await m_reverseConnectManager.WaitForConnection(
new Uri(m_session.Endpoint.EndpointUrl),
m_session.Endpoint.Server.ApplicationUri
).ConfigureAwait(false);
if (m_updateFromServer)
{
var endpoint = m_session.ConfiguredEndpoint;
await endpoint.UpdateFromServerAsync(
endpoint.EndpointUrl, connection,
endpoint.Description.SecurityMode,
endpoint.Description.SecurityPolicyUri).ConfigureAwait(false);
m_updateFromServer = false;
connection = null;
}
} while (connection == null);
session = await m_session.SessionFactory.RecreateAsync(m_session, connection).ConfigureAwait(false);
}
else
{
if (m_updateFromServer)
{
var endpoint = m_session.ConfiguredEndpoint;
await endpoint.UpdateFromServerAsync(
endpoint.EndpointUrl,
endpoint.Description.SecurityMode,
endpoint.Description.SecurityPolicyUri).ConfigureAwait(false);
m_updateFromServer = false;
}
session = await m_session.SessionFactory.RecreateAsync(m_session).ConfigureAwait(false);
}
m_session.Close();
m_session = session;
return true;
}
catch (ServiceResultException sre)
{
if (sre.InnerResult?.StatusCode == StatusCodes.BadSecurityChecksFailed)
{
// schedule endpoint update and retry
m_updateFromServer = true;
if (m_maxReconnectPeriod > MinReconnectPeriod &&
m_reconnectPeriod >= m_maxReconnectPeriod)
{
m_reconnectPeriod = m_baseReconnectPeriod;
}
Utils.LogError("Could not reconnect due to failed security check. Request endpoint update from server. {0}", sre.Message);
}
else
{
Utils.LogError("Could not reconnect the Session. {0}", sre.Message);
}
return false;
}
catch (Exception exception)
{
Utils.LogError("Could not reconnect the Session. {0}", exception.Message);
@@ -275,14 +503,31 @@ namespace Opc.Ua.Client
m_session.OperationTimeout = operationTimeout;
}
}
/// <summary>
/// Reset the timer and enter ready state.
/// </summary>
private void EnterReadyState()
{
m_reconnectTimer.Change(Timeout.Infinite, Timeout.Infinite);
m_state = ReconnectState.Ready;
m_cancelReconnect = false;
m_updateFromServer = false;
}
#endregion
#region Private Fields
private object m_lock = new object();
private ISession m_session;
private ReconnectState m_state;
private Random m_random;
private bool m_reconnectFailed;
private bool m_reconnectAbort;
private bool m_cancelReconnect;
private bool m_updateFromServer;
private int m_reconnectPeriod;
private int m_baseReconnectPeriod;
private int m_maxReconnectPeriod;
private Timer m_reconnectTimer;
private EventHandler m_callback;
private ReverseConnectManager m_reverseConnectManager;
Libraries/Opc.Ua.Client/Subscription.cs
+7 -3
View File
@@ -404,11 +404,11 @@ namespace Opc.Ua.Client
/// <summary>
/// If the available sequence numbers of a subscription
/// are republished or acknoledged after a transfer.
/// are republished or acknowledged after a transfer.
/// </summary>
/// <remarks>
/// Default <c>false</c>, set to <c>true</c> if no data loss is important
/// and available publish requests (sequence numbers) that were never acknoledged should be
/// and available publish requests (sequence numbers) that were never acknowledged should be
/// recovered with a republish. The setting is used after a subscription transfer.
/// </remarks>
[DataMember(Name = "RepublishAfterTransfer", Order = 15)]
@@ -1674,6 +1674,10 @@ namespace Opc.Ua.Client
Utils.LogError(e, "Error while raising PublishStateChanged event.");
}
}
// try to send a publish to recover stopped publishing.
int keepAliveInterval = (int)(Math.Min(m_currentPublishingInterval * m_currentKeepAliveCount, Int32.MaxValue));
m_session?.BeginPublish(Math.Min(keepAliveInterval, Int32.MaxValue / 3) * 3);
}
/// <summary>
@@ -1817,7 +1821,7 @@ namespace Opc.Ua.Client
{
if (m_minLifetimeInterval > 0 && m_minLifetimeInterval < m_session.SessionTimeout)
{
Utils.LogWarning("A smaller lifeTime {0}ms than session timeout {1}ms configured for subscription {2}.",
Utils.LogWarning("A smaller minLifetimeInterval {0}ms than session timeout {1}ms configured for subscription {2}.",
m_minLifetimeInterval, m_session.SessionTimeout, Id);
}
Stack/Opc.Ua.Core/Stack/Server/ServerBase.cs
+1 -3
View File
@@ -213,7 +213,7 @@ namespace Opc.Ua
if (ii.UriScheme == url.Scheme)
{
listener = ii;
break;
listener.CreateReverseConnection(url, timeout);
}
}
}
@@ -222,8 +222,6 @@ namespace Opc.Ua
{
throw new ArgumentException("No suitable listener found.", nameof(url));
}
listener.CreateReverseConnection(url, timeout);
}
/// <summary>
Stack/Opc.Ua.Core/Stack/Tcp/UaSCBinaryClientChannel.cs
+8 -6
View File
@@ -155,13 +155,15 @@ namespace Opc.Ua.Bindings
else
{
Socket = m_socketFactory.Create(this, BufferManager, Quotas.MaxBufferSize);
task = Task.Run(async () =>
await (Socket?.BeginConnect(
m_via, m_ConnectCallback, operation,
new CancellationTokenSource(timeout).Token) ?? Task.FromResult(false)).ConfigureAwait(false));
task = Task.Run(async () => {
using (var cts = new CancellationTokenSource(timeout))
{
await (Socket?.BeginConnect(m_via, m_ConnectCallback, operation, cts.Token) ?? Task.FromResult(false)).ConfigureAwait(false);
}
});
}
}
return m_handshakeOperation;
}
@@ -1272,7 +1274,7 @@ namespace Opc.Ua.Bindings
m_queuedOperations = null;
}
}
#endregion
#endregion
#region Message Processing
/// <summary>
Stack/Opc.Ua.Core/Stack/Transport/AsyncResultBase.cs
+7
View File
@@ -76,6 +76,13 @@ namespace Opc.Ua
// signal any waiting threads.
DisposeWaitHandle(true);
// dispose the cancellation token.
if (m_cts != null)
{
Utils.SilentDispose(m_cts);
m_cts = null;
}
}
}
#endregion
Stack/Opc.Ua.Core/Types/Encoders/JsonDecoder.cs
+8 -5
View File
@@ -1445,11 +1445,14 @@ namespace Opc.Ua
return new ExtensionObject(typeId, encodeable);
}
var ostrm = new MemoryStream();
using (var stream = new StreamWriter(ostrm))
using (JsonTextWriter writer = new JsonTextWriter(stream))
using (var ostrm = new MemoryStream())
{
EncodeAsJson(writer, token);
using (var stream = new StreamWriter(ostrm))
using (JsonTextWriter writer = new JsonTextWriter(stream))
{
EncodeAsJson(writer, token);
}
// Close the writer before retrieving the data
return new ExtensionObject(typeId, ostrm.ToArray());
}
}
@@ -1463,7 +1466,7 @@ namespace Opc.Ua
/// Reads an encodeable object from the stream.
/// </summary>
/// <param name="fieldName">The encodeable object field name</param>
/// <param name="systemType">The system type of the encopdeable object to be read</param>
/// <param name="systemType">The system type of the encodeable object to be read</param>
/// <param name="encodeableTypeId">The TypeId for the <see cref="IEncodeable"/> instance that will be read.</param>
/// <returns>An <see cref="IEncodeable"/> object that was read from the stream.</returns>
public IEncodeable ReadEncodeable(string fieldName, System.Type systemType, ExpandedNodeId encodeableTypeId = null)