/* ======================================================================== * 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 } }