scadalink-design/lmxproxy/src-reference/ZB.MOM.WW.LmxProxy.Host/Implementation/MxAccessClient.ReadWrite.cs

using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using Polly;
using ZB.MOM.WW.LmxProxy.Host.Domain;
using ZB.MOM.WW.LmxProxy.Host.Services;

namespace ZB.MOM.WW.LmxProxy.Host.Implementation
{
    /// <summary>
    ///     Read and write operations for MxAccessClient.
    /// </summary>
    public sealed partial class MxAccessClient
    {
        /// <inheritdoc />
        public async Task<Vtq> ReadAsync(string address, CancellationToken ct = default)
        {
            // Apply retry policy for read operations
            IAsyncPolicy<Vtq> policy = RetryPolicies.CreateReadPolicy<Vtq>();
            return await policy.ExecuteWithRetryAsync(async () =>
                {
                    ValidateConnection();
                    return await ReadSingleValueAsync(address, ct);
                }, $"Read-{address}");
        }

        /// <inheritdoc />
        public async Task<IReadOnlyDictionary<string, Vtq>> ReadBatchAsync(IEnumerable<string> addresses,
            CancellationToken ct = default)
        {
            var addressList = addresses.ToList();
            var results = new Dictionary<string, Vtq>(addressList.Count);

            // Create tasks for parallel reading
            IEnumerable<Task> tasks =
                addressList.Select(address => ReadAddressWithSemaphoreAsync(address, results, ct));

            await Task.WhenAll(tasks);
            return results;
        }

        /// <inheritdoc />
        public async Task WriteAsync(string address, object value, CancellationToken ct = default)
        {
            // Apply retry policy for write operations
            IAsyncPolicy policy = RetryPolicies.CreateWritePolicy();
            await policy.ExecuteWithRetryAsync(async () => { await WriteInternalAsync(address, value, ct); },
                $"Write-{address}");
        }

        /// <inheritdoc />
        public async Task WriteBatchAsync(IReadOnlyDictionary<string, object> values, CancellationToken ct = default)
        {
            // Create tasks for parallel writing
            IEnumerable<Task> tasks = values.Select(kvp => WriteAddressWithSemaphoreAsync(kvp.Key, kvp.Value, ct));

            await Task.WhenAll(tasks);
        }

        /// <inheritdoc />
        public async Task<bool> WriteBatchAndWaitAsync(
            IReadOnlyDictionary<string, object> values,
            string flagAddress,
            object flagValue,
            string responseAddress,
            object responseValue,
            CancellationToken ct = default)
        {
            // Write the batch values
            await WriteBatchAsync(values, ct);

            // Write the flag
            await WriteAsync(flagAddress, flagValue, ct);

            // Wait for the response
            return await WaitForResponseAsync(responseAddress, responseValue, ct);
        }

        #region Private Helper Methods

        /// <summary>
        ///     Validates that the client is connected.
        /// </summary>
        private void ValidateConnection()
        {
            if (!IsConnected)
            {
                throw new InvalidOperationException("Not connected to MxAccess");
            }
        }

        /// <summary>
        ///     Reads a single value from the specified address.
        /// </summary>
        private async Task<Vtq> ReadSingleValueAsync(string address, CancellationToken ct)
        {
            // MxAccess doesn't support direct read - we need to subscribe, get the value, then unsubscribe
            var tcs = new TaskCompletionSource<Vtq>();
            IAsyncDisposable? subscription = null;

            try
            {
                subscription = await SubscribeAsync(new[] { address }, (addr, vtq) => { tcs.TrySetResult(vtq); }, ct);

                return await WaitForReadResultAsync(tcs, ct);
            }
            finally
            {
                if (subscription != null)
                {
                    await subscription.DisposeAsync();
                }
            }
        }

        /// <summary>
        ///     Waits for a read result with timeout.
        /// </summary>
        private async Task<Vtq> WaitForReadResultAsync(TaskCompletionSource<Vtq> tcs, CancellationToken ct)
        {
            using (var cts = new CancellationTokenSource(TimeSpan.FromSeconds(_configuration.ReadTimeoutSeconds)))
            {
                using (ct.Register(() => cts.Cancel()))
                {
                    cts.Token.Register(() => tcs.TrySetException(new TimeoutException("Read timeout")));
                    return await tcs.Task;
                }
            }
        }

        /// <summary>
        ///     Reads an address with semaphore protection for batch operations.
        /// </summary>
        private async Task ReadAddressWithSemaphoreAsync(string address, Dictionary<string, Vtq> results,
            CancellationToken ct)
        {
            await _readSemaphore.WaitAsync(ct);
            try
            {
                Vtq vtq = await ReadAsync(address, ct);
                lock (results)
                {
                    results[address] = vtq;
                }
            }
            catch (Exception ex)
            {
                Logger.Warning(ex, "Failed to read {Address}", address);
                lock (results)
                {
                    results[address] = Vtq.Bad();
                }
            }
            finally
            {
                _readSemaphore.Release();
            }
        }

        /// <summary>
        ///     Internal write implementation.
        /// </summary>
        private async Task WriteInternalAsync(string address, object value, CancellationToken ct)
        {
            var tcs = new TaskCompletionSource<bool>();
            int itemHandle = await SetupWriteOperationAsync(address, value, tcs, ct);

            try
            {
                await WaitForWriteCompletionAsync(tcs, itemHandle, address, ct);
            }
            catch
            {
                await CleanupWriteOperationAsync(itemHandle);
                throw;
            }
        }

        /// <summary>
        ///     Sets up a write operation and returns the item handle.
        /// </summary>
        private async Task<int> SetupWriteOperationAsync(string address, object value, TaskCompletionSource<bool> tcs,
            CancellationToken ct)
        {
            return await Task.Run(() =>
            {
                lock (_lock)
                {
                    ValidateConnectionLocked();
                    return InitiateWriteOperation(address, value, tcs);
                }
            }, ct);
        }

