/* ========================================================================
* 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 System.Threading.Tasks;
namespace Opc.Ua.Client.Subscriptions.Engine
{
///
/// Abstraction for the V1 Subscription message cache so that the
/// bridge can feed translated notifications without a direct
/// assembly reference to Opc.Ua.Client.
///
internal interface ISubscriptionMessageSink
{
///
/// Stores a in the V1
/// subscription cache.
///
///
/// The currently available sequence numbers on the server.
///
///
/// The notification message to cache.
///
void SaveMessageInCache(
ArrayOf availableSequenceNumbers,
NotificationMessage message);
}
///
/// Bridges V2 subscription notification callbacks to V1 Subscription
/// event/cache mechanisms, enabling V1 Subscription objects to receive
/// notifications when the V2 engine is active.
///
///
/// The bridge converts V2 record-based notifications
/// (, )
/// into V1 instances and forwards
/// them to an , which is
/// typically implemented by the V1 Subscription class.
///
internal sealed class SubscriptionBridge : ISubscriptionNotificationHandler
{
private readonly ISubscriptionMessageSink m_messageSink;
///
/// Creates a new bridge for the specified V1 message sink.
///
///
/// The sink that receives translated V1 notification messages.
///
public SubscriptionBridge(ISubscriptionMessageSink messageSink)
{
m_messageSink = messageSink
?? throw new ArgumentNullException(nameof(messageSink));
}
///
public ValueTask OnDataChangeNotificationAsync(
ISubscription subscription,
uint sequenceNumber,
DateTime publishTime,
ReadOnlyMemory notification,
PublishState publishStateMask,
IReadOnlyList stringTable)
{
ReadOnlySpan span = notification.Span;
var monitoredItems =
new MonitoredItemNotification[span.Length];
DiagnosticInfo[]? diagnosticInfos = null;
for (int i = 0; i < span.Length; i++)
{
DataValueChange change = span[i];
uint clientHandle =
change.MonitoredItem?.ClientHandle ?? 0;
monitoredItems[i] = new MonitoredItemNotification
{
ClientHandle = clientHandle,
Value = change.Value
};
if (change.DiagnosticInfo != null)
{
diagnosticInfos ??= new DiagnosticInfo[span.Length];
diagnosticInfos[i] = change.DiagnosticInfo;
}
}
var dataChange = new DataChangeNotification
{
MonitoredItems = new ArrayOf(
monitoredItems),
DiagnosticInfos = diagnosticInfos != null
? new ArrayOf(diagnosticInfos)
: []
};
NotificationMessage message = BuildNotificationMessage(
sequenceNumber, publishTime, stringTable,
new ExtensionObject(dataChange));
m_messageSink.SaveMessageInCache(
[], message);
return default;
}
///
public ValueTask OnEventDataNotificationAsync(
ISubscription subscription,
uint sequenceNumber,
DateTime publishTime,
ReadOnlyMemory notification,
PublishState publishStateMask,
IReadOnlyList stringTable)
{
ReadOnlySpan span = notification.Span;
var events = new EventFieldList[span.Length];
for (int i = 0; i < span.Length; i++)
{
EventNotification evt = span[i];
uint clientHandle =
evt.MonitoredItem?.ClientHandle ?? 0;
events[i] = new EventFieldList
{
ClientHandle = clientHandle,
EventFields = evt.Fields
};
}
var eventList = new EventNotificationList
{
Events = new ArrayOf(events)
};
NotificationMessage message = BuildNotificationMessage(
sequenceNumber, publishTime, stringTable,
new ExtensionObject(eventList));
m_messageSink.SaveMessageInCache(
[], message);
return default;
}
///
public ValueTask OnKeepAliveNotificationAsync(
ISubscription subscription,
uint sequenceNumber,
DateTime publishTime,
PublishState publishStateMask)
{
// A keep-alive is a NotificationMessage with no
// notification data entries.
var message = new NotificationMessage
{
SequenceNumber = sequenceNumber,
PublishTime = publishTime
};
m_messageSink.SaveMessageInCache(
[], message);
return default;
}
///
public ValueTask OnSubscriptionStateChangedAsync(
ISubscription subscription,
SubscriptionState state,
PublishState publishStateMask,
System.Threading.CancellationToken ct = default)
{
// The bridge translates V2 notifications into V1 cache
// updates; the V1 subscription class drives its own
// state-change events via its existing PublishStateChanged /
// StateChanged event pipeline, so we have nothing to forward
// here.
return default;
}
///
/// Builds a containing a
/// single notification data extension object.
///
private static NotificationMessage BuildNotificationMessage(
uint sequenceNumber,
DateTime publishTime,
IReadOnlyList stringTable,
ExtensionObject notificationData)
{
var message = new NotificationMessage
{
SequenceNumber = sequenceNumber,
PublishTime = publishTime,
NotificationData = new ArrayOf(
new[] { notificationData })
};
if (stringTable != null && stringTable.Count > 0)
{
string[] table = new string[stringTable.Count];
for (int i = 0; i < stringTable.Count; i++)
{
table[i] = stringTable[i];
}
message.StringTable = new ArrayOf(table);
}
return message;
}
}
}