MessageProcessor.cs


namespace Opc.Ua.Cloud.Publisher
{
    using Microsoft.Extensions.Logging;
    using System;
    using System.Collections.Concurrent;
    using System.Collections.Generic;
    using System.Globalization;
    using System.IO;
    using System.Text;
    using System.Threading;
    using Opc.Ua.Cloud.Publisher.Interfaces;
    using Opc.Ua.Cloud.Publisher.Models;
    using System.Linq;
    using System.Threading.Tasks;

    public class MessageProcessor : IMessageProcessor
    {
        private static ulong _messageID = 0;
        private bool _batchEmpty = true;
        private bool _singleMessageSend = false;
        private int _messageClosingParenthesisSize = 2;
        DateTime _nextSendTime = DateTime.UtcNow;

        private Queue<long> _lastNotificationInBatch = new Queue<long>();
        private int _notificationsInBatch = 0;

        MemoryStream _batchBuffer = new MemoryStream();
        private static BlockingCollection<MessageProcessorModel> _monitoredItemsDataQueue;

        private Dictionary<ushort, string> _metadataMessages = new Dictionary<ushort, string>();
        private object _metadataMessagesLock = new object();

        private Dictionary<ushort, string> _dataSetWriters = new Dictionary<ushort, string>();
        private object _dataSetWritersLock = new object();

        private Timer _metadataTimer;
        private Timer _statusTimer;
        private bool _isRunning = false;

        private static ILogger _logger;
        private readonly IMessageEncoder _encoder;
        private readonly IMessagePublisher _sink;

        public MessageProcessor(
            IMessageEncoder encoder,
            ILoggerFactory loggerFactory,
            IMessagePublisher sink)
        {
            _logger = loggerFactory.CreateLogger("MessageProcessor");
            _encoder = encoder;
            _sink = sink;
        }

        public void ClearMetadataMessageCache()
        {
            lock (_metadataMessagesLock)
            {
                _metadataMessages.Clear();
            }

            lock (_dataSetWritersLock)
            {
                _dataSetWriters.Clear();
            }
        }

        public void Dispose()
        {
            _batchBuffer.Dispose();

            if (_monitoredItemsDataQueue != null)
            {
                // stop the processing loop and any producers cleanly before disposing the queue to avoid
                // ObjectDisposedException races in RunAsync (TryTake) and Enqueue (TryAdd) during shutdown
                _isRunning = false;

                try
                {
                    _monitoredItemsDataQueue.CompleteAdding();
                }
                catch (ObjectDisposedException)
                {
                    // already disposed; nothing to do
                }

                _monitoredItemsDataQueue.Dispose();
            }

            _metadataTimer?.Dispose();
            _statusTimer?.Dispose();
        }

        public static void Enqueue(MessageProcessorModel json)
        {
            // capture the reference locally as the static queue can be disposed/recreated concurrently during shutdown or reconfiguration
            BlockingCollection<MessageProcessorModel> queue = _monitoredItemsDataQueue;
            if (queue == null || queue.IsAddingCompleted)
            {
                return;
            }

            try
            {
                if (queue.TryAdd(json) == false)
                {
                    Diagnostics.Singleton.Info.EnqueueFailureCount++;

                    // log an error message for every 10K messages lost
                    if (Diagnostics.Singleton.Info.EnqueueFailureCount % 10000 == 0)
                    {
                        _logger.LogError($"The internal monitored item message queue is above its capacity of {queue.BoundedCapacity}. We have lost {Diagnostics.Singleton.Info.EnqueueFailureCount} monitored item notifications so far.");
                    }
                }
                else
                {
                    Diagnostics.Singleton.Info.EnqueueCount++;
                    Diagnostics.Singleton.Info.MonitoredItemsQueueCount = queue.Count;
                }
            }
            catch (ObjectDisposedException)
            {
                // the queue was disposed between the null/completed check and the add; drop the notification silently
            }
            catch (InvalidOperationException)
            {
                // adding was marked complete concurrently; drop the notification silently
            }
        }

