/* ======================================================================== * 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.IO; using Microsoft.Extensions.Logging; using Opc.Ua.PubSub.Configuration; using Opc.Ua.PubSub.PublishedData; using Opc.Ua.Security.Certificates; using Opc.Ua.Test; namespace Opc.Ua.PubSub { /// /// A class that runs an OPC UA PubSub application. /// public class UaPubSubApplication : IDisposable { private readonly List m_uaPubSubConnections; private readonly ITelemetryContext m_telemetry; private readonly ILogger m_logger; /// /// Event that is triggered when the receives a message via its active connections /// public event EventHandler? RawDataReceived; /// /// Event that is triggered when the receives and decodes subscribed DataSets /// public event EventHandler? DataReceived; /// /// Event that is triggered when the receives and decodes subscribed DataSet MetaData /// public event EventHandler? MetaDataReceived; /// /// Event that is triggered when the receives and decodes subscribed DataSet PublisherEndpoints /// public event EventHandler? PublisherEndpointsReceived; /// /// Event that is triggered before the configuration is updated with a new MetaData /// The configuration will not be updated if flag is set on true. /// public event EventHandler? ConfigurationUpdating; /// /// Event that is triggered when the receives and decodes subscribed DataSet MetaData /// public event EventHandler? DataSetWriterConfigurationReceived; /// /// Raised when the MQTT broker certificate is validated. /// /// /// Returns whether the broker certificate is valid and trusted. /// public ValidateBrokerCertificateHandler? OnValidateBrokerCertificate; /// /// Initializes a new instance of the class. /// /// The telemetry context to use to create obvservability instruments /// The current implementation of /// used by this instance of pub sub application /// The application id for instance. /// The optional /// used for timer and duration calculations. Defaults to /// when null. private UaPubSubApplication( ITelemetryContext telemetry, IUaPubSubDataStore? dataStore = null, string? applicationId = null, TimeProvider? timeProvider = null) { m_logger = telemetry.CreateLogger(); m_uaPubSubConnections = []; m_telemetry = telemetry; DataStore = dataStore ?? new UaPubSubDataStore(); TimeProvider = timeProvider ?? TimeProvider.System; if (!string.IsNullOrEmpty(applicationId)) { ApplicationId = applicationId!; } else { ApplicationId = $"opcua:{System.Net.Dns.GetHostName()}:{RandomSource.Default.NextInt32(int.MaxValue):D10}"; } DataCollector = new DataCollector(DataStore, m_telemetry); UaPubSubConfigurator = new UaPubSubConfigurator(m_telemetry); UaPubSubConfigurator.ConnectionAdded += UaPubSubConfigurator_ConnectionAdded; UaPubSubConfigurator.ConnectionRemoved += UaPubSubConfigurator_ConnectionRemoved; UaPubSubConfigurator.PublishedDataSetAdded += UaPubSubConfigurator_PublishedDataSetAdded; UaPubSubConfigurator.PublishedDataSetRemoved += UaPubSubConfigurator_PublishedDataSetRemoved; m_logger.LogInformation("An instance of UaPubSubApplication was created."); } /// /// The application id associated with the UA /// public string ApplicationId { get; set; } /// /// Get the list of SupportedTransportProfiles /// public static string[] SupportedTransportProfiles => [Profiles.PubSubUdpUadpTransport, Profiles.PubSubMqttJsonTransport, Profiles .PubSubMqttUadpTransport]; /// /// Get reference to the associated instance. /// public UaPubSubConfigurator UaPubSubConfigurator { get; } /// /// Get reference to current DataStore. Write here all node values needed to be /// published by this PubSubApplication /// public IUaPubSubDataStore DataStore { get; } /// /// Get the read only list of created for this /// Application instance /// public ArrayOf PubSubConnections => m_uaPubSubConnections.ToArrayOf(); /// /// Get reference to current configured DataCollector for this UaPubSubApplication /// internal DataCollector DataCollector { get; } /// /// Gets the used by this PubSub application and /// all components it owns (connections, publishers, discovery, metadata publishers, /// interval runners) for timer and duration calculations. /// public TimeProvider TimeProvider { get; } /// /// Creates a new and associates it with a /// custom implementation of . /// /// The current implementation of /// used by this instance of pub sub application /// The telemetry context to use to create obvservability instruments /// Optional for timer and /// duration calculations. Defaults to when null. /// New instance of public static UaPubSubApplication Create( IUaPubSubDataStore dataStore, ITelemetryContext telemetry, TimeProvider? timeProvider = null) { return Create( new PubSubConfigurationDataType { Enabled = true }, dataStore, telemetry, timeProvider); } /// /// Creates a new by loading the configuration parameters /// from the specified path. /// /// The path of the configuration path. /// The telemetry context to use to create obvservability instruments /// The current implementation of /// used by this instance of pub sub application /// Optional for timer and /// duration calculations. Defaults to when null. /// New instance of /// is null. /// public static UaPubSubApplication Create( string configFilePath, ITelemetryContext telemetry, IUaPubSubDataStore? dataStore = null, TimeProvider? timeProvider = null) { // validate input argument if (configFilePath == null) { throw new ArgumentNullException(nameof(configFilePath)); } if (!File.Exists(configFilePath)) { throw new ArgumentException( "The specified file {0} does not exist", configFilePath); } PubSubConfigurationDataType pubSubConfiguration = UaPubSubConfigurationHelper.LoadConfiguration(configFilePath, telemetry); return Create(pubSubConfiguration, dataStore, telemetry, timeProvider); } /// /// Creates a new by loading the configuration parameters from the /// specified parameter. /// /// Telemetry context to use /// Optional for timer and /// duration calculations. Defaults to when null. /// New instance of public static UaPubSubApplication Create( ITelemetryContext telemetry, TimeProvider? timeProvider = null) { return Create(null!, null, telemetry, timeProvider); } /// /// Creates a new by loading the configuration parameters from the /// specified parameter. /// /// The configuration object. /// Telemetry context to use /// Optional for timer and /// duration calculations. Defaults to when null. /// New instance of public static UaPubSubApplication Create( PubSubConfigurationDataType pubSubConfiguration, ITelemetryContext telemetry, TimeProvider? timeProvider = null) { return Create(pubSubConfiguration, null, telemetry, timeProvider); } /// /// Creates a new by loading the configuration parameters from the /// specified parameter. /// /// The configuration object. /// The current implementation of /// used by this instance of pub sub application /// Telemetry context to use /// Optional for timer and /// duration calculations. Defaults to when null. /// New instance of public static UaPubSubApplication Create( PubSubConfigurationDataType pubSubConfiguration, IUaPubSubDataStore? dataStore, ITelemetryContext telemetry, TimeProvider? timeProvider = null) { // if no argument received, start with empty configuration pubSubConfiguration ??= new PubSubConfigurationDataType { Enabled = true }; var uaPubSubApplication = new UaPubSubApplication( telemetry, dataStore, timeProvider: timeProvider); uaPubSubApplication.UaPubSubConfigurator.LoadConfiguration(pubSubConfiguration); return uaPubSubApplication; } /// /// Start Publish/Subscribe jobs associated with this instance /// public void Start() { m_logger.LogInformation("UaPubSubApplication is starting."); foreach (IUaPubSubConnection connection in m_uaPubSubConnections) { connection.Start(); } m_logger.LogInformation("UaPubSubApplication was started."); } /// /// Stop Publish/Subscribe jobs associated with this instance /// public void Stop() { m_logger.LogInformation("UaPubSubApplication is stopping."); foreach (IUaPubSubConnection connection in m_uaPubSubConnections) { connection.Stop(); } m_logger.LogInformation("UaPubSubApplication is stopped."); } /// /// Raise event /// internal void RaiseRawDataReceivedEvent(RawDataReceivedEventArgs e) { try { RawDataReceived?.Invoke(this, e); } catch (Exception ex) { m_logger.LogError(ex, "UaPubSubApplication.RaiseRawDataReceivedEvent"); } } /// /// Raise DataReceived event /// internal void RaiseDataReceivedEvent(SubscribedDataEventArgs e) { try { DataReceived?.Invoke(this, e); } catch (Exception ex) { m_logger.LogError(ex, "UaPubSubApplication.RaiseDataReceivedEvent"); } } /// /// Raise MetaDataReceived event /// internal void RaiseMetaDataReceivedEvent(SubscribedDataEventArgs e) { try { MetaDataReceived?.Invoke(this, e); } catch (Exception ex) { m_logger.LogError(ex, "UaPubSubApplication.RaiseMetaDataReceivedEvent"); } } /// /// Raise DatasetWriterConfigurationReceived event /// internal void RaiseDatasetWriterConfigurationReceivedEvent( DataSetWriterConfigurationEventArgs e) { try { DataSetWriterConfigurationReceived?.Invoke(this, e); } catch (Exception ex) { m_logger.LogError(ex, "UaPubSubApplication.DatasetWriterConfigurationReceivedEvent"); } } /// /// Raise PublisherEndpointsReceived event /// internal void RaisePublisherEndpointsReceivedEvent(PublisherEndpointsEventArgs e) { try { PublisherEndpointsReceived?.Invoke(this, e); } catch (Exception ex) { m_logger.LogError(ex, "UaPubSubApplication.RaisePublisherEndpointsReceivedEvent"); } } /// /// Raise event /// internal void RaiseConfigurationUpdatingEvent(ConfigurationUpdatingEventArgs e) { try { ConfigurationUpdating?.Invoke(this, e); } catch (Exception ex) { m_logger.LogError(ex, "UaPubSubApplication.RaiseConfigurationUpdatingEvent"); } } /// /// Handler for PublishedDataSetAdded event /// private void UaPubSubConfigurator_PublishedDataSetAdded( object? sender, PublishedDataSetEventArgs e) { DataCollector.AddPublishedDataSet(e.PublishedDataSetDataType); } /// /// Handler for PublishedDataSetRemoved event /// private void UaPubSubConfigurator_PublishedDataSetRemoved( object? sender, PublishedDataSetEventArgs e) { DataCollector.RemovePublishedDataSet(e.PublishedDataSetDataType); } /// /// Handler for ConnectionRemoved event /// private void UaPubSubConfigurator_ConnectionRemoved(object? sender, ConnectionEventArgs e) { IUaPubSubConnection? removedUaPubSubConnection = null; foreach (IUaPubSubConnection connection in m_uaPubSubConnections) { if (connection.PubSubConnectionConfiguration.Equals(e.PubSubConnectionDataType)) { removedUaPubSubConnection = connection; break; } } if (removedUaPubSubConnection != null) { m_uaPubSubConnections.Remove(removedUaPubSubConnection); removedUaPubSubConnection.Dispose(); } } /// /// Handler for ConnectionAdded event /// private void UaPubSubConfigurator_ConnectionAdded(object? sender, ConnectionEventArgs e) { m_uaPubSubConnections.Add(ObjectFactory.CreateConnection( this, e.PubSubConnectionDataType, m_telemetry)); } /// /// 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) { UaPubSubConfigurator.ConnectionAdded -= UaPubSubConfigurator_ConnectionAdded; UaPubSubConfigurator.ConnectionRemoved -= UaPubSubConfigurator_ConnectionRemoved; UaPubSubConfigurator.PublishedDataSetAdded -= UaPubSubConfigurator_PublishedDataSetAdded; UaPubSubConfigurator.PublishedDataSetRemoved -= UaPubSubConfigurator_PublishedDataSetRemoved; Stop(); // free managed resources foreach (IUaPubSubConnection connection in m_uaPubSubConnections) { connection.Dispose(); } m_uaPubSubConnections.Clear(); } } } /// /// A delegate which validates the MQTT broker certificate. /// /// The broker certificate. /// Returns whether the broker certificate is valid and trusted. public delegate bool ValidateBrokerCertificateHandler(Certificate brokerCertificate); }