/* ========================================================================
* 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.Linq;
using System.Threading;
using System.Threading.Tasks;
using Microsoft.Extensions.Logging;
using Opc.Ua.PubSub.Configuration;
using Opc.Ua.PubSub.PublishedData;
namespace Opc.Ua.PubSub
{
///
/// Abstract class that represents a working connection for PubSub
///
internal abstract class UaPubSubConnection : IUaPubSubConnection
{
private readonly List m_publishers;
protected TransportProtocol m_transportProtocol = TransportProtocol.NotAvailable;
///
/// Create new instance of UaPubSubConnection with PubSubConnectionDataType configuration data
///
internal UaPubSubConnection(
UaPubSubApplication parentUaPubSubApplication,
PubSubConnectionDataType pubSubConnectionDataType,
ITelemetryContext telemetry,
ILogger logger)
{
m_logger = logger;
Telemetry = telemetry;
// set the default message context that uses the GlobalContext
MessageContext = ServiceMessageContext.Create(Telemetry);
Application =
parentUaPubSubApplication ??
throw new ArgumentNullException(nameof(parentUaPubSubApplication));
Application.UaPubSubConfigurator.WriterGroupAdded
+= UaPubSubConfigurator_WriterGroupAdded;
PubSubConnectionConfiguration = pubSubConnectionDataType;
m_publishers = [];
if (string.IsNullOrEmpty(pubSubConnectionDataType.Name))
{
pubSubConnectionDataType.Name = "";
m_logger.LogInformation(
"UaPubSubConnection() received a PubSubConnectionDataType object without name. '' will be used");
}
}
///
/// Get the assigned transport protocol for this connection instance
///
public TransportProtocol TransportProtocol => m_transportProtocol;
///
/// Get the configuration object for this PubSub connection
///
public PubSubConnectionDataType PubSubConnectionConfiguration { get; }
///
/// Get reference to
///
public UaPubSubApplication Application { get; }
///
/// Get flag that indicates if the Connection is in running state
///
public bool IsRunning { get; private set; }
///
/// Get/Set the current
///
public IServiceMessageContext MessageContext { get; set; }
///
/// Get the list of current publishers associated with this connection
///
internal IReadOnlyCollection Publishers => m_publishers.AsReadOnly();
protected Lock Lock { get; } = new();
protected ITelemetryContext Telemetry { get; }
///
/// Releases all resources used by the current instance of the class.
///
public void Dispose()
{
Dispose(true);
GC.SuppressFinalize(this);
}
///
/// When overridden in a derived class, releases the unmanaged resources used by that class
/// and optionally releases the managed resources.
///
/// true to release both managed and unmanaged resources; false to release only unmanaged resources.
protected virtual void Dispose(bool disposing)
{
if (disposing)
{
Application.UaPubSubConfigurator.WriterGroupAdded
-= UaPubSubConfigurator_WriterGroupAdded;
Stop();
// free managed resources
foreach (UaPublisher publisher in m_publishers.OfType())
{
publisher.Dispose();
}
m_logger.LogInformation("Connection '{Name}' was disposed.", PubSubConnectionConfiguration.Name);
}
}
///
/// Start Publish/Subscribe jobs associated with this instance
///
public void Start()
{
InternalStart().Wait();
m_logger.LogInformation("Connection '{Name}' was started.", PubSubConnectionConfiguration.Name);
lock (Lock)
{
IsRunning = true;
foreach (IUaPublisher publisher in m_publishers)
{
publisher.Start();
}
}
}
///
/// Stop Publish/Subscribe jobs associated with this instance
///
public void Stop()
{
// Stop publishers and clear IsRunning first so that no new publish operations
// are started while the transport is being shut down.
lock (Lock)
{
IsRunning = false;
foreach (IUaPublisher publisher in m_publishers)
{
publisher.Stop();
}
}
InternalStop().Wait();
m_logger.LogInformation("Connection '{Name}' was stopped.", PubSubConnectionConfiguration.Name);
}
///
/// Determine if the connection has anything to publish -> at least one WriterDataSet is configured as enabled for current writer group
///
public bool CanPublish(WriterGroupDataType writerGroupConfiguration)
{
if (!IsRunning)
{
return false;
}
// check if connection status is operational
if (Application.UaPubSubConfigurator
.FindStateForObject(PubSubConnectionConfiguration) !=
PubSubState.Operational)
{
return false;
}
if (Application.UaPubSubConfigurator
.FindStateForObject(writerGroupConfiguration) != PubSubState.Operational)
{
return false;
}
foreach (DataSetWriterDataType writer in writerGroupConfiguration.DataSetWriters)
{
if (writer.Enabled)
{
return true;
}
}
return false;
}
///
/// Create the network messages built from the provided writerGroupConfiguration
///
/// The writer group configuration
/// The publish state for the writer group.
/// A list of the created from the provided writerGroupConfiguration.
public abstract IList? CreateNetworkMessages(
WriterGroupDataType writerGroupConfiguration,
WriterGroupPublishState state);
///
/// Publish the network message
///
/// The network message that needs to be published.
/// True if send was successful.
public abstract Task PublishNetworkMessageAsync(UaNetworkMessage networkMessage);
///
/// Get flag that indicates if all the network clients are connected
///
public abstract bool AreClientsConnected();
///
/// Get current list of Operational DataSetReaders available in this UaSubscriber component
///
public List GetOperationalDataSetReaders()
{
var readersList = new List();
if (Application.UaPubSubConfigurator
.FindStateForObject(PubSubConnectionConfiguration) !=
PubSubState.Operational)
{
return readersList;
}
foreach (ReaderGroupDataType readerGroup in PubSubConnectionConfiguration.ReaderGroups)
{
if (Application.UaPubSubConfigurator
.FindStateForObject(readerGroup) == PubSubState.Operational)
{
foreach (DataSetReaderDataType reader in readerGroup.DataSetReaders)
{
// check if the reader is properly configured to receive data
if (Application.UaPubSubConfigurator
.FindStateForObject(reader) == PubSubState.Operational)
{
readersList.Add(reader);
}
}
}
}
return readersList;
}
///
/// Perform specific Start tasks
///
protected abstract Task InternalStart();
///
/// Perform specific Stop tasks
///
protected abstract Task InternalStop();
///
/// Processes the decoded and
/// raises the or or or event.
///
/// The network message that was received.
/// The source of the received event.
protected void ProcessDecodedNetworkMessage(UaNetworkMessage networkMessage, string source)
{
if (networkMessage.IsMetaDataMessage)
{
// update configuration of the corresponding reader objects found in this connection configuration
foreach (DataSetReaderDataType reader in GetAllDataSetReaders())
{
bool raiseChangedEvent = false;
lock (Lock)
{
// check if reader's MetaData shall be updated
if (reader.DataSetWriterId != 0 &&
reader.DataSetWriterId == networkMessage.DataSetWriterId &&
(
reader.DataSetMetaData == null ||
!Utils.IsEqual(
reader.DataSetMetaData.ConfigurationVersion,
networkMessage.DataSetMetaData!.ConfigurationVersion)))
{
raiseChangedEvent = true;
}
}
if (raiseChangedEvent)
{
// raise event
var metaDataUpdatedEventArgs = new ConfigurationUpdatingEventArgs
{
ChangedProperty = ConfigurationProperty.DataSetMetaData,
Parent = reader,
NewValue = networkMessage.DataSetMetaData!,
Cancel = false
};
// raise the ConfigurationUpdating event and see if configuration shall be changed
Application.RaiseConfigurationUpdatingEvent(metaDataUpdatedEventArgs);
// check to see if the event handler canceled the save of new MetaData
if (!metaDataUpdatedEventArgs.Cancel)
{
m_logger.LogInformation(
"Connection '{Name}' - The MetaData is updated for DataSetReader '{ReaderName}' with DataSetWriterId={DataSetWriterId}",
source,
reader.Name,
networkMessage.DataSetWriterId);
lock (Lock)
{
reader.DataSetMetaData = networkMessage.DataSetMetaData!;
}
}
}
}
var subscribedDataEventArgs = new SubscribedDataEventArgs
{
NetworkMessage = networkMessage,
Source = source
};
// trigger notification for received DataSet MetaData
Application.RaiseMetaDataReceivedEvent(subscribedDataEventArgs);
m_logger.LogInformation(
"Connection '{Name}' - RaiseMetaDataReceivedEvent() with {Count} data set messages.",
source,
subscribedDataEventArgs.NetworkMessage.DataSetMessages.Count);
}
else if (networkMessage.DataSetMessages != null &&
networkMessage.DataSetMessages.Count > 0)
{
var subscribedDataEventArgs = new SubscribedDataEventArgs
{
NetworkMessage = networkMessage,
Source = source
};
//trigger notification for received subscribed DataSet
Application.RaiseDataReceivedEvent(subscribedDataEventArgs);
m_logger.LogInformation(
"Connection '{Source}' - RaiseNetworkMessageDataReceivedEvent() from source={Source}, with {Count} DataSets",
source, source, subscribedDataEventArgs.NetworkMessage.DataSetMessages.Count);
}
else if (networkMessage is Encoding.UadpNetworkMessage)
{
if (networkMessage is Encoding.UadpNetworkMessage uadpNetworkMessage)
{
if (uadpNetworkMessage.UADPDiscoveryType ==
UADPNetworkMessageDiscoveryType.DataSetWriterConfiguration &&
uadpNetworkMessage
.UADPNetworkMessageType == UADPNetworkMessageType.DiscoveryResponse)
{
var eventArgs = new DataSetWriterConfigurationEventArgs
{
DataSetWriterIds = uadpNetworkMessage.DataSetWriterIds!,
Source = source,
DataSetWriterConfiguration = uadpNetworkMessage
.DataSetWriterConfiguration!,
PublisherId = uadpNetworkMessage.PublisherId,
StatusCodes = uadpNetworkMessage.MessageStatusCodes!
};
//trigger notification for received configuration
Application.RaiseDatasetWriterConfigurationReceivedEvent(eventArgs);
m_logger.LogInformation(
"Connection '{Source}' - RaiseDataSetWriterConfigurationReceivedEvent() from source={Source}, with {Count} DataSetWriterConfiguration",
source, source, eventArgs.DataSetWriterIds!.Length);
}
else if (uadpNetworkMessage.UADPDiscoveryType ==
UADPNetworkMessageDiscoveryType.PublisherEndpoint &&
uadpNetworkMessage
.UADPNetworkMessageType == UADPNetworkMessageType.DiscoveryResponse)
{
var publisherEndpointsEventArgs = new PublisherEndpointsEventArgs
{
PublisherEndpoints = uadpNetworkMessage.PublisherEndpoints,
Source = source,
PublisherId = uadpNetworkMessage.PublisherId,
StatusCode = uadpNetworkMessage.PublisherProvideEndpoints
};
//trigger notification for received publisher endpoints
Application.RaisePublisherEndpointsReceivedEvent(
publisherEndpointsEventArgs);
m_logger.LogInformation(
"Connection '{Source}' - RaisePublisherEndpointsReceivedEvent() from source={Source}, with {Count} PublisherEndpoints",
source, source, publisherEndpointsEventArgs.PublisherEndpoints.Count);
}
}
}
}
///
/// Get all dataset readers defined for this UaSubscriber component
///
protected List GetAllDataSetReaders()
{
var readersList = new List();
foreach (ReaderGroupDataType readerGroup in PubSubConnectionConfiguration.ReaderGroups)
{
readersList.AddRange(readerGroup.DataSetReaders);
}
return readersList;
}
///
/// Get all dataset writers defined for this UaPublisher component
///
protected List GetWriterGroupsDataType()
{
var writerList = new List();
foreach (WriterGroupDataType writerGroup in PubSubConnectionConfiguration.WriterGroups)
{
writerList.AddRange(writerGroup.DataSetWriters);
}
return writerList;
}
///
/// Get data set writer discovery responses
///
protected IList GetDataSetWriterDiscoveryResponses(
ushort[] dataSetWriterIds)
{
var responses = new List();
var writerGroupsIds = PubSubConnectionConfiguration
.WriterGroups
.ToList()
.SelectMany(group => group.DataSetWriters.ToList())
.Select(writer => writer.DataSetWriterId)
.ToList();
foreach (ushort dataSetWriterId in dataSetWriterIds)
{
DataSetWriterConfigurationResponse response;
if (!writerGroupsIds.Contains(dataSetWriterId))
{
response = new DataSetWriterConfigurationResponse
{
DataSetWriterIds = [dataSetWriterId],
StatusCodes = [StatusCodes.BadNotFound]
};
}
else
{
response = new DataSetWriterConfigurationResponse
{
DataSetWriterIds = [dataSetWriterId],
StatusCodes = [StatusCodes.Good],
DataSetWriterConfig = PubSubConnectionConfiguration.WriterGroups.ToList()
.First(group =>
group.DataSetWriters.ToList()
.First(writer => writer.DataSetWriterId == dataSetWriterId) != null)
};
}
responses.Add(response);
}
return responses;
}
///
/// Get the maximum KeepAlive value from all present WriterGroups
///
protected double GetWriterGroupsMaxKeepAlive()
{
double maxKeepAlive = 0;
foreach (WriterGroupDataType writerGroup in PubSubConnectionConfiguration.WriterGroups)
{
if (maxKeepAlive < writerGroup.KeepAliveTime)
{
maxKeepAlive = writerGroup.KeepAliveTime;
}
}
return maxKeepAlive;
}
///
/// Create and return the current DataSet for the provided dataSetWriter according to current WriterGroupPublishState
///
protected DataSet? CreateDataSet(
DataSetWriterDataType dataSetWriter,
WriterGroupPublishState state)
{
DataSet? dataSet = null;
//check if dataSetWriter enabled
if (dataSetWriter.Enabled)
{
bool isDeltaFrame = state.IsDeltaFrame(dataSetWriter, out uint sequenceNumber);
// CollectData throws ArgumentException on null; existing behavior preserved.
dataSet = Application.DataCollector.CollectData(dataSetWriter.DataSetName!);
if (dataSet != null)
{
dataSet.SequenceNumber = sequenceNumber;
dataSet.IsDeltaFrame = isDeltaFrame;
if (isDeltaFrame)
{
dataSet = state.ExcludeUnchangedFields(dataSetWriter, dataSet);
}
}
}
return dataSet;
}
///
/// Handler for event.
///
private void UaPubSubConfigurator_WriterGroupAdded(object? sender, WriterGroupEventArgs e)
{
var pubSubConnectionDataType =
Application.UaPubSubConfigurator
.FindObjectById(e.ConnectionId) as PubSubConnectionDataType;
if (PubSubConnectionConfiguration == pubSubConnectionDataType)
{
var publisher = new UaPublisher(
this,
e.WriterGroupDataType,
Telemetry,
Application.TimeProvider);
m_publishers.Add(publisher);
}
}
#pragma warning disable IDE1006 // Naming Styles
protected ILogger m_logger { get; }
#pragma warning restore IDE1006 // Naming Styles
}
}