LineQualityResponder.cs

using System;
using System.Collections.ObjectModel;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using Microsoft.Extensions.Logging;
using OSDP.Net.Connections;
using OSDP.Net.Messages;
using OSDP.Net.Model;
using OSDP.Net.Model.ReplyData;
 
namespace OSDP.Net.LineQuality
{
    public sealed class LineQualityResponder
    {
        private static readonly TimeSpan IdleReadWait = TimeSpan.FromSeconds(1);
        private static readonly TimeSpan ReconnectDelay = TimeSpan.FromMilliseconds(250);

        private const int ReadFaultsBeforeReconnect = 3;
 
        private readonly IRetunableOsdpConnection _connection;
        private readonly byte _address;
        private readonly ILogger _logger;
 
        private DateTime _lastValidCommand = DateTime.MinValue;
        private int _consecutiveReadFaults;

        public LineQualityResponder(IRetunableOsdpConnection connection,
            byte address = LineQualityProtocol.TestAddress, ILoggerFactory loggerFactory = null)
        {
            _connection = connection ?? throw new ArgumentNullException(nameof(connection));
            _address = address;
            _logger = loggerFactory?.CreateLogger<LineQualityResponder>();
        }

        public TimeSpan AutoRevertTimeout { get; set; } = TimeSpan.FromSeconds(30);

        public TimeSpan BaudRateSettleDelay { get; set; } = TimeSpan.FromMilliseconds(100);

        public int CurrentBaudRate => _connection.BaudRate;

        public event EventHandler<LineQualityExchangeEventArgs> ExchangeCompleted;

        public event EventHandler<LineQualityBaudRateChangedEventArgs> BaudRateChanged;

        public async Task RunAsync(CancellationToken token)
        {
            if (!_connection.IsOpen)
            {
                await _connection.Open().ConfigureAwait(false);
            }
 
            // An ACU discards its buffers before each command because it drives a strict
            // command/reply cycle. A responder must not: the next command can already be arriving
            // while this reply is being written.
            _connection.DiscardBuffersBeforeWrite = false;
            _lastValidCommand = DateTime.UtcNow;
 
            _logger?.LogInformation(
                "Line quality responder listening at address {Address} on {Connection}, {BaudRate} baud",
                _address, _connection, _connection.BaudRate);
 
            while (!token.IsCancellationRequested)
            {
                var command = await ReadCommand(token).ConfigureAwait(false);
                if (command == null)
                {
                    RevertBaudRateIfIdle();
                    continue;
                }
 
                _lastValidCommand = DateTime.UtcNow;
 
                try
                {
                    await Respond(command, token).ConfigureAwait(false);
                }
                catch (OperationCanceledException)
                {
                    throw;
                }
                catch (Exception exception)
                {
                    _logger?.LogError(exception, "Failed to reply to a line quality command");
                }
            }
        }

        private async Task<IncomingMessage> ReadCommand(CancellationToken token)
        {
            var buffer = new Collection<byte>();
 
            if (!await WaitForStartOfMessage(buffer, token).ConfigureAwait(false))
            {
                return null;
            }
 
            if (!await Bus.WaitForMessageLength(_connection, buffer, token).ConfigureAwait(false))
            {
                return null;
            }
 
            if (!await Bus.WaitForRestOfMessage(_connection, buffer, Bus.ExtractMessageLength(buffer), token)
                    .ConfigureAwait(false))
            {
                return null;
            }
 
            IncomingMessage command;
            try
            {
                command = new IncomingMessage(buffer.ToArray(), ClearTextChannel.Instance);
            }
            catch (Exception exception)
            {
                // A frame mangled badly enough to fail parsing is indistinguishable from line
                // noise. Stay silent and let the controller record a timeout.
                _logger?.LogDebug(exception, "Discarded an unparsable frame");
                return null;
            }
 
            if (!command.IsDataCorrect)
            {
                _logger?.LogDebug("Discarded a frame that failed its integrity check");
                return null;
            }
 
            return command.Address == _address ? command : null;
        }

