dohertj2/lmxopcua
1
0
Fork 0
Code Issues 25 Pull Requests Actions Packages Projects Releases Wiki Activity

fix: preserve SDK 2.0 client value and lifetime contracts

Browse Source
This commit is contained in:
Joseph Doherty
2026-11-10 10:19:42 -05:00
parent 4400127810
commit 33c7dc4292
38 changed files with 591 additions and 186 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Client/ZB.MOM.WW.OtOpcUa.Client.CLI/Commands/HistoryReadCommand.cs
+4 -3
View File
@@ -1,3 +1,4 @@
using ZB.MOM.WW.OtOpcUa.Client.Shared.Helpers;
using System.Globalization;
using CliFx.Attributes;
using CliFx.Exceptions;
@@ -210,7 +211,7 @@ public class HistoryReadCommand : CommandBase
: StatusCode.IsBad(dv.StatusCode) ? "Bad"
: "Uncertain";
await console.Output.WriteLineAsync(
$"{dv.SourceTimestamp.ToString("O"),-35} {ZB.MOM.WW.OtOpcUa.Client.Shared.Helpers.ValueConverter.ToClrObject(dv.WrappedValue)?.ToString() ?? "(none)",-15} {status,-10} " +
$"{dv.SourceTimestamp.ToString("O"),-35} {ValueConverter.ToClrObject(dv.WrappedValue)?.ToString() ?? "(none)",-15} {status,-10} " +
$"{e.UpdateType,-8} {e.UserName ?? "",-20} {e.ModificationTime:O}");
}
@@ -244,7 +245,7 @@ public class HistoryReadCommand : CommandBase
: StatusCode.IsBad(dv.StatusCode) ? "Bad"
: "Uncertain";
await console.Output.WriteLineAsync(
$"{dv.SourceTimestamp.ToString("O"),-35} {ZB.MOM.WW.OtOpcUa.Client.Shared.Helpers.ValueConverter.ToClrObject(dv.WrappedValue),-15} {status}");
$"{dv.SourceTimestamp.ToString("O"),-35} {ValueConverter.ToClrObject(dv.WrappedValue),-15} {status}");
}
await console.Output.WriteLineAsync();
@@ -294,7 +295,7 @@ public class HistoryReadCommand : CommandBase
/// <summary>Formats one projected event field for the console: byte strings as hex, times as ISO 8601, null as <c>-</c>.</summary>
/// <param name="field">The field value.</param>
/// <returns>Its display text.</returns>
private static string FormatEventField(Variant field) => ZB.MOM.WW.OtOpcUa.Client.Shared.Helpers.ValueConverter.ToClrObject(field) switch
private static string FormatEventField(Variant field) => ValueConverter.ToClrObject(field) switch
{
null => "-",
byte[] bytes => Convert.ToHexString(bytes),
src/Client/ZB.MOM.WW.OtOpcUa.Client.CLI/Commands/HistoryUpdateCommand.cs
+2 -1
View File
@@ -1,3 +1,4 @@
using ZB.MOM.WW.OtOpcUa.Client.Shared.Helpers;
using System.Globalization;
using CliFx.Attributes;
using CliFx.Exceptions;
@@ -93,7 +94,7 @@ public class HistoryUpdateCommand : CommandBase
for (var i = 0; i < values.Count; i++)
{
var ts = ParseTime(times[i], "--time");
updates.Add(new DataValue(ZB.MOM.WW.OtOpcUa.Client.Shared.Helpers.ValueConverter.ToVariant(ParseValue(values[i], Type)), StatusCodes.Good, ts, ts));
updates.Add(new DataValue(ValueConverter.ToVariant(ParseValue(values[i], Type)), StatusCodes.Good, ts, ts));
}
return new UpdateDataDetails
src/Client/ZB.MOM.WW.OtOpcUa.Client.CLI/Commands/ReadCommand.cs
+2 -1
View File
@@ -1,3 +1,4 @@
using ZB.MOM.WW.OtOpcUa.Client.Shared.Helpers;
using CliFx.Attributes;
using CliFx.Exceptions;
using CliFx.Infrastructure;
@@ -64,7 +65,7 @@ public class ReadCommand : CommandBase
}
await console.Output.WriteLineAsync($"Node: {NodeId}");
await console.Output.WriteLineAsync($"Value: {ZB.MOM.WW.OtOpcUa.Client.Shared.Helpers.ValueConverter.ToClrObject(value.WrappedValue)}");
await console.Output.WriteLineAsync($"Value: {ValueConverter.ToClrObject(value.WrappedValue)}");
await console.Output.WriteLineAsync($"Status: {value.StatusCode}");
await console.Output.WriteLineAsync($"Source Time: {value.SourceTimestamp:O}");
await console.Output.WriteLineAsync($"Server Time: {value.ServerTimestamp:O}");
src/Client/ZB.MOM.WW.OtOpcUa.Client.CLI/Commands/SubscribeCommand.cs
+3 -2
View File
@@ -1,3 +1,4 @@
using ZB.MOM.WW.OtOpcUa.Client.Shared.Helpers;
using System.Collections.Concurrent;
using System.Threading.Channels;
using CliFx.Attributes;
@@ -123,14 +124,14 @@ public class SubscribeCommand : CommandBase
try
{
var key = e.NodeId.ToString();
lastStatus[key] = (e.Value.StatusCode, DateTime.UtcNow, ZB.MOM.WW.OtOpcUa.Client.Shared.Helpers.ValueConverter.ToClrObject(e.Value.WrappedValue));
lastStatus[key] = (e.Value.StatusCode, DateTime.UtcNow, ValueConverter.ToClrObject(e.Value.WrappedValue));
updateCount.AddOrUpdate(key, 1, (_, v) => v + 1);
if (!StatusCode.IsGood(e.Value.StatusCode))
everBad.TryAdd(key, 0);
if (!Quiet)
{
var line =
$"[{e.Value.SourceTimestamp:O}] {displayNameByNodeId.GetValueOrDefault(key, key)} = {ZB.MOM.WW.OtOpcUa.Client.Shared.Helpers.ValueConverter.ToClrObject(e.Value.WrappedValue)} ({e.Value.StatusCode})";
$"[{e.Value.SourceTimestamp:O}] {displayNameByNodeId.GetValueOrDefault(key, key)} = {ValueConverter.ToClrObject(e.Value.WrappedValue)} ({e.Value.StatusCode})";
outputChannel.Writer.TryWrite(line);
}
}
src/Client/ZB.MOM.WW.OtOpcUa.Client.CLI/Commands/WriteCommand.cs
+1 -1
View File
@@ -54,7 +54,7 @@ public class WriteCommand : CommandBase
// Read current value to determine type for conversion
var currentValue = await service.ReadValueAsync(nodeId, ct);
var typedValue = ValueConverter.ConvertValue(Value, ZB.MOM.WW.OtOpcUa.Client.Shared.Helpers.ValueConverter.ToClrObject(currentValue.WrappedValue));
var typedValue = ValueConverter.ConvertValue(Value, ValueConverter.ToClrObject(currentValue.WrappedValue));
var statusCode = await service.WriteValueAsync(nodeId, typedValue, ct);
src/Client/ZB.MOM.WW.OtOpcUa.Client.Shared/Adapters/DefaultApplicationConfigurationFactory.cs
+1 -1
View File
@@ -70,7 +70,7 @@ internal sealed class DefaultApplicationConfigurationFactory : IApplicationConfi
ApplicationType = ApplicationType.Client,
ApplicationConfiguration = config
};
await app.CheckApplicationInstanceCertificatesAsync(false, ct: ct);
await app.CheckApplicationInstanceCertificatesAsync(false, lifeTimeInMonths: 2048, ct: ct);
}
}
catch
src/Client/ZB.MOM.WW.OtOpcUa.Client.Shared/Adapters/DefaultSessionAdapter.cs
+17 -10
View File
@@ -115,18 +115,14 @@ internal sealed class DefaultSessionAdapter : ISessionAdapter
{
// Pass the caller's token so a cancelled browse does not block on the continuation
// round-trip.
var response = await _session.BrowseNextAsync(null, false, [ByteString.From(continuationPoint)], ct);
if (response.Results.Count == 0)
throw new ServiceResultException(response.ResponseHeader.ServiceResult, "BrowseNext response contained no results.");
var result = response.Results[0];
if (StatusCode.IsBad(result.StatusCode)) throw new ServiceResultException(result.StatusCode);
return (result.ContinuationPoint.IsNull ? null : result.ContinuationPoint.ToArray(), result.References);
var (_, nextCp, nextRefs) = await _session.BrowseNextAsync(null, false, ByteString.From(continuationPoint), ct);
return (nextCp.IsNull ? null : nextCp.ToArray(), nextRefs);
}
/// <inheritdoc />
public async Task ReleaseBrowseContinuationPointAsync(byte[] continuationPoint, CancellationToken ct)
{
await _session.BrowseNextAsync(null, true, [ByteString.From(continuationPoint)], ct).ConfigureAwait(false);
await _session.BrowseNextAsync(null, true, ByteString.From(continuationPoint), ct).ConfigureAwait(false);
}
/// <inheritdoc />
@@ -276,9 +272,20 @@ internal sealed class DefaultSessionAdapter : ISessionAdapter
if (results.Count > 0)
{
var result = results[0];
HistoryReadStatus.ThrowIfRefused(result.StatusCode, nodeId);
if (result.HistoryData.TryGetValue(out HistoryData? historyData))
allValues.AddRange(historyData.DataValues);
try
{
HistoryReadStatus.ThrowIfRefused(result.StatusCode, nodeId);
if (result.HistoryData.TryGetValue(out HistoryData? historyData))
allValues.AddRange(historyData.DataValues);
}
finally
{
// Keep the existing one-shot processed read; it owns no outstanding server cursor on return.
if (!result.ContinuationPoint.IsEmpty)
await HistoryReadPaging.ReleaseAsync(
(cp, release, token) => ReadHistoryPageAsync(details, nodeId, cp, release, TimestampsToReturn.Source, token),
result.ContinuationPoint.ToArray(), nodeId).ConfigureAwait(false);
}
}
return allValues;
src/Client/ZB.MOM.WW.OtOpcUa.Client.Shared/Adapters/DefaultSubscriptionAdapter.cs
+24 -28
View File
@@ -14,6 +14,8 @@ internal sealed class DefaultSubscriptionAdapter : ISubscriptionAdapter
private readonly Subscription _subscription;
private readonly Func<Subscription, CancellationToken, Task>? _detach;
private int _detached;
private int _finishing;
private readonly TaskCompletionSource _finished = new(TaskCreationOptions.RunContinuationsAsynchronously);
/// <summary>
/// Wraps a live OPC UA subscription so client code can manage monitored items through a testable abstraction.
@@ -47,7 +49,7 @@ internal sealed class DefaultSubscriptionAdapter : ISubscriptionAdapter
item.Notification += (_, e) =>
{
if (e.NotificationValue is MonitoredItemNotification notification)
if (Volatile.Read(ref _finishing) == 0 && e.NotificationValue is MonitoredItemNotification notification)
onDataChange(nodeId.ToString(), notification.Value);
};
@@ -90,7 +92,7 @@ internal sealed class DefaultSubscriptionAdapter : ISubscriptionAdapter
item.Notification += (_, e) =>
{
if (e.NotificationValue is EventFieldList eventFields) onEvent(eventFields);
if (Volatile.Read(ref _finishing) == 0 && e.NotificationValue is EventFieldList eventFields) onEvent(eventFields);
};
_subscription.AddItem(item);
@@ -112,19 +114,28 @@ internal sealed class DefaultSubscriptionAdapter : ISubscriptionAdapter
/// <inheritdoc />
public async Task DeleteAsync(CancellationToken ct)
{
if (Interlocked.Exchange(ref _finishing, 1) != 0)
{
await _finished.Task.WaitAsync(ct).ConfigureAwait(false);
return;
}
try
{
// Forward the caller's token so the delete can be cancelled.
await _subscription.DeleteAsync(true, ct);
}
catch (Exception ex)
{
Logger.Warning(ex, "Error deleting subscription");
}
try { await _subscription.DeleteAsync(true, ct).ConfigureAwait(false); }
catch (Exception ex) { Logger.Warning(ex, "Error deleting subscription"); }
// Detach whether or not the server-side delete succeeded: the adapter is finished with it either way.
await DetachAsync(ct);
_monitoredItems.Clear();
// A cancelled delete still has local resources to release. Cleanup gets its own bounded token.
using var cleanup = new CancellationTokenSource(TimeSpan.FromSeconds(5));
try { await DetachAsync(cleanup.Token).WaitAsync(cleanup.Token).ConfigureAwait(false); }
catch (Exception ex) { Logger.Warning(ex, "Error detaching subscription"); }
finally
{
_monitoredItems.Clear();
try { await Task.Run(_subscription.Dispose).WaitAsync(cleanup.Token).ConfigureAwait(false); }
catch (Exception ex) { Logger.Warning(ex, "Error disposing subscription resources"); }
}
}
finally { _finished.TrySetResult(); }
}
private async Task DetachAsync(CancellationToken ct)
@@ -145,21 +156,6 @@ internal sealed class DefaultSubscriptionAdapter : ISubscriptionAdapter
/// </summary>
public void Dispose()
{
// One bounded bridge (5 s) for delete + detach, so a dispose never blocks longer than before (review m6).
// DetachAsync swallows its own failures; a failed delete must not skip the detach.
try
{
DisposeBridge.Run(async ct =>
{
try { await _subscription.DeleteAsync(true, ct); }
catch (Exception ex) { Logger.Warning(ex, "Error deleting subscription"); }
await DetachAsync(ct);
});
}
catch
{
}
_monitoredItems.Clear();
DisposeBridge.Run(DeleteAsync);
}
}
src/Client/ZB.MOM.WW.OtOpcUa.Client.Shared/Helpers/HistoryReadPaging.cs
+5 -5
View File
@@ -61,10 +61,10 @@ internal static class HistoryReadPaging
if (all.Count >= maxValues) break;
}
}
catch (OperationCanceledException)
catch
{
// Cancelled between pages (or during page N+1) with a cursor outstanding: release it rather than leave it
// occupying one of the session's continuation-point slots, then surface the cancellation.
// A refused/failed/cancelled continuation still leaves the last cursor owned by this read.
// Best-effort release must preserve the original operation failure.
if (continuationPoint is { Length: > 0 }) await ReleaseAsync(readPage, continuationPoint, nodeId);
throw;
}
@@ -81,12 +81,12 @@ internal static class HistoryReadPaging
return all;
}
private static async Task ReleaseAsync(PageReader readPage, byte[] continuationPoint, NodeId nodeId)
internal static async Task ReleaseAsync(PageReader readPage, byte[] continuationPoint, NodeId nodeId)
{
try
{
using var cts = new CancellationTokenSource(ReleaseTimeout);
await readPage(continuationPoint, true, cts.Token).ConfigureAwait(false);
await readPage(continuationPoint, true, cts.Token).WaitAsync(cts.Token).ConfigureAwait(false);
}
catch (Exception ex)
{
src/Client/ZB.MOM.WW.OtOpcUa.Client.Shared/Helpers/ValueConverter.cs
+7 -5
View File
@@ -15,20 +15,22 @@ public static class ValueConverter
public static Variant ToVariant(object? value)
{
// Historically a CLR byte[] supplied to a client write/method was a scalar ByteString.
if (value is byte[] bytes) return Variant.From(ByteString.From(bytes.AsSpan()));
if (value is ByteString byteString) return Variant.From(byteString.Copy());
if (value is byte[] bytes && bytes.GetType() == typeof(byte[])) return Variant.From(ByteString.From(bytes.AsSpan()));
if (value is ByteString byteString) return Variant.From(byteString.IsNull ? default : byteString.Copy());
if (!VariantHelper.TryCastFrom(value, out Variant variant))
throw new ServiceResultException(StatusCodes.BadTypeMismatch, "Unsupported client value type or shape.");
if (variant.TypeInfo.IsMatrix)
throw new ServiceResultException(StatusCodes.BadTypeMismatch, "Unsupported client matrix shape.");
if (variant.TypeInfo.IsArray && variant.TryGetValue(out ArrayOf<ByteString> buffers))
return Variant.From(buffers.IsNull ? default : buffers.ConvertAll(b => b.IsNull ? default : b.Copy()));
// ArrayOf is read-only, but can wrap externally owned ReadOnlyMemory. Detach it too.
if (variant.TypeInfo.IsArray && variant.AsBoxedObject(Variant.BoxingBehavior.Legacy) is Array array)
if (value is not Array && variant.TypeInfo.IsArray && variant.AsBoxedObject(Variant.BoxingBehavior.Legacy) is Array array)
{
if (!VariantHelper.TryCastFrom(array, out variant))
throw new ServiceResultException(StatusCodes.BadTypeMismatch, "Unsupported client array type.");
}
else if (variant.TryGetValue(out ByteString retainedBytes))
variant = Variant.From(retainedBytes.Copy());
else if (variant.TypeInfo.IsScalar && variant.TryGetValue(out ByteString retainedBytes))
variant = Variant.From(retainedBytes.IsNull ? default : retainedBytes.Copy());
return variant;
}
src/Client/ZB.MOM.WW.OtOpcUa.Client.Shared/OpcUaClientService.cs
+23 -4
View File
@@ -310,7 +310,7 @@ public sealed class OpcUaClientService : IOpcUaClientService
}
var handle = await subscription.AddDataChangeMonitoredItemAsync(
nodeId, intervalMs, OnDataChangeNotification, ct);
nodeId, intervalMs, DataCallback(session, subscription), ct);
lock (_subscriptionLock)
{
@@ -394,7 +394,7 @@ public sealed class OpcUaClientService : IOpcUaClientService
var filter = CreateAlarmEventFilter();
await subscription.AddEventMonitoredItemAsync(
monitorNode, intervalMs, filter, OnAlarmEventNotification, ct);
monitorNode, intervalMs, filter, AlarmCallback(session, subscription), ct);
lock (_subscriptionLock)
{
@@ -989,7 +989,7 @@ public sealed class OpcUaClientService : IOpcUaClientService
}
var handle = await dataSubscription.AddDataChangeMonitoredItemAsync(
nodeId, intervalMs, OnDataChangeNotification, CancellationToken.None);
nodeId, intervalMs, DataCallback(session, dataSubscription), CancellationToken.None);
lock (_subscriptionLock)
{
@@ -1012,7 +1012,7 @@ public sealed class OpcUaClientService : IOpcUaClientService
lock (_subscriptionLock) _alarmSubscription = replayedAlarms;
var filter = CreateAlarmEventFilter();
await replayedAlarms.AddEventMonitoredItemAsync(
monitorNode, intervalMs, filter, OnAlarmEventNotification, CancellationToken.None);
monitorNode, intervalMs, filter, AlarmCallback(session, replayedAlarms), CancellationToken.None);
lock (_subscriptionLock)
{
@@ -1026,6 +1026,25 @@ public sealed class OpcUaClientService : IOpcUaClientService
}
}
private bool IsCurrentNotification(ISessionAdapter session, ISubscriptionAdapter subscription, bool alarm)
{
lock (_subscriptionLock)
return !_disposed && _state == ConnectionState.Connected && ReferenceEquals(_session, session)
&& ReferenceEquals(alarm ? _alarmSubscription : _dataSubscription, subscription);
}
private Action<string, DataValue> DataCallback(ISessionAdapter session, ISubscriptionAdapter subscription) =>
(nodeId, value) =>
{
if (IsCurrentNotification(session, subscription, false)) OnDataChangeNotification(nodeId, value);
};
private Action<EventFieldList> AlarmCallback(ISessionAdapter session, ISubscriptionAdapter subscription) =>
fields =>
{
if (IsCurrentNotification(session, subscription, true)) OnAlarmEventNotification(fields);
};
private void OnDataChangeNotification(string nodeId, DataValue value)
{
DataChanged?.Invoke(this, new DataChangedEventArgs(nodeId, value));
src/Client/ZB.MOM.WW.OtOpcUa.Client.UI/ViewModels/HistoryViewModel.cs
+2 -1
View File
@@ -1,3 +1,4 @@
using ZB.MOM.WW.OtOpcUa.Client.Shared.Helpers;
using System.Collections.ObjectModel;
using CommunityToolkit.Mvvm.ComponentModel;
using CommunityToolkit.Mvvm.Input;
@@ -97,7 +98,7 @@ public partial class HistoryViewModel : ObservableObject
{
foreach (var dv in values)
Results.Add(new HistoryValueViewModel(
Helpers.ValueFormatter.Format(ZB.MOM.WW.OtOpcUa.Client.Shared.Helpers.ValueConverter.ToClrObject(dv.WrappedValue)),
Helpers.ValueFormatter.Format(ValueConverter.ToClrObject(dv.WrappedValue)),
Helpers.StatusCodeFormatter.Format(dv.StatusCode),
dv.SourceTimestamp.ToString("O"),
dv.ServerTimestamp.ToString("O")));
src/Client/ZB.MOM.WW.OtOpcUa.Client.UI/ViewModels/ReadWriteViewModel.cs
+2 -1
View File
@@ -1,3 +1,4 @@
using ZB.MOM.WW.OtOpcUa.Client.Shared.Helpers;
using CommunityToolkit.Mvvm.ComponentModel;
using CommunityToolkit.Mvvm.Input;
using Opc.Ua;
@@ -76,7 +77,7 @@ public partial class ReadWriteViewModel : ObservableObject
_dispatcher.Post(() =>
{
CurrentValue = Helpers.ValueFormatter.Format(ZB.MOM.WW.OtOpcUa.Client.Shared.Helpers.ValueConverter.ToClrObject(dataValue.WrappedValue));
CurrentValue = Helpers.ValueFormatter.Format(ValueConverter.ToClrObject(dataValue.WrappedValue));
CurrentStatus = Helpers.StatusCodeFormatter.Format(dataValue.StatusCode);
SourceTimestamp = dataValue.SourceTimestamp.ToString("O");
ServerTimestamp = dataValue.ServerTimestamp.ToString("O");
src/Client/ZB.MOM.WW.OtOpcUa.Client.UI/ViewModels/SubscriptionsViewModel.cs
+3 -2
View File
@@ -1,3 +1,4 @@
using ZB.MOM.WW.OtOpcUa.Client.Shared.Helpers;
using System.Collections.ObjectModel;
using System.Collections.Specialized;
using CommunityToolkit.Mvvm.ComponentModel;
@@ -75,7 +76,7 @@ public partial class SubscriptionsViewModel : ObservableObject
if (!_rowsByNodeId.TryGetValue(e.NodeId, out var rows)) return;
foreach (var item in rows)
{
item.Value = Helpers.ValueFormatter.Format(ZB.MOM.WW.OtOpcUa.Client.Shared.Helpers.ValueConverter.ToClrObject(e.Value.WrappedValue));
item.Value = Helpers.ValueFormatter.Format(ValueConverter.ToClrObject(e.Value.WrappedValue));
item.Status = Helpers.StatusCodeFormatter.Format(e.Value.StatusCode);
item.Timestamp = e.Value.SourceTimestamp.ToString("O");
}
@@ -281,7 +282,7 @@ public partial class SubscriptionsViewModel : ObservableObject
// Read current value to determine target type
var currentDataValue = await _service.ReadValueAsync(nodeId);
var currentValue = ZB.MOM.WW.OtOpcUa.Client.Shared.Helpers.ValueConverter.ToClrObject(currentDataValue.WrappedValue);
var currentValue = ValueConverter.ToClrObject(currentDataValue.WrappedValue);
// Try parsing to the target type before writing
try