        /// <summary>
        ///     Validates connection while holding the lock.
        /// </summary>
        private void ValidateConnectionLocked()
        {
            if (!IsConnected || _lmxProxy == null)
            {
                throw new InvalidOperationException("Not connected to MxAccess");
            }
        }

        /// <summary>
        ///     Initiates a write operation and returns the item handle.
        /// </summary>
        private int InitiateWriteOperation(string address, object value, TaskCompletionSource<bool> tcs)
        {
            int itemHandle = 0;
            try
            {
                if (_lmxProxy == null)
                {
                    throw new InvalidOperationException("MxAccess proxy is not initialized");
                }

                // Add the item if not already added
                itemHandle = _lmxProxy.AddItem(_connectionHandle, address);

                // Advise the item to enable writing
                _lmxProxy.AdviseSupervisory(_connectionHandle, itemHandle);

                // Track the pending write operation
                TrackPendingWrite(address, itemHandle, tcs);

                // Write the value
                _lmxProxy.Write(_connectionHandle, itemHandle, value, -1); // -1 for no security

                return itemHandle;
            }
            catch (Exception ex)
            {
                CleanupFailedWrite(itemHandle);
                Logger.Error(ex, "Failed to write value to {Address}", address);
                throw;
            }
        }

        /// <summary>
        ///     Tracks a pending write operation.
        /// </summary>
        private void TrackPendingWrite(string address, int itemHandle, TaskCompletionSource<bool> tcs)
        {
            var writeOp = new WriteOperation
            {
                Address = address,
                ItemHandle = itemHandle,
                CompletionSource = tcs,
                StartTime = DateTime.UtcNow
            };
            _pendingWrites[itemHandle] = writeOp;
        }

        /// <summary>
        ///     Cleans up a failed write operation.
        /// </summary>
        private void CleanupFailedWrite(int itemHandle)
        {
            if (itemHandle > 0 && _lmxProxy != null)
            {
                try
                {
                    _lmxProxy.UnAdvise(_connectionHandle, itemHandle);
                    _lmxProxy.RemoveItem(_connectionHandle, itemHandle);
                    _pendingWrites.Remove(itemHandle);
                }
                catch
                {
                }
            }
        }

        /// <summary>
        ///     Waits for write completion with timeout.
        /// </summary>
        private async Task WaitForWriteCompletionAsync(TaskCompletionSource<bool> tcs, int itemHandle, string address,
            CancellationToken ct)
        {
            using (ct.Register(() => tcs.TrySetCanceled()))
            {
                var timeoutTask = Task.Delay(TimeSpan.FromSeconds(_configuration.WriteTimeoutSeconds), ct);
                Task? completedTask = await Task.WhenAny(tcs.Task, timeoutTask);

                if (completedTask == timeoutTask)
                {
                    await HandleWriteTimeoutAsync(itemHandle, address);
                }

                await tcs.Task; // This will throw if the write failed
            }
        }

        /// <summary>
        ///     Handles write timeout by cleaning up resources.
        /// </summary>
        private async Task HandleWriteTimeoutAsync(int itemHandle, string address)
        {
            await CleanupWriteOperationAsync(itemHandle);
            throw new TimeoutException($"Write operation to {address} timed out");
        }

        /// <summary>
        ///     Cleans up a write operation.
        /// </summary>
        private async Task CleanupWriteOperationAsync(int itemHandle)
        {
            await Task.Run(() =>
            {
                lock (_lock)
                {
                    if (_pendingWrites.ContainsKey(itemHandle))
                    {
                        _pendingWrites.Remove(itemHandle);
                        if (_lmxProxy != null)
                        {
                            try
                            {
                                _lmxProxy.UnAdvise(_connectionHandle, itemHandle);
                                _lmxProxy.RemoveItem(_connectionHandle, itemHandle);
                            }
                            catch
                            {
                            }
                        }
                    }
                }
            });
        }

        /// <summary>
        ///     Writes an address with semaphore protection for batch operations.
        /// </summary>
        private async Task WriteAddressWithSemaphoreAsync(string address, object value, CancellationToken ct)
        {
            await _writeSemaphore.WaitAsync(ct);
            try
            {
                await WriteAsync(address, value, ct);
            }
            finally
            {
                _writeSemaphore.Release();
            }
        }

        /// <summary>
        ///     Waits for a specific response value.
        /// </summary>
        private async Task<bool> WaitForResponseAsync(string responseAddress, object responseValue,
            CancellationToken ct)
        {
            var tcs = new TaskCompletionSource<bool>();
            IAsyncDisposable? subscription = null;

            try
            {
                subscription = await SubscribeAsync(new[] { responseAddress }, (addr, vtq) =>
                {
                    if (Equals(vtq.Value, responseValue))
                    {
                        tcs.TrySetResult(true);
                    }
                }, ct);

                // Wait for the response value
                using (ct.Register(() => tcs.TrySetResult(false)))
                {
                    return await tcs.Task;
                }
            }
            finally
            {
                if (subscription != null)
                {
                    await subscription.DisposeAsync();
                }
            }
        }

        /// <summary>
        ///     Gets a human-readable error message for a write error code.
        /// </summary>
        /// <param name="errorCode">The error code.</param>
        /// <returns>The error message.</returns>
        private static string GetWriteErrorMessage(int errorCode)
        {
            return errorCode switch
            {
                1008 => "User lacks proper security for write operation",
                1012 => "Secured write required",
                1013 => "Verified write required",
                _ => $"Unknown error code: {errorCode}"
            };
        }

        #endregion
    }
}