        private async Task<bool> WaitForStartOfMessage(Collection<byte> buffer, CancellationToken token)
        {
            var deadline = DateTime.UtcNow + IdleReadWait;
            var readBuffer = new byte[1];
 
            while (DateTime.UtcNow < deadline)
            {
                token.ThrowIfCancellationRequested();
 
                int bytesRead;
                try
                {
                    bytesRead = await ReadWithTimeout(readBuffer, IdleReadWait, token).ConfigureAwait(false);
                }
                catch (Exception exception)
                {
                    await HandleReadFault(exception, token).ConfigureAwait(false);
                    return false;
                }
 
                _consecutiveReadFaults = 0;
 
                // A genuinely idle line, which is the normal case between commands.
                if (bytesRead == 0) return false;
 
                if (readBuffer[0] != Message.StartOfMessage) continue;
 
                buffer.Add(readBuffer[0]);
                return true;
            }
 
            return false;
        }

        private async Task<int> ReadWithTimeout(byte[] buffer, TimeSpan timeout, CancellationToken token)
        {
            using var timeoutSource = new CancellationTokenSource(timeout);
            using var linkedSource =
                CancellationTokenSource.CreateLinkedTokenSource(token, timeoutSource.Token);
 
            try
            {
                return await _connection.ReadAsync(buffer, linkedSource.Token).ConfigureAwait(false);
            }
            catch (Exception exception) when (timeoutSource.IsCancellationRequested &&
                                              IsTimeoutSignal(exception))
            {
                // Our own read timeout expired: the line was simply quiet. Cancellation of the run
                // still propagates; anything else is a genuine fault and is left to the caller.
                token.ThrowIfCancellationRequested();
                return 0;
            }
        }

        private static bool IsTimeoutSignal(Exception exception) =>
            exception is OperationCanceledException || exception is TimeoutException;
 
        private async Task HandleReadFault(Exception exception, CancellationToken token)
        {
            // A read that fails while the responder is being shut down is a consequence of the
            // shutdown, not a fault worth reopening the port over.
            if (token.IsCancellationRequested) return;
 
            _consecutiveReadFaults++;
 
            _logger?.LogWarning(exception, "Serial read failed ({Count} in a row)",
                _consecutiveReadFaults);
 
            if (_consecutiveReadFaults < ReadFaultsBeforeReconnect) return;
 
            _consecutiveReadFaults = 0;
            await ReopenConnection(token).ConfigureAwait(false);
        }

        private async Task ReopenConnection(CancellationToken token)
        {
            int baudRate = _connection.BaudRate;
 
            try
            {
                await _connection.Close().ConfigureAwait(false);
                await Task.Delay(ReconnectDelay, token).ConfigureAwait(false);
                await _connection.Open().ConfigureAwait(false);
 
                _connection.SetBaudRate(baudRate);
                _connection.DiscardBuffersBeforeWrite = false;
 
                // Warning rather than Information: reopening a port is an abnormal event, and a
                // consumer that filters to Warning is exactly the one that needs to see it.
                _logger?.LogWarning(
                    "Reopened the connection at {BaudRate} baud to clear repeated read failures",
                    baudRate);
            }
            catch (OperationCanceledException)
            {
                throw;
            }
            catch (Exception exception)
            {
                _logger?.LogError(exception, "Could not reopen the connection after read failures");
            }
        }
 
        private async Task Respond(IncomingMessage command, CancellationToken token)
        {
            if (command.Type != (byte)CommandType.ManufacturerSpecific)
            {
                await SendReply(command, new Nak(ErrorCode.UnknownCommandCode), token).ConfigureAwait(false);
                return;
            }
 
            var payload = command.Payload.AsSpan();
 
            if (EchoRequest.TryParse(payload, out var echoRequest))
            {
                await HandleEcho(command, echoRequest, token).ConfigureAwait(false);
                return;
            }
 
            if (BaudRateChange.TryParse(payload, out var baudRateChange))
            {
                await HandleBaudRateChange(command, baudRateChange, token).ConfigureAwait(false);
                return;
            }
 
            await SendReply(command, new Nak(ErrorCode.UnknownCommandCode), token).ConfigureAwait(false);
        }
 