        public async Task RunAsync(CancellationToken cancellationToken = default)
        {
            if (_isRunning)
            {
                _logger.LogError("Message Processor is already running.");
                return;
            }

            Init();

            _isRunning = true;

            while (true)
            {
                try
                {
                    // read the next message from our queue
                    MessageProcessorModel messageData = new();
                    int timeout = CalculateBatchTimeout(cancellationToken);
                    bool gotItem = _monitoredItemsDataQueue.TryTake(out messageData, timeout, cancellationToken);
                    if (!gotItem)
                    {
                        // timeout or shutdown case (cancellation)
                        if (cancellationToken.IsCancellationRequested)
                        {
                            _logger.LogInformation($"Cancellation requested.");
                            _monitoredItemsDataQueue.CompleteAdding();
                            break;
                        }
                        else
                        {
                            // timeout (i.e. send interval reached). Check if there is something in the buffer and send it now
                            _logger.LogTrace($"Send interval reached at {_nextSendTime}");
                            if (!_batchEmpty)
                            {
                                // send what we have so far
                                await SendBatchAsync(FinishBatch()).ConfigureAwait(false);
                                continue;
                            }
                            else
                            {
                                // nothing to send, reset the clock and keep waiting
                                _logger.LogTrace("Adding {seconds} seconds to current nextSendTime {nextSendTime}...", Settings.Instance.DefaultSendIntervalSeconds, _nextSendTime);
                                _nextSendTime += TimeSpan.FromSeconds(Settings.Instance.DefaultSendIntervalSeconds);
                                continue;
                            }
                        }
                    }
                    else
                    {
                        Diagnostics.Singleton.Info.MonitoredItemsQueueCount = _monitoredItemsDataQueue.Count;
                    }

                    // check if we should send the new item straight away (single message send case or if there are events)
                    if (_singleMessageSend || (messageData.EventValues.Count > 0))
                    {
                        BatchMessage(await JsonEncodeMessageAsync(messageData).ConfigureAwait(false));
                        await SendBatchAsync(FinishBatch()).ConfigureAwait(false);
                    }
                    else
                    {
                        // batch message instead
                        string jsonMessage = await JsonEncodeMessageAsync(messageData).ConfigureAwait(false);
                        int jsonMessageSize = Encoding.UTF8.GetByteCount(jsonMessage);
                        uint hubMessageBufferSize = Settings.Instance.BrokerMessageSize > 0 ? Settings.Instance.BrokerMessageSize : Settings.HubMessageSizeMax;
                        int encodedMessagePropertiesLengthMax = 512;

                        // reduce the message payload by the space occupied by the message properties
                        hubMessageBufferSize -= (uint)encodedMessagePropertiesLengthMax;

                        // check if the message will fit into our batch in principle
                        if (jsonMessageSize > hubMessageBufferSize)
                        {
                            _logger.LogError($"Configured hub message size {hubMessageBufferSize} too small to even fit the generated telemetry message of {jsonMessageSize}. Please adjust. The telemetry message will be discarded!");
                            Diagnostics.Singleton.Info.TooLargeCount++;
                            continue;
                        }

                        // check if the message still fits into out batch, otherwise send what we have so far and start a new batch with the message
                        if ((_batchBuffer.Position + jsonMessageSize + _messageClosingParenthesisSize) < hubMessageBufferSize)
                        {
                            BatchMessage(jsonMessage);
                        }
                        else
                        {
                            await SendBatchAsync(FinishBatch()).ConfigureAwait(false);
                            BatchMessage(jsonMessage);
                        }
                    }
                }
                catch (Exception ex)
                {
                    if (ex is OperationCanceledException)
                    {
                        throw;
                    }
                    else if (ex is ObjectDisposedException || ex is InvalidOperationException)
                    {
                        // the queue has been disposed or marked as completed (e.g. during shutdown or
                        // reconfiguration); stop the loop instead of spinning and logging endlessly
                        _logger.LogInformation("Message queue is no longer available; stopping message processing loop.");
                        break;
                    }
                    else
                    {
                        _logger.LogError(ex, "Error while processing messages!");
                    }
                }
            }

            _isRunning = false;
        }

