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

Publish state change does not correctly handle subscription transferred case (#2281)

Browse Source 
If the subscription is transferred there was still try to publish on the subscription which is not active anymore and failed.
This commit is contained in:
bhnaphade
2023-08-25 20:10:22 +02:00
committed by GitHub
parent c75bf9be00
commit da5db41a34
2 changed files with 70 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Libraries/Opc.Ua.Client/Subscription.cs
+46 -26
View File
@@ -129,6 +129,19 @@ namespace Opc.Ua.Client
}
}
/// <summary>
/// Resets the state of the publish timer and associated message worker.
/// </summary>
private void ResetPublishTimerAndWorkerState()
{
// stop the publish timer.
Utils.SilentDispose(m_publishTimer);
m_publishTimer = null;
m_messageWorkerShutdownEvent.Set();
m_messageWorkerEvent.Set();
m_messageWorkerTask = null;
}
/// <summary>
/// Called by the .NET framework during deserialization.
/// </summary>
@@ -191,11 +204,7 @@ namespace Opc.Ua.Client
{
if (disposing)
{
Utils.SilentDispose(m_publishTimer);
m_publishTimer = null;
m_messageWorkerShutdownEvent.Set();
m_messageWorkerEvent.Set();
m_messageWorkerTask = null;
ResetPublishTimerAndWorkerState();
}
}
#endregion
@@ -1012,12 +1021,7 @@ namespace Opc.Ua.Client
lock (m_cache)
{
// stop the publish timer.
Utils.SilentDispose(m_publishTimer);
m_publishTimer = null;
m_messageWorkerShutdownEvent.Set();
m_messageWorkerEvent.Set();
m_messageWorkerTask = null;
ResetPublishTimerAndWorkerState();
}
// delete the subscription.
@@ -2180,21 +2184,6 @@ namespace Opc.Ua.Client
callback = m_publishStatusChanged;
}
if (callback != null)
{
if (publishStateChangedMask != PublishStateChangedMask.None)
{
try
{
callback(this, new PublishStateChangedEventArgs(publishStateChangedMask));
}
catch (Exception e)
{
Utils.LogError(e, "Error while raising PublishStateChanged event.");
}
}
}
// process new keep alive messages.
FastKeepAliveNotificationEventHandler keepAliveCallback = m_fastKeepAliveCallback;
if (keepAliveToProcess != null && keepAliveCallback != null)
@@ -2269,6 +2258,16 @@ namespace Opc.Ua.Client
{
Utils.LogWarning("StatusChangeNotification received with Status = {0} for SubscriptionId={1}.",
statusChanged.Status.ToString(), Id);
if (statusChanged.Status == StatusCodes.GoodSubscriptionTransferred)
{
publishStateChangedMask |= PublishStateChangedMask.Transferred;
ResetPublishTimerAndWorkerState();
}
else if (statusChanged.Status == StatusCodes.BadTimeout)
{
publishStateChangedMask |= PublishStateChangedMask.Timeout;
}
}
}
}
@@ -2283,6 +2282,17 @@ namespace Opc.Ua.Client
Id, noNotificationsReceived, MaxNotificationsPerPublish);
}
}
if ((callback != null) && (publishStateChangedMask != PublishStateChangedMask.None))
{
try
{
callback(this, new PublishStateChangedEventArgs(publishStateChangedMask));
}
catch (Exception e)
{
Utils.LogError(e, "Error while raising PublishStateChanged event.");
}
}
}
// do any re-publishes.
@@ -2806,6 +2816,16 @@ namespace Opc.Ua.Client
/// A republish for a missing message was issued.
/// </summary>
Republish = 0x08,
/// <summary>
/// The publishing was transferred to another node.
/// </summary>
Transferred = 0x10,
/// <summary>
/// The publishing was timed out
/// </summary>
Timeout = 0x20,
}
#endregion
Tests/Opc.Ua.Client.Tests/SubscriptionTest.cs
+24
View File
@@ -768,6 +768,7 @@ namespace Opc.Ua.Client.Tests
var originSubscriptionFastDataCounters = new int[kTestSubscriptions];
var targetSubscriptionCounters = new int[kTestSubscriptions];
var targetSubscriptionFastDataCounters = new int[kTestSubscriptions];
var originSubscriptionTransferred = new int[kTestSubscriptions];
var subscriptionTemplate = new Subscription(originSession.DefaultSubscription) {
PublishingInterval = 1_000,
LifetimeCount = 30,
@@ -781,6 +782,21 @@ namespace Opc.Ua.Client.Tests
originSubscriptions, originSubscriptionCounters, originSubscriptionFastDataCounters,
kTestSubscriptions, kQueueSize);
if(TransferType.KeepOpen == transferType)
{
foreach (var subscription in originSubscriptions)
{
subscription.PublishStatusChanged += (s, e) => {
TestContext.Out.WriteLine($"PublishStatusChanged: {s.Session.SessionId}-{s.Id}-{e.Status}");
if ((e.Status & PublishStateChangedMask.Transferred)!=0)
{
// subscription transferred
Interlocked.Increment(ref originSubscriptionTransferred[(int)s.Handle]);
}
};
}
}
// settle
await Task.Delay(kDelay).ConfigureAwait(false);
@@ -916,6 +932,14 @@ namespace Opc.Ua.Client.Tests
// wait for some events
await Task.Delay(kDelay).ConfigureAwait(false);
if (TransferType.KeepOpen == transferType)
{
foreach (var subscription in originSubscriptions)
{
// assert if originSubscriptionTransferred is incremented
Assert.AreEqual(1, originSubscriptionTransferred[(int)subscription.Handle]);
}
}
// stop publishing
foreach (var subscription in transferSubscriptions)
{