/* ========================================================================
* 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.Diagnostics.CodeAnalysis;
using System.IO;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using Microsoft.Extensions.Logging;
using Opc.Ua.Security.Certificates;
namespace Opc.Ua.Client
{
///
/// The implementation of a reverse connect client manager.
///
///
/// This reverse connect manager allows to register for reverse connections
/// with various strategies:
/// i) take any connection.
/// ii) filter for a specific application Uri and Url scheme.
/// iii) filter for the Url.
/// Second, any filter can be combined with the Once or Always flag.
///
public class ReverseConnectManager : IDisposable
{
///
/// A default value for reverse hello configurations, if undefined.
///
///
/// This value is used as wait timeout if the value is undefined by a caller.
///
public const int DefaultWaitTimeout = 20000;
///
/// Internal state of the reverse connect manager.
///
private enum ReverseConnectManagerState
{
New = 0,
Stopped = 1,
Started = 2,
Errored = 3
}
///
/// Internal state of the reverse connect host.
///
private enum ReverseConnectHostState
{
New = 0,
Closed = 1,
Open = 2,
Errored = 3
}
///
/// Specify the strategy for the reverse connect registration.
///
[Flags]
public enum ReverseConnectStrategy
{
///
/// Undefined strategy, defaults to Once.
///
Undefined = 0,
///
/// Remove entry after reverse connect callback.
///
Once = 1,
///
/// Always callback on matching url or uri.
///
Always = 2,
///
/// Flag for masking any connection.
///
Any = 0x80,
///
/// Respond to any incoming reverse connection,
/// remove entry after reverse connect callback.
///
AnyOnce = Any | Once,
///
/// Respond to any incoming reverse connection,
/// always callback.
///
AnyAlways = Any | Always
}
///
/// Entry for a client reverse connect registration.
///
private class ReverseConnectInfo
{
public ReverseConnectInfo(ReverseConnectHost reverseConnectHost, bool configEntry)
{
ReverseConnectHost = reverseConnectHost;
State = ReverseConnectHostState.New;
ConfigEntry = configEntry;
}
public ReverseConnectHost ReverseConnectHost;
public ReverseConnectHostState State;
public bool ConfigEntry;
}
///
/// Record to store information on a client
/// registration for a reverse connect event.
///
private class Registration
{
public Registration(
string? serverUri,
Uri endpointUrl,
EventHandler onConnectionWaiting)
: this(endpointUrl, onConnectionWaiting)
{
ServerUri = Utils.ReplaceLocalhost(serverUri);
}
///
/// Register with the server certificate to extract the application Uri.
///
///
/// The first Uri in the subject alternate name field is considered the application Uri.
///
/// The server certificate with the application Uri.
/// The endpoint Url of the server.
/// The connection to use.
public Registration(
Certificate serverCertificate,
Uri endpointUrl,
EventHandler onConnectionWaiting)
: this(endpointUrl, onConnectionWaiting)
{
IReadOnlyList serverUris =
X509Utils.GetApplicationUrisFromCertificate(serverCertificate);
ServerUri = serverUris.Count != 0 ? serverUris[0] : null;
}
private Registration(
Uri endpointUrl,
EventHandler onConnectionWaiting)
{
EndpointUrl = new Uri(Utils.ReplaceLocalhost(endpointUrl.ToString()));
OnConnectionWaiting = onConnectionWaiting;
ReverseConnectStrategy = ReverseConnectStrategy.Once;
}
public readonly string? ServerUri;
public readonly Uri EndpointUrl;
public readonly EventHandler OnConnectionWaiting;
public ReverseConnectStrategy ReverseConnectStrategy;
}
///
/// Obsolete default constructor
///
[Obsolete("Use ReverseConnectManager(ITelemetryContext) instead.")]
public ReverseConnectManager()
// Forwards null into a non-nullable telemetry parameter on the modern ctor;
// preserves the pre-nullable parameterless behavior of this obsolete API.
: this(null!)
{
}
///
/// Initializes the object with default values.
///
public ReverseConnectManager(ITelemetryContext telemetry)
: this(telemetry, timeProvider: null)
{
}
///
/// Initializes the object with default values.
///
/// The telemetry context to use to create obvservability instruments.
/// Optional time provider used for elapsed-time calculations and
/// async timeouts. Defaults to .
public ReverseConnectManager(ITelemetryContext telemetry, TimeProvider? timeProvider)
{
m_timeProvider = timeProvider ?? TimeProvider.System;
m_telemetry = telemetry;
m_logger = telemetry.CreateLogger();
m_state = ReverseConnectManagerState.New;
m_registrations = [];
m_endpointUrls = [];
m_cts = new CancellationTokenSource();
}
///
/// Dispose implementation.
///
public void Dispose()
{
Dispose(true);
GC.SuppressFinalize(this);
}
///
/// An overrideable version of the Dispose.
///
/// true to release both managed and unmanaged resources; false to release only unmanaged resources.
protected virtual void Dispose(bool disposing)
{
// close the watcher.
m_configurationWatcher?.Dispose();
m_configurationWatcher = null;
m_cts?.Dispose();
DisposeHosts();
}
///
/// Raised when the configuration changes.
///
/// The sender.
/// The instance containing the event data.
protected virtual async void OnConfigurationChangedAsync(
object? sender,
ConfigurationWatcherEventArgs args)
{
try
{
ApplicationConfiguration configuration = await ApplicationConfiguration
.LoadAsync(
new FileInfo(args.FilePath),
m_applicationType,
m_configType,
m_telemetry)
.ConfigureAwait(false);
OnUpdateConfiguration(configuration);
}
catch (Exception e)
{
m_logger.LogError(e, "Could not load updated configuration file from: {FilePath}", args.FilePath);
}
}
///
/// Called when the configuration is changed on disk.
///
/// The configuration.
[UnconditionalSuppressMessage("Trimming", "IL2074",
Justification = "The configuration type was loaded with PublicParameterlessConstructor, so GetType() is safe to store.")]
protected virtual void OnUpdateConfiguration(ApplicationConfiguration configuration)
{
// save types for config watcher
m_applicationType = configuration.ApplicationType;
m_configType = configuration.GetType();
// ClientConfiguration and ReverseConnect are nullable on ApplicationConfiguration,
// but the file watcher is only enabled for client configurations that include a
// populated ReverseConnect section, so both are guaranteed non-null here.
OnUpdateConfiguration(configuration.ClientConfiguration!.ReverseConnect!);
}
///
/// Called when the reverse connect configuration is changed.
///
///
/// An empty configuration or null stops service on all configured endpoints.
///
/// The client endpoint configuration.
protected virtual void OnUpdateConfiguration(
ReverseConnectClientConfiguration configuration)
{
bool restartService = false;
lock (m_lock)
{
if (m_configuration != null)
{
StopService();
m_configuration = null;
restartService = true;
}
m_configuration = configuration ?? new ReverseConnectClientConfiguration();
// clear configured endpoints
ClearEndpoints(true);
if (configuration?.ClientEndpoints != null)
{
foreach (ReverseConnectClientEndpoint endpoint in configuration.ClientEndpoints)
{
Uri? uri = Utils.ParseUri(endpoint.EndpointUrl);
if (uri != null)
{
AddEndpointInternal(uri, true);
}
}
}
if (restartService)
{
StartService();
}
}
}
///
/// Open host ports.
///
private void OpenHosts()
{
lock (m_lock)
{
foreach (KeyValuePair host in m_endpointUrls)
{
ReverseConnectInfo value = host.Value;
try
{
if (host.Value.State < ReverseConnectHostState.Open)
{
value.ReverseConnectHost.Open();
value.State = ReverseConnectHostState.Open;
}
}
catch (Exception e)
{
m_logger.LogError(e, "Failed to Open {Uri}.", host.Key);
value.State = ReverseConnectHostState.Errored;
}
}
}
}
///
/// Close host ports.
///
private void CloseHosts()
{
lock (m_lock)
{
foreach (KeyValuePair host in m_endpointUrls)
{
ReverseConnectInfo value = host.Value;
try
{
if (value.State == ReverseConnectHostState.Open)
{
value.ReverseConnectHost.Close();
value.State = ReverseConnectHostState.Closed;
}
}
catch (Exception e)
{
m_logger.LogError(e, "Failed to Close {Uri}.", host.Key);
value.State = ReverseConnectHostState.Errored;
}
}
}
}
///
/// Dispose the hosts;
///
private void DisposeHosts()
{
lock (m_lock)
{
CloseHosts();
m_endpointUrls.Clear();
}
}
///
/// Add endpoint for reverse connection.
///
/// is null.
///
public void AddEndpoint(Uri endpointUrl)
{
if (endpointUrl == null)
{
throw new ArgumentNullException(nameof(endpointUrl));
}
lock (m_lock)
{
if (m_state == ReverseConnectManagerState.Started)
{
throw new ServiceResultException(StatusCodes.BadInvalidState);
}
AddEndpointInternal(endpointUrl, false);
}
}
///
/// Starts the server application.
///
/// The configuration.
/// is null.
///
public void StartService(ApplicationConfiguration configuration)
{
if (configuration == null)
{
throw new ArgumentNullException(nameof(configuration));
}
lock (m_lock)
{
if (m_state == ReverseConnectManagerState.Started)
{
throw new ServiceResultException(StatusCodes.BadInvalidState);
}
try
{
OnUpdateConfiguration(configuration);
StartService();
// monitor the configuration file.
if (!string.IsNullOrEmpty(configuration.SourceFilePath))
{
m_configurationWatcher = new ConfigurationWatcher(configuration, m_telemetry);
m_configurationWatcher.Changed += OnConfigurationChangedAsync;
}
}
catch (Exception e)
{
m_logger.LogError(e, "Unexpected error starting reverse connect manager.");
m_state = ReverseConnectManagerState.Errored;
var error = ServiceResult.Create(
e,
StatusCodes.BadInternalError,
"Unexpected error starting application");
throw new ServiceResultException(error);
}
}
}
///
/// Starts the server application.
///
/// The configuration.
/// is null.
///
public void StartService(ReverseConnectClientConfiguration configuration)
{
if (configuration == null)
{
throw new ArgumentNullException(nameof(configuration));
}
lock (m_lock)
{
if (m_state == ReverseConnectManagerState.Started)
{
throw new ServiceResultException(StatusCodes.BadInvalidState);
}
try
{
m_configurationWatcher = null;
OnUpdateConfiguration(configuration);
OpenHosts();
m_state = ReverseConnectManagerState.Started;
}
catch (Exception e)
{
m_logger.LogError(e, "Unexpected error starting reverse connect manager.");
m_state = ReverseConnectManagerState.Errored;
var error = ServiceResult.Create(
e,
StatusCodes.BadInternalError,
"Unexpected error starting reverse connect manager");
throw new ServiceResultException(error);
}
}
}
///
/// Clears all waiting reverse connection handlers.
///
public void ClearWaitingConnections()
{
lock (m_registrationsLock)
{
m_registrations.Clear();
CancelAndRenewTokenSource();
}
}
///
/// Helper to wait for a reverse connection.
///
[Obsolete("Use WaitForConnectionAsync instead.")]
public Task WaitForConnection(
Uri endpointUrl,
string serverUri,
CancellationToken ct = default)
{
return WaitForConnectionAsync(endpointUrl, serverUri, ct);
}
///
/// Helper to wait for a reverse connection.
///
///
public async Task WaitForConnectionAsync(
Uri endpointUrl,
string? serverUri,
CancellationToken ct = default)
{
var tcs = new TaskCompletionSource();
int hashCode = RegisterWaitingConnection(
endpointUrl,
serverUri,
(sender, e) => tcs.TrySetResult(e),
ReverseConnectStrategy.Once);
async Task ListenForCancelAsync(CancellationToken ct)
{
if (ct == default)
{
int waitTimeout = m_configuration?.WaitTimeout ?? 20000;
if (waitTimeout <= 0)
{
waitTimeout = DefaultWaitTimeout;
}
await Task.Delay(waitTimeout, ct).ConfigureAwait(false);
}
else
{
await Task.Delay(-1, ct).ContinueWith(
_ => { },
ct,
TaskContinuationOptions.None,
TaskScheduler.Default).ConfigureAwait(false);
}
tcs.TrySetCanceled(ct);
}
await Task.WhenAny([tcs.Task, ListenForCancelAsync(ct)]).ConfigureAwait(false);
if (!tcs.Task.IsCompleted || tcs.Task.IsCanceled)
{
UnregisterWaitingConnection(hashCode);
throw new ServiceResultException(
StatusCodes.BadTimeout,
"Waiting for the reverse connection timed out.");
}
return await tcs.Task.ConfigureAwait(false);
}
///
/// Register for a waiting reverse connection.
///
/// The endpoint Url of the reverse connection.
/// Optional. The server application Uri of the reverse connection.
/// The callback
/// The reverse connect callback strategy.
/// is null.
public int RegisterWaitingConnection(
Uri endpointUrl,
string? serverUri,
EventHandler onConnectionWaiting,
ReverseConnectStrategy reverseConnectStrategy)
{
if (endpointUrl == null)
{
throw new ArgumentNullException(nameof(endpointUrl));
}
var registration = new Registration(serverUri, endpointUrl, onConnectionWaiting)
{
ReverseConnectStrategy = reverseConnectStrategy
};
lock (m_registrationsLock)
{
m_registrations.Add(registration);
CancelAndRenewTokenSource();
}
return registration.GetHashCode();
}
///
/// Unregister reverse connection callback.
///
/// The hashcode returned by the registration.
public void UnregisterWaitingConnection(int hashCode)
{
lock (m_registrationsLock)
{
Registration? toRemove = null;
foreach (Registration registration in m_registrations)
{
if (registration.GetHashCode() == hashCode)
{
toRemove = registration;
break;
}
}
if (toRemove != null)
{
m_registrations.Remove(toRemove);
CancelAndRenewTokenSource();
}
}
}
///
/// Called before the server stops
///
private void StopService()
{
ClearWaitingConnections();
lock (m_lock)
{
CloseHosts();
m_state = ReverseConnectManagerState.Stopped;
}
}
///
/// Called to start hosting the reverse connect ports.
///
private void StartService()
{
lock (m_lock)
{
OpenHosts();
m_state = ReverseConnectManagerState.Started;
}
}
///
/// Remove configuration endpoints from list.
///
private void ClearEndpoints(bool configEntry)
{
var newEndpointUrls = new Dictionary();
foreach (KeyValuePair endpoint in m_endpointUrls)
{
if (endpoint.Value.ConfigEntry != configEntry)
{
newEndpointUrls[endpoint.Key] = endpoint.Value;
}
}
m_endpointUrls = newEndpointUrls;
}
///
/// Add endpoint for reverse connection.
///
/// The endpoint Url of the reverse connect client endpoint.
/// Tf this is an entry in the application configuration.
private void AddEndpointInternal(Uri endpointUrl, bool configEntry)
{
var reverseConnectHost = new ReverseConnectHost(m_telemetry);
var info = new ReverseConnectInfo(reverseConnectHost, configEntry);
try
{
m_endpointUrls[endpointUrl] = info;
reverseConnectHost.CreateListener(
endpointUrl,
new ConnectionWaitingHandlerAsync(OnConnectionWaitingAsync),
new EventHandler(OnConnectionStatusChanged));
}
catch (ArgumentException ae)
{
m_logger.LogError(ae, "No listener was found for endpoint {EndpointUrl}.", endpointUrl);
info.State = ReverseConnectHostState.Errored;
}
}
///
/// Raised when a reverse connection is waiting,
/// finds and calls a waiting connection.
///
private async Task OnConnectionWaitingAsync(object sender, ConnectionWaitingEventArgs e)
{
long startTimestamp = m_timeProvider.GetTimestamp();
var holdTime = TimeSpan.FromMilliseconds(m_configuration?.HoldTime ?? 15000);
bool matched = MatchRegistration(sender, e);
while (!matched)
{
m_logger.LogInformation("Holding reverse connection: {ServerUri} {EndpointUrl}", e.ServerUri, e.EndpointUrl);
CancellationToken ct;
lock (m_registrationsLock)
{
ct = m_cts.Token;
}
TimeSpan delay = holdTime - m_timeProvider.GetElapsedTime(startTimestamp);
if (delay > TimeSpan.Zero)
{
await m_timeProvider.Delay(delay, ct)
.ContinueWith(tsk =>
{
if (tsk.IsCanceled)
{
matched = MatchRegistration(sender, e);
if (matched)
{
m_logger.LogInformation(
"Matched reverse connection {ServerUri} {EndpointUrl} after {Duration}ms",
e.ServerUri,
e.EndpointUrl,
(long)m_timeProvider.GetElapsedTime(startTimestamp).TotalMilliseconds);
}
}
},
default,
TaskContinuationOptions.None,
TaskScheduler.Default)
.ConfigureAwait(false);
}
break;
}
m_logger.LogInformation(
"{Action} reverse connection: {ServerUri} {EndpointUrl} after {Duration}ms",
e.Accepted ? "Accepted" : "Rejected",
e.ServerUri,
e.EndpointUrl,
(long)m_timeProvider.GetElapsedTime(startTimestamp).TotalMilliseconds);
}
///
/// Match the waiting connection with a registration, callback registration,
/// return if connection is accepted in event.
///
/// true if a match was found.
private bool MatchRegistration(object sender, ConnectionWaitingEventArgs e)
{
Registration? callbackRegistration = null;
bool found = false;
lock (m_registrationsLock)
{
// first try to match single registrations
foreach (Registration registration in m_registrations
.Where(r => (r.ReverseConnectStrategy & ReverseConnectStrategy.Any) == 0))
{
if (registration.EndpointUrl.Scheme
.Equals(e.EndpointUrl.Scheme, StringComparison.Ordinal) &&
(registration.ServerUri?
.Equals(e.ServerUri, StringComparison.Ordinal) == true ||
registration.EndpointUrl.Authority.Equals(e.EndpointUrl.Authority,
StringComparison.OrdinalIgnoreCase)))
{
callbackRegistration = registration;
e.Accepted = true;
found = true;
m_logger.LogInformation(
"Accepted reverse connection: {ServerUri} {EndpointUrl}",
e.ServerUri,
e.EndpointUrl);
break;
}
}
// now try any registrations.
if (callbackRegistration == null)
{
foreach (Registration registration in m_registrations.Where(r =>
(r.ReverseConnectStrategy & ReverseConnectStrategy.Any) != 0))
{
if (registration.EndpointUrl.Scheme
.Equals(e.EndpointUrl.Scheme, StringComparison.Ordinal))
{
callbackRegistration = registration;
e.Accepted = true;
found = true;
m_logger.LogInformation(
"Accept any reverse connection for approval: {ServerUri} {EndpointUrl}",
e.ServerUri,
e.EndpointUrl);
break;
}
}
}
if (callbackRegistration != null &&
(callbackRegistration.ReverseConnectStrategy &
ReverseConnectStrategy.Once) != 0)
{
m_registrations.Remove(callbackRegistration);
}
}
callbackRegistration?.OnConnectionWaiting?.Invoke(sender, e);
return found;
}
///
/// Raised when a connection status changes.
///
private void OnConnectionStatusChanged(object? sender, ConnectionStatusEventArgs e)
{
m_logger.LogInformation(
"Channel status: {EndpointUrl} {ChannelStatus} {Closed}",
e.EndpointUrl,
e.ChannelStatus,
e.Closed);
}
///
/// Renew the cancellation token after use.
///
private void CancelAndRenewTokenSource()
{
CancellationTokenSource cts = m_cts;
m_cts = new CancellationTokenSource();
cts.Cancel();
cts.Dispose();
}
[DynamicallyAccessedMembers(DynamicallyAccessedMemberTypes.PublicConstructors)]
private Type? m_configType;
private readonly Lock m_lock = new();
private readonly TimeProvider m_timeProvider;
private readonly ILogger m_logger;
private readonly ITelemetryContext m_telemetry;
private ConfigurationWatcher? m_configurationWatcher;
private ApplicationType m_applicationType;
private ReverseConnectClientConfiguration? m_configuration;
private Dictionary m_endpointUrls;
private ReverseConnectManagerState m_state;
private readonly List m_registrations;
private readonly Lock m_registrationsLock = new();
private CancellationTokenSource m_cts;
}
}