        private async Task HandleEcho(IncomingMessage command, EchoRequest request, CancellationToken token)
        {
            PayloadData reply;
 
            if (!LineQualityProtocol.IsSupportedPattern(request.Pattern))
            {
                reply = new EchoResponse(request.SequenceNumber, EchoStatus.UnsupportedPattern);
            }
            else if (request.DeclaredPayloadLength > LineQualityProtocol.MaxPayloadLength)
            {
                reply = new EchoResponse(request.SequenceNumber, EchoStatus.LengthError);
            }
            else if (!request.IsLengthConsistent)
            {
                // The header promised more data than the frame carried, yet the CRC passed. The
                // frame is internally inconsistent rather than corrupted in transit.
                reply = new Nak(ErrorCode.InvalidCommandLength);
            }
            else
            {
                // Echo the bytes as received. Regenerating them from the pattern would repair a
                // corrupted payload on the way back and hide the very errors being measured.
                reply = new EchoResponse(request.SequenceNumber, request.Data);
            }
 
            await SendReply(command, reply, token).ConfigureAwait(false);
 
            OnExchangeCompleted(new LineQualityExchangeEventArgs(request.SequenceNumber, request.Pattern,
                request.Data.Length, (reply as EchoResponse)?.Status, _connection.BaudRate));
        }
 
        private async Task HandleBaudRateChange(IncomingMessage command, BaudRateChange request,
            CancellationToken token)
        {
            int newBaudRate = 0;
            var status = BaudRateChangeStatus.UnsupportedRate;
 
            try
            {
                newBaudRate = LineQualityProtocol.ToBaudRate(request.BaudRateId);
                status = BaudRateChangeStatus.Success;
            }
            catch (ArgumentOutOfRangeException)
            {
                _logger?.LogWarning("Rejected unsupported baud rate ID {BaudRateId}", request.BaudRateId);
            }
 
            int writtenBytes = await SendReply(command,
                new BaudRateChangeAck(request.SequenceNumber, status), token).ConfigureAwait(false);
 
            if (status != BaudRateChangeStatus.Success) return;
 
            // The acknowledgment goes out at the old rate. Retuning before it has physically left
            // the line would corrupt its tail, so drain first, then observe the settle delay.
            await _connection.WaitForTransmitCompleteAsync(writtenBytes, token).ConfigureAwait(false);
            await Task.Delay(BaudRateSettleDelay, token).ConfigureAwait(false);
 
            _connection.SetBaudRate(newBaudRate);
            _lastValidCommand = DateTime.UtcNow;
 
            _logger?.LogInformation("Switched to {BaudRate} baud", newBaudRate);
            BaudRateChanged?.Invoke(this, new LineQualityBaudRateChangedEventArgs(newBaudRate, false));
        }
 
        private async Task<int> SendReply(IncomingMessage command, PayloadData reply, CancellationToken token)
        {
            var message = new OutgoingReply(command, reply).BuildMessage(ClearTextChannel.Instance);
 
            token.ThrowIfCancellationRequested();
            await _connection.WriteAsync(message).ConfigureAwait(false);
 
            return message.Length;
        }
 
        private void RevertBaudRateIfIdle()
        {
            if (AutoRevertTimeout == Timeout.InfiniteTimeSpan) return;
 
            int fallbackRate = LineQualityProtocol.ToBaudRate(LineQualityBaudRate.Baud9600);
            if (_connection.BaudRate == fallbackRate) return;
            if (DateTime.UtcNow - _lastValidCommand < AutoRevertTimeout) return;
 
            _logger?.LogWarning(
                "No valid command for {Timeout}; reverting from {Current} to {Fallback} baud",
                AutoRevertTimeout, _connection.BaudRate, fallbackRate);
 
            _connection.SetBaudRate(fallbackRate);
            _lastValidCommand = DateTime.UtcNow;
 
            BaudRateChanged?.Invoke(this, new LineQualityBaudRateChangedEventArgs(fallbackRate, true));
        }
 
        private void OnExchangeCompleted(LineQualityExchangeEventArgs args) =>
            ExchangeCompleted?.Invoke(this, args);
    }

    public class LineQualityExchangeEventArgs : EventArgs
    {
        public LineQualityExchangeEventArgs(byte sequenceNumber, TestPattern pattern, int payloadLength,
            EchoStatus? status, int baudRate)
        {
            SequenceNumber = sequenceNumber;
            Pattern = pattern;
            PayloadLength = payloadLength;
            Status = status;
            BaudRate = baudRate;
        }

        public byte SequenceNumber { get; }

        public TestPattern Pattern { get; }

        public int PayloadLength { get; }

        public EchoStatus? Status { get; }

        public int BaudRate { get; }
    }

    public class LineQualityBaudRateChangedEventArgs : EventArgs
    {
        public LineQualityBaudRateChangedEventArgs(int baudRate, bool wasAutoRevert)
        {
            BaudRate = baudRate;
            WasAutoRevert = wasAutoRevert;
        }

        public int BaudRate { get; }

        public bool WasAutoRevert { get; }
    }
}