        private void Init()
        {
            _logger.LogInformation($"Message processing configured with a send interval of {Settings.Instance.DefaultSendIntervalSeconds} sec and a message buffer size of {Settings.Instance.BrokerMessageSize} bytes.");

            // create the queue for monitored items
            _monitoredItemsDataQueue = new BlockingCollection<MessageProcessorModel>((int)Settings.Instance.InternalQueueCapacity);

            _singleMessageSend = Settings.Instance.DefaultSendIntervalSeconds == 0 && Settings.Instance.BrokerMessageSize == 0;

            InitBatch();

            // init our send time
            _nextSendTime = DateTime.UtcNow + TimeSpan.FromSeconds(Settings.Instance.DefaultSendIntervalSeconds);

            if (Settings.Instance.MetadataSendInterval != 0)
            {
                _metadataTimer = new Timer(SendMetadataOnTimerAsync, null, (int)Settings.Instance.MetadataSendInterval * 1000, (int)Settings.Instance.MetadataSendInterval * 1000);
            }

            if (Settings.Instance.SendUAStatus)
            {
                _statusTimer = new Timer(SendStatusOnTimerAsync, null, (int)Settings.Instance.DiagnosticsLoggingInterval * 1000, (int)Settings.Instance.DiagnosticsLoggingInterval * 1000);
            }
        }

        private void BatchMessage(string jsonMessage)
        {
            _batchBuffer.Write(Encoding.UTF8.GetBytes(jsonMessage));
            _batchBuffer.Write(Encoding.UTF8.GetBytes(","));

            _logger.LogDebug($"Batching message with size {Encoding.UTF8.GetByteCount(jsonMessage)}, size is now {_batchBuffer.Position - 1}.");

            _batchEmpty = false;

            _notificationsInBatch++;
        }

        private byte[] FinishBatch()
        {
            // remove the trailing comma and finish the JSON message
            _batchBuffer.Position -= 1;

            // when the NetworkMessage header is omitted, the message is just the JSON array of DataSetMessages,
            // so we only close the array; otherwise we also close the surrounding NetworkMessage object
            _batchBuffer.Write(Encoding.UTF8.GetBytes(Settings.Instance.OmitNetworkMessageHeader ? "]" : "]}"));

            _lastNotificationInBatch.Enqueue(_notificationsInBatch);

            // calc the average for the last 100 batches
            if (_lastNotificationInBatch.Count > 100)
            {
                _lastNotificationInBatch.Dequeue();
            }

            long sum = 0;
            foreach (long notificationInBatch in _lastNotificationInBatch)
            {
                sum += notificationInBatch;
            }

            Diagnostics.Singleton.Info.AverageNotificationsInBrokerMessage = sum / _lastNotificationInBatch.Count;

            return _batchBuffer.ToArray();
        }

        private async Task SendBatchAsync(byte[] bytesToSend)
        {
            if (await _sink.SendMessageAsync(bytesToSend).ConfigureAwait(false))
            {
                _logger.LogDebug($"Sent {bytesToSend.Length} bytes to broker!");
            }

            // reset our batch
            InitBatch();

            // reset our send time
            _nextSendTime = DateTime.UtcNow + TimeSpan.FromSeconds(Settings.Instance.DefaultSendIntervalSeconds);
        }

