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