lmxopcua/src/ZB.MOM.WW.LmxOpcUa.Host/MxAccess/MxAccessClient.Subscription.cs

using System;
using System.Threading.Tasks;
using Serilog;
using ZB.MOM.WW.LmxOpcUa.Host.Domain;

namespace ZB.MOM.WW.LmxOpcUa.Host.MxAccess
{
    public sealed partial class MxAccessClient
    {
        public async Task SubscribeAsync(string fullTagReference, Action<string, Vtq> callback)
        {
            _storedSubscriptions[fullTagReference] = callback;
            if (_state != ConnectionState.Connected) return;

            await SubscribeInternalAsync(fullTagReference);
        }

        public async Task UnsubscribeAsync(string fullTagReference)
        {
            _storedSubscriptions.TryRemove(fullTagReference, out _);

            // Don't unsubscribe the probe tag
            if (string.Equals(fullTagReference, _probeTag, StringComparison.OrdinalIgnoreCase))
                return;

            if (_addressToHandle.TryRemove(fullTagReference, out var itemHandle))
            {
                _handleToAddress.TryRemove(itemHandle, out _);

                if (_state == ConnectionState.Connected)
                {
                    await _staThread.RunAsync(() =>
                    {
                        try
                        {
                            _proxy.UnAdviseSupervisory(_connectionHandle, itemHandle);
                            _proxy.RemoveItem(_connectionHandle, itemHandle);
                        }
                        catch (Exception ex)
                        {
                            Log.Warning(ex, "Error unsubscribing {Address}", fullTagReference);
                        }
                    });
                }
            }
        }

        private async Task SubscribeInternalAsync(string address)
        {
            using var scope = _metrics.BeginOperation("Subscribe");
            try
            {
                var itemHandle = await _staThread.RunAsync(() =>
                {
                    var h = _proxy.AddItem(_connectionHandle, address);
                    _proxy.AdviseSupervisory(_connectionHandle, h);
                    return h;
                });

                _handleToAddress[itemHandle] = address;
                _addressToHandle[address] = itemHandle;
                Log.Debug("Subscribed to {Address} (handle={Handle})", address, itemHandle);
            }
            catch (Exception ex)
            {
                scope.SetSuccess(false);
                Log.Error(ex, "Failed to subscribe to {Address}", address);
                throw;
            }
        }

        private async Task ReplayStoredSubscriptionsAsync()
        {
            foreach (var kvp in _storedSubscriptions)
            {
                try
                {
                    await SubscribeInternalAsync(kvp.Key);
                }
                catch (Exception ex)
                {
                    Log.Warning(ex, "Failed to replay subscription for {Address}", kvp.Key);
                }
            }

            Log.Information("Replayed {Count} stored subscriptions", _storedSubscriptions.Count);
        }
    }
}