        private async void SendStatusOnTimerAsync(object state)
        {
            // Same pattern as SendMetadataOnTimerAsync: capture the timer, keep the stop inside the
            // try, and never let an exception escape this async void method - otherwise the timer
            // is left stopped and status messages cease for the process lifetime.
            Timer timer = _statusTimer;
            if (timer == null)
            {
                return;
            }

            try
            {
                // stop the timer while we're sending
                timer.Change(Timeout.Infinite, Timeout.Infinite);

                using (MemoryStream buffer = new MemoryStream())
                {
                    buffer.Write(Encoding.UTF8.GetBytes(_encoder.EncodeStatus(Interlocked.Increment(ref _messageID) - 1)));
                    if (await _sink.SendMetadataAsync(buffer.ToArray()).ConfigureAwait(false))
                    {
                        _logger.LogDebug($"Sent status message to broker!");
                    }
                }
            }
            catch (Exception ex)
            {
                _logger.LogError(ex, "Error while sending status message.");
            }
            finally
            {
                // restart the timer
                try
                {
                    _statusTimer?.Change((int)Settings.Instance.DiagnosticsLoggingInterval * 1000, (int)Settings.Instance.DiagnosticsLoggingInterval * 1000);
                }
                catch (ObjectDisposedException)
                {
                    // shutting down - nothing to restart
                }
            }
        }

        // Sends an OPC UA PubSub connection (discovery) message describing all DataSets currently being published.
        // It is sent whenever the set of DataSets changes (which also covers the initial publish), together with the
        // metadata messages - not on a periodic timer.
        private async Task SendConnectionMessageAsync()
        {
            try
            {
                Dictionary<ushort, string> dataSetWriters;
                lock (_dataSetWritersLock)
                {
                    dataSetWriters = new Dictionary<ushort, string>(_dataSetWriters);
                }

                using (MemoryStream buffer = new MemoryStream())
                {
                    buffer.Write(Encoding.UTF8.GetBytes(_encoder.EncodeConnection(Interlocked.Increment(ref _messageID) - 1, dataSetWriters)));
                    if (await _sink.SendMetadataAsync(buffer.ToArray()).ConfigureAwait(false))
                    {
                        _logger.LogDebug($"Sent connection message to broker!");
                    }
                }
            }
            catch (Exception ex)
            {
                _logger.LogError(ex, "Error while sending connection message.");
            }
        }

        private async void SendMetadataOnTimerAsync(object state)
        {
            // Capture the timer once: Dispose() (see Stop()) nulls the field, and this method is
            // async void, so a NullReferenceException here would be unobservable - the timer would
            // simply never be restarted and metadata would stop for the process lifetime.
            Timer timer = _metadataTimer;
            if (timer == null)
            {
                return;
            }

            try
            {
                // stop the timer while we're sending
                // NOTE: this MUST be inside the try. It used to sit above it, so any exception it
                // raised skipped the finally below and left the timer stopped forever - exactly one
                // metadata batch was sent after startup and then never again.
                timer.Change(Timeout.Infinite, Timeout.Infinite);

                KeyValuePair<ushort, string>[] currentMessages = null;
                lock (_metadataMessagesLock)
                {
                    if (_metadataMessages.Count > 0)
                    {
                        currentMessages = _metadataMessages.ToArray();
                    }
                }

                if (currentMessages != null)
                {
                    // don't even attempt the batch while the broker is down: with one metadata message per
                    // DataSetWriter this loop can be hundreds/thousands of publishes, which would just produce
                    // an equal number of failures and log entries
                    if (!Diagnostics.Singleton.Info.ConnectedToBroker)
                    {
                        _logger.LogInformation("Skipping sending {count} metadata message(s) as the broker is not connected.", currentMessages.Length);
                    }
                    else
                    {
                        int sent = 0;
                        int failed = 0;

                        foreach (KeyValuePair<ushort, string> metadataMessage in currentMessages)
                        {
                            // stop early if the connection drops part way through the batch
                            if (!Diagnostics.Singleton.Info.ConnectedToBroker)
                            {
                                _logger.LogInformation("Aborting metadata batch after {sent} of {total} message(s) as the broker connection was lost.", sent, currentMessages.Length);
                                break;
                            }

                            // Re-stamp the cached payload: it was encoded once, when this
                            // DataSetWriter was first seen, so resending it verbatim would repeat the
                            // original Timestamp and consumers would treat the republish as a
                            // duplicate rather than a fresh sample.
                            if (await SendMetadataMessageAsync(RefreshMetadataTimestamp(metadataMessage.Value), metadataMessage.Key).ConfigureAwait(false))
                            {
                                sent++;
                            }
                            else
                            {
                                failed++;
                            }

                            // throttle slightly so we don't overwhelm the broker's flow control / rate limits
                            // with a large burst of metadata publishes (which can cause it to drop the connection)
                            await Task.Delay(5).ConfigureAwait(false);
                        }

                        if (failed > 0)
                        {
                            // Count only what the sink confirmed it published, so this cannot report
                            // success while the broker is silently rejecting or dropping the messages.
                            _logger.LogWarning("Metadata send: {sent} of {total} message(s) reached topic {topic}, {failed} FAILED. See the errors above for the cause.", sent, currentMessages.Length, Settings.Instance.BrokerMetadataTopic, failed);
                        }
                        else if (sent > 0)
                        {
                            _logger.LogInformation("Sent {sent} metadata message(s) to topic {topic}; next send in {interval}s.", sent, Settings.Instance.BrokerMetadataTopic, Settings.Instance.MetadataSendInterval);
                        }
                    }
                }
                else
                {
                    _logger.LogInformation("No metadata messages to send yet; the cache is populated when telemetry for a DataSetWriter first arrives.");
                }
            }
            catch (Exception ex)
            {
                _logger.LogError(ex, "Error while sending metadata messages.");
            }
            finally
            {
                // restart the timer
                try
                {
                    _metadataTimer?.Change((int)Settings.Instance.MetadataSendInterval * 1000, (int)Settings.Instance.MetadataSendInterval * 1000);
                }
                catch (ObjectDisposedException)
                {
                    // shutting down - nothing to restart
                }
            }
        }

