diff --git a/Libraries/Opc.Ua.Client/Subscription.cs b/Libraries/Opc.Ua.Client/Subscription.cs index daefc23958..2787245bf8 100644 --- a/Libraries/Opc.Ua.Client/Subscription.cs +++ b/Libraries/Opc.Ua.Client/Subscription.cs @@ -129,6 +129,19 @@ namespace Opc.Ua.Client } } + /// + /// Resets the state of the publish timer and associated message worker. + /// + private void ResetPublishTimerAndWorkerState() + { + // stop the publish timer. + Utils.SilentDispose(m_publishTimer); + m_publishTimer = null; + m_messageWorkerShutdownEvent.Set(); + m_messageWorkerEvent.Set(); + m_messageWorkerTask = null; + } + /// /// Called by the .NET framework during deserialization. /// @@ -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. /// Republish = 0x08, + + /// + /// The publishing was transferred to another node. + /// + Transferred = 0x10, + + /// + /// The publishing was timed out + /// + Timeout = 0x20, } #endregion diff --git a/Tests/Opc.Ua.Client.Tests/SubscriptionTest.cs b/Tests/Opc.Ua.Client.Tests/SubscriptionTest.cs index f32f30e60a..393fa0f503 100644 --- a/Tests/Opc.Ua.Client.Tests/SubscriptionTest.cs +++ b/Tests/Opc.Ua.Client.Tests/SubscriptionTest.cs @@ -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) {