/* ========================================================================
* Copyright (c) 2005-2025 The OPC Foundation, Inc. All rights reserved.
*
* OPC Foundation MIT License 1.00
*
* Permission is hereby granted, free of charge, to any person
* obtaining a copy of this software and associated documentation
* files (the "Software"), to deal in the Software without
* restriction, including without limitation the rights to use,
* copy, modify, merge, publish, distribute, sublicense, and/or sell
* copies of the Software, and to permit persons to whom the
* Software is furnished to do so, subject to the following
* conditions:
*
* The above copyright notice and this permission notice shall be
* included in all copies or substantial portions of the Software.
* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
* EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES
* OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
* NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT
* HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY,
* WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
* FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR
* OTHER DEALINGS IN THE SOFTWARE.
*
* The complete license agreement can be found here:
* http://opcfoundation.org/License/MIT/1.00/
* ======================================================================*/
using System;
using System.Collections.Generic;
using Microsoft.Extensions.Logging;
namespace Opc.Ua.Server
{
///
/// Mangages a data value queue for a data change monitoredItem
///
public interface IDataChangeQueueHandler : IDisposable
{
///
/// Sets the queue size.
///
/// The new queue size.
/// Whether to discard the oldest values if the queue overflows.
/// Specifies which diagnostics which should be kept in the queue.
void SetQueueSize(uint queueSize, bool discardOldest, DiagnosticsMasks diagnosticsMasks);
///
/// Set the sampling interval of the queue
///
/// the sampling interval
void SetSamplingInterval(double samplingInterval);
///
/// Number of DataValues in the queue
///
int ItemsInQueue { get; }
///
/// Queues a value
///
/// the dataValue
/// the error
/// true of overflow occured
bool QueueValue(in DataValue value, ServiceResult error);
///
/// Dequeues the last item
///
/// true if an item was dequeued
bool PublishSingleValue(
out DataValue value,
out ServiceResult error,
bool noEventLog = false);
}
///
/// Mangages a data value queue for a data change monitoredItem
///
public class DataChangeQueueHandler : IDataChangeQueueHandler
{
///
/// Creates a new Queue
///
/// the id of the monitored item
/// true if a durable queue shall be created
/// the factory for
/// The telemetry context to use to create obvservability instruments
/// Handler for discarded values
public DataChangeQueueHandler(
uint monitoredItemId,
bool createDurable,
IMonitoredItemQueueFactory queueFactory,
ITelemetryContext telemetry,
Action? discardedValueHandler = null)
: this(monitoredItemId, createDurable, queueFactory, telemetry, discardedValueHandler, null)
{
}
///
/// Creates a new Queue with an explicit so
/// the per-sample timestamp used for sampling-interval throttling can
/// be mocked in tests.
///
public DataChangeQueueHandler(
uint monitoredItemId,
bool createDurable,
IMonitoredItemQueueFactory queueFactory,
ITelemetryContext telemetry,
Action? discardedValueHandler,
TimeProvider? timeProvider)
{
m_logger = telemetry.CreateLogger();
m_dataValueQueue = queueFactory.CreateDataChangeQueue(createDurable, monitoredItemId);
m_timeProvider = timeProvider ?? TimeProvider.System;
m_discardedValueHandler = discardedValueHandler!;
m_monitoredItemId = monitoredItemId;
m_discardOldest = false;
m_overflow = default;
m_overflowPending = false;
m_nextSampleTime = 0;
m_samplingInterval = 0;
}
///
/// Create a DatachangeQueueHandler from an existing queue
/// Used for restore after a server restart
///
public DataChangeQueueHandler(
IDataChangeMonitoredItemQueue dataValueQueue,
bool discardOldest,
double samplingInterval,
ITelemetryContext telemetry,
Action? discardedValueHandler = null)
: this(dataValueQueue, discardOldest, samplingInterval, telemetry, discardedValueHandler, null)
{
}
///
/// Create a DatachangeQueueHandler from an existing queue with an
/// explicit .
///
public DataChangeQueueHandler(
IDataChangeMonitoredItemQueue dataValueQueue,
bool discardOldest,
double samplingInterval,
ITelemetryContext telemetry,
Action? discardedValueHandler,
TimeProvider? timeProvider)
{
m_logger = telemetry.CreateLogger();
m_dataValueQueue = dataValueQueue;
m_monitoredItemId = dataValueQueue.QueueSize;
m_discardOldest = discardOldest;
m_discardedValueHandler = discardedValueHandler!;
m_nextSampleTime = 0;
m_overflow = default;
m_overflowPending = false;
m_timeProvider = timeProvider ?? TimeProvider.System;
SetSamplingInterval(samplingInterval);
}
///
/// Sets the queue size.
///
/// The new queue size.
/// Whether to discard the oldest values if the queue overflows.
/// Specifies which diagnostics which should be kept in the queue.
public void SetQueueSize(
uint queueSize,
bool discardOldest,
DiagnosticsMasks diagnosticsMasks)
{
bool queueErrors = (diagnosticsMasks & DiagnosticsMasks.OperationAll) != 0;
m_discardOldest = discardOldest;
// copy existing values.
List? existingValues = null;
List? existingErrors = null;
if (ItemsInQueue > 0)
{
existingValues = new List((int)queueSize);
existingErrors = new List((int)queueSize);
while (PublishSingleValue(out DataValue value, out ServiceResult error, true))
{
existingValues.Add(value);
existingErrors.Add(error);
}
}
m_dataValueQueue.ResetQueue(queueSize, queueErrors);
m_overflow = default;
m_overflowPending = false;
// requeue the data.
if (existingValues != null)
{
for (int ii = 0; ii < existingValues.Count; ii++)
{
Enqueue(existingValues[ii], existingErrors![ii]);
}
}
}
///
/// Set the sampling interval of the queue
///
/// the sampling interval
public void SetSamplingInterval(double samplingInterval)
{
// substract the previous sampling interval.
if (m_samplingInterval < m_nextSampleTime)
{
m_nextSampleTime -= m_samplingInterval;
}
// calculate the next sampling interval.
m_samplingInterval = (long)samplingInterval;
if (m_samplingInterval > 0)
{
m_nextSampleTime += m_samplingInterval;
}
else
{
m_nextSampleTime = 0;
}
}
///
/// Number of DataValues in the queue
///
public int ItemsInQueue => m_dataValueQueue.ItemsInQueue;
///
/// Queues a value
///
/// the dataValue
/// the error
/// true of overflow occured
public bool QueueValue(in DataValue value, ServiceResult error)
{
long now = m_timeProvider.GetTimestampMilliseconds();
if (m_dataValueQueue.ItemsInQueue > 0)
{
// check if too soon for another sample.
if (now < m_nextSampleTime)
{
if (m_logger.IsEnabled(LogLevel.Trace))
{
m_dataValueQueue.TryPeekLastValue(out DataValue overwrittenValue);
m_logger.LogTrace(
"OVERWRITTEN VALUE (TOO SOON FOR ANOTHER SAMPLE): Value={Value} CODE={Code}<{Code:X8}> SamplingInterval={SamplingInterval}" +
"QueueValueCall {Now} NextSampleTime {NextSampleTime}",
overwrittenValue.WrappedValue,
overwrittenValue.StatusCode.Code,
value.StatusCode.Code,
m_samplingInterval,
now,
m_nextSampleTime);
}
m_dataValueQueue.OverwriteLastValue(value, error);
m_discardedValueHandler?.Invoke();
return false;
}
}
// update next sample time.
if (m_nextSampleTime > 0)
{
long delta = now - m_nextSampleTime;
if (m_samplingInterval > 0 && delta >= 0)
{
m_nextSampleTime += ((delta / m_samplingInterval) + 1) * m_samplingInterval;
}
}
else
{
m_nextSampleTime = now + m_samplingInterval;
}
// queue next value.
return Enqueue(value, error);
}
///
/// Deques the last item
///
public bool PublishSingleValue(
out DataValue value,
out ServiceResult error,
bool noEventLog = false)
{
if (m_dataValueQueue.Dequeue(out value, out error))
{
if (m_overflowPending && m_overflow == value)
{
SetOverflowBit(ref value, ref error);
m_overflow = default;
m_overflowPending = false;
}
if (!noEventLog && m_logger.IsEnabled(LogLevel.Trace))
{
m_logger.LogTrace(
"DEQUEUE VALUE: Value={Value} CODE={Code}<{Code:X8}> OVERFLOW={Overflow}",
value.WrappedValue,
value.StatusCode.Code,
value.StatusCode.Code,
value.StatusCode.Overflow);
}
return true;
}
return false;
}
///
/// Enque value
///
/// true of overflow occured
///
private bool Enqueue(DataValue value, ServiceResult error)
{
// check for empty queue.
if (m_dataValueQueue.ItemsInQueue == 0)
{
if (m_logger.IsEnabled(LogLevel.Trace))
{
m_logger.LogTrace("ENQUEUE VALUE: Value={Value}", value.WrappedValue);
}
m_dataValueQueue.Enqueue(value, error);
return false;
}
// check if the latest value has initial dummy data
if (m_dataValueQueue.TryPeekLastValue(out DataValue lastValue) &&
lastValue.StatusCode == StatusCodes.BadWaitingForInitialData)
{
// overwrite the last value
m_dataValueQueue.OverwriteLastValue(value, error);
return false;
}
// check if queue is full.
if (m_dataValueQueue.ItemsInQueue == m_dataValueQueue.QueueSize)
{
m_discardedValueHandler?.Invoke();
if (!m_discardOldest)
{
if (m_dataValueQueue.TryPeekLastValue(out DataValue peekedLast))
{
ServerUtils.ReportDiscardedValue(
default,
m_monitoredItemId,
peekedLast);
}
//set overflow bit in newest value
m_overflow = value;
m_overflowPending = true;
// overwrite last value
m_dataValueQueue.OverwriteLastValue(value, error);
return true;
}
// remove oldest value.
if (m_dataValueQueue.Dequeue(out DataValue discardedValue, out _))
{
ServerUtils.ReportDiscardedValue(default, m_monitoredItemId, discardedValue);
}
else
{
throw new ServiceResultException(
StatusCodes.BadInternalError,
"Error queueing DataValue. DataValueQueue was full but it was not possible to discard the oldest value.");
}
//set overflow bit in oldest value
if (m_dataValueQueue.TryPeekOldestValue(out DataValue oldestValue))
{
m_overflow = oldestValue;
m_overflowPending = true;
}
m_dataValueQueue.Enqueue(value, error);
return true;
}
else if (m_logger.IsEnabled(LogLevel.Trace))
{
m_logger.LogTrace("ENQUEUE VALUE: Value={Value}", value.WrappedValue);
}
m_dataValueQueue.Enqueue(value, error);
return false;
}
///
/// Sets the overflow bit in the value and error.
///
/// The value to update.
/// The error to update.
private static void SetOverflowBit(ref DataValue value, ref ServiceResult error)
{
value = value.WithStatus(value.StatusCode.SetOverflow(true));
if (error != null)
{
// have to copy before updating because the ServiceResult is invariant.
error = new ServiceResult(
error.NamespaceUri,
error.StatusCode.SetOverflow(true),
error.LocalizedText,
error.AdditionalInfo,
error.InnerResult);
}
}
///
public void Dispose()
{
Dispose(true);
GC.SuppressFinalize(this);
}
///
/// Overridable method to dispose of resources.
///
protected virtual void Dispose(bool disposing)
{
if (disposing)
{
m_dataValueQueue?.Dispose();
}
}
private readonly IDataChangeMonitoredItemQueue m_dataValueQueue;
private readonly ILogger m_logger;
private readonly TimeProvider m_timeProvider;
private readonly uint m_monitoredItemId;
private bool m_discardOldest;
private long m_nextSampleTime;
private long m_samplingInterval;
private readonly Action m_discardedValueHandler;
private DataValue m_overflow;
private bool m_overflowPending;
}
}