        private async Task<string> JsonEncodeMessageAsync(MessageProcessorModel messageData)
        {
            ushort hash;
            string jsonMessage = _encoder.EncodePayload(messageData, out hash);

            if (Settings.Instance.SendUAConnection)
            {
                // remember the DataSetWriter (keyed by its DataSetWriterId) so the connection message can
                // advertise every DataSet this Publisher writes
                bool isNewDataSetWriter = false;
                lock (_dataSetWritersLock)
                {
                    if (!_dataSetWriters.ContainsKey(hash))
                    {
                        _dataSetWriters[hash] = messageData.ApplicationUri + ";" + messageData.ExpandedNodeId;
                        isNewDataSetWriter = true;
                    }
                }

                // send a connection (discovery) message whenever the set of DataSets changes, i.e. when a new
                // DataSet is published for the first time (this also covers the initial publish at the beginning)
                if (isNewDataSetWriter)
                {
                    await SendConnectionMessageAsync().ConfigureAwait(false);
                }
            }

            if (Settings.Instance.SendUAMetadata)
            {
                string metadataMessage = Settings.Instance.UseCloudEventsMetadataHeader
                    ? _encoder.EncodeCloudEventMetadata(messageData)
                    : _encoder.EncodeMetadata(messageData);

                bool isNewMetadata = false;
                lock (_metadataMessagesLock)
                {
                    if (!_metadataMessages.ContainsKey(hash))
                    {
                        _metadataMessages.Add(hash, metadataMessage);
                        isNewMetadata = true;
                    }
                }

                if (isNewMetadata)
                {
                    await SendMetadataMessageAsync(metadataMessage, hash).ConfigureAwait(false);
                }
            }

            Diagnostics.Singleton.Info.NumberOfEvents++;

            return jsonMessage;
        }

        // Sends a single metadata message, either as an OPC UA PubSub NetworkMessage (the network message header is
        // stitched in front of the payload) or as a CloudEvents message (payload-only, with the header information
        // carried in CloudEvents attributes / transport headers), depending on the UseCloudEventsMetadataHeader setting.
        // Returns whether the message actually reached the broker.
        private async Task<bool> SendMetadataMessageAsync(string metadataPayload, ushort dataSetWriterId)
        {
            ulong messageId = Interlocked.Increment(ref _messageID) - 1;

            byte[] bytesToSend;
            IReadOnlyDictionary<string, string> cloudEventAttributes = null;

            if (Settings.Instance.UseCloudEventsMetadataHeader)
            {
                bytesToSend = Encoding.UTF8.GetBytes(metadataPayload);
                cloudEventAttributes = _encoder.BuildCloudEventMetadataAttributes(messageId, dataSetWriterId);
            }
            else
            {
                using (MemoryStream buffer = new MemoryStream())
                {
                    buffer.Write(Encoding.UTF8.GetBytes(_encoder.EncodeHeader(messageId, true)));
                    buffer.Write(Encoding.UTF8.GetBytes(","));
                    buffer.Write(Encoding.UTF8.GetBytes(metadataPayload));
                    bytesToSend = buffer.ToArray();
                }
            }

            if (await _sink.SendMetadataAsync(bytesToSend, cloudEventAttributes).ConfigureAwait(false))
            {
                _logger.LogDebug($"Sent {bytesToSend.Length} metadata bytes to broker!");
                return true;
            }

            return false;
        }

        // Refreshes the "Timestamp" member of a cached metadata message.
        //
        // The cache holds the fully encoded JSON produced when the DataSetWriter was first seen, so
        // every periodic resend used to carry that original timestamp - only MessageId (added by the
        // network message header) differed. Consumers that key on time therefore saw the republished
        // metadata as a duplicate of a point they already had: InfluxDB overwrites a point with the
        // same measurement + tag set + timestamp, so the periodic sends were silently collapsing
        // into the original row instead of appearing as new ones.
        private static string RefreshMetadataTimestamp(string metadataPayload)
        {
            // The payload is the metadata object without its opening brace (it is stitched onto the
            // network message header), e.g.  "DataSetWriterId":123,"Timestamp":"...","MetaData":{...}
            const string timestampKey = "\"Timestamp\":\"";

            int start = metadataPayload.IndexOf(timestampKey, StringComparison.Ordinal);
            if (start < 0)
            {
                return metadataPayload;
            }

            int valueStart = start + timestampKey.Length;
            int valueEnd = metadataPayload.IndexOf('"', valueStart);
            if (valueEnd < 0)
            {
                return metadataPayload;
            }

            // Match the encoder's format (JsonEncoder.WriteDateTime uses ISO 8601 round-trip).
            string now = DateTime.UtcNow.ToString("yyyy-MM-ddTHH:mm:ss.fffffffZ", CultureInfo.InvariantCulture);

            return string.Concat(
                metadataPayload.AsSpan(0, valueStart),
                now,
                metadataPayload.AsSpan(valueEnd));
        }

        private int CalculateBatchTimeout(CancellationToken cancellationToken = default)
        {
            int timeout;

            // sanity check the send interval
            if (Settings.Instance.DefaultSendIntervalSeconds > 0)
            {
                TimeSpan timeTillNextSend = _nextSendTime.Subtract(DateTime.UtcNow);
                if (timeTillNextSend < TimeSpan.Zero)
                {
                    Diagnostics.Singleton.Info.MissedSendIntervalCount++;

                    // no wait if the send interval was missed
                    timeTillNextSend = TimeSpan.Zero;
                }

                long millisLong = (long)timeTillNextSend.TotalMilliseconds;
                if (millisLong < 0 || millisLong > int.MaxValue)
                {
                    timeout = 0;
                }
                else
                {
                    timeout = (int)millisLong;
                }
            }
            else
            {
                // no wait if shutdown is requested, else infinite wait if send interval is not set
                timeout = cancellationToken.IsCancellationRequested ? 0 : Timeout.Infinite;
            }

            return timeout;
        }

        private void InitBatch()
        {
            _batchEmpty = true;
            _batchBuffer.Position = 0;
            _batchBuffer.SetLength(0);
            _notificationsInBatch = 0;

            string pubSubJSONNetworkMessageHeader = _encoder.EncodeHeader(Interlocked.Increment(ref _messageID) - 1);

            _batchBuffer.Write(Encoding.UTF8.GetBytes(pubSubJSONNetworkMessageHeader));
        }
    }
}